diff --git a/docs/README.md b/docs/README.md index 9e65672..0a6e45e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -7,6 +7,7 @@ Completed work — architecture, APIs, and data models as built. - [`reference/api.md`](reference/api.md) — Go search API service - [`reference/app.md`](reference/app.md) — Ionic Vue mobile app - [`reference/lexicons.md`](reference/lexicons.md) — Tangled AT Protocol record types +- [`reference/resync.md`](reference/resync.md) — Backfill and repo-resync recovery playbook ## Specs diff --git a/docs/reference/resync.md b/docs/reference/resync.md new file mode 100644 index 0000000..3a1a3cf --- /dev/null +++ b/docs/reference/resync.md @@ -0,0 +1,185 @@ +--- +title: Backfill & Resync Playbook +updated: 2026-03-25 +--- + +Twister's search index has three recovery paths. Choose based on what broke. + +| Situation | Recovery path | +| ----------------------------------------------------- | -------------------------------------------- | +| FTS index corrupted or drifted from stored documents | `twister reindex` | +| Documents missing — never received via Tap | `twister backfill` + let the indexer consume | +| Documents missing — received but fields empty/wrong | `twister enrich` | +| Full index loss — DB dropped or migrated | backfill then reindex then enrich | +| Tap cursor too far ahead — events skipped after a gap | cursor reset via `sync_state` table | + +--- + +## Paths Overview + +**Tap** is the authoritative ingest and backfill path. Documents reach the index +when the `indexer` consumes events from Tap. Completeness depends on which DIDs +Tap is tracking. + +**Read-through indexing** closes gaps on demand: when the API fetches a record +not yet in the index, it enqueues a background job. This supplements Tap but is +not a substitute for it. + +**JetStream** feeds only the activity cache (`/activity`). It does not contribute +to the search index. + +--- + +## Commands + +### `twister indexer` + +Runs the Tap consumer. Must be running continuously for real-time indexing. +Persists cursor to `sync_state` table under consumer name `indexer-tap-v1`. + +### `twister backfill` + +Discovers users via follow graph from seed DIDs/handles, checks Tap status for +each, and registers untracked repos with Tap `/repos/add`. + +```sh +# dry-run first +twister backfill --seeds seeds.txt --max-hops 2 --dry-run + +# real run +twister backfill --seeds seeds.txt --max-hops 2 \ + --concurrency 5 --batch-size 10 --batch-delay 1s +``` + +Safe to re-run. Discovery deduplicates and `repos/add` is idempotent. + +### `twister reindex` + +Re-upserts stored documents into the FTS table and runs `optimize`. Does not +re-fetch from upstream — only re-processes what is already in the DB. + +```sh +twister reindex # all documents +twister reindex --collection sh.tangled.repo +twister reindex --did did:plc:abc123 +twister reindex --dry-run # preview without writing +``` + +Run this when: FTS results are stale after a schema migration, after a bulk +document import, or whenever search quality seems inconsistent with stored data. + +### `twister enrich` + +Resolves missing `author_handle`, `repo_name`, and `web_url` via XRPC for +documents already in the DB. + +```sh +twister enrich # all documents +twister enrich --collection sh.tangled.repo.issue +twister enrich --did did:plc:abc123 +twister enrich --dry-run +``` + +Run this when: search results show documents with empty author handles, or +after deploying enrichment logic changes. + +--- + +## Scenario Playbooks + +### FTS index out of sync + +Documents exist in the DB but search returns wrong/stale results. + +```sh +twister reindex --dry-run # confirm scope +twister reindex # re-upsert + FTS optimize +``` + +Verify with `GET /search?q=`. + +### Documents missing from search + +Fetch a known record directly. If it returns from `/actors/{handle}/repos/{repo}` +but does not appear in `/search`, the document was never indexed. + +1. Check if the DID is tracked by Tap. If not, run `backfill`: + + ```sh + twister backfill --seeds --max-hops 0 + ``` + +2. Once Tap is tracking the DID, the `indexer` will deliver historical events. + Monitor progress via `GET /admin/status` (requires `ENABLE_ADMIN_ENDPOINTS=true`). + +3. If you need the record indexed immediately, fetch it through the API — the + read-through indexer will enqueue it automatically. + +### Enrichment gaps + +Documents appear in search but `author_handle` or `repo_name` is empty. + +```sh +twister enrich --dry-run # preview what would be resolved +twister enrich # apply +twister reindex # re-sync FTS after field updates +``` + +### Full index recovery + +Use this sequence after a DB drop, migration to a new Turso database, or other +full-loss event. + +1. Confirm migrations ran: `twister api --local` performs `store.Migrate` on startup. +2. Register repos with Tap: + + ```sh + twister backfill --seeds seeds.txt --max-hops 2 --dry-run + twister backfill --seeds seeds.txt --max-hops 2 + ``` + +3. Start the indexer and let it consume: `twister indexer` +4. Once backfill is complete, enrich fields and re-sync FTS: + + ```sh + twister enrich + twister reindex + ``` + +5. Verify: `GET /admin/status` for cursor progress, `GET /readyz` for DB health. + +### Tap cursor reset + +If the indexer cursor is ahead of what Tap will deliver (e.g., after a Tap +instance reset), events will be skipped until the cursor catches up. + +To reset the cursor and reprocess from the beginning of Tap's retention window: + +```sql +DELETE FROM sync_state WHERE consumer_name = 'indexer-tap-v1'; +``` + +Then restart the `indexer`. It will start from the head of the stream and +process all events Tap delivers. + +> **Note:** This does not cause duplicate documents — `UpsertDocument` is +> idempotent. It may reprocess a large backlog depending on Tap retention. + +--- + +## Checking Status + +With `ENABLE_ADMIN_ENDPOINTS=true`: + +```sh +curl -H "Authorization: Bearer $ADMIN_AUTH_TOKEN" \ + http://localhost:8080/admin/status +``` + +Response includes: + +- `tap.cursor` — last Tap event ID processed by the indexer +- `tap.updated_at` — when the cursor was last advanced +- `jetstream.cursor` — JetStream timestamp cursor (activity cache only) +- `documents` — total searchable document count +- `pending_jobs` — read-through indexing jobs not yet processed diff --git a/docs/roadmap.md b/docs/roadmap.md index afabfb5..9477b65 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -26,8 +26,8 @@ Highest priority. This work blocks further investment in search quality and broa - [x] Add a JetStream cache consumer with a persisted timestamp cursor - [x] Seed the JetStream cursor to `now - 24h` on first boot and rewind slightly on reconnect - [x] Store and serve bounded recent activity from the local cache -- [ ] Keep Tap as the authoritative indexing and bulk backfill path -- [ ] Define a controlled backfill and repo-resync playbook for recovery (`docs/references/resync.md`) +- [x] Keep Tap as the authoritative indexing and bulk backfill path +- [x] Define a controlled backfill and repo-resync playbook for recovery (`docs/reference/resync.md`) ## API: Constellation Integration diff --git a/packages/api/README.md b/packages/api/README.md index 786b5fb..d262f9f 100644 --- a/packages/api/README.md +++ b/packages/api/README.md @@ -174,3 +174,13 @@ The API proxies all upstream AT Protocol and social-graph requests so the app ha | `GET /identity/did/{did}` | `https://plc.directory/{did}` or `/.well-known/did.json` | | `GET /backlinks/count` | Constellation `getBacklinksCount` (cached) | | `WS /activity/stream` | `wss://jetstream2.us-east.bsky.network/subscribe` | + +## Admin endpoints + +Available when `ENABLE_ADMIN_ENDPOINTS=true`. Require `Authorization: Bearer ` when +`ADMIN_AUTH_TOKEN` is set. + +| Route | Description | +| ---------------------- | -------------------------------------------------------- | +| `GET /admin/status` | Tap cursor, JetStream cursor, document count, job queue | +| `POST /admin/reindex` | Re-sync all (or filtered) documents into the FTS index | diff --git a/packages/api/internal/api/api.go b/packages/api/internal/api/api.go index ee9971c..c911417 100644 --- a/packages/api/internal/api/api.go +++ b/packages/api/internal/api/api.go @@ -18,7 +18,6 @@ import ( "tangled.org/desertthunder.dev/twister/internal/reindex" "tangled.org/desertthunder.dev/twister/internal/search" "tangled.org/desertthunder.dev/twister/internal/store" - "tangled.org/desertthunder.dev/twister/internal/view" "tangled.org/desertthunder.dev/twister/internal/xrpc" ) @@ -47,66 +46,8 @@ func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log } // Handler returns the HTTP handler with all routes registered. -// -// TODO: refactor this to have a func router() that returns [http.Handler] (mux), -// then registerXXX routes func (s *Server) Handler() http.Handler { - mux := http.NewServeMux() - - mux.HandleFunc("GET /healthz", s.handleHealthz) - mux.HandleFunc("GET /readyz", s.handleReadyz) - mux.HandleFunc("GET /oauth/client-metadata.json", s.handleOAuthClientMetadata) - mux.HandleFunc("GET /search", s.handleSearch) - mux.HandleFunc("GET /search/keyword", s.handleSearchKeyword) - - mux.HandleFunc("GET /documents/{id}", s.handleGetDocument) - mux.HandleFunc("GET /profiles/{did}/summary", s.handleProfileSummary) - - mux.HandleFunc("GET /backlinks/count", s.handleBacklinksCount) - mux.HandleFunc("GET /activity", s.handleActivity) - mux.HandleFunc("GET /activity/stream", s.handleActivityStream) - mux.HandleFunc("GET /identity/resolve", s.handleResolveHandle) - mux.HandleFunc("GET /identity/did/{did}", s.handleDidDocument) - mux.HandleFunc("GET /xrpc/knot/{knotHost}/{nsid}", s.handleKnotProxy) - mux.HandleFunc("GET /xrpc/pds/{pds}/{nsid}", s.handlePdsProxy) - mux.HandleFunc("GET /xrpc/bsky/{nsid}", s.handleBskyProxy) - - mux.HandleFunc("GET /actors/{handle}", s.handleGetActor) - mux.HandleFunc("GET /actors/{handle}/repos", s.handleListActorRepos) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}", s.handleGetActorRepo) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/tree", s.handleRepoTree) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/blob", s.handleRepoBlob) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/log", s.handleRepoLog) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/branches", s.handleRepoBranches) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/default-branch", s.handleRepoDefaultBranch) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/languages", s.handleRepoLanguages) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/tags", s.handleRepoTags) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/diff", s.handleRepoDiff) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/compare", s.handleRepoCompare) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/issues", s.handleRepoIssues) - mux.HandleFunc("GET /actors/{handle}/repos/{repo}/pulls", s.handleRepoPulls) - mux.HandleFunc("GET /actors/{handle}/issues", s.handleActorIssues) - mux.HandleFunc("GET /actors/{handle}/pulls", s.handleActorPulls) - mux.HandleFunc("GET /actors/{handle}/following", s.handleActorFollowing) - mux.HandleFunc("GET /actors/{handle}/strings", s.handleActorStrings) - mux.HandleFunc("GET /issues/{handle}/{rkey}", s.handleIssueDetail) - mux.HandleFunc("GET /issues/{handle}/{rkey}/comments", s.handleIssueComments) - mux.HandleFunc("GET /pulls/{handle}/{rkey}", s.handlePullDetail) - mux.HandleFunc("GET /pulls/{handle}/{rkey}/comments", s.handlePullComments) - - if s.cfg.EnableAdminEndpoints { - mux.HandleFunc("POST /admin/reindex", s.handleAdminReindex) - } - - site := view.Handler() - mux.Handle("GET /static/", site) - mux.Handle("GET /docs", site) - mux.Handle("GET /docs/search", site) - mux.Handle("GET /docs/documents", site) - mux.Handle("GET /docs/health", site) - mux.Handle("GET /{$}", site) - - return s.withMiddleware(mux) + return s.router() } // Run starts the HTTP server and blocks until ctx is cancelled. @@ -406,6 +347,79 @@ func (s *Server) handleProfileSummary(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, summary) } +const tapConsumerName = "indexer-tap-v1" + +// handleAdminStatus returns Tap cursor, JetStream cursor, document count, and pending job count. +// Route: GET /admin/status +func (s *Server) handleAdminStatus(w http.ResponseWriter, r *http.Request) { + if s.cfg.AdminAuthToken != "" { + token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + if token != s.cfg.AdminAuthToken { + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "invalid admin token")) + return + } + } + + tapState, err := s.store.GetSyncState(r.Context(), tapConsumerName) + if err != nil { + s.log.Error("admin status: get tap sync state failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to fetch tap sync state")) + return + } + + jsState, err := s.store.GetSyncState(r.Context(), jetstreamConsumerName) + if err != nil { + s.log.Error("admin status: get jetstream sync state failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to fetch jetstream sync state")) + return + } + + docs, err := s.store.CountDocuments(r.Context()) + if err != nil { + s.log.Error("admin status: count documents failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to count documents")) + return + } + + pending, err := s.store.CountPendingIndexingJobs(r.Context()) + if err != nil { + s.log.Error("admin status: count pending jobs failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to count pending jobs")) + return + } + + type syncStateJSON struct { + Cursor string `json:"cursor"` + HighWaterMark string `json:"high_water_mark,omitempty"` + UpdatedAt string `json:"updated_at,omitempty"` + } + + tapJSON := syncStateJSON{} + if tapState != nil { + tapJSON = syncStateJSON{ + Cursor: tapState.Cursor, + HighWaterMark: tapState.HighWaterMark, + UpdatedAt: tapState.UpdatedAt, + } + } + + jsJSON := syncStateJSON{} + if jsState != nil { + jsJSON = syncStateJSON{ + Cursor: jsState.Cursor, + HighWaterMark: jsState.HighWaterMark, + UpdatedAt: jsState.UpdatedAt, + } + } + + writeJSON(w, http.StatusOK, map[string]any{ + "tap": tapJSON, + "jetstream": jsJSON, + "documents": docs, + "pending_jobs": pending, + }) +} + // knownActivityParams is the whitelist of accepted query parameters for the activity endpoint. var knownActivityParams = map[string]bool{ "limit": true, "offset": true, @@ -477,4 +491,3 @@ func (s *Server) handleActivity(w http.ResponseWriter, r *http.Request) { "events": out, }) } - diff --git a/packages/api/internal/api/router.go b/packages/api/internal/api/router.go new file mode 100644 index 0000000..6b68790 --- /dev/null +++ b/packages/api/internal/api/router.go @@ -0,0 +1,118 @@ +package api + +import ( + "net/http" + + "tangled.org/desertthunder.dev/twister/internal/view" +) + +func (s *Server) router() http.Handler { + mux := http.NewServeMux() + + mux.HandleFunc("GET /healthz", s.handleHealthz) + mux.HandleFunc("GET /readyz", s.handleReadyz) + mux.HandleFunc("GET /oauth/client-metadata.json", s.handleOAuthClientMetadata) + + mux.HandleFunc("GET /documents/{id}", s.handleGetDocument) + mux.HandleFunc("GET /profiles/{did}/summary", s.handleProfileSummary) + mux.HandleFunc("GET /backlinks/count", s.handleBacklinksCount) + + s.registerSearchRoutes(mux) + s.registerActivityRoutes(mux) + s.registerIdentityRoutes(mux) + s.registerXRPCRoutes(mux) + s.registerActorRoutes(mux) + s.registerIssueRoutes(mux) + s.registerPullRoutes(mux) + s.registerAdminRoutes(mux) + s.registerSiteRoutes(mux) + + return s.withMiddleware(mux) +} + +func (s *Server) registerSearchRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /search", s.handleSearch) + mux.HandleFunc("GET /search/keyword", s.handleSearchKeyword) +} + +func (s *Server) registerActivityRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /activity", s.handleActivity) + mux.HandleFunc("GET /activity/stream", s.handleActivityStream) +} + +func (s *Server) registerIdentityRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /identity/resolve", s.handleResolveHandle) + mux.HandleFunc("GET /identity/did/{did}", s.handleDidDocument) +} + +func (s *Server) registerXRPCRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /xrpc/knot/{knotHost}/{nsid}", s.handleKnotProxy) + mux.HandleFunc("GET /xrpc/pds/{pds}/{nsid}", s.handlePdsProxy) + mux.HandleFunc("GET /xrpc/bsky/{nsid}", s.handleBskyProxy) +} + +func (s *Server) registerActorRepoRoutes(mux *http.ServeMux) { + const base = "GET /actors/{handle}/repos/{repo}" + + mux.HandleFunc(base, s.handleGetActorRepo) + mux.HandleFunc(base+"/tree", s.handleRepoTree) + mux.HandleFunc(base+"/blob", s.handleRepoBlob) + mux.HandleFunc(base+"/log", s.handleRepoLog) + mux.HandleFunc(base+"/branches", s.handleRepoBranches) + mux.HandleFunc(base+"/default-branch", s.handleRepoDefaultBranch) + mux.HandleFunc(base+"/languages", s.handleRepoLanguages) + mux.HandleFunc(base+"/tags", s.handleRepoTags) + mux.HandleFunc(base+"/diff", s.handleRepoDiff) + mux.HandleFunc(base+"/compare", s.handleRepoCompare) + mux.HandleFunc(base+"/issues", s.handleRepoIssues) + mux.HandleFunc(base+"/pulls", s.handleRepoPulls) +} + +func (s *Server) registerActorRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /actors/{handle}", s.handleGetActor) + mux.HandleFunc("GET /actors/{handle}/repos", s.handleListActorRepos) + mux.HandleFunc("GET /actors/{handle}/issues", s.handleActorIssues) + mux.HandleFunc("GET /actors/{handle}/pulls", s.handleActorPulls) + mux.HandleFunc("GET /actors/{handle}/following", s.handleActorFollowing) + mux.HandleFunc("GET /actors/{handle}/strings", s.handleActorStrings) + + s.registerActorRepoRoutes(mux) +} + +func (s *Server) registerIssueRoutes(mux *http.ServeMux) { + const base = "GET /issues/{handle}/{rkey}" + + mux.HandleFunc(base, s.handleIssueDetail) + mux.HandleFunc(base+"/comments", s.handleIssueComments) +} + +func (s *Server) registerPullRoutes(mux *http.ServeMux) { + const base = "GET /pulls/{handle}/{rkey}" + + mux.HandleFunc(base, s.handlePullDetail) + mux.HandleFunc(base+"/comments", s.handlePullComments) +} + +func (s *Server) registerAdminRoutes(mux *http.ServeMux) { + if !s.cfg.EnableAdminEndpoints { + return + } + mux.HandleFunc("GET /admin/status", s.handleAdminStatus) + mux.HandleFunc("POST /admin/reindex", s.handleAdminReindex) +} + +func (s *Server) registerDocsRoutes(mux *http.ServeMux, h http.Handler) { + mux.Handle("GET /docs", h) + mux.Handle("GET /docs/search", h) + mux.Handle("GET /docs/documents", h) + mux.Handle("GET /docs/health", h) +} + +func (s *Server) registerSiteRoutes(mux *http.ServeMux) { + site := view.Handler() + mux.Handle("GET /static/", site) + + s.registerDocsRoutes(mux, site) + + mux.Handle("GET /{$}", site) +}