refactor: Reorganise collection description storage (#1988)

## Relevant issue(s)

Resolves #1964 #1913 

## Description

Reorganises collection description storage so that it actually makes
sense.

The existing key/value setup has been replaced with the following keys:
```
[Collection.ID] => json // Json should be stored against an immutable index, i.e. not against `Name`
[Collection.Name] => [Collection.ID]  // This means collection names must be unique
```
We also need the following for the short-term, before
Collection.GlobalID gets added, as otherwise if a collection is updated
to a new schema version, we lose track of what collection it could be.
The locations that use this have been linked to
#1085
```
[Collection.SchemaVersionId]/[Collection.ID] => nil
```

With this change the storage setup should fully support multiple
collections from a single schema, however this is blocked off by
`setDefaultSchemaVersion` (it doesn't allow users to explicitly specify
a collection, so atm there is no way to try and do this). Later when we
introduce `patchCollection` we will need to be mindful of this and block
off user's ability to do this (if done before #1085).

core/key.go

@@ -11,6 +11,7 @@
 package core
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 
@@ -41,18 +42,18 @@ const (
 )
 
 const (
-	COLLECTION                        = "/collection/names"
-	COLLECTION_SCHEMA                 = "/collection/schema"
-	COLLECTION_SCHEMA_VERSION         = "/collection/version/v"
-	COLLECTION_SCHEMA_VERSION_HISTORY = "/collection/version/h"
-	COLLECTION_INDEX                  = "/collection/index"
-	SCHEMA_MIGRATION                  = "/schema/migration"
-	SCHEMA_VERSION                    = "/schema/version"
-	SEQ                               = "/seq"
-	PRIMARY_KEY                       = "/pk"
-	DATASTORE_DOC_VERSION_FIELD_ID    = "v"
-	REPLICATOR                        = "/replicator/id"
-	P2P_COLLECTION                    = "/p2p/collection"
+	COLLECTION                     = "/collection/id"
+	COLLECTION_NAME                = "/collection/name"
+	COLLECTION_SCHEMA_VERSION      = "/collection/version"
+	COLLECTION_INDEX               = "/collection/index"
+	SCHEMA_MIGRATION               = "/schema/migration"
+	SCHEMA_VERSION                 = "/schema/version/v"
+	SCHEMA_VERSION_HISTORY         = "/schema/version/h"
+	SEQ                            = "/seq"
+	PRIMARY_KEY                    = "/pk"
+	DATASTORE_DOC_VERSION_FIELD_ID = "v"
+	REPLICATOR                     = "/replicator/id"
+	P2P_COLLECTION                 = "/p2p/collection"
 )
 
 // Key is an interface that represents a key in the database.
@@ -99,26 +100,32 @@ type HeadStoreKey struct {
 
 var _ Key = (*HeadStoreKey)(nil)
 
-// CollectionKey points to the current/'head' SchemaVersionId for
-// the collection of the given name.
+// CollectionKey points to the json serialized description of the
+// the collection of the given ID.
 type CollectionKey struct {
-	CollectionName string
+	CollectionID uint32
 }
 
 var _ Key = (*CollectionKey)(nil)
 
-// CollectionSchemaKey points to the current/'head' SchemaVersionId for
-// the collection of the given schema id.
-type CollectionSchemaKey struct {
-	SchemaId string
+// CollectionNameKey points to the ID of the collection of the given
+// name.
+type CollectionNameKey struct {
+	Name string
 }
 
-var _ Key = (*CollectionSchemaKey)(nil)
+var _ Key = (*CollectionNameKey)(nil)
 
-// CollectionSchemaVersionKey points to schema of a collection at a given
-// version.
+// CollectionSchemaVersionKey points to nil, but the keys/prefix can be used
+// to get collections that are using, or have used a given schema version.
+//
+// If a collection is updated to a different schema version, the old entry(s)
+// of this key will be preserved.
+//
+// This key should be removed in https://github.com/sourcenetwork/defradb/issues/1085
 type CollectionSchemaVersionKey struct {
 	SchemaVersionId string
+	CollectionID    uint32
 }
 
 var _ Key = (*CollectionSchemaVersionKey)(nil)
@@ -255,21 +262,32 @@ func NewHeadStoreKey(key string) (HeadStoreKey, error) {
 
 // Returns a formatted collection key for the system data store.
 // It assumes the name of the collection is non-empty.
-func NewCollectionKey(name string) CollectionKey {
-	return CollectionKey{CollectionName: name}
+func NewCollectionKey(id uint32) CollectionKey {
+	return CollectionKey{CollectionID: id}
 }
 
-func NewCollectionSchemaKey(schemaId string) CollectionSchemaKey {
-	return CollectionSchemaKey{SchemaId: schemaId}
+func NewCollectionNameKey(name string) CollectionNameKey {
+	return CollectionNameKey{Name: name}
 }
 
-func NewCollectionSchemaVersionKey(schemaVersionId string) CollectionSchemaVersionKey {
-	return CollectionSchemaVersionKey{SchemaVersionId: schemaVersionId}
+func NewCollectionSchemaVersionKey(schemaVersionId string, collectionID uint32) CollectionSchemaVersionKey {
+	return CollectionSchemaVersionKey{
+		SchemaVersionId: schemaVersionId,
+		CollectionID:    collectionID,
+	}
 }
 
-func NewCollectionSchemaVersionKeyFromString(key string) CollectionSchemaVersionKey {
+func NewCollectionSchemaVersionKeyFromString(key string) (CollectionSchemaVersionKey, error) {
 	elements := strings.Split(key, "/")
-	return CollectionSchemaVersionKey{SchemaVersionId: elements[len(elements)-1]}
+	colID, err := strconv.Atoi(elements[len(elements)-1])
+	if err != nil {
+		return CollectionSchemaVersionKey{}, err
+	}
+
+	return CollectionSchemaVersionKey{
+		SchemaVersionId: elements[len(elements)-2],
+		CollectionID:    uint32(colID),
+	}, nil
 }
 
 // NewCollectionIndexKey creates a new CollectionIndexKey from a collection name and index name.
@@ -338,7 +356,7 @@ func NewSchemaVersionMigrationKey(schemaVersionID string) SchemaVersionMigration
 }
 
 func NewSchemaHistoryKeyFromString(keyString string) (SchemaHistoryKey, error) {
-	keyString = strings.TrimPrefix(keyString, COLLECTION_SCHEMA_VERSION_HISTORY+"/")
+	keyString = strings.TrimPrefix(keyString, SCHEMA_VERSION_HISTORY+"/")
 	elements := strings.Split(keyString, "/")
 	if len(elements) != 2 {
 		return SchemaHistoryKey{}, ErrInvalidKey
@@ -591,13 +609,7 @@ func (k PrimaryDataStoreKey) ToString() string {
 }
 
 func (k CollectionKey) ToString() string {
-	result := COLLECTION
-
-	if k.CollectionName != "" {
-		result = result + "/" + k.CollectionName
-	}
-
-	return result
+	return fmt.Sprintf("%s/%s", COLLECTION, strconv.Itoa(int(k.CollectionID)))
 }
 
 func (k CollectionKey) Bytes() []byte {
@@ -608,21 +620,15 @@ func (k CollectionKey) ToDS() ds.Key {
 	return ds.NewKey(k.ToString())
 }
 
-func (k CollectionSchemaKey) ToString() string {
-	result := COLLECTION_SCHEMA
-
-	if k.SchemaId != "" {
-		result = result + "/" + k.SchemaId
-	}
-
-	return result
+func (k CollectionNameKey) ToString() string {
+	return fmt.Sprintf("%s/%s", COLLECTION_NAME, k.Name)
 }
 
-func (k CollectionSchemaKey) Bytes() []byte {
+func (k CollectionNameKey) Bytes() []byte {
 	return []byte(k.ToString())
 }
 
-func (k CollectionSchemaKey) ToDS() ds.Key {
+func (k CollectionNameKey) ToDS() ds.Key {
 	return ds.NewKey(k.ToString())
 }
 
@@ -633,6 +639,10 @@ func (k CollectionSchemaVersionKey) ToString() string {
 		result = result + "/" + k.SchemaVersionId
 	}
 
+	if k.CollectionID != 0 {
+		result = fmt.Sprintf("%s/%s", result, strconv.Itoa(int(k.CollectionID)))
+	}
+
 	return result
 }
 
@@ -663,7 +673,7 @@ func (k SchemaVersionKey) ToDS() ds.Key {
 }
 
 func (k SchemaHistoryKey) ToString() string {
-	result := COLLECTION_SCHEMA_VERSION_HISTORY
+	result := SCHEMA_VERSION_HISTORY
 
 	if k.SchemaID != "" {
 		result = result + "/" + k.SchemaID

core/parser.go

@@ -54,5 +54,7 @@ type Parser interface {
 	ParseSDL(ctx context.Context, schemaString string) ([]client.CollectionDefinition, error)
 
 	// Adds the given schema to this parser's model.
+	//
+	// All collections should be provided, not just new/updated ones.
 	SetSchema(ctx context.Context, txn datastore.Txn, collections []client.CollectionDefinition) error
 }