feat: batched and chunked insertion+deletion of relation tuples

internal/e2e/transaction_cases_test.go

@@ -58,6 +58,51 @@ func runTransactionCases(c transactClient, m *namespaceTestManager) func(*testin
 			assert.Len(t, resp.RelationTuples, 0)
 		})
 
+		t.Run("case=large inserts and deletes", func(t *testing.T) {
+			ns := []*namespace.Namespace{
+				{Name: t.Name() + "1"},
+				{Name: t.Name() + "2"},
+			}
+			m.add(t, ns...)
+
+			var tuples []*ketoapi.RelationTuple
+			for range 12001 {
+				tuples = append(tuples, &ketoapi.RelationTuple{
+					Namespace: ns[0].Name,
+					Object:    "o",
+					Relation:  "rel",
+					SubjectSet: &ketoapi.SubjectSet{
+						Namespace: ns[1].Name,
+						Object:    "o",
+						Relation:  "rel",
+					},
+				},
+					&ketoapi.RelationTuple{
+						Namespace: ns[0].Name,
+						Object:    "o",
+						Relation:  "rel",
+						SubjectID: pointerx.Ptr("sid"),
+					},
+				)
+			}
+
+			c.transactTuples(t, tuples, nil)
+
+			resp := c.queryTuple(t, &ketoapi.RelationQuery{
+				Namespace: &ns[0].Name,
+			})
+			for i := range tuples {
+				assert.Contains(t, resp.RelationTuples, tuples[i])
+			}
+
+			c.transactTuples(t, nil, tuples)
+
+			resp = c.queryTuple(t, &ketoapi.RelationQuery{
+				Namespace: &ns[0].Name,
+			})
+			assert.Len(t, resp.RelationTuples, 0)
+		})
+
 		t.Run("case=expand-api-display-access docs code sample", func(t *testing.T) {
 			files := &namespace.Namespace{Name: t.Name() + "files"}
 			directories := &namespace.Namespace{Name: t.Name() + "directories"}

internal/persistence/sql/persister.go

@@ -6,11 +6,9 @@ package sql
 import (
 	"context"
 	"embed"
-	"reflect"
 
 	"github.com/gobuffalo/pop/v6"
 	"github.com/gofrs/uuid"
-	"github.com/ory/x/otelx"
 	"github.com/ory/x/popx"
 	"github.com/pkg/errors"
 
@@ -70,24 +68,6 @@ func (p *Persister) Connection(ctx context.Context) *pop.Connection {
 	return popx.GetConnection(ctx, p.conn.WithContext(ctx))
 }
 
-func (p *Persister) createWithNetwork(ctx context.Context, v interface{}) (err error) {
-	ctx, span := p.d.Tracer(ctx).Tracer().Start(ctx, "persistence.sql.createWithNetwork")
-	defer otelx.End(span, &err)
-
-	rv := reflect.ValueOf(v)
-
-	if rv.Kind() != reflect.Ptr && rv.Elem().Kind() != reflect.Struct {
-		panic("expected to get *struct in create")
-	}
-	nID := rv.Elem().FieldByName("NetworkID")
-	if !nID.IsValid() || !nID.CanSet() {
-		panic("expected struct to have a 'NetworkID uuid.UUID' field")
-	}
-	nID.Set(reflect.ValueOf(p.NetworkID(ctx)))
-
-	return p.Connection(ctx).Create(v)
-}
-
 func (p *Persister) queryWithNetwork(ctx context.Context) *pop.Query {
 	return p.Connection(ctx).Where("nid = ?", p.NetworkID(ctx))
 }

internal/persistence/sql/query_test.go

@@ -0,0 +1,129 @@
+package sql
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/gofrs/uuid"
+	"github.com/ory/x/uuidx"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ory/keto/internal/relationtuple"
+)
+
+func TestBuildDelete(t *testing.T) {
+	t.Parallel()
+	nid := uuidx.NewV4()
+
+	q, args, err := buildDelete(nid, nil)
+	assert.Error(t, err)
+	assert.Empty(t, q)
+	assert.Empty(t, args)
+
+	obj1, obj2, sub1, obj3 := uuidx.NewV4(), uuidx.NewV4(), uuidx.NewV4(), uuidx.NewV4()
+
+	q, args, err = buildDelete(nid, []*relationtuple.RelationTuple{
+		{
+			Namespace: "ns1",
+			Object:    obj1,
+			Relation:  "rel1",
+			Subject: &relationtuple.SubjectID{
+				ID: sub1,
+			},
+		},
+		{
+			Namespace: "ns2",
+			Object:    obj2,
+			Relation:  "rel2",
+			Subject: &relationtuple.SubjectSet{
+				Namespace: "ns3",
+				Object:    obj3,
+				Relation:  "rel3",
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	// parentheses are important here
+	assert.Equal(t, q, "DELETE FROM keto_relation_tuples WHERE ((namespace = ? AND object = ? AND relation = ? AND subject_id = ? AND subject_set_namespace IS NULL AND subject_set_object IS NULL AND subject_set_relation IS NULL) OR (namespace = ? AND object = ? AND relation = ? AND subject_id IS NULL AND subject_set_namespace = ? AND subject_set_object = ? AND subject_set_relation = ?)) AND nid = ?")
+	assert.Equal(t, []any{"ns1", obj1, "rel1", sub1, "ns2", obj2, "rel2", "ns3", obj3, "rel3", nid}, args)
+}
+
+func TestBuildInsert(t *testing.T) {
+	t.Parallel()
+	nid := uuidx.NewV4()
+
+	q, args, err := buildInsert(time.Now(), nid, nil)
+	assert.Error(t, err)
+	assert.Empty(t, q)
+	assert.Empty(t, args)
+
+	obj1, obj2, sub1, obj3 := uuidx.NewV4(), uuidx.NewV4(), uuidx.NewV4(), uuidx.NewV4()
+
+	now := time.Now()
+
+	q, args, err = buildInsert(now, nid, []*relationtuple.RelationTuple{
+		{
+			Namespace: "ns1",
+			Object:    obj1,
+			Relation:  "rel1",
+			Subject: &relationtuple.SubjectID{
+				ID: sub1,
+			},
+		},
+		{
+			Namespace: "ns2",
+			Object:    obj2,
+			Relation:  "rel2",
+			Subject: &relationtuple.SubjectSet{
+				Namespace: "ns3",
+				Object:    obj3,
+				Relation:  "rel3",
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	assert.Equal(t, q, "INSERT INTO keto_relation_tuples (shard_id, nid, namespace, object, relation, subject_id, subject_set_namespace, subject_set_object, subject_set_relation, commit_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?), (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
+	assert.Equal(t, []any{
+		args[0], // this is kind of cheating but we generate the shard id in the buildInsert function
+		nid,
+		"ns1",
+		obj1,
+		"rel1",
+		uuid.NullUUID{sub1, true},
+		sql.NullString{}, uuid.NullUUID{}, sql.NullString{},
+		now,
+
+		args[10], // again, cheating
+		nid,
+		"ns2",
+		obj2,
+		"rel2",
+		uuid.NullUUID{},
+		sql.NullString{"ns3", true}, uuid.NullUUID{obj3, true}, sql.NullString{"rel3", true},
+		now,
+	}, args)
+}
+
+func TestBuildInsertUUIDs(t *testing.T) {
+	t.Parallel()
+
+	nid := uuidx.NewV4()
+	foo, bar, baz := uuidx.NewV4(), uuidx.NewV4(), uuidx.NewV4()
+	uuids := []UUIDMapping{
+		{foo, "foo"},
+		{bar, "bar"},
+		{baz, "baz"},
+	}
+
+	q, args := buildInsertUUIDs(nid, uuids, "mysql")
+	assert.Equal(t, "INSERT IGNORE INTO keto_uuid_mappings (id, string_representation) VALUES (?,?),(?,?),(?,?)", q)
+	assert.Equal(t, []any{foo, "foo", bar, "bar", baz, "baz"}, args)
+
+	q, args = buildInsertUUIDs(nid, uuids, "anything else")
+	assert.Equal(t, "INSERT INTO keto_uuid_mappings (id, string_representation) VALUES (?,?),(?,?),(?,?) ON CONFLICT (id) DO NOTHING", q)
+	assert.Equal(t, []any{foo, "foo", bar, "bar", baz, "baz"}, args)
+}