WIP - Implement branchable collections

client/collection_description.go

@@ -88,6 +88,8 @@ type CollectionDescription struct {
 	// At the moment this can only be set to `false` if this collection sources its data from
 	// another collection/query (is a View).
 	IsMaterialized bool
+
+	IsBranchable bool
 }
 
 // QuerySource represents a collection data source from a query.
@@ -189,6 +191,7 @@ type collectionDescription struct {
 	RootID          uint32
 	SchemaVersionID string
 	IsMaterialized  bool
+	IsBranchable    bool
 	Policy          immutable.Option[PolicyDescription]
 	Indexes         []IndexDescription
 	Fields          []CollectionFieldDescription
@@ -209,6 +212,7 @@ func (c *CollectionDescription) UnmarshalJSON(bytes []byte) error {
 	c.RootID = descMap.RootID
 	c.SchemaVersionID = descMap.SchemaVersionID
 	c.IsMaterialized = descMap.IsMaterialized
+	c.IsBranchable = descMap.IsBranchable
 	c.Indexes = descMap.Indexes
 	c.Fields = descMap.Fields
 	c.Sources = make([]any, len(descMap.Sources))

internal/core/block/block.go

@@ -44,6 +44,7 @@ func init() {
 		&crdt.LWWRegDelta{},
 		&crdt.CompositeDAGDelta{},
 		&crdt.CounterDelta{},
+		&crdt.CollectionDelta{},
 	)
 
 	EncryptionSchema, EncryptionSchemaPrototype = mustSetSchema(

internal/core/crdt/collection.go

@@ -0,0 +1,74 @@
+// Copyright 2024 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 crdt
+
+import (
+	"context"
+
+	"github.com/sourcenetwork/defradb/datastore"
+	"github.com/sourcenetwork/defradb/internal/core"
+	"github.com/sourcenetwork/defradb/internal/keys"
+)
+
+type Collection struct {
+	store datastore.DSReaderWriter
+	// schemaVersionKey is the schema version datastore key at the time of commit.
+	//
+	// It can be used to identify the collection datastructure state at the time of commit.
+	schemaVersionKey keys.CollectionSchemaVersionKey
+}
+
+var _ core.ReplicatedData = (*Collection)(nil)
+
+func NewCollection(store datastore.DSReaderWriter, schemaVersionKey keys.CollectionSchemaVersionKey) *Collection {
+	return &Collection{
+		store:            store,
+		schemaVersionKey: schemaVersionKey,
+	}
+}
+
+func (c *Collection) Merge(ctx context.Context, other core.Delta) error {
+	// Collection merges don't actually need to do anything, as the delta is empty,
+	// and doc-level merges are handled by the document commits.
+
+	// todo - make sure this is well tested, you might be wrong!
+	return nil
+}
+
+func (c *Collection) Append() *CollectionDelta {
+	return &CollectionDelta{
+		SchemaVersionID: c.schemaVersionKey.SchemaVersionID,
+	}
+}
+
+type CollectionDelta struct {
+	Priority uint64
+	// todo: This is temporary pending a global collection id (link to ticket)
+	SchemaVersionID string
+}
+
+var _ core.Delta = (*CollectionDelta)(nil)
+
+func (delta *CollectionDelta) IPLDSchemaBytes() []byte {
+	return []byte(`
+	type CollectionDelta struct {
+		priority  		Int
+		schemaVersionID String
+	}`)
+}
+
+func (d *CollectionDelta) GetPriority() uint64 {
+	return d.Priority
+}
+
+func (d *CollectionDelta) SetPriority(priority uint64) {
+	d.Priority = priority
+}

internal/core/crdt/ipld_union.go

@@ -17,6 +17,7 @@ type CRDT struct {
 	LWWRegDelta       *LWWRegDelta
 	CompositeDAGDelta *CompositeDAGDelta
 	CounterDelta      *CounterDelta
+	CollectionDelta   *CollectionDelta
 }
 
 // NewCRDT returns a new CRDT.
@@ -28,6 +29,8 @@ func NewCRDT(delta core.Delta) CRDT {
 		return CRDT{CompositeDAGDelta: d}
 	case *CounterDelta:
 		return CRDT{CounterDelta: d}
+	case *CollectionDelta:
+		return CRDT{CollectionDelta: d}
 	}
 	return CRDT{}
 }
@@ -41,6 +44,7 @@ func (c CRDT) IPLDSchemaBytes() []byte {
 		| LWWRegDelta "lww"
 		| CompositeDAGDelta "composite"
 		| CounterDelta "counter"
+		| CollectionDelta "collection"
 	} representation keyed`)
 }
 
@@ -53,6 +57,8 @@ func (c CRDT) GetDelta() core.Delta {
 		return c.CompositeDAGDelta
 	case c.CounterDelta != nil:
 		return c.CounterDelta
+	case c.CollectionDelta != nil:
+		return c.CollectionDelta
 	}
 	return nil
 }
@@ -66,6 +72,8 @@ func (c CRDT) GetPriority() uint64 {
 		return c.CompositeDAGDelta.GetPriority()
 	case c.CounterDelta != nil:
 		return c.CounterDelta.GetPriority()
+	case c.CollectionDelta != nil:
+		return c.CollectionDelta.GetPriority()
 	}
 	return 0
 }
@@ -90,6 +98,8 @@ func (c CRDT) GetDocID() []byte {
 		return c.CompositeDAGDelta.DocID
 	case c.CounterDelta != nil:
 		return c.CounterDelta.DocID
+	case c.CollectionDelta != nil:
+		return nil
 	}
 	return nil
 }
@@ -103,6 +113,8 @@ func (c CRDT) GetSchemaVersionID() string {
 		return c.CompositeDAGDelta.SchemaVersionID
 	case c.CounterDelta != nil:
 		return c.CounterDelta.SchemaVersionID
+	case c.CollectionDelta != nil:
+		return c.CollectionDelta.SchemaVersionID
 	}
 	return ""
 }
@@ -135,6 +147,11 @@ func (c CRDT) Clone() CRDT {
 			Nonce:           c.CounterDelta.Nonce,
 			Data:            c.CounterDelta.Data,
 		}
+	case c.CollectionDelta != nil:
+		cloned.CollectionDelta = &CollectionDelta{
+			Priority:        c.CollectionDelta.Priority,
+			SchemaVersionID: c.CollectionDelta.SchemaVersionID,
+		}
 	}
 	return cloned
 }
@@ -172,3 +189,8 @@ func (c CRDT) SetData(data []byte) {
 func (c CRDT) IsComposite() bool {
 	return c.CompositeDAGDelta != nil
 }
+
+// IsCollection returns true if the CRDT is a collection CRDT.
+func (c CRDT) IsCollection() bool {
+	return c.CollectionDelta != nil
+}

internal/db/collection.go

@@ -711,6 +711,29 @@ func (c *collection) save(
 		doc.SetHead(link.Cid)
 	})
 
+	if c.def.Description.IsBranchable {
+		collectionCRDT := merklecrdt.NewMerkleCollection(
+			txn,
+			keys.NewCollectionSchemaVersionKey(c.Schema().VersionID, c.ID()),
+			keys.NewHeadstoreColKey(c.def.Description.RootID),
+		)
+
+		link, headNode, err := collectionCRDT.Save(ctx, []coreblock.DAGLink{{Link: link}})
+		if err != nil {
+			return err
+		}
+
+		updateEvent := event.Update{
+			Cid:        link.Cid,
+			SchemaRoot: c.Schema().Root,
+			Block:      headNode,
+		}
+
+		txn.OnSuccess(func() {
+			c.db.events.Publish(event.NewMessage(event.UpdateName, updateEvent))
+		})
+	}
+
 	return nil
 }
 

internal/db/definition_validation.go

@@ -146,6 +146,7 @@ var collectionUpdateValidators = append(
 		validateIDExists,
 		validateSchemaVersionIDNotMutated,
 		validateCollectionNotRemoved,
+		validateCollectionIsBranchable,
 	),
 	globalValidators...,
 )
@@ -1036,3 +1037,19 @@ func validateCollectionFieldDefaultValue(
 
 	return nil
 }
+func validateCollectionIsBranchable(
+	ctx context.Context,
+	db *db,
+	newState *definitionState,
+	oldState *definitionState,
+) error {
+	for _, newCol := range newState.collections {
+		oldCol := oldState.collectionsByID[newCol.ID]
+
+		if newCol.IsBranchable != oldCol.IsBranchable {
+			return NewErrColMutatingIsBranchable(newCol.Name.Value())
+		}
+	}
+
+	return nil
+}

internal/db/errors.go

@@ -104,6 +104,7 @@ const (
 	errFailedToHandleEncKeysReceivedEvent       string = "failed to handle encryption-keys-received event"
 	errSelfReferenceWithoutSelf                 string = "must specify 'Self' kind for self referencing relations"
 	errColNotMaterialized                       string = "non-materialized collections are not supported"
+	errColMutatingIsBranchable                  string = "mutating IsBranchable is not supported"
 	errMaterializedViewAndACPNotSupported       string = "materialized views do not support ACP"
 	errInvalidDefaultFieldValue                 string = "default field value is invalid"
 )
@@ -152,6 +153,7 @@ var (
 	ErrContextDone                              = errors.New("context done")
 	ErrFailedToRetryDoc                         = errors.New("failed to retry doc")
 	ErrTimeoutDocRetry                          = errors.New("timeout while retrying doc")
+	ErrColMutatingIsBranchable                  = errors.New(errColMutatingIsBranchable)
 )
 
 // NewErrFailedToGetHeads returns a new error indicating that the heads of a document
@@ -676,6 +678,13 @@ func NewErrColNotMaterialized(collection string) error {
 	)
 }
 
+func NewErrColMutatingIsBranchable(collection string) error {
+	return errors.New(
+		errColMutatingIsBranchable,
+		errors.NewKV("Collection", collection),
+	)
+}
+
 func NewErrMaterializedViewAndACPNotSupported(collection string) error {
 	return errors.New(
 		errMaterializedViewAndACPNotSupported,

internal/db/fetcher/dag.go

@@ -31,13 +31,18 @@ type HeadFetcher struct {
 func (hf *HeadFetcher) Start(
 	ctx context.Context,
 	txn datastore.Txn,
-	prefix keys.HeadstoreDocKey,
+	prefix immutable.Option[keys.HeadstoreKey],
 	fieldId immutable.Option[string],
 ) error {
 	hf.fieldId = fieldId
 
+	var prefixString string
+	if prefix.HasValue() {
+		prefixString = prefix.Value().ToString()
+	}
+
 	q := dsq.Query{
-		Prefix: prefix.ToString(),
+		Prefix: prefixString,
 		Orders: []dsq.Order{dsq.OrderByKey{}},
 	}
 
@@ -64,17 +69,32 @@ func (hf *HeadFetcher) FetchNext() (*cid.Cid, error) {
 		return nil, nil
 	}
 
-	headStoreKey, err := keys.NewHeadstoreDocKey(res.Key)
+	headStoreKey, err := keys.NewHeadstoreKey(res.Key)
 	if err != nil {
 		return nil, err
 	}
 
-	if hf.fieldId.HasValue() && hf.fieldId.Value() != headStoreKey.FieldID {
-		// FieldIds do not match, continue to next row
-		return hf.FetchNext()
+	if hf.fieldId.HasValue() {
+		switch typedHeadStoreKey := headStoreKey.(type) {
+		case keys.HeadstoreDocKey:
+			if hf.fieldId.Value() != typedHeadStoreKey.FieldID {
+				// FieldIds do not match, continue to next row
+				return hf.FetchNext()
+			}
+
+			return &typedHeadStoreKey.Cid, nil
+
+		case keys.HeadstoreColKey:
+			if hf.fieldId.Value() == "" {
+				return &typedHeadStoreKey.Cid, nil
+			} else {
+				return nil, nil
+			}
+		}
 	}
 
-	return &headStoreKey.Cid, nil
+	cid := headStoreKey.GetCid()
+	return &cid, nil
 }
 
 func (hf *HeadFetcher) Close() error {

internal/db/merge.go

@@ -411,7 +411,7 @@ func decryptBlock(
 ) (*coreblock.Block, error) {
 	_, encryptor := encryption.EnsureContextWithEncryptor(ctx)
 
-	if block.Delta.IsComposite() {
+	if block.Delta.IsComposite() || block.Delta.IsCollection() {
 		// for composite blocks there is nothing to decrypt
 		return block, nil
 	}

internal/keys/headstore.go

@@ -10,6 +10,30 @@
 
 package keys
 
+import (
+	"strings"
+
+	"github.com/ipfs/go-cid"
+)
+
 const (
 	HEADSTORE_DOC = "/d"
+	HEADSTORE_COL = "/c"
 )
+
+type HeadstoreKey interface {
+	Walkable
+	GetCid() cid.Cid
+	WithCid(c cid.Cid) HeadstoreKey
+}
+
+func NewHeadstoreKey(key string) (HeadstoreKey, error) {
+	switch {
+	case strings.HasPrefix(key, HEADSTORE_DOC):
+		return NewHeadstoreDocKey(key)
+	case strings.HasPrefix(key, HEADSTORE_COL):
+		return NewHeadstoreColKeyFromString(key)
+	default:
+		return nil, ErrInvalidKey
+	}
+}