From 99b5cd776e20f3428a794a4b8e607118d9f402f4 Mon Sep 17 00:00:00 2001 From: Owais Jamil Date: Mon, 23 Mar 2026 22:20:33 -0500 Subject: [PATCH] build: switch to fts5 --- packages/api/internal/backfill/backfill.go | 46 +++++-- .../api/internal/backfill/backfill_test.go | 112 +++++++++++++++++- packages/api/internal/search/search.go | 72 +++++++---- packages/api/internal/search/search_test.go | 81 +++++++++++++ packages/api/internal/store/db.go | 38 +++++- packages/api/internal/store/db_test.go | 44 +++++++ .../internal/store/migrations/001_initial.sql | 9 -- .../store/migrations/003_documents_fts5.sql | 23 ++++ packages/api/internal/store/sql_store.go | 50 +++++++- packages/api/internal/store/store_test.go | 2 +- packages/api/main.go | 8 +- 11 files changed, 429 insertions(+), 56 deletions(-) create mode 100644 packages/api/internal/search/search_test.go create mode 100644 packages/api/internal/store/db_test.go create mode 100644 packages/api/internal/store/migrations/003_documents_fts5.sql diff --git a/packages/api/internal/backfill/backfill.go b/packages/api/internal/backfill/backfill.go index 3a02102..7acae38 100644 --- a/packages/api/internal/backfill/backfill.go +++ b/packages/api/internal/backfill/backfill.go @@ -84,11 +84,17 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { alreadyTracked := 0 inProgress := 0 + statusFailures := 0 toSubmit := make([]string, 0, len(discovered)) for _, user := range discovered { status, err := r.tap.RepoStatus(ctx, user.DID) if err != nil { - return fmt.Errorf("tap info for %s: %w", user.DID, err) + statusFailures++ + r.log.Warn("tap classification failed", + slog.String("did", user.DID), + slog.String("error", err.Error()), + ) + continue } if status.Tracked && status.Backfilled { alreadyTracked++ @@ -104,10 +110,12 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { r.log.Info("tap classification complete", slog.Int("already_tracked", alreadyTracked), slog.Int("backfill_in_progress", inProgress), + slog.Int("status_failures", statusFailures), slog.Int("to_submit", len(toSubmit)), ) submitted := 0 + submitFailures := 0 for i := 0; i < len(toSubmit); i += opts.BatchSize { end := i + opts.BatchSize if end > len(toSubmit) { @@ -115,15 +123,33 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { } batch := toSubmit[i:end] if err := r.tap.AddRepos(ctx, batch); err != nil { - return fmt.Errorf("submit batch %d-%d: %w", i, end, err) + r.log.Warn("tap batch submission failed", + slog.Int("batch_start", i), + slog.Int("batch_end", end), + slog.Int("batch_size", len(batch)), + slog.String("error", err.Error()), + ) + for _, did := range batch { + if err := r.tap.AddRepos(ctx, []string{did}); err != nil { + submitFailures++ + r.log.Warn("tap repo submission failed", + slog.String("did", did), + slog.String("error", err.Error()), + ) + continue + } + submitted++ + r.log.Info("submitted Tap repo", slog.String("did", did), slog.Int("submitted_total", submitted)) + } + } else { + submitted += len(batch) + r.log.Info("submitted Tap batch", + slog.Int("batch_start", i), + slog.Int("batch_end", end), + slog.Int("batch_size", len(batch)), + slog.Int("submitted_total", submitted), + ) } - submitted += len(batch) - r.log.Info("submitted Tap batch", - slog.Int("batch_start", i), - slog.Int("batch_end", end), - slog.Int("batch_size", len(batch)), - slog.Int("submitted_total", submitted), - ) if end < len(toSubmit) && opts.BatchDelay > 0 { select { case <-ctx.Done(): @@ -138,6 +164,8 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { slog.Int("already_tracked", alreadyTracked), slog.Int("backfill_in_progress", inProgress), slog.Int("submitted", submitted), + slog.Int("status_failures", statusFailures), + slog.Int("submit_failures", submitFailures), ) return nil } diff --git a/packages/api/internal/backfill/backfill_test.go b/packages/api/internal/backfill/backfill_test.go index 62acbc2..dfc6b62 100644 --- a/packages/api/internal/backfill/backfill_test.go +++ b/packages/api/internal/backfill/backfill_test.go @@ -2,10 +2,12 @@ package backfill import ( "context" + "errors" "io" "log/slog" "os" "path/filepath" + "strings" "testing" ) @@ -26,11 +28,16 @@ func (f *fakeFollowFetcher) ListFollowSubjects(_ context.Context, did string) ([ } type fakeTapAdmin struct { - statuses map[string]RepoStatus - added [][]string + statuses map[string]RepoStatus + statusErrs map[string]error + added [][]string + addReposError func(dids []string) error } func (f *fakeTapAdmin) RepoStatus(_ context.Context, did string) (RepoStatus, error) { + if err, ok := f.statusErrs[did]; ok { + return RepoStatus{}, err + } if status, ok := f.statuses[did]; ok { return status, nil } @@ -38,6 +45,11 @@ func (f *fakeTapAdmin) RepoStatus(_ context.Context, did string) (RepoStatus, er } func (f *fakeTapAdmin) AddRepos(_ context.Context, dids []string) error { + if f.addReposError != nil { + if err := f.addReposError(dids); err != nil { + return err + } + } batch := make([]string, len(dids)) copy(batch, dids) f.added = append(f.added, batch) @@ -144,3 +156,99 @@ func TestRunner_SkipsInProgressBackfills(t *testing.T) { t.Fatalf("expected no submission for in-progress did, got %#v", tap.added) } } + +func TestRunner_ContinuesWhenRepoStatusFails(t *testing.T) { + st := &fakeStore{ + collaborators: map[string][]string{ + "did:plc:seed": {"did:plc:good", "did:plc:bad"}, + }, + } + follows := &fakeFollowFetcher{follows: map[string][]string{}} + tap := &fakeTapAdmin{ + statuses: map[string]RepoStatus{}, + statusErrs: map[string]error{"did:plc:bad": errors.New("tap info request failed: status 502")}, + } + resolver := &fakeResolver{mapping: map[string]string{"alice.tangled.sh": "did:plc:seed"}} + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + r := NewRunnerWithDeps(st, tap, resolver, follows, log) + + dir := t.TempDir() + seedsPath := filepath.Join(dir, "seeds.txt") + if err := os.WriteFile(seedsPath, []byte("alice.tangled.sh\n"), 0o644); err != nil { + t.Fatalf("write seeds: %v", err) + } + + err := r.Run(context.Background(), Options{ + SeedsPath: seedsPath, + MaxHops: 1, + Concurrency: 1, + BatchSize: 10, + }) + if err != nil { + t.Fatalf("run backfill: %v", err) + } + + if len(tap.added) != 1 { + t.Fatalf("expected one submission batch, got %d", len(tap.added)) + } + if len(tap.added[0]) != 2 { + t.Fatalf("expected seed and good DID submitted, got %#v", tap.added[0]) + } + for _, did := range tap.added[0] { + if did == "did:plc:bad" { + t.Fatalf("did with status error should have been skipped, got %#v", tap.added[0]) + } + } +} + +func TestRunner_FallsBackToSingleRepoSubmissionOnBatchFailure(t *testing.T) { + st := &fakeStore{ + collaborators: map[string][]string{ + "did:plc:seed": {"did:plc:good", "did:plc:bad"}, + }, + } + follows := &fakeFollowFetcher{follows: map[string][]string{}} + tap := &fakeTapAdmin{ + statuses: map[string]RepoStatus{}, + addReposError: func(dids []string) error { + if len(dids) > 1 { + return errors.New("repos add failed: status 502") + } + if len(dids) == 1 && strings.Contains(dids[0], "bad") { + return errors.New("repos add failed: status 502") + } + return nil + }, + } + resolver := &fakeResolver{mapping: map[string]string{"alice.tangled.sh": "did:plc:seed"}} + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + r := NewRunnerWithDeps(st, tap, resolver, follows, log) + + dir := t.TempDir() + seedsPath := filepath.Join(dir, "seeds.txt") + if err := os.WriteFile(seedsPath, []byte("alice.tangled.sh\n"), 0o644); err != nil { + t.Fatalf("write seeds: %v", err) + } + + err := r.Run(context.Background(), Options{ + SeedsPath: seedsPath, + MaxHops: 1, + Concurrency: 1, + BatchSize: 10, + }) + if err != nil { + t.Fatalf("run backfill: %v", err) + } + + if len(tap.added) != 2 { + t.Fatalf("expected successful individual fallbacks only, got %#v", tap.added) + } + for _, batch := range tap.added { + if len(batch) != 1 { + t.Fatalf("expected only single-DID successful submissions after batch fallback, got %#v", tap.added) + } + if batch[0] == "did:plc:bad" { + t.Fatalf("bad DID should not have been successfully submitted, got %#v", tap.added) + } + } +} diff --git a/packages/api/internal/search/search.go b/packages/api/internal/search/search.go index 73ad9c0..4fce8a7 100644 --- a/packages/api/internal/search/search.go +++ b/packages/api/internal/search/search.go @@ -24,20 +24,20 @@ type Params struct { // Result is a single search hit. type Result struct { - ID string `json:"id"` - Collection string `json:"collection"` - RecordType string `json:"record_type"` - Title string `json:"title"` - BodySnippet string `json:"body_snippet,omitempty"` - Summary string `json:"summary,omitempty"` - RepoName string `json:"repo_name,omitempty"` - AuthorHandle string `json:"author_handle,omitempty"` - DID string `json:"did"` - ATURI string `json:"at_uri"` - Score float64 `json:"score"` + ID string `json:"id"` + Collection string `json:"collection"` + RecordType string `json:"record_type"` + Title string `json:"title"` + BodySnippet string `json:"body_snippet,omitempty"` + Summary string `json:"summary,omitempty"` + RepoName string `json:"repo_name,omitempty"` + AuthorHandle string `json:"author_handle,omitempty"` + DID string `json:"did"` + ATURI string `json:"at_uri"` + Score float64 `json:"score"` MatchedBy []string `json:"matched_by"` - CreatedAt string `json:"created_at,omitempty"` - UpdatedAt string `json:"updated_at,omitempty"` + CreatedAt string `json:"created_at,omitempty"` + UpdatedAt string `json:"updated_at,omitempty"` } // Response is the search API response envelope. @@ -67,6 +67,8 @@ func (r *Repository) Ping(ctx context.Context) error { // Keyword runs a full-text keyword search. func (r *Repository) Keyword(ctx context.Context, p Params) (*Response, error) { + ftsQuery := toFTS5Query(p.Query) + // Build filter conditions beyond the base FTS match. var filters []string var filterArgs []any @@ -108,40 +110,41 @@ func (r *Repository) Keyword(ctx context.Context, p Params) (*Response, error) { filterArgs = append(filterArgs, p.State) } - where := "fts_match(d.title, d.body, d.summary, d.repo_name, d.author_handle, d.tags_json, ?) AND d.deleted_at IS NULL" + where := "documents_fts MATCH ? AND d.deleted_at IS NULL" if len(filters) > 0 { where += " AND " + strings.Join(filters, " AND ") } // Count total matching documents. - countSQL := fmt.Sprintf("SELECT COUNT(*) FROM documents d %s WHERE %s", join, where) - countArgs := append([]any{p.Query}, filterArgs...) + countSQL := fmt.Sprintf("SELECT COUNT(*) FROM documents_fts JOIN documents d ON d.id = documents_fts.id %s WHERE %s", join, where) + countArgs := append([]any{ftsQuery}, filterArgs...) var total int if err := r.db.QueryRowContext(ctx, countSQL, countArgs...).Scan(&total); err != nil { - return nil, fmt.Errorf("count: %w", err) + return nil, explainNativeFTSError("count", err) } // Fetch results with score and snippet. resultsSQL := fmt.Sprintf(` SELECT d.id, d.title, d.summary, d.repo_name, d.author_handle, d.did, d.at_uri, d.collection, d.record_type, d.created_at, d.updated_at, - fts_score(d.title, d.body, d.summary, d.repo_name, d.author_handle, d.tags_json, ?) AS score, - fts_highlight(d.body, '', '', ?) AS body_snippet - FROM documents d + -bm25(documents_fts, 0.0, 3.0, 1.0, 1.5, 2.5, 2.0, 1.2) AS score, + snippet(documents_fts, 2, '', '', '...', 20) AS body_snippet + FROM documents_fts + JOIN documents d ON d.id = documents_fts.id %s WHERE %s ORDER BY score DESC LIMIT ? OFFSET ?`, join, where) - resultsArgs := make([]any, 0, 3+len(filterArgs)+2) - resultsArgs = append(resultsArgs, p.Query, p.Query, p.Query) // score, highlight, match + resultsArgs := make([]any, 0, 1+len(filterArgs)+2) + resultsArgs = append(resultsArgs, ftsQuery) resultsArgs = append(resultsArgs, filterArgs...) resultsArgs = append(resultsArgs, p.Limit, p.Offset) rows, err := r.db.QueryContext(ctx, resultsSQL, resultsArgs...) if err != nil { - return nil, fmt.Errorf("search: %w", err) + return nil, explainNativeFTSError("search", err) } defer rows.Close() @@ -182,3 +185,26 @@ func (r *Repository) Keyword(ctx context.Context, p Params) (*Response, error) { Results: results, }, nil } + +func explainNativeFTSError(op string, err error) error { + msg := err.Error() + if strings.Contains(msg, "no such table: documents_fts") || + strings.Contains(msg, "no such module: fts5") { + return fmt.Errorf("%s: SQLite FTS5 is unavailable on this database; ensure the FTS5 migration succeeded and that Turso SQLite extensions are enabled for this database/group: %w", op, err) + } + return fmt.Errorf("%s: %w", op, err) +} + +func toFTS5Query(raw string) string { + parts := strings.Fields(raw) + if len(parts) == 0 { + return `""` + } + + quoted := make([]string, 0, len(parts)) + for _, part := range parts { + part = strings.ReplaceAll(part, `"`, `""`) + quoted = append(quoted, `"`+part+`"`) + } + return strings.Join(quoted, " OR ") +} diff --git a/packages/api/internal/search/search_test.go b/packages/api/internal/search/search_test.go new file mode 100644 index 0000000..be72324 --- /dev/null +++ b/packages/api/internal/search/search_test.go @@ -0,0 +1,81 @@ +package search_test + +import ( + "context" + "os" + "path/filepath" + "testing" + + "tangled.org/desertthunder.dev/twister/internal/search" + "tangled.org/desertthunder.dev/twister/internal/store" +) + +func TestKeywordSearchUsesFTS5Index(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "search.db") + url := "file:" + dbPath + + db, err := store.Open(url, "") + if err != nil { + t.Fatalf("open: %v", err) + } + t.Cleanup(func() { + _ = db.Close() + _ = os.Remove(dbPath) + }) + + if err := store.Migrate(db, url); err != nil { + t.Fatalf("migrate: %v", err) + } + + st := store.New(db) + repo := search.NewRepository(db) + ctx := context.Background() + + doc := &store.Document{ + ID: "did:plc:abc|sh.tangled.repo|desert", + DID: "did:plc:abc", + Collection: "sh.tangled.repo", + RKey: "desert", + ATURI: "at://did:plc:abc/sh.tangled.repo/desert", + CID: "bafyreidesert", + RecordType: "repo", + Title: "desert-runner", + Body: "desert search repository", + Summary: "index me", + RepoName: "desert-runner", + AuthorHandle: "owais.tangled.org", + } + if err := st.UpsertDocument(ctx, doc); err != nil { + t.Fatalf("upsert doc: %v", err) + } + + resp, err := repo.Keyword(ctx, search.Params{Query: "desert", Limit: 10}) + if err != nil { + t.Fatalf("keyword search: %v", err) + } + if resp.Total != 1 { + t.Fatalf("total: got %d want 1", resp.Total) + } + if len(resp.Results) != 1 { + t.Fatalf("results length: got %d want 1", len(resp.Results)) + } + if resp.Results[0].ID != doc.ID { + t.Fatalf("result id: got %q want %q", resp.Results[0].ID, doc.ID) + } + if resp.Results[0].BodySnippet == "" { + t.Fatal("expected body snippet") + } + + if err := st.MarkDeleted(ctx, doc.ID); err != nil { + t.Fatalf("mark deleted: %v", err) + } + + resp, err = repo.Keyword(ctx, search.Params{Query: "desert", Limit: 10}) + if err != nil { + t.Fatalf("keyword search after delete: %v", err) + } + if resp.Total != 0 || len(resp.Results) != 0 { + t.Fatalf("expected no results after delete, got total=%d len=%d", resp.Total, len(resp.Results)) + } +} diff --git a/packages/api/internal/store/db.go b/packages/api/internal/store/db.go index e0b4024..72e91b3 100644 --- a/packages/api/internal/store/db.go +++ b/packages/api/internal/store/db.go @@ -17,6 +17,11 @@ var migrationsFS embed.FS var extensionMigrationNoticeLogged bool +type migrationMode struct { + allowTursoExtensionSkip bool + targetDescription string +} + // Open establishes a connection to the database. // For remote Turso URLs (libsql:// or https://) it uses the libsql-client-go driver. // For local file: URLs it uses the pure-Go SQLite driver (no CGo required). @@ -46,7 +51,11 @@ func driverAndDSN(url, token string) (driver, dsn string) { } // Migrate runs all embedded SQL migration files in order. -func Migrate(db *sql.DB) error { +func Migrate(db *sql.DB, url string) error { + mode := migrationMode{ + allowTursoExtensionSkip: strings.HasPrefix(url, "file:"), + targetDescription: migrationTargetDescription(url), + } entries, err := migrationsFS.ReadDir("migrations") if err != nil { return fmt.Errorf("read migrations dir: %w", err) @@ -62,7 +71,7 @@ func Migrate(db *sql.DB) error { if err != nil { return fmt.Errorf("read migration %s: %w", entry.Name(), err) } - if err := execMigration(db, entry.Name(), string(data)); err != nil { + if err := execMigration(db, entry.Name(), string(data), mode); err != nil { return err } slog.Info("migration applied", "file", entry.Name()) @@ -70,26 +79,43 @@ func Migrate(db *sql.DB) error { return nil } -func execMigration(db *sql.DB, name, content string) error { +func execMigration(db *sql.DB, name, content string, mode migrationMode) error { for _, stmt := range splitStatements(content) { if _, err := db.Exec(stmt); err != nil { upper := strings.ToUpper(stmt) - if strings.Contains(upper, "USING FTS") || strings.Contains(upper, "LIBSQL_VECTOR_IDX") { + if strings.Contains(upper, "LIBSQL_VECTOR_IDX") { if !extensionMigrationNoticeLogged { extensionMigrationNoticeLogged = true - slog.Info("migration: skipping Turso extension indexes in this environment", + slog.Info("migration: skipping unsupported extension index", "migration", name, - "reason", "database engine does not support Turso-specific FTS/vector DDL", + "reason", "database engine does not support vector index DDL in this environment", ) } continue } + if strings.Contains(upper, "CREATE VIRTUAL TABLE") && strings.Contains(upper, "USING FTS5") { + return fmt.Errorf( + "migration %s: SQLite FTS5 statement failed on %s: %w\nstatement: %s\nhint: this app uses SQLite FTS5 on Turso Cloud. Enable SQLite extensions for the Turso group/database before rerunning the service", + name, mode.targetDescription, err, stmt, + ) + } return fmt.Errorf("migration %s: exec failed: %w\nstatement: %s", name, err, stmt) } } return nil } +func migrationTargetDescription(url string) string { + switch { + case strings.HasPrefix(url, "file:"): + return "local SQLite" + case strings.HasPrefix(url, "libsql://"), strings.HasPrefix(url, "https://"): + return "remote Turso/libSQL" + default: + return "database" + } +} + func splitStatements(content string) []string { var stmts []string for _, s := range strings.Split(content, ";") { diff --git a/packages/api/internal/store/db_test.go b/packages/api/internal/store/db_test.go new file mode 100644 index 0000000..cf3bf43 --- /dev/null +++ b/packages/api/internal/store/db_test.go @@ -0,0 +1,44 @@ +package store + +import ( + "database/sql" + "strings" + "testing" + + _ "modernc.org/sqlite" +) + +func TestExecMigrationSkipsTursoExtensionDDLForLocalSQLite(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + err = execMigration(db, "003_documents_fts5.sql", "CREATE VIRTUAL TABLE documents_fts USING fts5(title);", migrationMode{ + allowTursoExtensionSkip: true, + targetDescription: "local SQLite", + }) + if err != nil { + t.Fatalf("expected local SQLite migration to create FTS5 table: %v", err) + } +} + +func TestExecMigrationFailsForRemoteWhenNativeFTSUnavailable(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + err = execMigration(db, "003_documents_fts5.sql", "CREATE VIRTUAL TABLE documents_fts USING fts5(", migrationMode{ + allowTursoExtensionSkip: false, + targetDescription: "remote Turso/libSQL", + }) + if err == nil { + t.Fatal("expected remote migration to fail when FTS5 is unavailable") + } + if !strings.Contains(err.Error(), "uses SQLite FTS5 on Turso Cloud") { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/packages/api/internal/store/migrations/001_initial.sql b/packages/api/internal/store/migrations/001_initial.sql index 267f742..1da7a1a 100644 --- a/packages/api/internal/store/migrations/001_initial.sql +++ b/packages/api/internal/store/migrations/001_initial.sql @@ -32,15 +32,6 @@ CREATE INDEX IF NOT EXISTS idx_documents_created_at ON documents(created_at); CREATE INDEX IF NOT EXISTS idx_documents_deleted_at ON documents(deleted_at); -CREATE INDEX IF NOT EXISTS idx_documents_fts ON documents USING fts ( - title WITH tokenizer=default, - body WITH tokenizer=default, - summary WITH tokenizer=default, - repo_name WITH tokenizer=simple, - author_handle WITH tokenizer=raw, - tags_json WITH tokenizer=simple -) WITH (weights='title=3.0,repo_name=2.5,author_handle=2.0,summary=1.5,tags_json=1.2,body=1.0'); - CREATE TABLE IF NOT EXISTS sync_state ( consumer_name TEXT PRIMARY KEY, cursor TEXT NOT NULL, diff --git a/packages/api/internal/store/migrations/003_documents_fts5.sql b/packages/api/internal/store/migrations/003_documents_fts5.sql new file mode 100644 index 0000000..f4963d7 --- /dev/null +++ b/packages/api/internal/store/migrations/003_documents_fts5.sql @@ -0,0 +1,23 @@ +CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts USING fts5 ( + id UNINDEXED, + title, + body, + summary, + repo_name, + author_handle, + tags_json, + tokenize = 'unicode61' +); + +DELETE FROM documents_fts; + +INSERT INTO documents_fts (id, title, body, summary, repo_name, author_handle, tags_json) +SELECT d.id, + COALESCE(d.title, ''), + COALESCE(d.body, ''), + COALESCE(d.summary, ''), + COALESCE(d.repo_name, ''), + COALESCE(d.author_handle, ''), + COALESCE(d.tags_json, '') +FROM documents d +WHERE d.deleted_at IS NULL; diff --git a/packages/api/internal/store/sql_store.go b/packages/api/internal/store/sql_store.go index 332b126..b8c8ab0 100644 --- a/packages/api/internal/store/sql_store.go +++ b/packages/api/internal/store/sql_store.go @@ -20,7 +20,13 @@ func New(db *sql.DB) Store { func (s *SQLStore) UpsertDocument(ctx context.Context, doc *Document) error { doc.IndexedAt = time.Now().UTC().Format(time.RFC3339) - _, err := s.db.ExecContext(ctx, ` + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin upsert document tx: %w", err) + } + defer tx.Rollback() + + _, err = tx.ExecContext(ctx, ` INSERT INTO documents ( id, did, collection, rkey, at_uri, cid, record_type, title, body, summary, repo_did, repo_name, author_handle, @@ -52,6 +58,12 @@ func (s *SQLStore) UpsertDocument(ctx context.Context, doc *Document) error { if err != nil { return fmt.Errorf("upsert document: %w", err) } + if err := syncDocumentFTS(ctx, tx, doc); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit upsert document tx: %w", err) + } return nil } @@ -74,11 +86,23 @@ func (s *SQLStore) GetDocument(ctx context.Context, id string) (*Document, error func (s *SQLStore) MarkDeleted(ctx context.Context, id string) error { now := time.Now().UTC().Format(time.RFC3339) - _, err := s.db.ExecContext(ctx, + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin mark deleted tx: %w", err) + } + defer tx.Rollback() + + _, err = tx.ExecContext(ctx, `UPDATE documents SET deleted_at = ? WHERE id = ?`, now, id) if err != nil { return fmt.Errorf("mark deleted: %w", err) } + if _, err := tx.ExecContext(ctx, `DELETE FROM documents_fts WHERE id = ?`, id); err != nil { + return fmt.Errorf("delete document from fts: %w", err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit mark deleted tx: %w", err) + } return nil } @@ -289,3 +313,25 @@ func nullableStr(s string) any { } return s } + +type execer interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +func syncDocumentFTS(ctx context.Context, db execer, doc *Document) error { + if _, err := db.ExecContext(ctx, `DELETE FROM documents_fts WHERE id = ?`, doc.ID); err != nil { + return fmt.Errorf("delete document from fts: %w", err) + } + if doc.DeletedAt != "" { + return nil + } + _, err := db.ExecContext(ctx, ` + INSERT INTO documents_fts (id, title, body, summary, repo_name, author_handle, tags_json) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + doc.ID, doc.Title, doc.Body, doc.Summary, doc.RepoName, doc.AuthorHandle, doc.TagsJSON, + ) + if err != nil { + return fmt.Errorf("insert document into fts: %w", err) + } + return nil +} diff --git a/packages/api/internal/store/store_test.go b/packages/api/internal/store/store_test.go index f1ee2ff..8303dea 100644 --- a/packages/api/internal/store/store_test.go +++ b/packages/api/internal/store/store_test.go @@ -24,7 +24,7 @@ func TestIntegration(t *testing.T) { os.Remove(dbPath) }) - if err := store.Migrate(db); err != nil { + if err := store.Migrate(db, url); err != nil { t.Fatalf("migrate: %v", err) } diff --git a/packages/api/main.go b/packages/api/main.go index 0ba2eeb..841e936 100644 --- a/packages/api/main.go +++ b/packages/api/main.go @@ -4,8 +4,8 @@ import ( "context" "fmt" "log/slog" - "os" "net/http" + "os" "os/signal" "syscall" "time" @@ -82,7 +82,7 @@ func newAPICmd() *cobra.Command { } defer db.Close() - if err := store.Migrate(db); err != nil { + if err := store.Migrate(db, cfg.TursoURL); err != nil { return fmt.Errorf("migrate database: %w", err) } @@ -125,7 +125,7 @@ func newIndexerCmd() *cobra.Command { } defer db.Close() - if err := store.Migrate(db); err != nil { + if err := store.Migrate(db, cfg.TursoURL); err != nil { return fmt.Errorf("migrate database: %w", err) } @@ -220,7 +220,7 @@ func newBackfillCmd() *cobra.Command { } defer db.Close() - if err := store.Migrate(db); err != nil { + if err := store.Migrate(db, cfg.TursoURL); err != nil { return fmt.Errorf("migrate database: %w", err) } -- 2.51.2