Add extended relationships filtering to watch API

e2e/go.mod

@@ -5,7 +5,7 @@ go 1.21
 toolchain go1.21.1
 
 require (
-	github.com/authzed/authzed-go v0.10.1
+	github.com/authzed/authzed-go v0.10.2-0.20240206183056-781a5f5d1b3c
 	github.com/authzed/grpcutil v0.0.0-20240123092924-129dc0a6a6e1
 	github.com/authzed/spicedb v1.29.1
 	github.com/brianvoe/gofakeit/v6 v6.23.2
@@ -50,7 +50,7 @@ require (
 	go.opentelemetry.io/otel/metric v1.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
+	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect
 	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/net v0.20.0 // indirect
 	golang.org/x/sync v0.6.0 // indirect

e2e/go.sum

@@ -20,8 +20,8 @@ github.com/Masterminds/squirrel v1.5.4 h1:uUcX/aBc8O7Fg9kaISIUsHXdKuqehiXAMQTYX8
 github.com/Masterminds/squirrel v1.5.4/go.mod h1:NNaOrjSoIDfDA40n7sr2tPNZRfjzjA400rg+riTZj10=
 github.com/antlr/antlr4/runtime/Go/antlr/v4 v4.0.0-20230512164433-5d1fd1a340c9 h1:goHVqTbFX3AIo0tzGr14pgfAW2ZfPChKO21Z9MGf/gk=
 github.com/antlr/antlr4/runtime/Go/antlr/v4 v4.0.0-20230512164433-5d1fd1a340c9/go.mod h1:pSwJ0fSY5KhvocuWSx4fz3BA8OrA1bQn+K1Eli3BRwM=
-github.com/authzed/authzed-go v0.10.1 h1:0aX2Ox9PPPknID92kLs/FnmhCmfl6Ni16v3ZTLsds5M=
-github.com/authzed/authzed-go v0.10.1/go.mod h1:ZsaFPCiMjwT0jLW0gCyYzh3elHqhKDDGGRySyykXwqc=
+github.com/authzed/authzed-go v0.10.2-0.20240206183056-781a5f5d1b3c h1:w71KppS/+mCDkNXR8vmwXGVt/1dLt+axcIq+qK7f7To=
+github.com/authzed/authzed-go v0.10.2-0.20240206183056-781a5f5d1b3c/go.mod h1:bS4eeTw/ZpCunZHePrt1MAcvOggAwL8Djh8cx+CR33g=
 github.com/authzed/cel-go v0.17.5 h1:lfpkNrR99B5QRHg5qdG9oLu/kguVlZC68VJuMk8tH9Y=
 github.com/authzed/cel-go v0.17.5/go.mod h1:XL/zEq5hKGVF8aOdMbG7w+BQPihLjY2W8N+UIygDA2I=
 github.com/authzed/grpcutil v0.0.0-20240123092924-129dc0a6a6e1 h1:zBfQzia6Hz45pJBeURTrv1b6HezmejB6UmiGuBilHZM=
@@ -256,8 +256,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
 golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI=
-golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo=
+golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 h1:hNQpMuAJe5CtcUqCXaWga3FHu+kQvCqcsoVaQgSV60o=
+golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=

internal/datastore/common/sql.go

@@ -351,47 +351,49 @@ func (sqf SchemaQueryFilterer) MustFilterWithRelationshipsFilter(filter datastor
 }
 
 func (sqf SchemaQueryFilterer) FilterWithRelationshipsFilter(filter datastore.RelationshipsFilter) (SchemaQueryFilterer, error) {
+	copy := sqf
+
 	if filter.OptionalResourceType != "" {
-		sqf = sqf.FilterToResourceType(filter.OptionalResourceType)
+		copy = copy.FilterToResourceType(filter.OptionalResourceType)
 	}
 
 	if filter.OptionalResourceRelation != "" {
-		sqf = sqf.FilterToRelation(filter.OptionalResourceRelation)
+		copy = copy.FilterToRelation(filter.OptionalResourceRelation)
 	}
 
 	if len(filter.OptionalResourceIds) > 0 && filter.OptionalResourceIDPrefix != "" {
-		return sqf, spiceerrors.MustBugf("cannot filter by both resource IDs and ID prefix")
+		return copy, spiceerrors.MustBugf("cannot filter by both resource IDs and ID prefix")
 	}
 
 	if len(filter.OptionalResourceIds) > 0 {
-		usqf, err := sqf.FilterToResourceIDs(filter.OptionalResourceIds)
+		usqf, err := copy.FilterToResourceIDs(filter.OptionalResourceIds)
 		if err != nil {
-			return sqf, err
+			return copy, err
 		}
-		sqf = usqf
+		copy = usqf
 	}
 
 	if len(filter.OptionalResourceIDPrefix) > 0 {
-		usqf, err := sqf.FilterWithResourceIDPrefix(filter.OptionalResourceIDPrefix)
+		usqf, err := copy.FilterWithResourceIDPrefix(filter.OptionalResourceIDPrefix)
 		if err != nil {
-			return sqf, err
+			return copy, err
 		}
-		sqf = usqf
+		copy = usqf
 	}
 
 	if len(filter.OptionalSubjectsSelectors) > 0 {
-		usqf, err := sqf.FilterWithSubjectsSelectors(filter.OptionalSubjectsSelectors...)
+		usqf, err := copy.FilterWithSubjectsSelectors(filter.OptionalSubjectsSelectors...)
 		if err != nil {
-			return sqf, err
+			return copy, err
 		}
-		sqf = usqf
+		copy = usqf
 	}
 
 	if filter.OptionalCaveatName != "" {
-		sqf = sqf.FilterWithCaveatName(filter.OptionalCaveatName)
+		copy = copy.FilterWithCaveatName(filter.OptionalCaveatName)
 	}
 
-	return sqf, nil
+	return copy, nil
 }
 
 // MustFilterWithSubjectsSelectors returns a new SchemaQueryFilterer that is limited to resources with

internal/datastore/crdb/readwrite.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 
 	sq "github.com/Masterminds/squirrel"
 	v1 "github.com/authzed/authzed-go/proto/authzed/api/v1"
@@ -211,6 +212,14 @@ func (rwt *crdbReadWriteTXN) DeleteRelationships(ctx context.Context, filter *v1
 	if filter.OptionalRelation != "" {
 		query = query.Where(sq.Eq{colRelation: filter.OptionalRelation})
 	}
+	if filter.OptionalResourceIdPrefix != "" {
+		if strings.Contains(filter.OptionalResourceIdPrefix, "%") {
+			return false, fmt.Errorf("unable to delete relationships with a prefix containing the %% character")
+		}
+
+		query = query.Where(sq.Like{colObjectID: filter.OptionalResourceIdPrefix + "%"})
+	}
+
 	rwt.addOverlapKey(filter.ResourceType)
 
 	// Add clauses for the SubjectFilter

internal/datastore/memdb/readonly.go

@@ -8,14 +8,13 @@ import (
 	"sort"
 	"strings"
 
-	"github.com/authzed/spicedb/pkg/spiceerrors"
-
 	"github.com/hashicorp/go-memdb"
 
 	"github.com/authzed/spicedb/internal/datastore/common"
 	"github.com/authzed/spicedb/pkg/datastore"
 	"github.com/authzed/spicedb/pkg/datastore/options"
 	core "github.com/authzed/spicedb/pkg/proto/core/v1"
+	"github.com/authzed/spicedb/pkg/spiceerrors"
 )
 
 type txFactory func() (*memdb.Txn, error)
@@ -264,8 +263,12 @@ func (r *memdbReader) mustLock() {
 }
 
 func iteratorForFilter(txn *memdb.Txn, filter datastore.RelationshipsFilter) (memdb.ResultIterator, error) {
+	// "_prefix" is a specialized index suffix used by github.com/hashicorp/go-memdb to match on
+	// a prefix of a string.
+	// See: https://github.com/hashicorp/go-memdb/blob/9940d4a14258e3b887bfb4bc6ebc28f65461a01c/txn.go#L531
 	index := indexNamespace + "_prefix"
-	args := []any{}
+
+	var args []any
 	if filter.OptionalResourceType != "" {
 		args = append(args, filter.OptionalResourceType)
 		index = indexNamespace

internal/datastore/mysql/readwrite.go

@@ -226,6 +226,13 @@ func (rwt *mysqlReadWriteTXN) DeleteRelationships(ctx context.Context, filter *v
 	if filter.OptionalRelation != "" {
 		query = query.Where(sq.Eq{colRelation: filter.OptionalRelation})
 	}
+	if filter.OptionalResourceIdPrefix != "" {
+		if strings.Contains(filter.OptionalResourceIdPrefix, "%") {
+			return false, fmt.Errorf("unable to delete relationships with a prefix containing the %% character")
+		}
+
+		query = query.Where(sq.Like{colObjectID: filter.OptionalResourceIdPrefix + "%"})
+	}
 
 	// Add clauses for the SubjectFilter
 	if subjectFilter := filter.OptionalSubjectFilter; subjectFilter != nil {

internal/datastore/postgres/readwrite.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/authzed/spicedb/pkg/datastore/options"
 	"github.com/authzed/spicedb/pkg/spiceerrors"
@@ -288,13 +289,24 @@ func (rwt *pgReadWriteTXN) DeleteRelationships(ctx context.Context, filter *v1.R
 
 func (rwt *pgReadWriteTXN) deleteRelationshipsWithLimit(ctx context.Context, filter *v1.RelationshipFilter, limit uint64) (bool, error) {
 	// Construct a select query for the relationships to be removed.
-	query := selectForDelete.Where(sq.Eq{colNamespace: filter.ResourceType})
+	query := selectForDelete
+
+	if filter.ResourceType != "" {
+		query = query.Where(sq.Eq{colNamespace: filter.ResourceType})
+	}
 	if filter.OptionalResourceId != "" {
 		query = query.Where(sq.Eq{colObjectID: filter.OptionalResourceId})
 	}
 	if filter.OptionalRelation != "" {
 		query = query.Where(sq.Eq{colRelation: filter.OptionalRelation})
 	}
+	if filter.OptionalResourceIdPrefix != "" {
+		if strings.Contains(filter.OptionalResourceIdPrefix, "%") {
+			return false, fmt.Errorf("unable to delete relationships with a prefix containing the %% character")
+		}
+
+		query = query.Where(sq.Like{colObjectID: filter.OptionalResourceIdPrefix + "%"})
+	}
 
 	// Add clauses for the SubjectFilter
 	if subjectFilter := filter.OptionalSubjectFilter; subjectFilter != nil {
@@ -315,16 +327,14 @@ func (rwt *pgReadWriteTXN) deleteRelationshipsWithLimit(ctx context.Context, fil
 	}
 
 	args = append(args, rwt.newXID)
-	if len(args) != 3 {
-		return false, spiceerrors.MustBugf("expected 3 arguments, got %d", len(args))
-	}
 
 	// Construct a CTE to update the relationships as removed.
 	cteSQL := fmt.Sprintf(
-		"WITH found_tuples AS (%s)\nUPDATE %s SET %s = $3 WHERE (%s, %s, %s, %s, %s, %s) IN (select * from found_tuples)",
+		"WITH found_tuples AS (%s)\nUPDATE %s SET %s = $%d WHERE (%s, %s, %s, %s, %s, %s) IN (select * from found_tuples)",
 		selectSQL,
 		tableTuple,
 		colDeletedXid,
+		len(args),
 		colNamespace,
 		colObjectID,
 		colRelation,
@@ -353,6 +363,13 @@ func (rwt *pgReadWriteTXN) deleteRelationships(ctx context.Context, filter *v1.R
 	if filter.OptionalRelation != "" {
 		query = query.Where(sq.Eq{colRelation: filter.OptionalRelation})
 	}
+	if filter.OptionalResourceIdPrefix != "" {
+		if strings.Contains(filter.OptionalResourceIdPrefix, "%") {
+			return fmt.Errorf("unable to delete relationships with a prefix containing the %% character")
+		}
+
+		query = query.Where(sq.Like{colObjectID: filter.OptionalResourceIdPrefix + "%"})
+	}
 
 	// Add clauses for the SubjectFilter
 	if subjectFilter := filter.OptionalSubjectFilter; subjectFilter != nil {

internal/datastore/spanner/readwrite.go

@@ -3,6 +3,7 @@ package spanner
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"cloud.google.com/go/spanner"
 	sq "github.com/Masterminds/squirrel"
@@ -122,7 +123,11 @@ func deleteWithFilter(ctx context.Context, rwt *spanner.ReadWriteTransaction, fi
 
 func deleteWithFilterAndLimit(ctx context.Context, rwt *spanner.ReadWriteTransaction, filter *v1.RelationshipFilter, disableStats bool, delLimit uint64) (int64, error) {
 	query := queryTuplesForDelete
-	query = applyFilterToQuery(query, filter)
+	filteredQuery, err := applyFilterToQuery(query, filter)
+	if err != nil {
+		return -1, err
+	}
+	query = filteredQuery
 	query = query.Limit(delLimit)
 
 	sql, args, err := query.ToSql()
@@ -169,7 +174,11 @@ func deleteWithFilterAndLimit(ctx context.Context, rwt *spanner.ReadWriteTransac
 
 func deleteWithFilterAndNoLimit(ctx context.Context, rwt *spanner.ReadWriteTransaction, filter *v1.RelationshipFilter, disableStats bool) (int64, error) {
 	query := sql.Delete(tableRelationship)
-	query = applyFilterToQuery(query, filter)
+	filteredQuery, err := applyFilterToQuery(query, filter)
+	if err != nil {
+		return -1, err
+	}
+	query = filteredQuery
 
 	sql, args, err := query.ToSql()
 	if err != nil {
@@ -184,7 +193,7 @@ type builder[T any] interface {
 	Where(pred interface{}, args ...interface{}) T
 }
 
-func applyFilterToQuery[T builder[T]](query T, filter *v1.RelationshipFilter) T {
+func applyFilterToQuery[T builder[T]](query T, filter *v1.RelationshipFilter) (T, error) {
 	// Add clauses for the ResourceFilter
 	if filter.ResourceType != "" {
 		query = query.Where(sq.Eq{colNamespace: filter.ResourceType})
@@ -195,6 +204,13 @@ func applyFilterToQuery[T builder[T]](query T, filter *v1.RelationshipFilter) T
 	if filter.OptionalRelation != "" {
 		query = query.Where(sq.Eq{colRelation: filter.OptionalRelation})
 	}
+	if filter.OptionalResourceIdPrefix != "" {
+		if strings.Contains(filter.OptionalResourceIdPrefix, "%") {
+			return query, fmt.Errorf("unable to delete relationships with a prefix containing the %% character")
+		}
+
+		query = query.Where(sq.Like{colObjectID: filter.OptionalResourceIdPrefix + "%"})
+	}
 
 	// Add clauses for the SubjectFilter
 	if subjectFilter := filter.OptionalSubjectFilter; subjectFilter != nil {
@@ -207,7 +223,7 @@ func applyFilterToQuery[T builder[T]](query T, filter *v1.RelationshipFilter) T
 		}
 	}
 
-	return query
+	return query, nil
 }
 
 func upsertVals(r *core.RelationTuple) []any {

internal/services/v1/experimental.go

@@ -693,5 +693,6 @@ func decodeCursor(ds datastore.Datastore, encoded *v1.Cursor) (datastore.Revisio
 		return datastore.NoRevision, "", nil, errors.New("malformed cursor: invalid encoded relation tuple")
 	}
 
+	// Returns the current namespace and the cursor.
 	return atRevision, decoded.GetV1().GetSections()[0], cur, nil
 }