diff --git a/internal/store/pg/vault_documents.go b/internal/store/pg/vault_documents.go index cacb19d2..13aca493 100644 --- a/internal/store/pg/vault_documents.go +++ b/internal/store/pg/vault_documents.go @@ -145,7 +145,7 @@ func (s *PGVaultStore) GetDocument(ctx context.Context, tenantID, agentID, path return nil, fmt.Errorf("vault get document: tenant: %w", err) } - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE tenant_id = $1 AND path = $2` args := []any{tid, path} p := 3 @@ -197,10 +197,10 @@ func (s *PGVaultStore) GetDocumentByID(ctx context.Context, tenantID, id string) } var row vaultDocRow err = s.db.QueryRowContext(ctx, ` - SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE id = $1 AND tenant_id = $2`, uid, tid, ).Scan(&row.ID, &row.TenantID, &row.AgentID, &row.TeamID, &row.Scope, &row.CustomScope, - &row.Path, &row.Title, &row.DocType, &row.ContentHash, &row.Summary, + &row.Path, &row.PathBasename, &row.Title, &row.DocType, &row.ContentHash, &row.Summary, &row.MetaJSON, &row.CreatedAt, &row.UpdatedAt) if err != nil { return nil, err @@ -225,7 +225,7 @@ func (s *PGVaultStore) GetDocumentsByIDs(ctx context.Context, tenantID string, d end := min(start+chunkSize, len(docIDs)) var scanned []vaultDocRow if err := pkgSqlxDB.SelectContext(ctx, &scanned, - `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE id = ANY($1) AND tenant_id = $2`, pqStringArray(docIDs[start:end]), tid); err != nil { return nil, err @@ -244,7 +244,7 @@ func (s *PGVaultStore) GetDocumentByBasename(ctx context.Context, tenantID, agen if err != nil { return nil, fmt.Errorf("vault get by basename: tenant: %w", err) } - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE tenant_id = $1 AND path_basename = lower($2)` args := []any{tid, basename} @@ -317,7 +317,7 @@ func (s *PGVaultStore) ListDocuments(ctx context.Context, tenantID, agentID stri return nil, fmt.Errorf("vault list documents: tenant: %w", err) } - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE tenant_id = $1` args := []any{tid} p := 2 @@ -532,7 +532,7 @@ func (tf searchTeamFilter) append(q string, args []any, p int) (string, []any, i } func (s *PGVaultStore) ftsSearch(ctx context.Context, query string, tenantID uuid.UUID, agentID *uuid.UUID, tf searchTeamFilter, scope string, docTypes []string, limit int) ([]store.VaultSearchResult, error) { - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at, + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at, ts_rank(tsv, plainto_tsquery('simple', $1)) AS score FROM vault_documents WHERE tenant_id = $2 AND tsv @@ plainto_tsquery('simple', $1)` @@ -570,7 +570,7 @@ func (s *PGVaultStore) ftsSearch(ctx context.Context, query string, tenantID uui func (s *PGVaultStore) vectorSearch(ctx context.Context, embedding []float32, tenantID uuid.UUID, agentID *uuid.UUID, tf searchTeamFilter, scope string, docTypes []string, limit int) ([]store.VaultSearchResult, error) { vecStr := vectorToString(embedding) - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at, + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at, 1 - (embedding <=> $1) AS score FROM vault_documents WHERE tenant_id = $2 AND embedding IS NOT NULL` diff --git a/internal/store/pg/vault_links.go b/internal/store/pg/vault_links.go index 0a487d33..65516b34 100644 --- a/internal/store/pg/vault_links.go +++ b/internal/store/pg/vault_links.go @@ -3,6 +3,7 @@ package pg import ( "context" "database/sql" + "encoding/json" "fmt" "strings" "time" @@ -75,7 +76,8 @@ func (s *PGVaultStore) CreateLinks(ctx context.Context, links []store.VaultLink) for start := 0; start < len(valid); start += chunkSize { end := min(start+chunkSize, len(valid)) chunk := valid[start:end] - const paramsPerRow = 6 // id, from_doc_id, to_doc_id, link_type, context, created_at + // id, from_doc_id, to_doc_id, link_type, context, metadata, created_at + const paramsPerRow = 7 args := make([]any, 0, len(chunk)*paramsPerRow) ph := make([]string, 0, len(chunk)) now := time.Now().UTC() @@ -88,17 +90,25 @@ func (s *PGVaultStore) CreateLinks(ctx context.Context, links []store.VaultLink) if err != nil { return fmt.Errorf("vault batch create links: to_doc_id: %w", err) } + metaJSON, merr := json.Marshal(l.Metadata) + if merr != nil || len(metaJSON) == 0 { + metaJSON = []byte("{}") + } b := i * paramsPerRow - ph = append(ph, fmt.Sprintf("($%d,$%d,$%d,$%d,$%d,$%d)", b+1, b+2, b+3, b+4, b+5, b+6)) + ph = append(ph, fmt.Sprintf("($%d,$%d,$%d,$%d,$%d,$%d,$%d)", + b+1, b+2, b+3, b+4, b+5, b+6, b+7)) args = append(args, uuid.Must(uuid.NewV7()), fromID, toID, - l.LinkType, l.Context, now, + l.LinkType, l.Context, metaJSON, now, ) } - q := `INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, created_at) + // Single-winner metadata (Phase 03): ON CONFLICT replaces metadata. + q := `INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, metadata, created_at) VALUES ` + strings.Join(ph, ",") + ` - ON CONFLICT (from_doc_id, to_doc_id, link_type) DO UPDATE SET context = EXCLUDED.context` + ON CONFLICT (from_doc_id, to_doc_id, link_type) DO UPDATE SET + context = EXCLUDED.context, + metadata = EXCLUDED.metadata` if _, err := s.db.ExecContext(ctx, q, args...); err != nil { return fmt.Errorf("vault batch create links: %w", err) } @@ -143,16 +153,21 @@ func (s *PGVaultStore) CreateLink(ctx context.Context, link *store.VaultLink) er return fmt.Errorf("vault link: documents belong to different teams") } + metaJSON, mErr := json.Marshal(link.Metadata) + if mErr != nil || len(metaJSON) == 0 { + metaJSON = []byte("{}") + } id := uuid.Must(uuid.NewV7()) now := time.Now().UTC() var actualID uuid.UUID err = s.db.QueryRowContext(ctx, ` - INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, created_at) - VALUES ($1, $2, $3, $4, $5, $6) + INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, metadata, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (from_doc_id, to_doc_id, link_type) DO UPDATE SET - context = EXCLUDED.context + context = EXCLUDED.context, + metadata = EXCLUDED.metadata RETURNING id`, - id, fromID, toID, link.LinkType, link.Context, now, + id, fromID, toID, link.LinkType, link.Context, metaJSON, now, ).Scan(&actualID) if err != nil { return fmt.Errorf("vault create link: %w", err) @@ -189,7 +204,7 @@ func (s *PGVaultStore) GetOutLinks(ctx context.Context, tenantID, docID string) return nil, fmt.Errorf("vault get out links: tenant: %w", err) } rows, err := s.db.QueryContext(ctx, ` - SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.created_at + SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.metadata, vl.created_at FROM vault_links vl JOIN vault_documents vd ON vl.from_doc_id = vd.id WHERE vl.from_doc_id = $1 AND vd.tenant_id = $2 @@ -211,7 +226,7 @@ func (s *PGVaultStore) GetOutLinksBatch(ctx context.Context, tenantID string, do return nil, fmt.Errorf("vault get out links batch: %w", err) } rows, err := s.db.QueryContext(ctx, ` - SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.created_at + SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.metadata, vl.created_at FROM vault_links vl JOIN vault_documents vd ON vl.from_doc_id = vd.id WHERE vl.from_doc_id = ANY($1) AND vd.tenant_id = $2 @@ -334,12 +349,16 @@ func scanVaultLinks(rows *sql.Rows) ([]store.VaultLink, error) { for rows.Next() { var l store.VaultLink var id, fromID, toID uuid.UUID - if err := rows.Scan(&id, &fromID, &toID, &l.LinkType, &l.Context, &l.CreatedAt); err != nil { + var metaJSON []byte + if err := rows.Scan(&id, &fromID, &toID, &l.LinkType, &l.Context, &metaJSON, &l.CreatedAt); err != nil { return nil, err } l.ID = id.String() l.FromDocID = fromID.String() l.ToDocID = toID.String() + if len(metaJSON) > 0 { + _ = json.Unmarshal(metaJSON, &l.Metadata) + } links = append(links, l) } return links, rows.Err() diff --git a/internal/store/pg/vault_scan_rows.go b/internal/store/pg/vault_scan_rows.go index 96ea271c..7b4a6390 100644 --- a/internal/store/pg/vault_scan_rows.go +++ b/internal/store/pg/vault_scan_rows.go @@ -12,36 +12,38 @@ import ( // IDs are scanned as uuid.UUID then converted to string for VaultDocument. // Metadata is scanned as raw JSON then unmarshalled post-scan. type vaultDocRow struct { - ID uuid.UUID `db:"id"` - TenantID uuid.UUID `db:"tenant_id"` - AgentID *uuid.UUID `db:"agent_id"` - TeamID *uuid.UUID `db:"team_id"` - Scope string `db:"scope"` - CustomScope *string `db:"custom_scope"` - Path string `db:"path"` - Title string `db:"title"` - DocType string `db:"doc_type"` - ContentHash string `db:"content_hash"` - Summary string `db:"summary"` - MetaJSON []byte `db:"metadata"` - CreatedAt time.Time `db:"created_at"` - UpdatedAt time.Time `db:"updated_at"` + ID uuid.UUID `db:"id"` + TenantID uuid.UUID `db:"tenant_id"` + AgentID *uuid.UUID `db:"agent_id"` + TeamID *uuid.UUID `db:"team_id"` + Scope string `db:"scope"` + CustomScope *string `db:"custom_scope"` + Path string `db:"path"` + PathBasename string `db:"path_basename"` + Title string `db:"title"` + DocType string `db:"doc_type"` + ContentHash string `db:"content_hash"` + Summary string `db:"summary"` + MetaJSON []byte `db:"metadata"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` } // toVaultDocument converts a vaultDocRow to store.VaultDocument. func (r *vaultDocRow) toVaultDocument() store.VaultDocument { doc := store.VaultDocument{ - ID: r.ID.String(), - TenantID: r.TenantID.String(), - Scope: r.Scope, - CustomScope: r.CustomScope, - Path: r.Path, - Title: r.Title, - DocType: r.DocType, - ContentHash: r.ContentHash, - Summary: r.Summary, - CreatedAt: r.CreatedAt, - UpdatedAt: r.UpdatedAt, + ID: r.ID.String(), + TenantID: r.TenantID.String(), + Scope: r.Scope, + CustomScope: r.CustomScope, + Path: r.Path, + PathBasename: r.PathBasename, + Title: r.Title, + DocType: r.DocType, + ContentHash: r.ContentHash, + Summary: r.Summary, + CreatedAt: r.CreatedAt, + UpdatedAt: r.UpdatedAt, } if r.AgentID != nil { s := r.AgentID.String() diff --git a/internal/store/pg/vault_source_cleanup.go b/internal/store/pg/vault_source_cleanup.go new file mode 100644 index 00000000..e973da12 --- /dev/null +++ b/internal/store/pg/vault_source_cleanup.go @@ -0,0 +1,154 @@ +package pg + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// DeleteLinksBySource removes vault_links rows whose metadata->>'source' equals +// the given source key (e.g. "task:{uuid}", "delegation:{uuid}"). Used by +// Phase 04/05 cleanup paths (DetachFileFromTask, DeleteTask, bulk deletion) +// to surgically remove auto-links without touching classify-owned links. +// +// Tenant isolation is enforced by joining vault_documents on tenant_id so a +// guessable source string can't delete another tenant's links. +func (s *PGVaultStore) DeleteLinksBySource(ctx context.Context, tenantID, source string) (int64, error) { + tid, err := parseUUID(tenantID) + if err != nil { + return 0, fmt.Errorf("delete links by source: tenant: %w", err) + } + res, err := s.db.ExecContext(ctx, ` + DELETE FROM vault_links vl + USING vault_documents vd + WHERE vl.metadata->>'source' = $1 + AND vd.tenant_id = $2 + AND (vl.from_doc_id = vd.id OR vl.to_doc_id = vd.id) + `, source, tid) + if err != nil { + return 0, fmt.Errorf("delete links by source: %w", err) + } + n, _ := res.RowsAffected() + return n, nil +} + +// BatchFindByDelegationIDs returns vault docs sharing any of the given +// delegation_ids, keyed by delegation_id. Each bucket is capped at `limit`. +// excludeDocIDs prevents self-link emission when the caller's source docs +// appear in the result set. Uses ROW_NUMBER() PARTITION BY delegation_id +// over the partial index idx_vault_docs_delegation added by migration 000048. +func (s *PGVaultStore) BatchFindByDelegationIDs( + ctx context.Context, + tenantID string, + delegationIDs []string, + limit int, + excludeDocIDs []string, +) (map[string][]store.VaultDocument, error) { + if len(delegationIDs) == 0 || limit <= 0 { + return nil, nil + } + tid, err := parseUUID(tenantID) + if err != nil { + return nil, fmt.Errorf("batch find by delegation: tenant: %w", err) + } + + // Exclude doc IDs are real UUIDs — parse strictly. + excludeUUIDs := make([]string, 0, len(excludeDocIDs)) + for _, id := range excludeDocIDs { + if id == "" { + continue + } + docID, err := parseUUID(id) + if err != nil { + return nil, fmt.Errorf("batch find by delegation: exclude doc_id: %w", err) + } + excludeUUIDs = append(excludeUUIDs, docID.String()) + } + + q := ` +WITH ranked AS ( + SELECT + vd.id, vd.tenant_id, vd.agent_id, vd.team_id, vd.scope, vd.custom_scope, + vd.path, vd.path_basename, vd.title, vd.doc_type, vd.content_hash, + vd.summary, vd.metadata, vd.created_at, vd.updated_at, + vd.metadata->>'delegation_id' AS deleg_id, + ROW_NUMBER() OVER ( + PARTITION BY vd.metadata->>'delegation_id' + ORDER BY vd.created_at DESC, vd.id DESC + ) AS rn + FROM vault_documents vd + WHERE vd.tenant_id = $1 + AND vd.metadata ? 'delegation_id' + AND vd.metadata->>'delegation_id' = ANY($2) +` + args := []any{tid, pqStringArray(delegationIDs)} + if len(excludeUUIDs) > 0 { + q += fmt.Sprintf(" AND NOT (vd.id = ANY($%d::uuid[]))\n", len(args)+1) + args = append(args, pqStringArray(excludeUUIDs)) + } + q += `) +SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, + title, doc_type, content_hash, summary, metadata, created_at, updated_at, deleg_id +FROM ranked +WHERE rn <= $` + fmt.Sprintf("%d", len(args)+1) + ` +ORDER BY deleg_id, created_at DESC +` + args = append(args, limit) + + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("batch find by delegation: %w", err) + } + defer rows.Close() + + out := make(map[string][]store.VaultDocument, len(delegationIDs)) + for rows.Next() { + var ( + id, tenantIDVal uuid.UUID + agentID, teamID *uuid.UUID + customScope *string + metaJSON []byte + delegID string + path, pathBase string + scope, title, docTyp string + contentHash, summary string + ) + doc := store.VaultDocument{} + if err := rows.Scan( + &id, &tenantIDVal, &agentID, &teamID, &scope, &customScope, + &path, &pathBase, &title, &docTyp, &contentHash, + &summary, &metaJSON, &doc.CreatedAt, &doc.UpdatedAt, &delegID, + ); err != nil { + return nil, err + } + doc.ID = id.String() + doc.TenantID = tenantIDVal.String() + if agentID != nil { + v := agentID.String() + doc.AgentID = &v + } + if teamID != nil { + v := teamID.String() + doc.TeamID = &v + } + doc.Scope = scope + doc.CustomScope = customScope + doc.Path = path + doc.PathBasename = pathBase + doc.Title = title + doc.DocType = docTyp + doc.ContentHash = contentHash + doc.Summary = summary + if len(metaJSON) > 0 { + _ = json.Unmarshal(metaJSON, &doc.Metadata) + } + out[delegID] = append(out[delegID], doc) + } + if err := rows.Err(); err != nil { + return nil, err + } + return out, nil +} diff --git a/internal/store/sqlitestore/basename_helper.go b/internal/store/sqlitestore/basename_helper.go new file mode 100644 index 00000000..a3bdf8e6 --- /dev/null +++ b/internal/store/sqlitestore/basename_helper.go @@ -0,0 +1,27 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "path/filepath" + "strings" +) + +// ComputeAttachmentBaseName derives the canonical basename used for both +// team_task_attachments.base_name and vault_documents.path_basename on SQLite. +// +// PG GENERATES these columns via `lower(regexp_replace(path, '.+/', ''))`. +// modernc.org/sqlite (bundled) ships no regexp_replace, so SQLite callers +// must compute the value app-side at INSERT/UPDATE time. This helper is the +// single source of truth — call it from workspace_interceptor, team_tasks_create, +// vault UpsertDocument, and v15→v16 migration backfill. +// +// Pure string manipulation; safe to invoke from PG code paths too where it +// becomes a harmless no-op (PG's GENERATED column overrides anything the Go +// caller supplies). +func ComputeAttachmentBaseName(path string) string { + if path == "" { + return "" + } + return strings.ToLower(filepath.Base(path)) +} diff --git a/internal/store/sqlitestore/basename_helper_test.go b/internal/store/sqlitestore/basename_helper_test.go new file mode 100644 index 00000000..9a86c291 --- /dev/null +++ b/internal/store/sqlitestore/basename_helper_test.go @@ -0,0 +1,24 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import "testing" + +func TestComputeAttachmentBaseName(t *testing.T) { + cases := []struct { + in, want string + }{ + {"", ""}, + {"file.md", "file.md"}, + {"path/to/file.md", "file.md"}, + {"/abs/path/to/IMAGE.PNG", "image.png"}, + {"path/to/Mixed-Case.TXT", "mixed-case.txt"}, + {"no-slash-just-name.pdf", "no-slash-just-name.pdf"}, + {"deep/nested/folder/doc.docx", "doc.docx"}, + } + for _, c := range cases { + if got := ComputeAttachmentBaseName(c.in); got != c.want { + t.Errorf("ComputeAttachmentBaseName(%q) = %q; want %q", c.in, got, c.want) + } + } +} diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index 2b53e969..61c0c0f3 100644 --- a/internal/store/sqlitestore/schema.go +++ b/internal/store/sqlitestore/schema.go @@ -3,6 +3,7 @@ package sqlitestore import ( + "context" "database/sql" _ "embed" "errors" @@ -15,7 +16,7 @@ var schemaSQL string // SchemaVersion is the current SQLite schema version. // Bump this when adding new migration steps below. -const SchemaVersion = 15 +const SchemaVersion = 16 // migrations maps version → SQL to apply when upgrading FROM that version. // schema.sql always represents the LATEST full schema (for fresh DBs). @@ -412,6 +413,104 @@ CREATE INDEX IF NOT EXISTS idx_vault_docs_team ON vault_documents(team_id);`, ); CREATE UNIQUE INDEX IF NOT EXISTS uq_cron_jobs_agent_tenant_name ON cron_jobs(agent_id, tenant_id, name);`, + + // Version 15 → 16: vault media linking schema + // (team_task_attachments.base_name, vault_documents.path_basename, + // vault_links.metadata, auto-linking indexes). + // Backfill of base_name / path_basename happens in backfillV16 below + // (Go-loop — SQLite has no regexp_replace in modernc.org/sqlite bundle). + 15: `ALTER TABLE team_task_attachments ADD COLUMN base_name TEXT NOT NULL DEFAULT ''; +CREATE INDEX IF NOT EXISTS idx_tta_tenant_basename + ON team_task_attachments(tenant_id, base_name); + +ALTER TABLE vault_documents ADD COLUMN path_basename TEXT NOT NULL DEFAULT ''; +CREATE INDEX IF NOT EXISTS idx_vault_docs_basename + ON vault_documents(tenant_id, path_basename); + +ALTER TABLE vault_links ADD COLUMN metadata TEXT NOT NULL DEFAULT '{}'; +CREATE INDEX IF NOT EXISTS idx_vault_links_source + ON vault_links(json_extract(metadata, '$.source')) + WHERE json_extract(metadata, '$.source') IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_vault_docs_delegation + ON vault_documents(json_extract(metadata, '$.delegation_id')) + WHERE json_extract(metadata, '$.delegation_id') IS NOT NULL;`, +} + +// backfillV16 populates base_name / path_basename for rows that existed +// before the v15 → v16 migration. Idempotent — re-running on already-filled +// rows is a no-op thanks to the WHERE base_name = '' filter. +func backfillV16(ctx context.Context, db *sql.DB) error { + type row struct{ id, path string } + + // ---- team_task_attachments ---- + attRows, err := collectIDPath(ctx, db, + `SELECT id, path FROM team_task_attachments WHERE base_name = ''`) + if err != nil { + return fmt.Errorf("v16 scan attachments: %w", err) + } + if err := updateBaseNames(ctx, db, + `UPDATE team_task_attachments SET base_name = ? WHERE id = ?`, attRows); err != nil { + return fmt.Errorf("v16 update attachments: %w", err) + } + + // ---- vault_documents ---- + docRows, err := collectIDPath(ctx, db, + `SELECT id, path FROM vault_documents WHERE path_basename = ''`) + if err != nil { + return fmt.Errorf("v16 scan vault_docs: %w", err) + } + if err := updateBaseNames(ctx, db, + `UPDATE vault_documents SET path_basename = ? WHERE id = ?`, docRows); err != nil { + return fmt.Errorf("v16 update vault_docs: %w", err) + } + return nil +} + +// collectIDPath reads (id, path) tuples for a SELECT that returns exactly +// those two columns. Separated so backfillV16 stays readable. +func collectIDPath(ctx context.Context, db *sql.DB, q string) ([][2]string, error) { + rows, err := db.QueryContext(ctx, q) + if err != nil { + return nil, err + } + defer rows.Close() + var out [][2]string + for rows.Next() { + var id, path string + if err := rows.Scan(&id, &path); err != nil { + return nil, err + } + out = append(out, [2]string{id, path}) + } + return out, rows.Err() +} + +// updateBaseNames runs a prepared UPDATE inside one transaction for all rows. +// No-op when rows is empty. The prepared statement form keeps SQLite happy +// on larger backfills (<= ~10k legacy rows on typical desktop lite DBs). +func updateBaseNames(ctx context.Context, db *sql.DB, updateSQL string, rows [][2]string) error { + if len(rows) == 0 { + return nil + } + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + stmt, err := tx.PrepareContext(ctx, updateSQL) + if err != nil { + _ = tx.Rollback() + return err + } + defer stmt.Close() + for _, r := range rows { + bn := ComputeAttachmentBaseName(r[1]) + if _, err := stmt.ExecContext(ctx, bn, r[0]); err != nil { + _ = tx.Rollback() + return fmt.Errorf("update %s: %w", r[0], err) + } + } + return tx.Commit() } // EnsureSchema creates tables if they don't exist and applies incremental migrations. @@ -479,6 +578,14 @@ func EnsureSchema(db *sql.DB) error { if err := tx.Commit(); err != nil { return fmt.Errorf("commit migration v%d: %w", v, err) } + // Post-SQL backfill hooks for migrations needing app-side logic. + // modernc.org/sqlite lacks regexp_replace, so the v15 → v16 + // basename columns must be populated via a Go loop. + if v == 15 { + if err := backfillV16(context.Background(), db); err != nil { + return fmt.Errorf("v16 backfill: %w", err) + } + } slog.Info("sqlite: applied migration", "version", v+1) } } diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index 7efb3241..4064d720 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/internal/store/sqlitestore/schema.sql @@ -939,6 +939,7 @@ CREATE TABLE IF NOT EXISTS team_task_attachments ( team_id TEXT NOT NULL REFERENCES agent_teams(id) ON DELETE CASCADE, chat_id VARCHAR(255) NOT NULL DEFAULT '', path TEXT NOT NULL, + base_name TEXT NOT NULL DEFAULT '', -- app-populated lowercased basename; PG equivalent is GENERATED file_size BIGINT NOT NULL DEFAULT 0, mime_type VARCHAR(100) DEFAULT '', created_by_agent_id TEXT REFERENCES agents(id), @@ -953,6 +954,7 @@ CREATE TABLE IF NOT EXISTS team_task_attachments ( CREATE INDEX IF NOT EXISTS idx_tta_task ON team_task_attachments(task_id); CREATE INDEX IF NOT EXISTS idx_tta_team ON team_task_attachments(team_id); CREATE INDEX IF NOT EXISTS idx_team_task_attachments_tenant ON team_task_attachments(tenant_id); +CREATE INDEX IF NOT EXISTS idx_tta_tenant_basename ON team_task_attachments(tenant_id, base_name); -- ============================================================ -- Table: team_user_grants @@ -1515,20 +1517,21 @@ CREATE INDEX IF NOT EXISTS idx_scuc_binary ON secure_cli_user_credentials(binary -- ============================================================ CREATE TABLE IF NOT EXISTS vault_documents ( - id TEXT NOT NULL PRIMARY KEY, - tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, - agent_id TEXT REFERENCES agents(id) ON DELETE SET NULL, - team_id TEXT REFERENCES agent_teams(id) ON DELETE SET NULL, - scope TEXT NOT NULL DEFAULT 'personal', - custom_scope TEXT, - path TEXT NOT NULL, - title TEXT NOT NULL DEFAULT '', - doc_type TEXT NOT NULL DEFAULT 'note', - content_hash TEXT NOT NULL DEFAULT '', - summary TEXT NOT NULL DEFAULT '', - metadata TEXT DEFAULT '{}', - created_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), - updated_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + agent_id TEXT REFERENCES agents(id) ON DELETE SET NULL, + team_id TEXT REFERENCES agent_teams(id) ON DELETE SET NULL, + scope TEXT NOT NULL DEFAULT 'personal', + custom_scope TEXT, + path TEXT NOT NULL, + path_basename TEXT NOT NULL DEFAULT '', -- app-populated lowercased basename; PG equivalent is GENERATED + title TEXT NOT NULL DEFAULT '', + doc_type TEXT NOT NULL DEFAULT 'note', + content_hash TEXT NOT NULL DEFAULT '', + summary TEXT NOT NULL DEFAULT '', + metadata TEXT DEFAULT '{}', + created_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + updated_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) ); -- SQLite prohibits expressions in inline UNIQUE constraints; use a unique index instead. CREATE UNIQUE INDEX IF NOT EXISTS idx_vault_docs_unique_path @@ -1538,6 +1541,10 @@ CREATE INDEX IF NOT EXISTS idx_vault_docs_agent_scope ON vault_documents(agent_i CREATE INDEX IF NOT EXISTS idx_vault_docs_type ON vault_documents(agent_id, doc_type); CREATE INDEX IF NOT EXISTS idx_vault_docs_hash ON vault_documents(content_hash); CREATE INDEX IF NOT EXISTS idx_vault_docs_team ON vault_documents(team_id); +CREATE INDEX IF NOT EXISTS idx_vault_docs_basename ON vault_documents(tenant_id, path_basename); +CREATE INDEX IF NOT EXISTS idx_vault_docs_delegation + ON vault_documents(json_extract(metadata, '$.delegation_id')) + WHERE json_extract(metadata, '$.delegation_id') IS NOT NULL; -- ============================================================ -- Table: vault_links (V3 wikilink edges) @@ -1549,6 +1556,7 @@ CREATE TABLE IF NOT EXISTS vault_links ( to_doc_id TEXT NOT NULL REFERENCES vault_documents(id) ON DELETE CASCADE, link_type TEXT NOT NULL DEFAULT 'wikilink', context TEXT NOT NULL DEFAULT '', + metadata TEXT NOT NULL DEFAULT '{}', custom_scope TEXT, created_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), UNIQUE(from_doc_id, to_doc_id, link_type) @@ -1556,3 +1564,6 @@ CREATE TABLE IF NOT EXISTS vault_links ( CREATE INDEX IF NOT EXISTS idx_vault_links_from ON vault_links(from_doc_id); CREATE INDEX IF NOT EXISTS idx_vault_links_to ON vault_links(to_doc_id); +CREATE INDEX IF NOT EXISTS idx_vault_links_source + ON vault_links(json_extract(metadata, '$.source')) + WHERE json_extract(metadata, '$.source') IS NOT NULL; diff --git a/internal/store/sqlitestore/schema_migration_test.go b/internal/store/sqlitestore/schema_migration_test.go index e3a9f6c9..addd3abe 100644 --- a/internal/store/sqlitestore/schema_migration_test.go +++ b/internal/store/sqlitestore/schema_migration_test.go @@ -154,6 +154,86 @@ func openTestDBAtVersion(t *testing.T, targetVersion int) *sql.DB { // Undo columns added by migrations after targetVersion. // SQLite DROP COLUMN support varies, so recreate affected tables. + + // Phase 03 (v15 → v16) adds: + // - team_task_attachments.base_name + // - vault_documents.path_basename + // - vault_links.metadata + // Strip these when the test targets any version < 16 so the v13 migration + // (which recreates vault_documents via SELECT *) doesn't hit a column + // count mismatch. + if targetVersion < 16 { + // Recreate vault_documents without path_basename. + db.Exec(`CREATE TABLE vault_documents_v15 AS SELECT + id, tenant_id, agent_id, team_id, scope, custom_scope, path, + title, doc_type, content_hash, summary, metadata, + created_at, updated_at + FROM vault_documents`) + db.Exec(`DROP TABLE vault_documents`) + db.Exec(`CREATE TABLE vault_documents ( + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + agent_id TEXT REFERENCES agents(id) ON DELETE SET NULL, + team_id TEXT REFERENCES agent_teams(id) ON DELETE SET NULL, + scope TEXT NOT NULL DEFAULT 'personal', + custom_scope TEXT, + path TEXT NOT NULL, + title TEXT NOT NULL DEFAULT '', + doc_type TEXT NOT NULL DEFAULT 'note', + content_hash TEXT NOT NULL DEFAULT '', + summary TEXT NOT NULL DEFAULT '', + metadata TEXT DEFAULT '{}', + created_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + updated_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + )`) + db.Exec(`INSERT INTO vault_documents SELECT * FROM vault_documents_v15`) + db.Exec(`DROP TABLE vault_documents_v15`) + + // Recreate team_task_attachments without base_name. + db.Exec(`CREATE TABLE team_task_attachments_v15 AS SELECT + id, task_id, team_id, chat_id, path, file_size, mime_type, + created_by_agent_id, created_by_sender_id, metadata, custom_scope, + tenant_id, created_at + FROM team_task_attachments`) + db.Exec(`DROP TABLE team_task_attachments`) + db.Exec(`CREATE TABLE team_task_attachments ( + id TEXT NOT NULL PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES team_tasks(id) ON DELETE CASCADE, + team_id TEXT NOT NULL REFERENCES agent_teams(id) ON DELETE CASCADE, + chat_id VARCHAR(255) NOT NULL DEFAULT '', + path TEXT NOT NULL, + file_size BIGINT NOT NULL DEFAULT 0, + mime_type VARCHAR(100) DEFAULT '', + created_by_agent_id TEXT REFERENCES agents(id), + created_by_sender_id VARCHAR(255) DEFAULT '', + metadata TEXT NOT NULL DEFAULT '{}', + custom_scope TEXT, + tenant_id TEXT NOT NULL REFERENCES tenants(id), + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + UNIQUE(task_id, path) + )`) + db.Exec(`INSERT INTO team_task_attachments SELECT * FROM team_task_attachments_v15`) + db.Exec(`DROP TABLE team_task_attachments_v15`) + + // Recreate vault_links without metadata column. + db.Exec(`CREATE TABLE vault_links_v15 AS SELECT + id, from_doc_id, to_doc_id, link_type, context, custom_scope, created_at + FROM vault_links`) + db.Exec(`DROP TABLE vault_links`) + db.Exec(`CREATE TABLE vault_links ( + id TEXT NOT NULL PRIMARY KEY, + from_doc_id TEXT NOT NULL REFERENCES vault_documents(id) ON DELETE CASCADE, + to_doc_id TEXT NOT NULL REFERENCES vault_documents(id) ON DELETE CASCADE, + link_type TEXT NOT NULL DEFAULT 'wikilink', + context TEXT NOT NULL DEFAULT '', + custom_scope TEXT, + created_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + UNIQUE(from_doc_id, to_doc_id, link_type) + )`) + db.Exec(`INSERT INTO vault_links SELECT * FROM vault_links_v15`) + db.Exec(`DROP TABLE vault_links_v15`) + } + if targetVersion <= 11 { // Migration 12 adds recall_count, recall_score, last_recalled_at. // Recreate episodic_summaries without those columns. diff --git a/internal/store/sqlitestore/vault_documents.go b/internal/store/sqlitestore/vault_documents.go index 7852340d..31148fe9 100644 --- a/internal/store/sqlitestore/vault_documents.go +++ b/internal/store/sqlitestore/vault_documents.go @@ -63,21 +63,28 @@ func (s *SQLiteVaultStore) UpsertDocument(ctx context.Context, doc *store.VaultD if doc.AgentID != nil && *doc.AgentID != "" { agentIDVal = *doc.AgentID } + // SQLite has no GENERATED column equivalent via modernc driver; compute + // the basename app-side. PG auto-populates via GENERATED so this is a + // no-op on PG callers that share the struct. + if doc.PathBasename == "" && doc.Path != "" { + doc.PathBasename = ComputeAttachmentBaseName(doc.Path) + } err = s.db.QueryRowContext(ctx, ` INSERT INTO vault_documents - (id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + (id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (tenant_id, COALESCE(agent_id,''), COALESCE(team_id,''), scope, path) DO UPDATE SET - title = excluded.title, - doc_type = excluded.doc_type, - content_hash = excluded.content_hash, - summary = excluded.summary, - metadata = excluded.metadata, - tenant_id = excluded.tenant_id, - updated_at = excluded.updated_at + path_basename = excluded.path_basename, + title = excluded.title, + doc_type = excluded.doc_type, + content_hash = excluded.content_hash, + summary = excluded.summary, + metadata = excluded.metadata, + tenant_id = excluded.tenant_id, + updated_at = excluded.updated_at RETURNING id`, id, doc.TenantID, agentIDVal, doc.TeamID, doc.Scope, doc.CustomScope, - doc.Path, doc.Title, doc.DocType, doc.ContentHash, doc.Summary, string(meta), now, now, + doc.Path, doc.PathBasename, doc.Title, doc.DocType, doc.ContentHash, doc.Summary, string(meta), now, now, ).Scan(&doc.ID) if err != nil { return fmt.Errorf("vault upsert document: %w", err) @@ -89,7 +96,7 @@ func (s *SQLiteVaultStore) UpsertDocument(ctx context.Context, doc *store.VaultD // Empty agentID means no agent filter. // Team scoping via RunContext: present+TeamID → filter; present+empty → personal; nil → any match. func (s *SQLiteVaultStore) GetDocument(ctx context.Context, tenantID, agentID, path string) (*store.VaultDocument, error) { - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE tenant_id = ? AND path = ?` args := []any{tenantID, path} @@ -114,7 +121,7 @@ func (s *SQLiteVaultStore) GetDocument(ctx context.Context, tenantID, agentID, p // GetDocumentByID retrieves a vault document by ID with tenant isolation. func (s *SQLiteVaultStore) GetDocumentByID(ctx context.Context, tenantID, id string) (*store.VaultDocument, error) { row := s.db.QueryRowContext(ctx, ` - SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE id = ? AND tenant_id = ?`, id, tenantID) return scanVaultDoc(row) } @@ -136,7 +143,7 @@ func (s *SQLiteVaultStore) GetDocumentsByIDs(ctx context.Context, tenantID strin args[i] = id } args = append(args, tenantID) - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE id IN (` + strings.Join(ph, ",") + `) AND tenant_id = ?` rows, err := s.db.QueryContext(ctx, q, args...) if err != nil { @@ -160,7 +167,7 @@ func (s *SQLiteVaultStore) GetDocumentsByIDs(ctx context.Context, tenantID strin // GetDocumentByBasename finds a document by path basename (case-insensitive). func (s *SQLiteVaultStore) GetDocumentByBasename(ctx context.Context, tenantID, agentID, basename string) (*store.VaultDocument, error) { - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE tenant_id = ? AND lower(replace(path, rtrim(path, replace(path, '/', '')), '')) = lower(?)` args := []any{tenantID, basename} @@ -200,7 +207,7 @@ func (s *SQLiteVaultStore) DeleteDocument(ctx context.Context, tenantID, agentID // ListDocuments returns vault documents with optional scope/type filters. func (s *SQLiteVaultStore) ListDocuments(ctx context.Context, tenantID, agentID string, opts store.VaultListOptions) ([]store.VaultDocument, error) { - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE tenant_id = ?` args := []any{tenantID} @@ -318,7 +325,7 @@ func (s *SQLiteVaultStore) Search(ctx context.Context, opts store.VaultSearchOpt maxResults = 10 } - q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, title, doc_type, content_hash, summary, metadata, created_at, updated_at + q := `SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, title, doc_type, content_hash, summary, metadata, created_at, updated_at FROM vault_documents WHERE tenant_id = ? AND (title LIKE ? ESCAPE '\' OR path LIKE ? ESCAPE '\')` @@ -383,7 +390,7 @@ func scanVaultDoc(row *sql.Row) (*store.VaultDocument, error) { var agentID *string ca, ua := &sqliteTime{}, &sqliteTime{} err := row.Scan(&doc.ID, &doc.TenantID, &agentID, &doc.TeamID, &doc.Scope, &doc.CustomScope, - &doc.Path, &doc.Title, &doc.DocType, &doc.ContentHash, &doc.Summary, &meta, ca, ua) + &doc.Path, &doc.PathBasename, &doc.Title, &doc.DocType, &doc.ContentHash, &doc.Summary, &meta, ca, ua) if err != nil { return nil, err } diff --git a/internal/store/sqlitestore/vault_links.go b/internal/store/sqlitestore/vault_links.go index ff78953c..8692e17d 100644 --- a/internal/store/sqlitestore/vault_links.go +++ b/internal/store/sqlitestore/vault_links.go @@ -5,6 +5,7 @@ package sqlitestore import ( "context" "database/sql" + "encoding/json" "fmt" "strings" "time" @@ -82,16 +83,26 @@ func (s *SQLiteVaultStore) CreateLinks(ctx context.Context, links []store.VaultL for start := 0; start < len(valid); start += chunkSize { end := min(start+chunkSize, len(valid)) chunk := valid[start:end] - const cols = 6 + // id, from_doc_id, to_doc_id, link_type, context, metadata, created_at + const cols = 7 iArgs := make([]any, 0, len(chunk)*cols) iPh := make([]string, 0, len(chunk)) for _, l := range chunk { - iPh = append(iPh, "(?,?,?,?,?,?)") - iArgs = append(iArgs, uuid.Must(uuid.NewV7()).String(), l.FromDocID, l.ToDocID, l.LinkType, l.Context, now) + metaJSON, mErr := json.Marshal(l.Metadata) + if mErr != nil || len(metaJSON) == 0 { + metaJSON = []byte("{}") + } + iPh = append(iPh, "(?,?,?,?,?,?,?)") + iArgs = append(iArgs, + uuid.Must(uuid.NewV7()).String(), + l.FromDocID, l.ToDocID, + l.LinkType, l.Context, string(metaJSON), now) } - q := `INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, created_at) + q := `INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, metadata, created_at) VALUES ` + strings.Join(iPh, ",") + ` - ON CONFLICT (from_doc_id, to_doc_id, link_type) DO UPDATE SET context = excluded.context` + ON CONFLICT (from_doc_id, to_doc_id, link_type) DO UPDATE SET + context = excluded.context, + metadata = excluded.metadata` if _, err := s.db.ExecContext(ctx, q, iArgs...); err != nil { return fmt.Errorf("vault batch create links: %w", err) } @@ -99,7 +110,7 @@ func (s *SQLiteVaultStore) CreateLinks(ctx context.Context, links []store.VaultL return nil } -// CreateLink inserts a vault link, updating context on conflict. +// CreateLink inserts a vault link, updating context+metadata on conflict. // Validates same-tenant + same-team boundary before insert. func (s *SQLiteVaultStore) CreateLink(ctx context.Context, link *store.VaultLink) error { // Verify both docs exist and belong to same tenant + team boundary. @@ -124,15 +135,20 @@ func (s *SQLiteVaultStore) CreateLink(ctx context.Context, link *store.VaultLink return fmt.Errorf("vault link: documents belong to different teams") } + metaJSON, mErr := json.Marshal(link.Metadata) + if mErr != nil || len(metaJSON) == 0 { + metaJSON = []byte("{}") + } id := uuid.Must(uuid.NewV7()).String() now := time.Now().UTC().Format(time.RFC3339Nano) err = s.db.QueryRowContext(ctx, ` - INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, created_at) - VALUES (?, ?, ?, ?, ?, ?) + INSERT INTO vault_links (id, from_doc_id, to_doc_id, link_type, context, metadata, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (from_doc_id, to_doc_id, link_type) DO UPDATE SET - context = excluded.context + context = excluded.context, + metadata = excluded.metadata RETURNING id`, - id, link.FromDocID, link.ToDocID, link.LinkType, link.Context, now, + id, link.FromDocID, link.ToDocID, link.LinkType, link.Context, string(metaJSON), now, ).Scan(&link.ID) if err != nil { return fmt.Errorf("vault create link: %w", err) @@ -152,7 +168,7 @@ func (s *SQLiteVaultStore) DeleteLink(ctx context.Context, tenantID, id string) // GetOutLinks returns all links originating from a document, scoped by tenant (F9). func (s *SQLiteVaultStore) GetOutLinks(ctx context.Context, tenantID, docID string) ([]store.VaultLink, error) { rows, err := s.db.QueryContext(ctx, ` - SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.created_at + SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.metadata, vl.created_at FROM vault_links vl JOIN vault_documents d ON d.id = vl.from_doc_id WHERE vl.from_doc_id = ? AND d.tenant_id = ? @@ -176,7 +192,7 @@ func (s *SQLiteVaultStore) GetOutLinksBatch(ctx context.Context, tenantID string } args = append(args, tenantID) rows, err := s.db.QueryContext(ctx, - `SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.created_at + `SELECT vl.id, vl.from_doc_id, vl.to_doc_id, vl.link_type, vl.context, vl.metadata, vl.created_at FROM vault_links vl JOIN vault_documents d ON d.id = vl.from_doc_id WHERE vl.from_doc_id IN (`+ph+`) AND d.tenant_id = ? @@ -260,11 +276,15 @@ func scanVaultLinkRows(rows *sql.Rows) ([]store.VaultLink, error) { var links []store.VaultLink for rows.Next() { var l store.VaultLink + var metaJSON []byte ca := &sqliteTime{} - if err := rows.Scan(&l.ID, &l.FromDocID, &l.ToDocID, &l.LinkType, &l.Context, ca); err != nil { + if err := rows.Scan(&l.ID, &l.FromDocID, &l.ToDocID, &l.LinkType, &l.Context, &metaJSON, ca); err != nil { return nil, err } l.CreatedAt = ca.Time + if len(metaJSON) > 2 { + _ = json.Unmarshal(metaJSON, &l.Metadata) + } links = append(links, l) } return links, rows.Err() diff --git a/internal/store/sqlitestore/vault_source_cleanup.go b/internal/store/sqlitestore/vault_source_cleanup.go new file mode 100644 index 00000000..eb7ac24e --- /dev/null +++ b/internal/store/sqlitestore/vault_source_cleanup.go @@ -0,0 +1,116 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// DeleteLinksBySource removes vault_links rows whose metadata->>'source' +// equals the given source key. Mirrors PG variant — tenant isolation via +// subquery on vault_documents since SQLite DELETE lacks USING. +func (s *SQLiteVaultStore) DeleteLinksBySource(ctx context.Context, tenantID, source string) (int64, error) { + res, err := s.db.ExecContext(ctx, ` + DELETE FROM vault_links + WHERE json_extract(metadata, '$.source') = ? + AND ( + from_doc_id IN (SELECT id FROM vault_documents WHERE tenant_id = ?) + OR to_doc_id IN (SELECT id FROM vault_documents WHERE tenant_id = ?) + ) + `, source, tenantID, tenantID) + if err != nil { + return 0, fmt.Errorf("delete links by source: %w", err) + } + n, _ := res.RowsAffected() + return n, nil +} + +// BatchFindByDelegationIDs returns vault docs sharing any of the given +// delegation_ids, grouped by delegation_id, capped per bucket at `limit`. +// Uses the SQLite partial index idx_vault_docs_delegation added by the v15 +// → v16 migration (json_extract based). +func (s *SQLiteVaultStore) BatchFindByDelegationIDs( + ctx context.Context, + tenantID string, + delegationIDs []string, + limit int, + excludeDocIDs []string, +) (map[string][]store.VaultDocument, error) { + if len(delegationIDs) == 0 || limit <= 0 { + return nil, nil + } + + delegPH := make([]string, len(delegationIDs)) + args := []any{tenantID} + for i, d := range delegationIDs { + delegPH[i] = "?" + args = append(args, d) + } + var excludeClause string + if len(excludeDocIDs) > 0 { + excludePH := make([]string, len(excludeDocIDs)) + for i, id := range excludeDocIDs { + excludePH[i] = "?" + args = append(args, id) + } + excludeClause = " AND id NOT IN (" + strings.Join(excludePH, ",") + ")" + } + args = append(args, limit) + + q := fmt.Sprintf(` +WITH ranked AS ( + SELECT + id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, + title, doc_type, content_hash, summary, metadata, created_at, updated_at, + json_extract(metadata, '$.delegation_id') AS deleg_id, + ROW_NUMBER() OVER ( + PARTITION BY json_extract(metadata, '$.delegation_id') + ORDER BY created_at DESC, id DESC + ) AS rn + FROM vault_documents + WHERE tenant_id = ? + AND json_extract(metadata, '$.delegation_id') IN (%s) + %s +) +SELECT id, tenant_id, agent_id, team_id, scope, custom_scope, path, path_basename, + title, doc_type, content_hash, summary, metadata, created_at, updated_at, deleg_id +FROM ranked +WHERE rn <= ? +ORDER BY deleg_id, created_at DESC +`, strings.Join(delegPH, ","), excludeClause) + + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("batch find by delegation: %w", err) + } + defer rows.Close() + + out := make(map[string][]store.VaultDocument, len(delegationIDs)) + for rows.Next() { + var doc store.VaultDocument + var agentID *string + var metaJSON []byte + var delegID string + ca, ua := &sqliteTime{}, &sqliteTime{} + if err := rows.Scan( + &doc.ID, &doc.TenantID, &agentID, &doc.TeamID, &doc.Scope, &doc.CustomScope, + &doc.Path, &doc.PathBasename, &doc.Title, &doc.DocType, &doc.ContentHash, + &doc.Summary, &metaJSON, ca, ua, &delegID, + ); err != nil { + return nil, err + } + doc.AgentID = agentID + doc.CreatedAt = ca.Time + doc.UpdatedAt = ua.Time + if len(metaJSON) > 2 { + _ = json.Unmarshal(metaJSON, &doc.Metadata) + } + out[delegID] = append(out[delegID], doc) + } + return out, rows.Err() +} diff --git a/internal/store/vault_store.go b/internal/store/vault_store.go index 7c09a9bc..be4ba77c 100644 --- a/internal/store/vault_store.go +++ b/internal/store/vault_store.go @@ -7,30 +7,32 @@ import ( // VaultDocument is a registered document in the Knowledge Vault. type VaultDocument struct { - ID string `json:"id" db:"id"` - TenantID string `json:"tenant_id" db:"tenant_id"` - AgentID *string `json:"agent_id,omitempty" db:"agent_id"` - TeamID *string `json:"team_id,omitempty" db:"team_id"` - Scope string `json:"scope" db:"scope"` // personal, team, shared - CustomScope *string `json:"custom_scope,omitempty" db:"custom_scope"` - Path string `json:"path" db:"path"` // workspace-relative path - Title string `json:"title" db:"title"` - DocType string `json:"doc_type" db:"doc_type"` // context, memory, note, skill, episodic, media - ContentHash string `json:"content_hash" db:"content_hash"` // SHA-256 hex digest - Summary string `json:"summary" db:"summary"` // LLM-generated summary for richer embedding/search - Metadata map[string]any `json:"metadata,omitempty" db:"metadata"` - CreatedAt time.Time `json:"created_at" db:"created_at"` - UpdatedAt time.Time `json:"updated_at" db:"updated_at"` + ID string `json:"id" db:"id"` + TenantID string `json:"tenant_id" db:"tenant_id"` + AgentID *string `json:"agent_id,omitempty" db:"agent_id"` + TeamID *string `json:"team_id,omitempty" db:"team_id"` + Scope string `json:"scope" db:"scope"` // personal, team, shared + CustomScope *string `json:"custom_scope,omitempty" db:"custom_scope"` + Path string `json:"path" db:"path"` // workspace-relative path + PathBasename string `json:"path_basename,omitempty" db:"path_basename"` // lowercased basename (PG GENERATED, SQLite app-populated) + Title string `json:"title" db:"title"` + DocType string `json:"doc_type" db:"doc_type"` // context, memory, note, skill, episodic, media, document + ContentHash string `json:"content_hash" db:"content_hash"` // SHA-256 hex digest + Summary string `json:"summary" db:"summary"` // LLM-generated or synthesized summary for embedding/search + Metadata map[string]any `json:"metadata,omitempty" db:"metadata"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` } // VaultLink is a directed link between two vault documents. type VaultLink struct { - ID string `json:"id" db:"id"` - FromDocID string `json:"from_doc_id" db:"from_doc_id"` - ToDocID string `json:"to_doc_id" db:"to_doc_id"` - LinkType string `json:"link_type" db:"link_type"` // wikilink, reference, etc. - Context string `json:"context" db:"context"` // surrounding text snippet - CreatedAt time.Time `json:"created_at" db:"created_at"` + ID string `json:"id" db:"id"` + FromDocID string `json:"from_doc_id" db:"from_doc_id"` + ToDocID string `json:"to_doc_id" db:"to_doc_id"` + LinkType string `json:"link_type" db:"link_type"` // wikilink, reference, task_attachment, delegation_attachment, ... + Context string `json:"context" db:"context"` // surrounding text snippet or source reference + Metadata map[string]any `json:"metadata,omitempty" db:"metadata"` // {"source": "task:{id}"} etc., used by cleanup paths + CreatedAt time.Time `json:"created_at" db:"created_at"` } // VaultBacklink is an enriched backlink with source doc metadata (single JOIN query). @@ -102,12 +104,33 @@ type VaultStore interface { DeleteDocLinksByType(ctx context.Context, tenantID, docID, linkType string) error DeleteDocLinksByTypes(ctx context.Context, tenantID, docID string, types []string) error + // DeleteLinksBySource removes vault_links rows where metadata->>'source' + // equals the given source key (e.g. "task:{uuid}", "delegation:{uuid}"). + // Used by cleanup paths (DetachFileFromTask, DeleteTask, bulk task delete) + // to surgically remove Phase-04/05 auto-links without touching classify- + // owned links. Returns the number of rows deleted. Tenant isolation is + // enforced by joining vault_documents on tenant_id. + DeleteLinksBySource(ctx context.Context, tenantID, source string) (int64, error) + // Enrichment // UpdateSummaryAndReembed updates summary text and re-generates embedding from title+path+summary. UpdateSummaryAndReembed(ctx context.Context, tenantID, docID, summary string) error // FindSimilarDocs finds documents with similar embeddings to the given docID. // Returns top-N neighbors excluding the source doc. Score = cosine similarity. FindSimilarDocs(ctx context.Context, tenantID, agentID, docID string, limit int) ([]VaultSearchResult, error) + // BatchFindByDelegationIDs returns vault docs sharing any of the given + // delegation_ids in their metadata, keyed by delegation_id. Each + // delegation's bucket is capped at `limit` (ordered by created_at DESC). + // excludeDocIDs is applied as a NOT-IN filter to prevent self-links. + // Single SQL query — uses ROW_NUMBER() PARTITION BY delegation_id over + // the partial index added by migration 000048. + BatchFindByDelegationIDs( + ctx context.Context, + tenantID string, + delegationIDs []string, + limit int, + excludeDocIDs []string, + ) (map[string][]VaultDocument, error) // Embedding SetEmbeddingProvider(provider EmbeddingProvider) diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index 7ba0c007..42e3e2e5 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 47 +const RequiredSchemaVersion uint = 48 diff --git a/migrations/000048_vault_media_linking.down.sql b/migrations/000048_vault_media_linking.down.sql new file mode 100644 index 00000000..c227217a --- /dev/null +++ b/migrations/000048_vault_media_linking.down.sql @@ -0,0 +1,19 @@ +-- Reverse Phase 03 schema changes. + +DROP INDEX IF EXISTS idx_vault_docs_delegation; + +ALTER TABLE vault_links DROP CONSTRAINT IF EXISTS vault_links_to_doc_id_fkey; +ALTER TABLE vault_links + ADD CONSTRAINT vault_links_to_doc_id_fkey + FOREIGN KEY (to_doc_id) REFERENCES vault_documents(id); + +ALTER TABLE vault_links DROP CONSTRAINT IF EXISTS vault_links_from_doc_id_fkey; +ALTER TABLE vault_links + ADD CONSTRAINT vault_links_from_doc_id_fkey + FOREIGN KEY (from_doc_id) REFERENCES vault_documents(id); + +DROP INDEX IF EXISTS idx_vault_links_source; +ALTER TABLE vault_links DROP COLUMN IF EXISTS metadata; + +DROP INDEX IF EXISTS idx_tta_tenant_basename; +ALTER TABLE team_task_attachments DROP COLUMN IF EXISTS base_name; diff --git a/migrations/000048_vault_media_linking.up.sql b/migrations/000048_vault_media_linking.up.sql new file mode 100644 index 00000000..36805991 --- /dev/null +++ b/migrations/000048_vault_media_linking.up.sql @@ -0,0 +1,37 @@ +-- Phase 03: vault media linking foundation. +-- Adds base_name index on task attachments, metadata column on vault_links +-- for cleanup tracking, repairs missing CASCADE FKs on vault_links, and a +-- partial index for delegation lookup inside vault_documents.metadata. + +-- 1. GENERATED base_name on team_task_attachments (mirrors vault_documents.path_basename). +ALTER TABLE team_task_attachments + ADD COLUMN IF NOT EXISTS base_name TEXT + GENERATED ALWAYS AS (lower(regexp_replace(path, '.+/', ''))) STORED; + +CREATE INDEX IF NOT EXISTS idx_tta_tenant_basename + ON team_task_attachments(tenant_id, base_name); + +-- 2. metadata column on vault_links for cleanup tracking (task:{id}, delegation:{id}). +ALTER TABLE vault_links + ADD COLUMN IF NOT EXISTS metadata JSONB NOT NULL DEFAULT '{}'; + +CREATE INDEX IF NOT EXISTS idx_vault_links_source + ON vault_links((metadata->>'source')) + WHERE metadata ? 'source'; + +-- 3. Fix missing CASCADE on vault_links FKs (SQLite already has CASCADE at schema.sql:1548-1549). +ALTER TABLE vault_links DROP CONSTRAINT IF EXISTS vault_links_from_doc_id_fkey; +ALTER TABLE vault_links + ADD CONSTRAINT vault_links_from_doc_id_fkey + FOREIGN KEY (from_doc_id) REFERENCES vault_documents(id) ON DELETE CASCADE; + +ALTER TABLE vault_links DROP CONSTRAINT IF EXISTS vault_links_to_doc_id_fkey; +ALTER TABLE vault_links + ADD CONSTRAINT vault_links_to_doc_id_fkey + FOREIGN KEY (to_doc_id) REFERENCES vault_documents(id) ON DELETE CASCADE; + +-- 4. Partial index for delegation-id lookup in vault_documents.metadata. +-- Keeps index small — only rows explicitly tagged with a delegation_id. +CREATE INDEX IF NOT EXISTS idx_vault_docs_delegation + ON vault_documents((metadata->>'delegation_id')) + WHERE metadata ? 'delegation_id';