chore: Strip DSKey prefixes and simplify NewDataStoreKey

core/errors.go

@@ -0,0 +1,18 @@
+// Copyright 2022 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package core
+
+import "github.com/sourcenetwork/defradb/errors"
+
+var (
+	ErrEmptyKey   = errors.New("received empty key string")
+	ErrInvalidKey = errors.New("invalid key string")
+)

core/key.go

@@ -105,29 +105,37 @@ var _ Key = (*SequenceKey)(nil)
 // /[CollectionId]/[InstanceType]/[DocKey]/[FieldId]
 //
 // Any properties before the above (assuming a '/' deliminator) are ignored
-func NewDataStoreKey(key string) DataStoreKey {
+func NewDataStoreKey(key string) (DataStoreKey, error) {
 	dataStoreKey := DataStoreKey{}
 	if key == "" {
-		return dataStoreKey
+		return dataStoreKey, errors.WithStack(ErrEmptyKey)
 	}
 
-	elements := strings.Split(key, "/")
+	elements := strings.Split(strings.TrimPrefix(key, "/"), "/")
 
-	if isDataObjectMarker(elements) {
-		return DataStoreKey{
-			CollectionId: elements[3],
-			InstanceType: ValueKey,
-			DocKey:       elements[5],
-		}
+	numberOfElements := len(elements)
+
+	// With less than 3 or more than 4 elements, we know it's an invalid key
+	if numberOfElements < 3 || numberOfElements > 4 {
+		return dataStoreKey, errors.WithStack(ErrInvalidKey)
 	}
 
-	numberOfElements := len(elements)
-	return DataStoreKey{
-		CollectionId: elements[numberOfElements-4],
-		InstanceType: InstanceType(elements[numberOfElements-3]),
-		DocKey:       elements[numberOfElements-2],
-		FieldId:      elements[numberOfElements-1],
+	dataStoreKey.CollectionId = elements[0]
+	dataStoreKey.InstanceType = InstanceType(elements[1])
+	dataStoreKey.DocKey = elements[2]
+	if numberOfElements == 4 {
+		dataStoreKey.FieldId = elements[3]
 	}
+
+	return dataStoreKey, nil
+}
+
+func MustNewDataStoreKey(key string) DataStoreKey {
+	dsKey, err := NewDataStoreKey(key)
+	if err != nil {
+		panic(err)
+	}
+	return dsKey
 }
 
 func DataStoreKeyFromDocKey(dockey client.DocKey) DataStoreKey {
@@ -450,21 +458,3 @@ func bytesPrefixEnd(b []byte) []byte {
 	// maximal byte string (i.e. already \xff...).
 	return b
 }
-
-func isDataObjectMarker(elements []string) bool {
-	numElements := len(elements)
-	// lenght is 6 because it has no FieldID
-	if numElements != 6 {
-		return false
-	}
-	if elements[1] != "db" {
-		return false
-	}
-	if elements[2] != "data" {
-		return false
-	}
-	if elements[4] != "v" {
-		return false
-	}
-	return true
-}

core/key_test.go

@@ -19,11 +19,13 @@ import (
 func TestNewDataStoreKey_ReturnsEmptyStruct_GivenEmptyString(t *testing.T) {
 	inputString := ""
 
-	result := NewDataStoreKey(inputString)
+	result, err := NewDataStoreKey(inputString)
+
 	resultString := result.ToString()
 
 	assert.Equal(t, DataStoreKey{}, result)
 	assert.Equal(t, "", resultString)
+	assert.ErrorIs(t, ErrEmptyKey, err)
 }
 
 func TestNewDataStoreKey_ReturnsCollectionIdAndIndexIdAndDocKeyAndFieldIdAndInstanceType_GivenFourItemsWithType(
@@ -35,7 +37,10 @@ func TestNewDataStoreKey_ReturnsCollectionIdAndIndexIdAndDocKeyAndFieldIdAndInst
 	collectionId := "1"
 	inputString := collectionId + "/" + instanceType + "/" + docKey + "/" + fieldId
 
-	result := NewDataStoreKey(inputString)
+	result, err := NewDataStoreKey(inputString)
+	if err != nil {
+		t.Error(err)
+	}
 	resultString := result.ToString()
 
 	assert.Equal(
@@ -48,3 +53,57 @@ func TestNewDataStoreKey_ReturnsCollectionIdAndIndexIdAndDocKeyAndFieldIdAndInst
 		result)
 	assert.Equal(t, "/"+collectionId+"/"+instanceType+"/"+docKey+"/"+fieldId, resultString)
 }
+
+func TestNewDataStoreKey_ReturnsEmptyStruct_GivenAStringWithMissingElements(t *testing.T) {
+	inputString := "/0/v"
+
+	_, err := NewDataStoreKey(inputString)
+
+	assert.ErrorIs(t, ErrInvalidKey, err)
+}
+
+func TestNewDataStoreKey_GivenAShortObjectMarker(t *testing.T) {
+	instanceType := "anyType"
+	docKey := "docKey"
+	collectionId := "1"
+	inputString := collectionId + "/" + instanceType + "/" + docKey
+
+	result, err := NewDataStoreKey(inputString)
+	if err != nil {
+		t.Error(err)
+	}
+	resultString := result.ToString()
+
+	assert.Equal(
+		t,
+		DataStoreKey{
+			CollectionId: collectionId,
+			DocKey:       docKey,
+			InstanceType: InstanceType(instanceType)},
+		result)
+	assert.Equal(t, "/"+collectionId+"/"+instanceType+"/"+docKey, resultString)
+}
+
+func TestNewDataStoreKey_GivenAStringWithExtraPrefixes(t *testing.T) {
+	instanceType := "anyType"
+	fieldId := "f1"
+	docKey := "docKey"
+	collectionId := "1"
+	inputString := "/db/my_database_name/data/" + collectionId + "/" + instanceType + "/" + docKey + "/" + fieldId
+
+	_, err := NewDataStoreKey(inputString)
+
+	assert.ErrorIs(t, ErrInvalidKey, err)
+}
+
+func TestNewDataStoreKey_GivenAStringWithExtraSuffix(t *testing.T) {
+	instanceType := "anyType"
+	fieldId := "f1"
+	docKey := "docKey"
+	collectionId := "1"
+	inputString := "/db/data/" + collectionId + "/" + instanceType + "/" + docKey + "/" + fieldId + "/version_number"
+
+	_, err := NewDataStoreKey(inputString)
+
+	assert.ErrorIs(t, ErrInvalidKey, err)
+}

datastore/wrappedStore.go

@@ -217,13 +217,36 @@ func (w *wrappedIterator) IteratePrefix(
 	startPrefix ds.Key,
 	endPrefix ds.Key,
 ) (dsq.Results, error) {
-	return w.iterator.IteratePrefix(
+	results, err := w.iterator.IteratePrefix(
 		ctx,
 		w.transform.ConvertKey(startPrefix),
 		w.transform.ConvertKey(endPrefix),
 	)
+	if err != nil {
+		return nil, err
+	}
+	return &wrappedResults{
+		Results:   results,
+		transform: w.transform,
+	}, nil
 }
 
 func (w *wrappedIterator) Close() error {
 	return w.iterator.Close()
 }
+
+type wrappedResults struct {
+	dsq.Results
+	transform ktds.KeyTransform
+}
+
+func (wr *wrappedResults) NextSync() (dsq.Result, bool) {
+	r, ok := wr.Results.NextSync()
+	if !ok {
+		return r, false
+	}
+	if r.Error == nil {
+		r.Entry.Key = wr.transform.InvertKey(ds.RawKey(r.Entry.Key)).String()
+	}
+	return r, true
+}

db/collection.go

@@ -745,7 +745,7 @@ func (c *collection) deleteWithPrefix(ctx context.Context, txn datastore.Txn, ke
 	}
 
 	if key.InstanceType == core.ValueKey {
-		err := txn.Datastore().Delete(ctx, core.NewDataStoreKey(key.ToString()).ToDS())
+		err := txn.Datastore().Delete(ctx, key.ToDS())
 		if err != nil {
 			return false, err
 		}
@@ -758,7 +758,12 @@ func (c *collection) deleteWithPrefix(ctx context.Context, txn datastore.Txn, ke
 			return false, err
 		}
 
-		err = txn.Datastore().Delete(ctx, core.NewDataStoreKey(e.Key).ToDS())
+		dsKey, err := core.NewDataStoreKey(e.Key)
+		if err != nil {
+			return false, err
+		}
+
+		err = txn.Datastore().Delete(ctx, dsKey.ToDS())
 		if err != nil {
 			return false, err
 		}

db/fetcher/fetcher.go

@@ -245,8 +245,13 @@ func (df *DocumentFetcher) nextKV() (iterDone bool, kv *core.KeyValue, err error
 		return true, nil, err
 	}
 
+	dsKey, err := core.NewDataStoreKey(res.Key)
+	if err != nil {
+		return true, nil, err
+	}
+
 	kv = &core.KeyValue{
-		Key:   core.NewDataStoreKey(res.Key),
+		Key:   dsKey,
 		Value: res.Value,
 	}
 	return false, kv, nil

merkle/crdt/factory_test.go

@@ -127,7 +127,7 @@ func TestFactoryInstanceMissing(t *testing.T) {
 	m := newStores()
 	f := NewFactory(m)
 
-	_, err := f.Instance("", client.EmptyUpdateChannel, client.LWW_REGISTER, core.NewDataStoreKey("/1/0/MyKey"))
+	_, err := f.Instance("", client.EmptyUpdateChannel, client.LWW_REGISTER, core.MustNewDataStoreKey("/1/0/MyKey"))
 	assert.Equal(t, err, ErrFactoryTypeNoExist)
 }
 
@@ -137,7 +137,7 @@ func TestBlankFactoryInstanceWithLWWRegister(t *testing.T) {
 	f1.Register(client.LWW_REGISTER, &lwwFactoryFn)
 	f := f1.WithStores(m)
 
-	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.LWW_REGISTER, core.NewDataStoreKey("/1/0/MyKey"))
+	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.LWW_REGISTER, core.MustNewDataStoreKey("/1/0/MyKey"))
 	assert.NoError(t, err)
 
 	_, ok := crdt.(*MerkleLWWRegister)
@@ -150,7 +150,7 @@ func TestBlankFactoryInstanceWithCompositeRegister(t *testing.T) {
 	f1.Register(client.COMPOSITE, &compFactoryFn)
 	f := f1.WithStores(m)
 
-	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.COMPOSITE, core.NewDataStoreKey("/1/0/MyKey"))
+	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.COMPOSITE, core.MustNewDataStoreKey("/1/0/MyKey"))
 	assert.NoError(t, err)
 
 	_, ok := crdt.(*MerkleCompositeDAG)
@@ -162,7 +162,7 @@ func TestFullFactoryInstanceLWWRegister(t *testing.T) {
 	f := NewFactory(m)
 	f.Register(client.LWW_REGISTER, &lwwFactoryFn)
 
-	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.LWW_REGISTER, core.NewDataStoreKey("/1/0/MyKey"))
+	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.LWW_REGISTER, core.MustNewDataStoreKey("/1/0/MyKey"))
 	assert.NoError(t, err)
 
 	_, ok := crdt.(*MerkleLWWRegister)
@@ -174,7 +174,7 @@ func TestFullFactoryInstanceCompositeRegister(t *testing.T) {
 	f := NewFactory(m)
 	f.Register(client.COMPOSITE, &compFactoryFn)
 
-	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.COMPOSITE, core.NewDataStoreKey("/1/0/MyKey"))
+	crdt, err := f.Instance("", client.EmptyUpdateChannel, client.COMPOSITE, core.MustNewDataStoreKey("/1/0/MyKey"))
 	assert.NoError(t, err)
 
 	_, ok := crdt.(*MerkleCompositeDAG)
@@ -185,7 +185,7 @@ func TestLWWRegisterFactoryFn(t *testing.T) {
 	ctx := context.Background()
 	m := newStores()
 	f := NewFactory(m) // here factory is only needed to satisfy datastore.MultiStore interface
-	crdt := lwwFactoryFn(f, "", client.EmptyUpdateChannel)(core.NewDataStoreKey("/1/0/MyKey"))
+	crdt := lwwFactoryFn(f, "", client.EmptyUpdateChannel)(core.MustNewDataStoreKey("/1/0/MyKey"))
 
 	lwwreg, ok := crdt.(*MerkleLWWRegister)
 	assert.True(t, ok)
@@ -198,7 +198,7 @@ func TestCompositeRegisterFactoryFn(t *testing.T) {
 	ctx := context.Background()
 	m := newStores()
 	f := NewFactory(m) // here factory is only needed to satisfy datastore.MultiStore interface
-	crdt := compFactoryFn(f, "", client.EmptyUpdateChannel)(core.NewDataStoreKey("/1/0/MyKey"))
+	crdt := compFactoryFn(f, "", client.EmptyUpdateChannel)(core.MustNewDataStoreKey("/1/0/MyKey"))
 
 	merkleReg, ok := crdt.(*MerkleCompositeDAG)
 	assert.True(t, ok)