chore(data): resolve five data-plane hygiene items from index audit (#178)

server/internal/api/api.go

@@ -65,7 +65,7 @@ type MemoryService interface {
 	Restore(context.Context, memory.LifecycleCommand) (*memory.MutationResult, error)
 	Forget(context.Context, memory.ForgetCommand) (*memory.ForgetResult, error)
 	CreateRelation(context.Context, memory.CreateRelationCommand) (*memory.CreateRelationResult, error)
-	ListRelations(context.Context, memory.ListRelationsQuery) ([]memory.Relation, error)
+	ListRelations(context.Context, memory.ListRelationsQuery) (*memory.ListRelationsResult, error)
 }
 
 // DurableContextService is the scoped durable-context port (mem#70). Handlers

server/internal/api/handlers_memory.go

@@ -763,18 +763,21 @@ func (s *Server) handleListMemoryRelations(w http.ResponseWriter, r *http.Reques
 	}
 
 	tok := r.Context().Value(ctxToken).(*auth.Token)
-	relations, err := s.Memory.ListRelations(r.Context(), memory.ListRelationsQuery{
+	result, err := s.Memory.ListRelations(r.Context(), memory.ListRelationsQuery{
 		WorkspaceID:  currentWorkspace(r).ID,
 		MemoryID:     id,
 		Direction:    direction,
 		RelationType: relationType,
 		AllowedPaths: tok.Paths,
 		Limit:        limit,
+		Cursor:       r.URL.Query().Get("cursor"),
 	})
 	if err != nil {
 		switch {
 		case errors.Is(err, memory.ErrInvalidCommand):
 			writeError(w, http.StatusBadRequest, "invalid_relation_query", err.Error())
+		case errors.Is(err, memory.ErrInvalidCursor):
+			writeError(w, http.StatusBadRequest, "invalid_cursor", err.Error())
 		case errors.Is(err, memory.ErrNotFound):
 			writeError(w, http.StatusNotFound, "not_found", "memory not found")
 		case errors.Is(err, memory.ErrForgotten):
@@ -789,8 +792,12 @@ func (s *Server) handleListMemoryRelations(w http.ResponseWriter, r *http.Reques
 		}
 		return
 	}
-	if relations == nil {
-		relations = []memory.Relation{}
+	resp := map[string]any{"relations": result.Relations}
+	if result.Relations == nil {
+		resp["relations"] = []memory.Relation{}
 	}
-	writeJSON(w, http.StatusOK, map[string]any{"relations": relations})
+	if result.NextCursor != "" {
+		resp["next_cursor"] = result.NextCursor
+	}
+	writeJSON(w, http.StatusOK, resp)
 }

server/internal/api/handlers_memory_test.go

@@ -107,10 +107,10 @@ func (s *memoryServiceStub) CreateRelation(
 func (s *memoryServiceStub) ListRelations(
 	_ context.Context,
 	q memory.ListRelationsQuery,
-) ([]memory.Relation, error) {
+) (*memory.ListRelationsResult, error) {
 	s.calls++
 	s.listRelationsQuery = q
-	return s.relations, s.controlErr
+	return &memory.ListRelationsResult{Relations: s.relations}, s.controlErr
 }
 
 func memoryHandlerContext(req *http.Request, paths []string) (*http.Request, uuid.UUID, uuid.UUID, uuid.UUID) {

server/internal/db/migrations/0026_data_plane_hygiene.sql

@@ -0,0 +1,85 @@
+-- +goose Up
+-- Data-plane hygiene from the index audit (#178).
+-- Three independent fixes bundled into one migration because each is a single
+-- DDL statement and none warrants its own schema version.
+
+-- Item 1: embeddings_text uniqueness on (file_id, chunk_index).
+-- The write path (indexer.go) DELETEs all chunks for a file before re-inserting,
+-- so duplicates should never exist in practice. Deduplicate defensively before
+-- adding the constraint: if any duplicates survived, keep the row with the
+-- smallest id (earliest insert).
+-- +goose StatementBegin
+DELETE FROM embeddings_text
+ WHERE id NOT IN (
+    SELECT DISTINCT ON (file_id, chunk_index) id
+      FROM embeddings_text
+     ORDER BY file_id, chunk_index, id
+ );
+-- +goose StatementEnd
+
+-- +goose StatementBegin
+ALTER TABLE embeddings_text
+    ADD CONSTRAINT uq_embeddings_text_file_chunk UNIQUE (file_id, chunk_index);
+-- +goose StatementEnd
+
+-- Item 2: partial indexes for ON DELETE SET NULL lookups on memories.
+-- Without these, every file or user delete takes a RowExclusiveLock on memories
+-- and performs a sequential scan to find the rows to null out.
+-- +goose StatementBegin
+CREATE INDEX IF NOT EXISTS idx_memories_source_file_id
+    ON memories (source_file_id) WHERE source_file_id IS NOT NULL;
+-- +goose StatementEnd
+
+-- +goose StatementBegin
+CREATE INDEX IF NOT EXISTS idx_memories_created_by_user_id
+    ON memories (created_by_user_id) WHERE created_by_user_id IS NOT NULL;
+-- +goose StatementEnd
+
+-- Item 3: memory_relations FKs must cascade with memories.
+-- memories.workspace_id is ON DELETE CASCADE, so a workspace delete removes
+-- memories rows. Without matching cascade on memory_relations, the delete then
+-- fails on any edge touching those memories. Align the referential actions.
+-- +goose StatementBegin
+ALTER TABLE memory_relations
+    DROP CONSTRAINT IF EXISTS memory_relations_workspace_id_fkey,
+    DROP CONSTRAINT IF EXISTS memory_relations_source_id_fkey,
+    DROP CONSTRAINT IF EXISTS memory_relations_target_id_fkey;
+-- +goose StatementEnd
+
+-- +goose StatementBegin
+ALTER TABLE memory_relations
+    ADD CONSTRAINT memory_relations_workspace_id_fkey
+        FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE,
+    ADD CONSTRAINT memory_relations_source_id_fkey
+        FOREIGN KEY (source_id) REFERENCES memories(id) ON DELETE CASCADE,
+    ADD CONSTRAINT memory_relations_target_id_fkey
+        FOREIGN KEY (target_id) REFERENCES memories(id) ON DELETE CASCADE;
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+ALTER TABLE memory_relations
+    DROP CONSTRAINT IF EXISTS memory_relations_workspace_id_fkey,
+    DROP CONSTRAINT IF EXISTS memory_relations_source_id_fkey,
+    DROP CONSTRAINT IF EXISTS memory_relations_target_id_fkey;
+-- +goose StatementEnd
+
+-- +goose StatementBegin
+ALTER TABLE memory_relations
+    ADD CONSTRAINT memory_relations_workspace_id_fkey
+        FOREIGN KEY (workspace_id) REFERENCES workspaces(id),
+    ADD CONSTRAINT memory_relations_source_id_fkey
+        FOREIGN KEY (source_id) REFERENCES memories(id),
+    ADD CONSTRAINT memory_relations_target_id_fkey
+        FOREIGN KEY (target_id) REFERENCES memories(id);
+-- +goose StatementEnd
+
+-- +goose StatementBegin
+DROP INDEX IF EXISTS idx_memories_created_by_user_id;
+DROP INDEX IF EXISTS idx_memories_source_file_id;
+-- +goose StatementEnd
+
+-- +goose StatementBegin
+ALTER TABLE embeddings_text
+    DROP CONSTRAINT IF EXISTS uq_embeddings_text_file_chunk;
+-- +goose StatementEnd

server/internal/db/migrations_test.go

@@ -0,0 +1,93 @@
+package db
+
+import (
+	"context"
+	"errors"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func TestEmbeddingsTextUniqueConstraint(t *testing.T) {
+	dsn := os.Getenv("MEM_TEST_DB")
+	if dsn == "" {
+		t.Skip("MEM_TEST_DB not set; skipping DB integration test")
+	}
+	config, err := pgxpool.ParseConfig(dsn)
+	if err != nil {
+		t.Fatalf("parse MEM_TEST_DB: %v", err)
+	}
+	if !strings.HasSuffix(config.ConnConfig.Database, "_test") {
+		t.Fatalf("refusing to modify non-test database %q", config.ConnConfig.Database)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	pool, err := pgxpool.NewWithConfig(ctx, config)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer pool.Close()
+
+	db := &DB{Pool: pool, url: dsn}
+	if err := db.Migrate(ctx); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+
+	var userID, workspaceID, fileID uuid.UUID
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO users (email, password_hash) VALUES ('unique-ctest@test', 'x')
+		ON CONFLICT (email) DO UPDATE SET email = EXCLUDED.email
+		RETURNING id
+	`).Scan(&userID); err != nil {
+		t.Fatalf("seed user: %v", err)
+	}
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO workspaces (name, resource_owner_user_id)
+		VALUES ('unique-ctest-ws', $1)
+		ON CONFLICT (resource_owner_user_id) DO UPDATE SET name = EXCLUDED.name
+		RETURNING id
+	`).Scan(&workspaceID); err != nil {
+		t.Fatalf("seed workspace: %v", err)
+	}
+	if err := pool.QueryRow(ctx, `
+		INSERT INTO files (user_id, name, path, size, sha256, mime, storage_key)
+		VALUES ($1, 'unique-ctest.txt', '/unique-ctest.txt', 0, '', 'text/plain', 'test://unique')
+		RETURNING id
+	`, userID).Scan(&fileID); err != nil {
+		t.Fatalf("seed file: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM files WHERE id = $1`, fileID)
+	})
+
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO embeddings_text (file_id, chunk_index, chunk_text, provider)
+		VALUES ($1, 0, 'chunk zero', 'test')
+	`, fileID); err != nil {
+		t.Fatalf("first insert: %v", err)
+	}
+
+	_, err = pool.Exec(ctx, `
+		INSERT INTO embeddings_text (file_id, chunk_index, chunk_text, provider)
+		VALUES ($1, 0, 'duplicate chunk', 'test')
+	`, fileID)
+	if err == nil {
+		t.Fatal("expected duplicate (file_id, chunk_index) to be rejected")
+	}
+	var pgErr *pgconn.PgError
+	if !errors.As(err, &pgErr) || pgErr.Code != "23505" {
+		t.Fatalf("expected unique violation 23505, got %v", err)
+	}
+
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(),
+			`DELETE FROM embeddings_text WHERE file_id = $1`, fileID)
+	})
+}

server/internal/indexgeneration/service.go

@@ -287,19 +287,33 @@ func (s *Service) List(ctx context.Context, workspaceID uuid.UUID, limit int) ([
 	}
 	defer rows.Close()
 	out := make([]Build, 0, limit)
+	buildIDs := make([]uuid.UUID, 0, limit)
 	for rows.Next() {
 		build, err := scanBuild(rows)
 		if err != nil {
 			return nil, fmt.Errorf("%w: scan build: %v", ErrUnavailable, err)
 		}
-		generations, err := listGenerations(ctx, s.pool, workspaceID, build.ID)
-		if err != nil {
-			return nil, err
-		}
-		build.Generations = generations
 		out = append(out, *build)
+		buildIDs = append(buildIDs, build.ID)
 	}
-	return out, rows.Err()
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	if len(buildIDs) == 0 {
+		return out, nil
+	}
+	generations, err := listGenerationsForBuilds(ctx, s.pool, workspaceID, buildIDs)
+	if err != nil {
+		return nil, err
+	}
+	byBuild := make(map[uuid.UUID][]Generation, len(buildIDs))
+	for _, g := range generations {
+		byBuild[g.BuildID] = append(byBuild[g.BuildID], g)
+	}
+	for i := range out {
+		out[i].Generations = byBuild[out[i].ID]
+	}
+	return out, nil
 }
 
 func (s *Service) Cancel(

server/internal/indexgeneration/store.go

@@ -106,6 +106,42 @@ func listGenerations(
 	return out, rows.Err()
 }
 
+func listGenerationsForBuilds(
+	ctx context.Context,
+	q queryer,
+	workspaceID uuid.UUID,
+	buildIDs []uuid.UUID,
+) ([]Generation, error) {
+	rows, err := q.Query(ctx, `
+		SELECT id, build_id, workspace_id, route_kind, provider,
+		       model_revision, output_dimension, pipeline_revision,
+		       profile_id, profile_revision, state, created_at, updated_at
+		  FROM index_generations
+		 WHERE workspace_id = $1 AND build_id = ANY($2::uuid[])
+		 ORDER BY build_id, route_kind
+	`, workspaceID, buildIDs)
+	if err != nil {
+		return nil, fmt.Errorf("%w: list route generations batch: %v", ErrUnavailable, err)
+	}
+	defer rows.Close()
+	var out []Generation
+	for rows.Next() {
+		var generation Generation
+		if err := rows.Scan(
+			&generation.ID, &generation.BuildID, &generation.WorkspaceID,
+			&generation.RouteKind, &generation.Provider,
+			&generation.ModelRevision, &generation.OutputDimension,
+			&generation.PipelineRevision, &generation.ProfileID,
+			&generation.ProfileRevision, &generation.State,
+			&generation.CreatedAt, &generation.UpdatedAt,
+		); err != nil {
+			return nil, fmt.Errorf("%w: scan route generation: %v", ErrUnavailable, err)
+		}
+		out = append(out, generation)
+	}
+	return out, rows.Err()
+}
+
 func getInflightIdentity(
 	ctx context.Context,
 	q queryer,

server/internal/memory/memory_integration_test.go

@@ -1224,37 +1224,37 @@ func TestMemoryPostgres(t *testing.T) {
 			t.Fatalf("superseded markers = %+v", superseded)
 		}
 
-		outbound, err := service.ListRelations(ctx, ListRelationsQuery{
+		outboundResult, err := service.ListRelations(ctx, ListRelationsQuery{
 			WorkspaceID:  workspaceA,
 			MemoryID:     fresh.Memory.ID,
 			Direction:    "source",
 			AllowedPaths: []string{scope},
 		})
-		if err != nil || len(outbound) != 1 ||
-			outbound[0].TargetID != old.Memory.ID ||
-			outbound[0].RelationType != RelSupersedes ||
-			outbound[0].Reason != "decision updated" {
-			t.Fatalf("outbound relations = %+v err=%v", outbound, err)
+		if err != nil || len(outboundResult.Relations) != 1 ||
+			outboundResult.Relations[0].TargetID != old.Memory.ID ||
+			outboundResult.Relations[0].RelationType != RelSupersedes ||
+			outboundResult.Relations[0].Reason != "decision updated" {
+			t.Fatalf("outbound relations = %+v err=%v", outboundResult, err)
 		}
-		inbound, err := service.ListRelations(ctx, ListRelationsQuery{
+		inboundResult, err := service.ListRelations(ctx, ListRelationsQuery{
 			WorkspaceID:  workspaceA,
 			MemoryID:     old.Memory.ID,
 			Direction:    "target",
 			RelationType: RelSupersedes,
 			AllowedPaths: []string{scope},
 		})
-		if err != nil || len(inbound) != 1 || inbound[0].SourceID != fresh.Memory.ID {
-			t.Fatalf("inbound relations = %+v err=%v", inbound, err)
+		if err != nil || len(inboundResult.Relations) != 1 || inboundResult.Relations[0].SourceID != fresh.Memory.ID {
+			t.Fatalf("inbound relations = %+v err=%v", inboundResult, err)
 		}
-		typed, err := service.ListRelations(ctx, ListRelationsQuery{
+		typedResult, err := service.ListRelations(ctx, ListRelationsQuery{
 			WorkspaceID:  workspaceA,
 			MemoryID:     old.Memory.ID,
 			Direction:    "target",
 			RelationType: RelCorrects,
 			AllowedPaths: []string{scope},
 		})
-		if err != nil || len(typed) != 0 {
-			t.Fatalf("type-filtered relations = %+v err=%v", typed, err)
+		if err != nil || len(typedResult.Relations) != 0 {
+			t.Fatalf("type-filtered relations = %+v err=%v", typedResult, err)
 		}
 
 		// Anchors the caller cannot read are hidden as not found.