feat(store): vault media linking schema migration + document support

- Add migration 000048: vault_documents.path_basename + vault_links.metadata (PG)
- Update SQLite schema.sql: add path_basename, metadata columns + v15→v16 markers
- Add schema.go backfill: loop-patch v15→v16 + basename extraction for existing docs
- Update vault_documents store: SELECT/INSERT with path_basename (PG + SQLite)
- Update vault_links store: SELECT/INSERT with metadata (PG + SQLite)
- Add vault_source_cleanup: helper to purge orphaned sources + links by source ID
- Add basename_helper.go + test: extract filename from path for DB sync
- Update version.go: RequiredSchemaVersion → 48, SQLite SchemaVersion → 16
- Update vault_store.go: DeleteLinksBySource + BatchFindByDelegationIDs interface
This commit is contained in:
viettranx committed 2026-04-11 21:22:23 +07:00
1 parent 5623f97961
commit 2644a08b2b
16 files changed
+757 -111

No files matched your search

+8 -8
View File
@@ -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`
+31 -12
View File
@@ -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()
+27 -25
View File
@@ -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()
+154
View File
@@ -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
}
@@ -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))
}
@@ -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)
}
}
}
+108 -1
View File
@@ -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)
}
}
+25 -14
View File
@@ -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;
@@ -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.
+24 -17
View File
@@ -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
}
+33 -13
View File
@@ -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()
@@ -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()
}
+43 -20
View File
@@ -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)
+1 -1
View File
@@ -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
@@ -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;
@@ -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';