diff --git a/server/internal/api/api.go b/server/internal/api/api.go index e8c6d07..35cfedf 100644 --- a/server/internal/api/api.go +++ b/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 diff --git a/server/internal/api/handlers_memory.go b/server/internal/api/handlers_memory.go index 879e1d4..4020bb3 100644 --- a/server/internal/api/handlers_memory.go +++ b/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) } diff --git a/server/internal/api/handlers_memory_test.go b/server/internal/api/handlers_memory_test.go index 54887a4..0fc2e6d 100644 --- a/server/internal/api/handlers_memory_test.go +++ b/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) { diff --git a/server/internal/db/migrations/0026_data_plane_hygiene.sql b/server/internal/db/migrations/0026_data_plane_hygiene.sql new file mode 100644 index 0000000..d1db867 --- /dev/null +++ b/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 diff --git a/server/internal/db/migrations_test.go b/server/internal/db/migrations_test.go new file mode 100644 index 0000000..e38e7aa --- /dev/null +++ b/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) + }) +} diff --git a/server/internal/indexgeneration/service.go b/server/internal/indexgeneration/service.go index a8bb35b..4220936 100644 --- a/server/internal/indexgeneration/service.go +++ b/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( diff --git a/server/internal/indexgeneration/store.go b/server/internal/indexgeneration/store.go index 245236e..8e06ba2 100644 --- a/server/internal/indexgeneration/store.go +++ b/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, diff --git a/server/internal/memory/memory_integration_test.go b/server/internal/memory/memory_integration_test.go index fbe694a..d6d291e 100644 --- a/server/internal/memory/memory_integration_test.go +++ b/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. diff --git a/server/internal/memory/relation.go b/server/internal/memory/relation.go index b9c238f..58c8ad2 100644 --- a/server/internal/memory/relation.go +++ b/server/internal/memory/relation.go @@ -2,6 +2,9 @@ package memory import ( "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" "errors" "fmt" "strings" @@ -79,6 +82,13 @@ type ListRelationsQuery struct { RelationType string // optional filter AllowedPaths []string Limit int + Cursor string // opaque keyset cursor from a prior ListRelationsResult +} + +// ListRelationsResult is a keyset-paginated page of relations. +type ListRelationsResult struct { + Relations []Relation `json:"relations"` + NextCursor string `json:"next_cursor,omitempty"` } // validateCreateRelationCommand checks the command fields without touching the database. @@ -235,8 +245,9 @@ func validateListRelationsQuery(q ListRelationsQuery) error { } // ListRelations returns relations for a given memory. Direction controls -// whether MemoryID is matched as source or target. -func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Relation, error) { +// whether MemoryID is matched as source or target. Results are keyset-paginated +// using the same cursor shape as memory/list.go. +func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) (*ListRelationsResult, error) { if s == nil || s.pool == nil { return nil, fmt.Errorf("memory service is not configured") } @@ -259,9 +270,6 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re return nil, err } - // The anchor memory must exist, be visible under the caller's allowed - // paths, and not be forgotten — mirroring Get — so listing relations never - // leaks the existence of a memory the caller cannot read. args := []any{q.WorkspaceID, q.MemoryID} where := []string{"m.workspace_id = $1", "m.id = $2"} args, where = appendPathFilters(args, where, "m.path", "/", allowed) @@ -280,6 +288,20 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re return nil, ErrForgotten } + filterHash, err := relationFilterHash(q.WorkspaceID, q.MemoryID, direction, q.RelationType) + if err != nil { + return nil, fmt.Errorf("list relations: filter hash: %w", err) + } + + var cursor *decodedListCursor + if strings.TrimSpace(q.Cursor) != "" { + decoded, err := decodeListCursor(q.Cursor, filterHash) + if err != nil { + return nil, err + } + cursor = &decoded + } + args = []any{q.WorkspaceID, q.MemoryID} var dirColumn string if direction == "source" { @@ -296,7 +318,15 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re args = append(args, relType) where = append(where, fmt.Sprintf("r.relation_type = $%d", len(args))) } - args = append(args, q.Limit) + if cursor != nil { + args = append(args, cursor.createdAt, cursor.id) + timeArg, idArg := len(args)-1, len(args) + where = append(where, fmt.Sprintf( + "(r.created_at < $%d OR (r.created_at = $%d AND r.id < $%d))", + timeArg, timeArg, idArg, + )) + } + args = append(args, q.Limit+1) limitIdx := len(args) sql := fmt.Sprintf(` @@ -313,7 +343,7 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re } defer rows.Close() - out := make([]Relation, 0, q.Limit) + out := make([]Relation, 0, q.Limit+1) for rows.Next() { var rel Relation if err := rows.Scan(&rel.ID, &rel.WorkspaceID, &rel.SourceID, &rel.TargetID, @@ -322,7 +352,43 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re } out = append(out, rel) } - return out, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + + result := &ListRelationsResult{Relations: out} + if len(out) <= q.Limit { + return result, nil + } + result.Relations = out[:q.Limit] + last := result.Relations[len(result.Relations)-1] + result.NextCursor, err = encodeListCursor(last.CreatedAt, last.ID, filterHash) + if err != nil { + return nil, fmt.Errorf("list relations: encode cursor: %w", err) + } + return result, nil +} + +type relationFilterFingerprint struct { + WorkspaceID string `json:"workspace_id"` + MemoryID string `json:"memory_id"` + Direction string `json:"direction"` + RelationType string `json:"relation_type"` +} + +func relationFilterHash(workspaceID, memoryID uuid.UUID, direction, relationType string) (string, error) { + payload := relationFilterFingerprint{ + WorkspaceID: workspaceID.String(), + MemoryID: memoryID.String(), + Direction: direction, + RelationType: strings.ToLower(strings.TrimSpace(relationType)), + } + encoded, err := json.Marshal(payload) + if err != nil { + return "", err + } + sum := sha256.Sum256(encoded) + return hex.EncodeToString(sum[:]), nil } // IsSuperseded returns true if the given memory has been superseded or corrected @@ -410,8 +476,6 @@ func (s *Service) loadMemoryInTx(ctx context.Context, tx pgx.Tx, workspaceID, me // from target and looks for source. The exact edge being written is excluded // so an idempotent replay of an existing edge is not reported as a cycle. func (s *Service) wouldCycle(ctx context.Context, tx pgx.Tx, workspaceID, sourceID, targetID uuid.UUID) (bool, error) { - // BFS from target following supersedes/corrects edges in their forward - // direction: if we ever reach sourceID, there is a cycle. var found bool err := tx.QueryRow(ctx, ` WITH RECURSIVE chain(id) AS (