fix: Make peers sync secondary index

client/db.go

@@ -215,6 +215,15 @@ type Store interface {
 	// GetAllIndexes returns all the indexes that currently exist within this [Store].
 	GetAllIndexes(context.Context) (map[CollectionName][]IndexDescription, error)
 
+	// CreateDocIndex creates an index for the given document.
+	CreateDocIndex(context.Context, Collection, *Document) error
+
+	// UpdateDocIndex updates the index for the given document.
+	UpdateDocIndex(ctx context.Context, col Collection, oldDoc, newDoc *Document) error
+
+	// DeleteDocIndex deletes the index for the given document.
+	DeleteDocIndex(context.Context, Collection, *Document) error
+
 	// ExecRequest executes the given GQL request against the [Store].
 	ExecRequest(context.Context, string) *RequestResult
 }

client/index.go

@@ -10,6 +10,12 @@
 
 package client
 
+import (
+	"context"
+
+	"github.com/sourcenetwork/defradb/datastore"
+)
+
 // IndexFieldDescription describes how a field is being indexed.
 type IndexedFieldDescription struct {
 	// Name contains the name of the field.
@@ -30,6 +36,20 @@ type IndexDescription struct {
 	Unique bool
 }
 
+// CollectionIndex is an interface for indexing documents in a collection.
+type CollectionIndex interface {
+	// Save indexes a document by storing it
+	Save(context.Context, datastore.Txn, *Document) error
+	// Update updates an existing document in the index
+	Update(context.Context, datastore.Txn, *Document, *Document) error
+	// Delete deletes an existing document from the index
+	Delete(context.Context, datastore.Txn, *Document) error
+	// Name returns the name of the index
+	Name() string
+	// Description returns the description of the index
+	Description() IndexDescription
+}
+
 // CollectIndexedFields returns all fields that are indexed by all collection indexes.
 func (d CollectionDefinition) CollectIndexedFields() []FieldDefinition {
 	fieldsMap := make(map[string]bool)

db/collection_index.go

@@ -57,8 +57,8 @@
 	return col.DropIndex(ctx, indexName)
 }
 
-// getAllIndexes returns all the indexes in the database.
-func (db *db) getAllIndexes(
+// getAllIndexDescriptions returns all the index descriptions in the database.
+func (db *db) getAllIndexDescriptions(
 	ctx context.Context,
 	txn datastore.Txn,
 ) (map[client.CollectionName][]client.IndexDescription, error) {
@@ -107,6 +107,26 @@
 	return indexDescriptions, nil
 }
 
+func (db *db) getCollectionIndexes(
+	ctx context.Context,
+	txn datastore.Txn,
+	col client.Collection,
+) ([]CollectionIndex, error) {
+	indexDescriptions, err := db.fetchCollectionIndexDescriptions(ctx, txn, col.ID())
+	if err != nil {
+		return nil, err
+	}
+	colIndexes := make([]CollectionIndex, 0, len(indexDescriptions))
+	for _, indexDesc := range indexDescriptions {
+		index, err := NewCollectionIndex(col, indexDesc)
+		if err != nil {
+			return nil, err
+		}
+		colIndexes = append(colIndexes, index)
+	}
+	return colIndexes, nil
+}
+
 func (c *collection) indexNewDoc(ctx context.Context, txn datastore.Txn, doc *client.Document) error {
 	err := c.loadIndexes(ctx, txn)
 	if err != nil {
@@ -133,7 +153,8 @@
 	oldDoc, err := c.get(
 		ctx,
 		txn,
-		c.getPrimaryKeyFromDocID(doc.ID()), c.Definition().CollectIndexedFields(),
+		c.getPrimaryKeyFromDocID(doc.ID()),
+		c.Definition().CollectIndexedFields(),
 		false,
 	)
 	if err != nil {

db/index.go

@@ -25,16 +25,9 @@
 // It abstracts away common index functionality to be implemented
 // by different index types: non-unique, unique, and composite
 type CollectionIndex interface {
-	// Save indexes a document by storing it
-	Save(context.Context, datastore.Txn, *client.Document) error
-	// Update updates an existing document in the index
-	Update(context.Context, datastore.Txn, *client.Document, *client.Document) error
+	client.CollectionIndex
 	// RemoveAll removes all documents from the index
 	RemoveAll(context.Context, datastore.Txn) error
-	// Name returns the name of the index
-	Name() string
-	// Description returns the description of the index
-	Description() client.IndexDescription
 }
 
 func canConvertIndexFieldValue[T any](val any) bool {
@@ -248,6 +241,14 @@
 	return index.Save(ctx, txn, newDoc)
 }
 
+func (index *collectionSimpleIndex) Delete(
+	ctx context.Context,
+	txn datastore.Txn,
+	doc *client.Document,
+) error {
+	return index.deleteDocIndex(ctx, txn, doc)
+}
+
 func (index *collectionSimpleIndex) deleteDocIndex(
 	ctx context.Context,
 	txn datastore.Txn,
@@ -358,6 +359,14 @@
 	return key, val, nil
 }
 
+func (index *collectionUniqueIndex) Delete(
+	ctx context.Context,
+	txn datastore.Txn,
+	doc *client.Document,
+) error {
+	return index.deleteDocIndex(ctx, txn, doc)
+}
+
 func (index *collectionUniqueIndex) Update(
 	ctx context.Context,
 	txn datastore.Txn,

db/index_test.go

@@ -263,7 +263,7 @@ func (f *indexTestFixture) createCollectionIndexFor(
 }
 
 func (f *indexTestFixture) getAllIndexes() (map[client.CollectionName][]client.IndexDescription, error) {
-	return f.db.getAllIndexes(f.ctx, f.txn)
+	return f.db.getAllIndexDescriptions(f.ctx, f.txn)
 }
 
 func (f *indexTestFixture) getCollectionIndexes(colID uint32) ([]client.IndexDescription, error) {

db/txn_db.go

@@ -174,14 +174,154 @@
 	}
 	defer txn.Discard(ctx)
 
-	return db.getAllIndexes(ctx, txn)
+	return db.getAllIndexDescriptions(ctx, txn)
 }
 
 // GetAllIndexes gets all the indexes in the database.
 func (db *explicitTxnDB) GetAllIndexes(
 	ctx context.Context,
 ) (map[client.CollectionName][]client.IndexDescription, error) {
-	return db.getAllIndexes(ctx, db.txn)
+	return db.getAllIndexDescriptions(ctx, db.txn)
+}
+
+// CreateDocIndex creates a new index for the given document.
+func (db *implicitTxnDB) CreateDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	doc *client.Document,
+) error {
+	txn, err := db.NewTxn(ctx, true)
+	if err != nil {
+		return err
+	}
+	defer txn.Discard(ctx)
+
+	indexes, err := db.getCollectionIndexes(ctx, txn, col)
+	if err != nil {
+		return err
+	}
+
+	for _, index := range indexes {
+		err := index.Save(ctx, txn, doc)
+		if err != nil {
+			return err
+		}
+	}
+	return txn.Commit(ctx)
+}
+
+// CreateDocIndex creates a new index for the given document.
+func (db *explicitTxnDB) CreateDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	doc *client.Document,
+) error {
+	indexes, err := db.getCollectionIndexes(ctx, db.txn, col)
+	if err != nil {
+		return err
+	}
+
+	for _, index := range indexes {
+		err := index.Save(ctx, db.txn, doc)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// UpdateDocIndex updates the indexes for the given document.
+func (db *implicitTxnDB) UpdateDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	oldDoc *client.Document,
+	newDoc *client.Document,
+) error {
+	txn, err := db.NewTxn(ctx, true)
+	if err != nil {
+		return err
+	}
+	defer txn.Discard(ctx)
+
+	indexes, err := db.getCollectionIndexes(ctx, txn, col)
+	if err != nil {
+		return err
+	}
+
+	for _, index := range indexes {
+		err := index.Update(ctx, txn, oldDoc, newDoc)
+		if err != nil {
+			return err
+		}
+	}
+	return txn.Commit(ctx)
+}
+
+// UpdateDocIndex updates the indexes for the given document.
+func (db *explicitTxnDB) UpdateDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	oldDoc *client.Document,
+	newDoc *client.Document,
+) error {
+	indexes, err := db.getCollectionIndexes(ctx, db.txn, col)
+	if err != nil {
+		return err
+	}
+
+	for _, index := range indexes {
+		err := index.Update(ctx, db.txn, oldDoc, newDoc)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// DeleteDocIndex deletes the indexes for the given document.
+func (db *implicitTxnDB) DeleteDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	doc *client.Document,
+) error {
+	txn, err := db.NewTxn(ctx, true)
+	if err != nil {
+		return err
+	}
+	defer txn.Discard(ctx)
+
+	indexes, err := db.getCollectionIndexes(ctx, txn, col)
+	if err != nil {
+		return err
+	}
+
+	for _, index := range indexes {
+		err := index.Delete(ctx, txn, doc)
+		if err != nil {
+			return err
+		}
+	}
+	return txn.Commit(ctx)
+}
+
+// DeleteDocIndex deletes the indexes for the given document.
+func (db *explicitTxnDB) DeleteDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	doc *client.Document,
+) error {
+	indexes, err := db.getCollectionIndexes(ctx, db.txn, col)
+	if err != nil {
+		return err
+	}
+
+	for _, index := range indexes {
+		err := index.Delete(ctx, db.txn, doc)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 // AddSchema takes the provided GQL schema in SDL format, and applies it to the database,

http/client.go

@@ -344,7 +344,7 @@
 		return result
 	}
 	if res.Header.Get("Content-Type") == "text/event-stream" {
-		result.Pub = c.execRequestSubscription(ctx, res.Body)
+		result.Pub = c.execRequestSubscription(res.Body)
 		return result
 	}
 	// ignore close errors because they have
@@ -367,7 +367,7 @@
 	return result
 }
 
-func (c *Client) execRequestSubscription(ctx context.Context, r io.ReadCloser) *events.Publisher[events.Update] {
+func (c *Client) execRequestSubscription(r io.ReadCloser) *events.Publisher[events.Update] {
 	pubCh := events.New[events.Update](0, 0)
 	pub, err := events.NewPublisher[events.Update](pubCh, 0)
 	if err != nil {
@@ -434,3 +434,28 @@
 func (c *Client) MaxTxnRetries() int {
 	panic("client side database")
 }
+
+func (c *Client) CreateDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	doc *client.Document,
+) error {
+	panic("client side database")
+}
+
+func (c *Client) UpdateDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	oldDoc *client.Document,
+	newDoc *client.Document,
+) error {
+	panic("client side database")
+}
+
+func (w *Client) DeleteDocIndex(
+	ctx context.Context,
+	col client.Collection,
+	newDoc *client.Document,
+) error {
+	panic("client side database")
+}