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)
}