diff --git a/README.md b/README.md index 423b2ca..4762676 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,15 @@ To run the API against remote Turso instead: just api-dev remote ``` +Run the API smoke checks from the repo root: + +```bash +uv run --project packages/scripts/api twister-api-smoke +``` + +To verify admin endpoints as well, ensure `ADMIN_AUTH_TOKEN` is present in the +environment before running the smoke script. + To run the indexer in local file mode as well: ```bash @@ -91,6 +100,26 @@ If you want the app to call the local API, set this in `apps/twisted/.env`: VITE_TWISTER_API_BASE_URL=http://localhost:8080 ``` +### Local API DB + +The experimental local API database lives at `packages/api/twister-dev.db`. +Treat it as disposable unless you explicitly back it up. + +Operational rules: + +1. Stop the API before copying or restoring the file. +2. Copy `twister-dev.db` and any matching `-wal` or `-shm` sidecars together. +3. Prefer restore-or-rebuild over manual repair if the DB looks suspect. +4. Let the file grow during experiments, then compact or delete it afterward. + +Useful local commands: + +```bash +cd packages/api +du -h twister-dev.db* +ls -lh twister-dev.db* +``` + ## Infrastructure Setup ### Turso @@ -139,6 +168,11 @@ Set these API-specific variables: - `HTTP_BIND_ADDR` - `SEARCH_DEFAULT_LIMIT` - `SEARCH_MAX_LIMIT` +- `ENABLE_ADMIN_ENDPOINTS` +- `ADMIN_AUTH_TOKEN` +- `READ_THROUGH_MODE` +- `READ_THROUGH_COLLECTIONS` +- `READ_THROUGH_MAX_ATTEMPTS` Set these indexer-specific variables: @@ -157,7 +191,3 @@ For a brand-new environment: 3. Verify API readiness and indexer health. 4. Run `twister backfill` with your seed file. 5. Treat the environment as search-ready only after historical backfill completes. - -## Docs - -- Index: [`docs/README.md`](docs/README.md) diff --git a/apps/twisted/README.md b/apps/twisted/README.md index ef253c8..8e7a36c 100644 --- a/apps/twisted/README.md +++ b/apps/twisted/README.md @@ -6,7 +6,7 @@ Ionic Vue client for Twisted — a Tangled browser and search app for Android & - Node.js 20+ - pnpm -- The Twister API running locally (see `packages/api/README.md`) +- The Twister API running locally ## Running locally diff --git a/docs/adr/storage.md b/docs/adr/storage.md index 832cad0..8b4d253 100644 --- a/docs/adr/storage.md +++ b/docs/adr/storage.md @@ -127,7 +127,8 @@ Operational rules: 3. Prefer restore-or-rebuild over repair if the DB becomes suspect. 4. Allow the file to grow during active experiments, then compact or delete it afterward. -The concrete local backup, restore, and disk-growth procedures live in [packages/api/README.md](/Users/owais/Projects/Twisted/packages/api/README.md). +The concrete local backup, restore, and disk-growth procedures live in +[docs/reference/api.md](/Users/owais/Projects/Twisted/docs/reference/api.md). ## Migration Path To Production Turso diff --git a/docs/reference/api.md b/docs/reference/api.md index 7eb3a71..89c7664 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -61,8 +61,28 @@ Response includes query metadata, total count, and an array of results each cont When `ENABLE_ADMIN_ENDPOINTS=true` with a configured `ADMIN_AUTH_TOKEN`: +- **`GET /admin/status`** — Tap cursor, JetStream cursor, document count, and + read-through queue status +- **`GET /admin/indexing/jobs`** — List queue rows, filtered by `status`, + `source`, or `document` +- **`GET /admin/indexing/audit`** — List append-only audit rows, filtered by + `source`, `decision`, or `document` +- **`POST /admin/indexing/enqueue`** — Queue a single record by explicit body - **`POST /admin/reindex`** — Trigger FTS re-sync +### Smoke Checks + +Smoke checks for the API surface live in `packages/scripts/api/`. + +From the repo root: + +```sh +uv run --project packages/scripts/api twister-api-smoke +``` + +If `ADMIN_AUTH_TOKEN` is present in the environment, the smoke script can also +verify `GET /admin/status`. + ### Static Content The API also serves a search site with live search and API documentation at `/` and `/docs*`, built with Alpine.js (no build step, embedded in `internal/view/`). @@ -81,6 +101,12 @@ Turso (libSQL) with the following tables: **record_state** — Issue and PR state cache (open/closed/merged). Keyed by subject AT-URI. +**indexing_jobs** — Durable read-through/admin queue with status, lease owner, +lease expiry, retry counters, and terminal states (`failed`, `dead_letter`). + +**indexing_audit** — Append-only record of enqueue decisions, retries, skips, +completions, and dead letters. + **document_embeddings** — Vector storage (768-dim F32_BLOB with DiskANN cosine index). Schema ready but not yet populated. **embedding_jobs** — Async embedding job queue. Schema ready but worker not yet active. @@ -94,9 +120,15 @@ The indexer connects to Tap via WebSocket, consuming AT Protocol record events i 3. Normalize into a document (extract title, body, summary, metadata) 4. Optionally enrich via XRPC (resolve author handle, repo name, web URL) 5. Upsert into the database (auto-syncs FTS) -6. Advance cursor and acknowledge to Tap +6. Persist the Tap cursor and then acknowledge the event -The indexer resumes from its last cursor on restart (no duplicate processing). It logs status every 30 seconds and uses exponential backoff (1s–5s) for transient failures. +The indexer resumes from its last cursor on restart and replays idempotently. +It logs status every 30 seconds and uses exponential backoff (1s–5s) for +transient failures. + +Read-through indexing is `missing` by default. Only allowed collections can be +queued, detail reads queue single focal records, and bulk list handlers no +longer enqueue whole collections. ## Record Normalizers @@ -140,6 +172,9 @@ All configuration is via environment variables (with `.env` file support): | `TAP_URL` | — | Tap WebSocket URL | | `TAP_AUTH_PASSWORD` | — | Tap admin password | | `INDEXED_COLLECTIONS` | all | Collection allowlist (CSV, supports wildcards) | +| `READ_THROUGH_MODE` | missing | `off`, `missing`, or `broad` | +| `READ_THROUGH_COLLECTIONS` | `INDEXED_COLLECTIONS` | Read-through allowlist | +| `READ_THROUGH_MAX_ATTEMPTS`| 5 | Retries before `dead_letter` | | `HTTP_BIND_ADDR` | `:8080` | API server bind address | | `INDEXER_HEALTH_ADDR` | `:9090` | Indexer health probe address | | `LOG_LEVEL` | info | debug/info/warn/error | @@ -159,3 +194,23 @@ Deployed on Railway with three services: - **tap** — Tap instance (external dependency) All services share the same Turso database. The API and indexer are separate deployments of the same binary with different subcommands. + +## Experimental Local DB + +The local development database lives at `packages/api/twister-dev.db` when the +API runs with `--local`. + +Operational rules: + +1. Stop the API before backup or restore. +2. Copy `twister-dev.db` and any matching `-wal` or `-shm` files together. +3. Prefer restore-or-rebuild over repair if the file becomes suspect. +4. Let the DB grow during active experiments, then compact or delete it later. + +Useful local inspection: + +```sh +cd packages/api +du -h twister-dev.db* +ls -lh twister-dev.db* +``` diff --git a/docs/reference/resync.md b/docs/reference/resync.md index 3a1a3cf..0513e44 100644 --- a/docs/reference/resync.md +++ b/docs/reference/resync.md @@ -21,9 +21,9 @@ Twister's search index has three recovery paths. Choose based on what broke. 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. +**Read-through indexing** now runs in `missing` mode by default: when the API +fetches a record that is absent or stale, and the collection is allowed, it +enqueues a background job. Bulk list reads no longer enqueue entire collections. **JetStream** feeds only the activity cache (`/activity`). It does not contribute to the search index. @@ -110,10 +110,11 @@ but does not appear in `/search`, the document was never indexed. ``` 2. Once Tap is tracking the DID, the `indexer` will deliver historical events. - Monitor progress via `GET /admin/status` (requires `ENABLE_ADMIN_ENDPOINTS=true`). + Monitor progress via `GET /admin/status` and inspect backlog or failures with + `GET /admin/indexing/jobs` and `GET /admin/indexing/audit`. -3. If you need the record indexed immediately, fetch it through the API — the - read-through indexer will enqueue it automatically. +3. If you need the record indexed immediately, fetch the detail endpoint through + the API or enqueue it explicitly with `POST /admin/indexing/enqueue`. ### Enrichment gaps @@ -178,8 +179,9 @@ curl -H "Authorization: Bearer $ADMIN_AUTH_TOKEN" \ 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 +- `tap.cursor` and `tap.updated_at` +- `jetstream.cursor` and `jetstream.updated_at` +- `documents` +- `read_through.pending`, `processing`, `completed`, `failed`, `dead_letter` +- `read_through.oldest_pending_age_s` and `oldest_running_age_s` +- `read_through.last_completed_at` and `last_processed_at` diff --git a/docs/specs/search.md b/docs/specs/search.md index 45b8074..fb5efd2 100644 --- a/docs/specs/search.md +++ b/docs/specs/search.md @@ -96,24 +96,29 @@ Acceptance: - the production backend choice is documented with explicit tradeoffs - the chosen production backend has a migration path from the experimental local setup -The concrete local DB operating procedure lives in `packages/api/README.md`. +The concrete local DB operating procedure lives in `docs/reference/api.md`. The production migration path is documented in `docs/adr/storage.md`. ### Read-Through Indexing -When the API fetches a repo, issue, PR, profile, or similar record directly from upstream, it should enqueue background indexing work if that record is not already searchable. Tap remains the primary ingest path; read-through indexing only closes gaps. +When the API fetches a repo, issue, PR, profile, or similar detail record +directly from upstream, it should enqueue background indexing work only when +that record is missing or stale. Tap remains the primary ingest path; +read-through indexing only closes gaps. Requirements: - add a durable job table for on-demand indexing - deduplicate jobs by stable document identity - reuse the existing normalization and upsert path -- trigger jobs from the handlers that already fetch upstream records +- trigger jobs from detail handlers that already fetch upstream records +- do not enqueue whole collections from list or browse handlers Acceptance: - a fetched-but-missing record becomes searchable shortly after the first successful API read - repeated page views do not create unbounded duplicate work +- queue state and terminal failures are inspectable through admin endpoints - failures are visible through logs and smoke tests ### Activity Cache @@ -215,7 +220,7 @@ Acceptance: } ``` -## Pragmatic Search Strategy +## Search Strategy Indexing via Tap is useful but has proven unreliable for maintaining complete, up-to-date coverage. The approach: diff --git a/packages/api/README.md b/packages/api/README.md index d262f9f..94b78b6 100644 --- a/packages/api/README.md +++ b/packages/api/README.md @@ -1,186 +1,3 @@ -# Twister +# Twister API -Tap-based indexing and search API for Tangled. Acts as a proxy layer between the Twisted app and all upstream AT Protocol services (knots, PDS, Bluesky, Constellation, Jetstream). - -## Requirements - -- Go 1.25+ -- A Turso database (or local SQLite for development) - -## Running locally - -```sh -cd packages/api - -# Start the API server with a local SQLite database (twister-dev.db) -go run . api --local -``` - -The server listens on `:8080` by default. Logs are printed as text when `--local` is set. - -## API Smoke Tests - -Smoke checks for the API surface live in a uv-managed Python project at -`packages/scripts/api/`. - -From the repo root: - -```sh -uv run --project packages/scripts/api twister-api-smoke -``` - -Optional base URL override: - -```sh -TWISTER_API_BASE_URL=http://localhost:8080 \ - uv run --project packages/scripts/api twister-api-smoke -``` - -## Experimental Local DB Operations - -The experimental local database lives at `packages/api/twister-dev.db` when you run Twister from `packages/api` with `--local`. - -This database is for local experimentation only. Treat it as disposable unless you explicitly back it up. - -### Backup - -Recommended procedure: - -1. Stop the Twister process using the local DB. -2. Copy the database file and any SQLite sidecar files if they exist. - -Example: - -```sh -cd packages/api -mkdir -p backups -timestamp="$(date +%Y%m%d-%H%M%S)" -cp twister-dev.db "backups/twister-dev-${timestamp}.db" -test -f twister-dev.db-wal && cp twister-dev.db-wal "backups/twister-dev-${timestamp}.db-wal" -test -f twister-dev.db-shm && cp twister-dev.db-shm "backups/twister-dev-${timestamp}.db-shm" -``` - -For this experimental DB, stop-and-copy is preferred over hot backup complexity. - -### Restore - -Recommended procedure: - -1. Stop the Twister process. -2. Move the current local DB aside if you want to keep it. -3. Copy the backup file back to `twister-dev.db`. -4. Restore matching `-wal` and `-shm` files only if they were captured with the same backup set. - -Example: - -```sh -cd packages/api -mv twister-dev.db "twister-dev.db.broken.$(date +%Y%m%d-%H%M%S)" 2>/dev/null || true -cp backups/twister-dev-YYYYMMDD-HHMMSS.db twister-dev.db -``` - -After restore, restart Twister and let the app run migrations normally. - -### Disk Growth - -The local DB will grow during experimentation because of: - -- indexed documents -- FTS tables -- activity cache rows -- repeated backfill or reindex runs - -Recommended operating procedure: - -1. Check file growth periodically. -2. Delete and rebuild the experimental DB freely when the dataset is no longer useful. -3. Run `VACUUM` only when you intentionally want to compact a long-lived local DB. -4. Keep old backups out of the repo and rotate them manually. - -Example inspection commands: - -```sh -cd packages/api -du -h twister-dev.db* -ls -lh twister-dev.db* -``` - -For experimental use, the simplest policy is usually: - -- back up anything worth keeping -- remove the DB when the experiment is over -- let Twister rebuild from migrations and backfill paths - -### Failure Recovery Rule - -If the experimental DB becomes suspicious or inconsistent, prefer restore-or-rebuild over manual repair. This is a developer convenience database, not the source of truth. - -## Environment variables - -Copy `.env.example` to `.env` in the repo root (or `packages/api/`). The server loads `.env`, `../.env`, and `../../.env` automatically. - -| Variable | Default | Description | -| -------------------------- | -------------------------------------- | ------------------------------------------------------- | -| `TURSO_DATABASE_URL` | — | Turso/libSQL connection URL (required unless `--local`) | -| `TURSO_AUTH_TOKEN` | — | Auth token (required for non-file URLs) | -| `HTTP_BIND_ADDR` | `:8080` | Address the HTTP server listens on | -| `LOG_LEVEL` | `info` | Log level (`debug`, `info`, `warn`, `error`) | -| `LOG_FORMAT` | `json` | Log format (`json` or `text`) | -| `SEARCH_DEFAULT_LIMIT` | `20` | Default result count for search | -| `SEARCH_MAX_LIMIT` | `100` | Maximum result count for search | -| `ENABLE_ADMIN_ENDPOINTS` | `false` | Expose `/admin/*` endpoints | -| `ADMIN_AUTH_TOKEN` | — | Bearer token required for admin endpoints | -| `CONSTELLATION_URL` | `https://constellation.microcosm.blue` | Constellation API base URL | -| `CONSTELLATION_USER_AGENT` | `twister/1.0 …` | User-Agent sent to Constellation | -| `TAP_URL` | — | Tap firehose URL (indexer only) | -| `TAP_AUTH_PASSWORD` | — | Tap auth password (indexer only) | -| `INDEXED_COLLECTIONS` | — | Comma-separated AT collections to index | - -## CLI commands - -```sh -twister api # Start the HTTP API server -twister indexer # Start the Tap firehose consumer -twister backfill # Seed the index from upstream APIs -twister reindex # Re-process existing documents (re-syncs FTS) -twister enrich # Backfill RepoName, AuthorHandle, WebURL on existing documents -``` - -### enrich - -Resolves missing `author_handle`, `repo_name`, and `web_url` fields on documents already -in the database. Run this after deploying enrichment changes or when search results show -documents with empty author handles. - -```sh -twister enrich --local # all documents -twister enrich --local --collection sh.tangled.repo -twister enrich --local --did did:plc:abc123 -twister enrich --local --dry-run # preview without writing -``` - -Flags: `--collection`, `--did`, `--document`, `--dry-run`, `--concurrency` (default 5). - -## Proxy endpoints - -The API proxies all upstream AT Protocol and social-graph requests so the app has a single origin: - -| Route | Upstream | -| ------------------------------- | ------------------------------------------------------------- | -| `GET /proxy/knot/{host}/{nsid}` | `https://{host}/xrpc/{nsid}` | -| `GET /proxy/pds/{host}/{nsid}` | `https://{host}/xrpc/{nsid}` | -| `GET /proxy/bsky/{nsid}` | `https://public.api.bsky.app/xrpc/{nsid}` | -| `GET /identity/resolve` | `https://bsky.social/xrpc/com.atproto.identity.resolveHandle` | -| `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 | +This package's operational docs live in [`doc.go`](./doc.go). diff --git a/packages/api/doc.go b/packages/api/doc.go new file mode 100644 index 0000000..22a9226 --- /dev/null +++ b/packages/api/doc.go @@ -0,0 +1,149 @@ +// Twister is the Tap-backed indexing and search API for Tangled. +// +// It proxies upstream AT Protocol services such as knots, PDS endpoints, +// Bluesky, Constellation, and Jetstream so the app can use a single origin. +// +// Requirements +// +// - Go 1.25+ +// - A Turso database, or local SQLite for development +// +// Running locally +// +// cd packages/api +// go run . api --local +// +// The local API listens on :8080 by default and uses packages/api/twister-dev.db. +// Logs are printed as text when --local is set. +// +// # API smoke tests +// +// Smoke checks live in packages/scripts/api/. From the repo root: +// +// uv run --project packages/scripts/api twister-api-smoke +// +// Optional base URL override: +// +// TWISTER_API_BASE_URL=http://localhost:8080 \ +// uv run --project packages/scripts/api twister-api-smoke +// +// # Experimental local DB operations +// +// The experimental local database lives at packages/api/twister-dev.db when +// you run Twister with --local. Treat it as disposable unless you explicitly +// back it up. +// +// Backup: +// +// 1. Stop the Twister process using the local DB. +// 2. Copy the database file and any SQLite sidecar files if they exist. +// +// Example: +// +// cd packages/api +// mkdir -p backups +// timestamp="$(date +%Y%m%d-%H%M%S)" +// cp twister-dev.db "backups/twister-dev-${timestamp}.db" +// test -f twister-dev.db-wal && cp twister-dev.db-wal "backups/twister-dev-${timestamp}.db-wal" +// test -f twister-dev.db-shm && cp twister-dev.db-shm "backups/twister-dev-${timestamp}.db-shm" +// +// Restore: +// +// 1. Stop the Twister process. +// 2. Move the current local DB aside if you want to keep it. +// 3. Copy the backup file back to twister-dev.db. +// 4. Restore matching -wal and -shm files only if they came from the same set. +// +// Example: +// +// cd packages/api +// mv twister-dev.db "twister-dev.db.broken.$(date +%Y%m%d-%H%M%S)" 2>/dev/null || true +// cp backups/twister-dev-YYYYMMDD-HHMMSS.db twister-dev.db +// +// Disk growth: +// +// The local DB grows because of indexed documents, FTS tables, activity cache +// rows, and repeated backfill or reindex runs. +// +// Recommended operating procedure: +// +// 1. Check file growth periodically. +// 2. Delete and rebuild the DB freely when the dataset is no longer useful. +// 3. Run VACUUM only when you intentionally want to compact a long-lived DB. +// 4. Keep old backups out of the repo and rotate them manually. +// +// Inspection commands: +// +// cd packages/api +// du -h twister-dev.db* +// ls -lh twister-dev.db* +// +// Failure recovery: prefer restore-or-rebuild over manual repair if the +// experimental DB becomes +// suspicious or inconsistent. It is a developer convenience database, not the +// source of truth. +// +// # Environment variables +// +// Copy .env.example to .env in the repo root or packages/api/. The server loads +// .env, ../.env, and ../../.env automatically. +// +// - TURSO_DATABASE_URL: Turso/libSQL connection URL, required unless --local +// - TURSO_AUTH_TOKEN: auth token, required for non-file URLs +// - HTTP_BIND_ADDR: default :8080 +// - LOG_LEVEL: debug, info, warn, or error; default info +// - LOG_FORMAT: json or text; default json +// - SEARCH_DEFAULT_LIMIT: default 20 +// - SEARCH_MAX_LIMIT: default 100 +// - ENABLE_ADMIN_ENDPOINTS: default false +// - ADMIN_AUTH_TOKEN: bearer token for admin endpoints +// - CONSTELLATION_URL: default https://constellation.microcosm.blue +// - CONSTELLATION_USER_AGENT: user-agent sent to Constellation +// - TAP_URL: Tap firehose URL, indexer only +// - TAP_AUTH_PASSWORD: Tap auth password, indexer only +// - INDEXED_COLLECTIONS: comma-separated AT collections to index +// - READ_THROUGH_MODE: off, missing, or broad; default missing +// - READ_THROUGH_COLLECTIONS: read-through allowlist, default INDEXED_COLLECTIONS +// - READ_THROUGH_MAX_ATTEMPTS: max retries before dead_letter, default 5 +// +// CLI commands +// +// twister api +// twister indexer +// twister backfill +// twister reindex +// twister enrich +// +// Enrich: +// +// Resolves missing author_handle, repo_name, and web_url fields on documents +// already in the database. +// +// twister enrich --local +// twister enrich --local --collection sh.tangled.repo +// twister enrich --local --did did:plc:abc123 +// twister enrich --local --dry-run +// +// Flags: --collection, --did, --document, --dry-run, --concurrency (default 5). +// +// Proxy endpoints +// +// - GET /proxy/knot/{host}/{nsid} -> https://{host}/xrpc/{nsid} +// - GET /proxy/pds/{host}/{nsid} -> https://{host}/xrpc/{nsid} +// - GET /proxy/bsky/{nsid} -> https://public.api.bsky.app/xrpc/{nsid} +// - GET /identity/resolve -> https://bsky.social/xrpc/com.atproto.identity.resolveHandle +// - 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 +// +// Admin routes require ENABLE_ADMIN_ENDPOINTS=true. If ADMIN_AUTH_TOKEN is set, +// requests must send Authorization: Bearer . +// +// - GET /admin/status: cursor state, queue counts, oldest ages, last activity +// - GET /admin/indexing/jobs: inspect queue rows by status, source, or document +// - GET /admin/indexing/audit: inspect append-only indexing audit rows +// - POST /admin/indexing/enqueue: queue one explicit record for indexing +// - POST /admin/reindex: re-sync all or filtered documents into the FTS index +package main diff --git a/packages/api/internal/api/actors.go b/packages/api/internal/api/actors.go index 6c335dd..fabf99a 100644 --- a/packages/api/internal/api/actors.go +++ b/packages/api/internal/api/actors.go @@ -427,7 +427,7 @@ func (s *Server) handleRepoIssues(w http.ResponseWriter, r *http.Request) { return } - issues, stateMap, err := s.fetchIssuesAndStates(r, repo.PDS, repo.DID) + issues, stateMap, _, err := s.fetchIssuesAndStates(r, repo.PDS, repo.DID) if err != nil { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch issues")) return @@ -465,7 +465,7 @@ func (s *Server) handleRepoPulls(w http.ResponseWriter, r *http.Request) { return } - pulls, statusMap, err := s.fetchPullsAndStatuses(r, repo.PDS, repo.DID) + pulls, statusMap, _, err := s.fetchPullsAndStatuses(r, repo.PDS, repo.DID) if err != nil { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch pulls")) return @@ -504,7 +504,7 @@ func (s *Server) handleActorIssues(w http.ResponseWriter, r *http.Request) { return } - issues, stateMap, err := s.fetchIssuesAndStates(r, actor.PDS, actor.DID) + issues, stateMap, _, err := s.fetchIssuesAndStates(r, actor.PDS, actor.DID) if err != nil { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch issues")) return @@ -535,7 +535,7 @@ func (s *Server) handleActorPulls(w http.ResponseWriter, r *http.Request) { return } - pulls, statusMap, err := s.fetchPullsAndStatuses(r, actor.PDS, actor.DID) + pulls, statusMap, _, err := s.fetchPullsAndStatuses(r, actor.PDS, actor.DID) if err != nil { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch pulls")) return @@ -632,11 +632,17 @@ func (s *Server) handleIssueDetail(w http.ResponseWriter, r *http.Request) { } s.enqueueXRPCRecord(r.Context(), rec.URI, rec.CID, rec.Value) - _, stateMap, err := s.fetchIssuesAndStates(r, actor.PDS, actor.DID) + _, stateMap, states, err := s.fetchIssuesAndStates(r, actor.PDS, actor.DID) if err != nil { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch issue states")) return } + for _, entry := range states { + if issue, _ := entry.Value["issue"].(string); issue == rec.URI { + s.syncStateEntry(r.Context(), entry) + break + } + } writeJSON(w, http.StatusOK, issueEntry{ recordEntry: recordEntry{URI: rec.URI, CID: rec.CID, Value: rec.Value}, @@ -703,11 +709,17 @@ func (s *Server) handlePullDetail(w http.ResponseWriter, r *http.Request) { } s.enqueueXRPCRecord(r.Context(), rec.URI, rec.CID, rec.Value) - _, statusMap, err := s.fetchPullsAndStatuses(r, actor.PDS, actor.DID) + _, statusMap, statuses, err := s.fetchPullsAndStatuses(r, actor.PDS, actor.DID) if err != nil { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch pull statuses")) return } + for _, entry := range statuses { + if pull, _ := entry.Value["pull"].(string); pull == rec.URI { + s.syncStateEntry(r.Context(), entry) + break + } + } writeJSON(w, http.StatusOK, pullEntry{ recordEntry: recordEntry{URI: rec.URI, CID: rec.CID, Value: rec.Value}, @@ -755,7 +767,9 @@ func (s *Server) handlePullComments(w http.ResponseWriter, r *http.Request) { }) } -func (s *Server) fetchIssuesAndStates(r *http.Request, pds, did string) ([]xrpc.ListRecordEntry, map[string]string, error) { +func (s *Server) fetchIssuesAndStates( + r *http.Request, pds, did string, +) ([]xrpc.ListRecordEntry, map[string]string, []xrpc.ListRecordEntry, error) { issueCh := make(chan []xrpc.ListRecordEntry, 1) stateCh := make(chan []xrpc.ListRecordEntry, 1) errCh := make(chan error, 2) @@ -786,12 +800,11 @@ func (s *Server) fetchIssuesAndStates(r *http.Request, pds, did string) ([]xrpc. case e := <-stateCh: states = e case err := <-errCh: - return nil, nil, err + return nil, nil, nil, err } } stateMap := make(map[string]string, len(states)) - s.enqueueXRPCList(r.Context(), states) for _, e := range states { issueURI, _ := e.Value["issue"].(string) state, _ := e.Value["state"].(string) @@ -800,10 +813,12 @@ func (s *Server) fetchIssuesAndStates(r *http.Request, pds, did string) ([]xrpc. } } - return issues, stateMap, nil + return issues, stateMap, states, nil } -func (s *Server) fetchPullsAndStatuses(r *http.Request, pds, did string) ([]xrpc.ListRecordEntry, map[string]string, error) { +func (s *Server) fetchPullsAndStatuses( + r *http.Request, pds, did string, +) ([]xrpc.ListRecordEntry, map[string]string, []xrpc.ListRecordEntry, error) { pullCh := make(chan []xrpc.ListRecordEntry, 1) statusCh := make(chan []xrpc.ListRecordEntry, 1) errCh := make(chan error, 2) @@ -834,12 +849,11 @@ func (s *Server) fetchPullsAndStatuses(r *http.Request, pds, did string) ([]xrpc case e := <-statusCh: statuses = e case err := <-errCh: - return nil, nil, err + return nil, nil, nil, err } } statusMap := make(map[string]string, len(statuses)) - s.enqueueXRPCList(r.Context(), statuses) for _, e := range statuses { pullURI, _ := e.Value["pull"].(string) status, _ := e.Value["status"].(string) @@ -848,7 +862,7 @@ func (s *Server) fetchPullsAndStatuses(r *http.Request, pds, did string) ([]xrpc } } - return pulls, statusMap, nil + return pulls, statusMap, statuses, nil } type bskyProfileResponse struct { diff --git a/packages/api/internal/api/admin_indexing.go b/packages/api/internal/api/admin_indexing.go new file mode 100644 index 0000000..63f403c --- /dev/null +++ b/packages/api/internal/api/admin_indexing.go @@ -0,0 +1,97 @@ +package api + +import ( + "encoding/json" + "net/http" + + "tangled.org/desertthunder.dev/twister/internal/store" +) + +func (s *Server) handleAdminIndexingJobs(w http.ResponseWriter, r *http.Request) { + if !s.authorizeAdmin(w, r) { + return + } + limit, err := intParam(r, "limit", 50) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "limit must be numeric")) + return + } + offset, err := intParam(r, "offset", 0) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "offset must be numeric")) + return + } + jobs, err := s.store.ListIndexingJobs(r.Context(), store.IndexingJobFilter{ + Status: r.URL.Query().Get("status"), + Source: r.URL.Query().Get("source"), + DocumentID: r.URL.Query().Get("document"), + Limit: limit, + Offset: offset, + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to list jobs")) + return + } + writeJSON(w, http.StatusOK, map[string]any{"jobs": jobs, "limit": limit, "offset": offset}) +} + +func (s *Server) handleAdminIndexingAudit(w http.ResponseWriter, r *http.Request) { + if !s.authorizeAdmin(w, r) { + return + } + limit, err := intParam(r, "limit", 50) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "limit must be numeric")) + return + } + offset, err := intParam(r, "offset", 0) + if err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "offset must be numeric")) + return + } + entries, err := s.store.ListIndexingAudit(r.Context(), store.IndexingAuditFilter{ + Source: r.URL.Query().Get("source"), + Decision: r.URL.Query().Get("decision"), + DocumentID: r.URL.Query().Get("document"), + Limit: limit, + Offset: offset, + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to list audit")) + return + } + writeJSON(w, http.StatusOK, map[string]any{"audit": entries, "limit": limit, "offset": offset}) +} + +func (s *Server) handleAdminIndexingEnqueue(w http.ResponseWriter, r *http.Request) { + if !s.authorizeAdmin(w, r) { + return + } + var body struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value map[string]any `json:"value"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_body", "expected JSON body")) + return + } + if body.URI == "" || body.CID == "" || body.Value == nil { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_body", "uri, cid, and value are required")) + return + } + s.enqueueRecordForIndexing(r.Context(), store.IndexSourceAdmin, body.URI, body.CID, body.Value) + writeJSON(w, http.StatusAccepted, map[string]string{"status": "accepted"}) +} + +func (s *Server) authorizeAdmin(w http.ResponseWriter, r *http.Request) bool { + if s.cfg.AdminAuthToken == "" { + return true + } + token := r.Header.Get("Authorization") + if token == "Bearer "+s.cfg.AdminAuthToken { + return true + } + writeJSON(w, http.StatusUnauthorized, errorBody("unauthorized", "invalid admin token")) + return false +} diff --git a/packages/api/internal/api/api.go b/packages/api/internal/api/api.go index c911417..07ef0af 100644 --- a/packages/api/internal/api/api.go +++ b/packages/api/internal/api/api.go @@ -14,6 +14,7 @@ import ( "tangled.org/desertthunder.dev/twister/internal/config" "tangled.org/desertthunder.dev/twister/internal/constellation" + idx "tangled.org/desertthunder.dev/twister/internal/index" "tangled.org/desertthunder.dev/twister/internal/normalize" "tangled.org/desertthunder.dev/twister/internal/reindex" "tangled.org/desertthunder.dev/twister/internal/search" @@ -30,10 +31,15 @@ type Server struct { constellation *constellation.Client xrpc *xrpc.Client registry *normalize.Registry + policy idx.Policy + processor *idx.Processor + workerID string } // New creates a new API server. func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log *slog.Logger, constellation *constellation.Client, xrpcClient *xrpc.Client) *Server { + registry := normalize.NewRegistry() + policy := idx.NewPolicy(cfg.IndexedCollections, cfg.ReadThroughCollections, cfg.ReadThroughMode) return &Server{ search: searchRepo, store: st, @@ -41,7 +47,10 @@ func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log log: log, constellation: constellation, xrpc: xrpcClient, - registry: normalize.NewRegistry(), + registry: registry, + policy: policy, + processor: idx.NewProcessor(st, registry, xrpcClient, policy, log), + workerID: newWorkerID("api"), } } @@ -349,7 +358,7 @@ func (s *Server) handleProfileSummary(w http.ResponseWriter, r *http.Request) { const tapConsumerName = "indexer-tap-v1" -// handleAdminStatus returns Tap cursor, JetStream cursor, document count, and pending job count. +// handleAdminStatus returns cursor and queue status for indexing subsystems. // Route: GET /admin/status func (s *Server) handleAdminStatus(w http.ResponseWriter, r *http.Request) { if s.cfg.AdminAuthToken != "" { @@ -381,10 +390,10 @@ func (s *Server) handleAdminStatus(w http.ResponseWriter, r *http.Request) { return } - pending, err := s.store.CountPendingIndexingJobs(r.Context()) + stats, err := s.store.GetIndexingJobStats(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")) + s.log.Error("admin status: queue stats failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusInternalServerError, errorBody("db_error", "failed to fetch queue stats")) return } @@ -413,10 +422,20 @@ func (s *Server) handleAdminStatus(w http.ResponseWriter, r *http.Request) { } writeJSON(w, http.StatusOK, map[string]any{ - "tap": tapJSON, - "jetstream": jsJSON, - "documents": docs, - "pending_jobs": pending, + "tap": tapJSON, + "jetstream": jsJSON, + "documents": docs, + "read_through": map[string]any{ + "pending": stats.Pending, + "processing": stats.Processing, + "completed": stats.Completed, + "failed": stats.Failed, + "dead_letter": stats.DeadLetter, + "oldest_pending_age_s": ageSeconds(stats.OldestPendingAt), + "oldest_running_age_s": ageSeconds(stats.OldestRunningAt), + "last_completed_at": stats.LastCompletedAt, + "last_processed_at": stats.LastProcessedAt, + }, }) } diff --git a/packages/api/internal/api/helpers.go b/packages/api/internal/api/helpers.go index 2c5a4a4..e60d8f0 100644 --- a/packages/api/internal/api/helpers.go +++ b/packages/api/internal/api/helpers.go @@ -149,3 +149,18 @@ func truncateErr(err error) string { } return msg } + +func newWorkerID(prefix string) string { + return fmt.Sprintf("%s-%d", prefix, time.Now().UnixNano()) +} + +func ageSeconds(ts string) int64 { + if ts == "" { + return 0 + } + parsed, err := time.Parse(time.RFC3339, ts) + if err != nil { + return 0 + } + return int64(time.Since(parsed).Seconds()) +} diff --git a/packages/api/internal/api/readthrough.go b/packages/api/internal/api/readthrough.go index 4b40dfb..3f3c75a 100644 --- a/packages/api/internal/api/readthrough.go +++ b/packages/api/internal/api/readthrough.go @@ -8,6 +8,7 @@ import ( "sync" "time" + idx "tangled.org/desertthunder.dev/twister/internal/index" "tangled.org/desertthunder.dev/twister/internal/normalize" "tangled.org/desertthunder.dev/twister/internal/store" "tangled.org/desertthunder.dev/twister/internal/xrpc" @@ -15,11 +16,16 @@ import ( const ( readThroughIdlePoll = 1 * time.Second + readThroughLeaseDuration = 30 * time.Second readThroughStatusInterval = 30 * time.Second - maxIndexingAttempts = 10 ) func (s *Server) runReadThroughIndexer(ctx context.Context) { + if s.policy.ReadThroughMode() == idx.ReadThroughOff { + s.log.Info("read-through indexer worker disabled") + return + } + ticker := time.NewTicker(readThroughIdlePoll) defer ticker.Stop() @@ -34,14 +40,11 @@ func (s *Server) runReadThroughIndexer(ctx context.Context) { return } - job, err := s.store.ClaimIndexingJob(ctx) + leaseUntil := time.Now().UTC().Add(readThroughLeaseDuration).Format(time.RFC3339) + job, err := s.store.ClaimIndexingJob(ctx, s.workerID, leaseUntil) if err != nil { s.log.Warn("read-through claim failed", slog.String("error", err.Error())) - select { - case <-ctx.Done(): - return - case <-ticker.C: - } + <-ticker.C continue } if job == nil { @@ -53,35 +56,11 @@ func (s *Server) runReadThroughIndexer(ctx context.Context) { continue } - if err := s.processReadThroughJob(ctx, job); err != nil { - if job.Attempts+1 >= maxIndexingAttempts { - s.log.Error("read-through job exceeded max attempts; discarding", - slog.String("document_id", job.DocumentID), - slog.Int("attempts", job.Attempts+1), - slog.String("last_error", err.Error()), - ) - _ = s.store.CompleteIndexingJob(ctx, job.DocumentID) - continue - } - nextDelay := retryDelay(job.Attempts + 1) - nextAt := time.Now().UTC().Add(nextDelay).Format(time.RFC3339) - retryErr := s.store.RetryIndexingJob(ctx, job.DocumentID, nextAt, truncateErr(err)) - if retryErr != nil { - s.log.Error("read-through retry update failed", - slog.String("document_id", job.DocumentID), - slog.String("error", retryErr.Error()), - ) - continue - } - s.log.Warn("read-through job failed; scheduled retry", - slog.String("document_id", job.DocumentID), - slog.Int("attempt", job.Attempts+1), - slog.Duration("retry_in", nextDelay), - slog.String("error", err.Error()), - ) + result, err := s.processReadThroughJob(ctx, job) + if err != nil { + s.handleReadThroughFailure(ctx, job, err) continue } - if err := s.store.CompleteIndexingJob(ctx, job.DocumentID); err != nil { s.log.Error("read-through complete failed", slog.String("document_id", job.DocumentID), @@ -90,7 +69,14 @@ func (s *Server) runReadThroughIndexer(ctx context.Context) { continue } - s.log.Debug("read-through job completed", slog.String("document_id", job.DocumentID)) + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: job.Source, + DocumentID: job.DocumentID, + Collection: job.Collection, + CID: job.CID, + Decision: result.Decision, + Attempt: job.Attempts, + }) mu.Lock() processedTick++ mu.Unlock() @@ -110,23 +96,41 @@ func (s *Server) runIndexerStatusLogger(ctx context.Context, mu *sync.Mutex, pro *processedTick = 0 mu.Unlock() - pending, err := s.store.CountPendingIndexingJobs(ctx) + stats, err := s.store.GetIndexingJobStats(ctx) if err != nil { - s.log.Warn("read-through status: count failed", slog.String("error", err.Error())) + s.log.Warn("read-through status: stats failed", slog.String("error", err.Error())) continue } s.log.Info("read-through indexer status", slog.Int64("jobs_processed", n), - slog.Int64("jobs_pending", pending), + slog.Int64("pending", stats.Pending), + slog.Int64("processing", stats.Processing), + slog.Int64("failed", stats.Failed), + slog.Int64("dead_letter", stats.DeadLetter), ) } } } -func (s *Server) processReadThroughJob(ctx context.Context, job *store.IndexingJob) error { +func (s *Server) processReadThroughJob( + ctx context.Context, job *store.IndexingJob, +) (*idx.Result, error) { + doc, err := s.store.GetDocument(ctx, job.DocumentID) + if err != nil { + return nil, fmt.Errorf("get document: %w", err) + } + if doc != nil && doc.CID == job.CID && doc.DeletedAt == "" { + return &idx.Result{ + Decision: "skip_already_indexed", + DocumentID: job.DocumentID, + Collection: job.Collection, + CID: job.CID, + }, nil + } + record := map[string]any{} if err := json.Unmarshal([]byte(job.RecordJSON), &record); err != nil { - return fmt.Errorf("decode record json: %w", err) + return nil, &idx.PermanentError{Decision: "decode_record_json", Err: err} } event := normalize.TapRecordEvent{ @@ -140,146 +144,189 @@ func (s *Server) processReadThroughJob(ctx context.Context, job *store.IndexingJ Record: record, }, } + return s.processor.ProcessRecord(ctx, job.Source, event) +} - if handler, ok := s.registry.StateHandler(job.Collection); ok { - update, err := handler.HandleState(event) - if err != nil { - return fmt.Errorf("state normalize: %w", err) - } - if err := s.store.UpdateRecordState(ctx, update.SubjectURI, update.State); err != nil { - return fmt.Errorf("update state: %w", err) - } - return nil - } - - adapter, ok := s.registry.Adapter(job.Collection) - if !ok { - return nil +func (s *Server) handleReadThroughFailure( + ctx context.Context, job *store.IndexingJob, err error, +) { + if perr, ok := idx.IsPermanent(err); ok { + _ = s.store.FailIndexingJob(ctx, job.DocumentID, store.IndexingJobDeadLetter, truncateErr(perr)) + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: job.Source, + DocumentID: job.DocumentID, + Collection: job.Collection, + CID: job.CID, + Decision: perr.Decision, + Attempt: job.Attempts + 1, + Error: perr.Error(), + }) + return } - doc, err := adapter.Normalize(event) - if err != nil { - return fmt.Errorf("normalize record: %w", err) + if job.Attempts+1 >= s.cfg.ReadThroughMaxAttempts { + _ = s.store.FailIndexingJob(ctx, job.DocumentID, store.IndexingJobDeadLetter, truncateErr(err)) + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: job.Source, + DocumentID: job.DocumentID, + Collection: job.Collection, + CID: job.CID, + Decision: "dead_letter", + Attempt: job.Attempts + 1, + Error: err.Error(), + }) + return } - handle, err := s.store.GetIdentityHandle(ctx, job.DID) - if err != nil { - return fmt.Errorf("lookup identity handle: %w", err) - } - if handle != "" { - doc.AuthorHandle = handle - if doc.RecordType == "profile" { - doc.Title = handle - } + nextDelay := retryDelay(job.Attempts + 1) + nextAt := time.Now().UTC().Add(nextDelay).Format(time.RFC3339) + if retryErr := s.store.RetryIndexingJob(ctx, job.DocumentID, nextAt, truncateErr(err)); retryErr != nil { + s.log.Error("read-through retry update failed", + slog.String("document_id", job.DocumentID), + slog.String("error", retryErr.Error()), + ) + return } + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: job.Source, + DocumentID: job.DocumentID, + Collection: job.Collection, + CID: job.CID, + Decision: "retry_scheduled", + Attempt: job.Attempts + 1, + Error: err.Error(), + }) +} - s.enrichDocument(ctx, doc, record) - - if err := s.store.UpsertDocument(ctx, doc); err != nil { - return fmt.Errorf("upsert document: %w", err) - } +func (s *Server) enqueueXRPCRecord(ctx context.Context, uri, cid string, value map[string]any) { + s.enqueueRecordForIndexing(ctx, store.IndexSourceReadThrough, uri, cid, value) +} - return nil +func (s *Server) enqueueXRPCList(context.Context, []xrpc.ListRecordEntry) { } -// enrichDocument fills RepoName, AuthorHandle, and WebURL via XRPC when possible. -// Failures are logged but never block indexing. -func (s *Server) enrichDocument(ctx context.Context, doc *store.Document, record map[string]any) { - if s.xrpc == nil { +func (s *Server) enqueueRecordForIndexing( + ctx context.Context, source, uri, cid string, value map[string]any, +) { + did, collection, rkey, err := normalize.ParseATURI(uri) + if err != nil { + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: source, + DocumentID: uri, + Collection: collection, + CID: cid, + Decision: "skip_invalid_uri", + Error: err.Error(), + }) return } - if doc.RepoDID != "" && doc.RepoName == "" { - repoURI := repoURIFromRecord(record) - if repoURI != "" { - _, _, repoRKey, err := normalize.ParseATURI(repoURI) - if err == nil && repoRKey != "" { - name, err := s.xrpc.ResolveRepoName(ctx, doc.RepoDID, repoRKey) - if err == nil { - doc.RepoName = name - } else { - s.log.Debug("read-through enrich: resolve repo name failed", - slog.String("doc_id", doc.ID), - slog.String("repo_did", doc.RepoDID), - slog.String("error", err.Error()), - ) - } - } - } - } - - if doc.AuthorHandle == "" && doc.DID != "" { - info, err := s.xrpc.ResolveIdentity(ctx, doc.DID) - if err == nil && info.Handle != "" { - doc.AuthorHandle = info.Handle - if doc.RecordType == "profile" { - doc.Title = info.Handle - } - } else if err != nil { - s.log.Debug("read-through enrich: resolve author handle failed", - slog.String("doc_id", doc.ID), - slog.String("did", doc.DID), - slog.String("error", err.Error()), - ) - } - } - - if doc.WebURL == "" { - ownerHandle := doc.AuthorHandle - if doc.RepoDID != "" && doc.RepoDID != doc.DID { - if h, err := s.store.GetIdentityHandle(ctx, doc.RepoDID); err == nil && h != "" { - ownerHandle = h - } else if info, err := s.xrpc.ResolveIdentity(ctx, doc.RepoDID); err == nil && info.Handle != "" { - ownerHandle = info.Handle - } - } - doc.WebURL = xrpc.BuildWebURL(ownerHandle, doc.RepoName, doc.RecordType, doc.RKey) + documentID := normalize.StableID(did, collection, rkey) + if source == store.IndexSourceReadThrough && s.policy.ReadThroughMode() == idx.ReadThroughOff { + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: source, DocumentID: documentID, Collection: collection, CID: cid, + Decision: "skip_mode_off", + }) + return } -} - -// repoURIFromRecord extracts the repo AT-URI from common record fields. -// Issues store it in rec["repo"]; pulls store it in rec["target"]["repo"]. -func repoURIFromRecord(record map[string]any) string { - if uri, _ := record["repo"].(string); uri != "" { - return uri + if !s.policy.Allows(source, collection) { + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: source, DocumentID: documentID, Collection: collection, CID: cid, + Decision: "skip_collection", + }) + return } - if target, _ := record["target"].(map[string]any); target != nil { - if uri, _ := target["repo"].(string); uri != "" { - return uri + if source == store.IndexSourceReadThrough && s.policy.ReadThroughMode() == idx.ReadThroughMissing { + if s.shouldSkipReadThrough(ctx, documentID, cid) { + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: source, DocumentID: documentID, Collection: collection, CID: cid, + Decision: "skip_already_indexed", + }) + return } } - return "" -} -func (s *Server) enqueueXRPCRecord(ctx context.Context, uri, cid string, value map[string]any) { - did, collection, rkey, err := normalize.ParseATURI(uri) - if err != nil { - s.log.Debug("read-through skip invalid at-uri", slog.String("uri", uri), slog.String("error", err.Error())) - return - } payload, err := json.Marshal(value) if err != nil { - s.log.Debug("read-through skip unmarshalable record", slog.String("uri", uri), slog.String("error", err.Error())) + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: source, DocumentID: documentID, Collection: collection, CID: cid, + Decision: "skip_unmarshalable_record", Error: err.Error(), + }) return } input := store.IndexingJobInput{ - DocumentID: normalize.StableID(did, collection, rkey), + DocumentID: documentID, DID: did, Collection: collection, RKey: rkey, CID: cid, RecordJSON: string(payload), + Source: source, } if err := s.store.EnqueueIndexingJob(ctx, input); err != nil { - s.log.Warn("enqueue read-through indexing job failed", - slog.String("document_id", input.DocumentID), + s.log.Warn("enqueue indexing job failed", + slog.String("document_id", documentID), slog.String("error", err.Error()), ) + return + } + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: source, DocumentID: documentID, Collection: collection, CID: cid, + Decision: "enqueued", + }) +} + +func (s *Server) shouldSkipReadThrough(ctx context.Context, documentID, cid string) bool { + doc, err := s.store.GetDocument(ctx, documentID) + if err == nil && doc != nil && doc.CID == cid && doc.DeletedAt == "" { + return true + } + job, err := s.store.GetIndexingJob(ctx, documentID) + if err != nil || job == nil { + return false + } + return job.CID == cid +} + +func (s *Server) syncStateEntry(ctx context.Context, entry xrpc.ListRecordEntry) { + did, collection, rkey, err := normalize.ParseATURI(entry.URI) + if err != nil { + return + } + event := normalize.TapRecordEvent{ + Type: "record", + Record: &normalize.TapRecord{ + DID: did, + Collection: collection, + RKey: rkey, + Action: "create", + CID: entry.CID, + Record: entry.Value, + }, + } + result, err := s.processor.ProcessRecord(ctx, store.IndexSourceReadThrough, event) + if err != nil { + s.handleReadThroughFailure(ctx, &store.IndexingJob{ + DocumentID: normalize.StableID(did, collection, rkey), + Collection: collection, + CID: entry.CID, + Source: store.IndexSourceReadThrough, + }, err) + return + } + if result != nil { + s.appendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: store.IndexSourceReadThrough, + DocumentID: result.DocumentID, + Collection: result.Collection, + CID: result.CID, + Decision: result.Decision, + }) } } -func (s *Server) enqueueXRPCList(ctx context.Context, entries []xrpc.ListRecordEntry) { - for _, e := range entries { - s.enqueueXRPCRecord(ctx, e.URI, e.CID, e.Value) +func (s *Server) appendIndexingAudit(ctx context.Context, input store.IndexingAuditInput) { + if err := s.store.AppendIndexingAudit(ctx, input); err != nil { + s.log.Debug("append indexing audit failed", slog.String("error", err.Error())) } } diff --git a/packages/api/internal/api/readthrough_test.go b/packages/api/internal/api/readthrough_test.go new file mode 100644 index 0000000..0895db7 --- /dev/null +++ b/packages/api/internal/api/readthrough_test.go @@ -0,0 +1,156 @@ +package api + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "tangled.org/desertthunder.dev/twister/internal/config" + idx "tangled.org/desertthunder.dev/twister/internal/index" + "tangled.org/desertthunder.dev/twister/internal/normalize" + "tangled.org/desertthunder.dev/twister/internal/store" + "tangled.org/desertthunder.dev/twister/internal/xrpc" +) + +func newAPITestServer(st *apiTestStore, client *xrpc.Client) *Server { + cfg := &config.Config{ + IndexedCollections: "sh.tangled.*", + ReadThroughCollections: "sh.tangled.*", + ReadThroughMode: "missing", + ReadThroughMaxAttempts: 5, + } + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + policy := idx.NewPolicy(cfg.IndexedCollections, cfg.ReadThroughCollections, cfg.ReadThroughMode) + registry := normalize.NewRegistry() + return &Server{ + store: st, cfg: cfg, log: log, xrpc: client, registry: registry, + policy: policy, processor: idx.NewProcessor(st, registry, client, policy, log), workerID: "api-test", + } +} + +func TestEnqueueXRPCRecordSkipsExistingDocument(t *testing.T) { + st := newAPITestStore() + st.docs["did:plc:alice|sh.tangled.repo|repo1"] = &store.Document{ + ID: "did:plc:alice|sh.tangled.repo|repo1", CID: "cid-1", + } + srv := newAPITestServer(st, nil) + + srv.enqueueXRPCRecord(context.Background(), "at://did:plc:alice/sh.tangled.repo/repo1", "cid-1", map[string]any{"name": "repo1"}) + + if len(st.jobs) != 0 { + t.Fatalf("expected no jobs, got %#v", st.jobs) + } + if len(st.audits) == 0 || st.audits[0].Decision != "skip_already_indexed" { + t.Fatalf("unexpected audit rows: %#v", st.audits) + } +} + +func TestEnqueueXRPCRecordOnlyQueuesOncePerCID(t *testing.T) { + st := newAPITestStore() + srv := newAPITestServer(st, nil) + uri := "at://did:plc:alice/sh.tangled.repo/repo1" + + srv.enqueueXRPCRecord(context.Background(), uri, "cid-1", map[string]any{"name": "repo1"}) + srv.enqueueXRPCRecord(context.Background(), uri, "cid-1", map[string]any{"name": "repo1"}) + + if len(st.jobs) != 1 { + t.Fatalf("expected one queued job, got %#v", st.jobs) + } + if st.jobs["did:plc:alice|sh.tangled.repo|repo1"].Status != store.IndexingJobPending { + t.Fatalf("unexpected job state: %#v", st.jobs["did:plc:alice|sh.tangled.repo|repo1"]) + } +} + +func TestHandleActorFollowingDoesNotEnqueueBulkList(t *testing.T) { + var upstream *httptest.Server + upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/xrpc/com.atproto.identity.resolveHandle": + _ = json.NewEncoder(w).Encode(map[string]string{"did": "did:plc:alice"}) + case r.URL.Path == "/did%3Aplc%3Aalice" || r.URL.Path == "/did:plc:alice": + _ = json.NewEncoder(w).Encode(map[string]any{ + "id": "did:plc:alice", + "alsoKnownAs": []string{"at://alice.tangled.org"}, + "service": []map[string]string{{ + "type": "AtprotoPersonalDataServer", "serviceEndpoint": upstream.URL, + }}, + }) + case r.URL.Path == "/xrpc/com.atproto.repo.listRecords": + _ = json.NewEncoder(w).Encode(map[string]any{ + "records": []map[string]any{{ + "uri": "at://did:plc:alice/sh.tangled.graph.follow/1", + "cid": "cid-1", + "value": map[string]any{"subject": "did:plc:bob"}, + }}, + }) + default: + http.NotFound(w, r) + } + })) + defer upstream.Close() + + client := xrpc.NewClient( + xrpc.WithHTTPClient(upstream.Client()), + xrpc.WithIdentityService(upstream.URL), + xrpc.WithPLCDirectory(upstream.URL), + ) + st := newAPITestStore() + srv := newAPITestServer(st, client) + mux := http.NewServeMux() + mux.HandleFunc("GET /actors/{handle}/following", srv.handleActorFollowing) + + req := httptest.NewRequest(http.MethodGet, "/actors/alice.tangled.org/following", nil) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) + } + if len(st.jobs) != 0 { + t.Fatalf("expected no queued jobs from list handler, got %#v", st.jobs) + } +} + +func TestDeadLetterJobVisibleThroughAdminEndpoints(t *testing.T) { + st := newAPITestStore() + job := &store.IndexingJob{ + DocumentID: "did:plc:alice|sh.tangled.repo|repo1", + Collection: "sh.tangled.repo", + CID: "cid-1", + Source: store.IndexSourceReadThrough, + Status: store.IndexingJobProcessing, + } + st.jobs[job.DocumentID] = job + srv := newAPITestServer(st, nil) + + srv.handleReadThroughFailure(context.Background(), job, &idx.PermanentError{ + Decision: "normalize_failed", + Err: errors.New("boom"), + }) + + req := httptest.NewRequest(http.MethodGet, "/admin/indexing/jobs", nil) + rec := httptest.NewRecorder() + srv.handleAdminIndexingJobs(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("jobs status: got %d body=%s", rec.Code, rec.Body.String()) + } + if st.jobs[job.DocumentID].Status != store.IndexingJobDeadLetter { + t.Fatalf("expected dead letter status, got %#v", st.jobs[job.DocumentID]) + } + + req = httptest.NewRequest(http.MethodGet, "/admin/indexing/audit?document="+job.DocumentID, nil) + rec = httptest.NewRecorder() + srv.handleAdminIndexingAudit(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("audit status: got %d body=%s", rec.Code, rec.Body.String()) + } + if !bytes.Contains(rec.Body.Bytes(), []byte("normalize_failed")) { + t.Fatalf("expected normalize_failed in audit body: %s", rec.Body.String()) + } +} diff --git a/packages/api/internal/api/router.go b/packages/api/internal/api/router.go index 701fca9..548cec2 100644 --- a/packages/api/internal/api/router.go +++ b/packages/api/internal/api/router.go @@ -98,6 +98,9 @@ func (s *Server) registerAdminRoutes(mux *http.ServeMux) { return } mux.HandleFunc("GET /admin/status", s.handleAdminStatus) + mux.HandleFunc("GET /admin/indexing/jobs", s.handleAdminIndexingJobs) + mux.HandleFunc("GET /admin/indexing/audit", s.handleAdminIndexingAudit) + mux.HandleFunc("POST /admin/indexing/enqueue", s.handleAdminIndexingEnqueue) mux.HandleFunc("POST /admin/reindex", s.handleAdminReindex) } diff --git a/packages/api/internal/api/test_store_test.go b/packages/api/internal/api/test_store_test.go new file mode 100644 index 0000000..25067b5 --- /dev/null +++ b/packages/api/internal/api/test_store_test.go @@ -0,0 +1,165 @@ +package api + +import ( + "context" + "sort" + "time" + + "tangled.org/desertthunder.dev/twister/internal/store" +) + +type apiTestStore struct { + docs map[string]*store.Document + jobs map[string]*store.IndexingJob + audits []*store.IndexingAuditEntry + handles map[string]string +} + +func newAPITestStore() *apiTestStore { + return &apiTestStore{ + docs: map[string]*store.Document{}, + jobs: map[string]*store.IndexingJob{}, + handles: map[string]string{}, + } +} + +func (s *apiTestStore) UpsertDocument(_ context.Context, doc *store.Document) error { + clone := *doc + s.docs[doc.ID] = &clone + return nil +} +func (s *apiTestStore) GetDocument(_ context.Context, id string) (*store.Document, error) { + return s.docs[id], nil +} +func (s *apiTestStore) MarkDeleted(_ context.Context, id string) error { + if doc := s.docs[id]; doc != nil { + doc.DeletedAt = time.Now().UTC().Format(time.RFC3339) + } + return nil +} +func (s *apiTestStore) ListDocuments(_ context.Context, _ store.DocumentFilter) ([]*store.Document, error) { + return nil, nil +} +func (s *apiTestStore) OptimizeFTS(_ context.Context) error { return nil } +func (s *apiTestStore) GetSyncState(_ context.Context, _ string) (*store.SyncState, error) { + return nil, nil +} +func (s *apiTestStore) SetSyncState(_ context.Context, _, _ string) error { return nil } +func (s *apiTestStore) UpdateRecordState(_ context.Context, _, _ string) error { return nil } +func (s *apiTestStore) UpsertIdentityHandle(_ context.Context, did, handle string, _ bool, _ string) error { + s.handles[did] = handle + return nil +} +func (s *apiTestStore) GetIdentityHandle(_ context.Context, did string) (string, error) { + return s.handles[did], nil +} +func (s *apiTestStore) GetIndexingJob(_ context.Context, documentID string) (*store.IndexingJob, error) { + return s.jobs[documentID], nil +} +func (s *apiTestStore) EnqueueIndexingJob(_ context.Context, input store.IndexingJobInput) error { + s.jobs[input.DocumentID] = &store.IndexingJob{ + DocumentID: input.DocumentID, DID: input.DID, Collection: input.Collection, + RKey: input.RKey, CID: input.CID, RecordJSON: input.RecordJSON, Source: input.Source, + Status: store.IndexingJobPending, ScheduledAt: time.Now().UTC().Format(time.RFC3339), + UpdatedAt: time.Now().UTC().Format(time.RFC3339), + } + return nil +} +func (s *apiTestStore) ClaimIndexingJob(context.Context, string, string) (*store.IndexingJob, error) { + return nil, nil +} +func (s *apiTestStore) CompleteIndexingJob(_ context.Context, documentID string) error { + if j := s.jobs[documentID]; j != nil { + j.Status = store.IndexingJobCompleted + } + return nil +} +func (s *apiTestStore) RetryIndexingJob(_ context.Context, documentID, nextAt, lastError string) error { + if j := s.jobs[documentID]; j != nil { + j.Status, j.ScheduledAt, j.LastError = store.IndexingJobPending, nextAt, lastError + j.Attempts++ + } + return nil +} +func (s *apiTestStore) FailIndexingJob(_ context.Context, documentID, status, lastError string) error { + if j := s.jobs[documentID]; j != nil { + j.Status, j.LastError = status, lastError + j.Attempts++ + } + return nil +} +func (s *apiTestStore) ListIndexingJobs(_ context.Context, filter store.IndexingJobFilter) ([]*store.IndexingJob, error) { + out := []*store.IndexingJob{} + for _, job := range s.jobs { + if filter.DocumentID != "" && job.DocumentID != filter.DocumentID { + continue + } + if filter.Status != "" && job.Status != filter.Status { + continue + } + if filter.Source != "" && job.Source != filter.Source { + continue + } + clone := *job + out = append(out, &clone) + } + sort.Slice(out, func(i, j int) bool { return out[i].DocumentID < out[j].DocumentID }) + return out, nil +} +func (s *apiTestStore) GetIndexingJobStats(_ context.Context) (*store.IndexingJobStats, error) { + stats := &store.IndexingJobStats{} + for _, job := range s.jobs { + switch job.Status { + case store.IndexingJobPending: + stats.Pending++ + case store.IndexingJobProcessing: + stats.Processing++ + case store.IndexingJobCompleted: + stats.Completed++ + case store.IndexingJobFailed: + stats.Failed++ + case store.IndexingJobDeadLetter: + stats.DeadLetter++ + } + } + return stats, nil +} +func (s *apiTestStore) AppendIndexingAudit(_ context.Context, input store.IndexingAuditInput) error { + s.audits = append(s.audits, &store.IndexingAuditEntry{ + ID: int64(len(s.audits) + 1), Source: input.Source, DocumentID: input.DocumentID, + Collection: input.Collection, CID: input.CID, Decision: input.Decision, + Attempt: input.Attempt, Error: input.Error, CreatedAt: time.Now().UTC().Format(time.RFC3339), + }) + return nil +} +func (s *apiTestStore) ListIndexingAudit(_ context.Context, filter store.IndexingAuditFilter) ([]*store.IndexingAuditEntry, error) { + out := []*store.IndexingAuditEntry{} + for _, entry := range s.audits { + if filter.DocumentID != "" && entry.DocumentID != filter.DocumentID { + continue + } + if filter.Source != "" && entry.Source != filter.Source { + continue + } + if filter.Decision != "" && entry.Decision != filter.Decision { + continue + } + out = append(out, entry) + } + return out, nil +} +func (s *apiTestStore) GetFollowSubjects(context.Context, string) ([]string, error) { return nil, nil } +func (s *apiTestStore) GetRepoCollaborators(context.Context, string) ([]string, error) { + return nil, nil +} +func (s *apiTestStore) CountDocuments(_ context.Context) (int64, error) { + return int64(len(s.docs)), nil +} +func (s *apiTestStore) CountPendingIndexingJobs(_ context.Context) (int64, error) { return 0, nil } +func (s *apiTestStore) InsertJetstreamEvent(context.Context, *store.JetstreamEvent, int) error { + return nil +} +func (s *apiTestStore) ListJetstreamEvents(context.Context, store.JetstreamEventFilter) ([]*store.JetstreamEvent, error) { + return nil, nil +} +func (s *apiTestStore) Ping(context.Context) error { return nil } diff --git a/packages/api/internal/config/config.go b/packages/api/internal/config/config.go index 80b4f65..784bf40 100644 --- a/packages/api/internal/config/config.go +++ b/packages/api/internal/config/config.go @@ -12,34 +12,37 @@ import ( ) type Config struct { - TursoURL string - TursoToken string - TapURL string - TapAuthPassword string - IndexedCollections string - SearchDefaultLimit int - SearchMaxLimit int - SearchDefaultMode string - HTTPBindAddr string - IndexerHealthAddr string - LogLevel string - LogFormat string - EnableAdminEndpoints bool - AdminAuthToken string - EnableIngestEnrichment bool - PLCDirectoryURL string - IdentityServiceURL string - XRPCTimeout time.Duration - ConstellationURL string - ConstellationUserAgent string - ConstellationTimeout time.Duration - ConstellationCacheTTL time.Duration - OAuthClientID string - OAuthRedirectURIs []string - JetstreamURL string + TursoURL string + TursoToken string + TapURL string + TapAuthPassword string + IndexedCollections string + ReadThroughMode string + ReadThroughCollections string + ReadThroughMaxAttempts int + SearchDefaultLimit int + SearchMaxLimit int + SearchDefaultMode string + HTTPBindAddr string + IndexerHealthAddr string + LogLevel string + LogFormat string + EnableAdminEndpoints bool + AdminAuthToken string + EnableIngestEnrichment bool + PLCDirectoryURL string + IdentityServiceURL string + XRPCTimeout time.Duration + ConstellationURL string + ConstellationUserAgent string + ConstellationTimeout time.Duration + ConstellationCacheTTL time.Duration + OAuthClientID string + OAuthRedirectURIs []string + JetstreamURL string JetstreamWantedCollections string - ActivityMaxEvents int - ActivityRewindDuration time.Duration + ActivityMaxEvents int + ActivityRewindDuration time.Duration } type LoadOptions struct { @@ -51,28 +54,31 @@ func Load(opts LoadOptions) (*Config, error) { loadDotEnv() cfg := &Config{ - TursoURL: os.Getenv("TURSO_DATABASE_URL"), - TursoToken: os.Getenv("TURSO_AUTH_TOKEN"), - TapURL: os.Getenv("TAP_URL"), - TapAuthPassword: os.Getenv("TAP_AUTH_PASSWORD"), - IndexedCollections: os.Getenv("INDEXED_COLLECTIONS"), - SearchDefaultMode: envOrDefault("SEARCH_DEFAULT_MODE", "keyword"), - HTTPBindAddr: envOrDefault("HTTP_BIND_ADDR", ":8080"), - IndexerHealthAddr: envOrDefault("INDEXER_HEALTH_ADDR", ":9090"), - LogLevel: envOrDefault("LOG_LEVEL", "info"), - LogFormat: envOrDefault("LOG_FORMAT", "json"), - AdminAuthToken: os.Getenv("ADMIN_AUTH_TOKEN"), - SearchDefaultLimit: envInt("SEARCH_DEFAULT_LIMIT", 20), - SearchMaxLimit: envInt("SEARCH_MAX_LIMIT", 100), - EnableAdminEndpoints: envBool("ENABLE_ADMIN_ENDPOINTS", false), - EnableIngestEnrichment: envBool("ENABLE_INGEST_ENRICHMENT", true), - PLCDirectoryURL: envOrDefault("PLC_DIRECTORY_URL", "https://plc.directory"), - IdentityServiceURL: envOrDefault("IDENTITY_SERVICE_URL", "https://public.api.bsky.app"), - XRPCTimeout: envDuration("XRPC_TIMEOUT", 15*time.Second), - ConstellationURL: envOrDefault("CONSTELLATION_URL", "https://constellation.microcosm.blue"), - ConstellationUserAgent: envOrDefault("CONSTELLATION_USER_AGENT", "twister/1.0 (https://tangled.org/desertthunder.dev/twisted; Owais )"), - ConstellationTimeout: envDuration("CONSTELLATION_TIMEOUT", 10*time.Second), - ConstellationCacheTTL: envDuration("CONSTELLATION_CACHE_TTL", 5*time.Minute), + TursoURL: os.Getenv("TURSO_DATABASE_URL"), + TursoToken: os.Getenv("TURSO_AUTH_TOKEN"), + TapURL: os.Getenv("TAP_URL"), + TapAuthPassword: os.Getenv("TAP_AUTH_PASSWORD"), + IndexedCollections: os.Getenv("INDEXED_COLLECTIONS"), + ReadThroughMode: envOrDefault("READ_THROUGH_MODE", "missing"), + ReadThroughCollections: envOrDefault("READ_THROUGH_COLLECTIONS", os.Getenv("INDEXED_COLLECTIONS")), + ReadThroughMaxAttempts: envInt("READ_THROUGH_MAX_ATTEMPTS", 5), + SearchDefaultMode: envOrDefault("SEARCH_DEFAULT_MODE", "keyword"), + HTTPBindAddr: envOrDefault("HTTP_BIND_ADDR", ":8080"), + IndexerHealthAddr: envOrDefault("INDEXER_HEALTH_ADDR", ":9090"), + LogLevel: envOrDefault("LOG_LEVEL", "info"), + LogFormat: envOrDefault("LOG_FORMAT", "json"), + AdminAuthToken: os.Getenv("ADMIN_AUTH_TOKEN"), + SearchDefaultLimit: envInt("SEARCH_DEFAULT_LIMIT", 20), + SearchMaxLimit: envInt("SEARCH_MAX_LIMIT", 100), + EnableAdminEndpoints: envBool("ENABLE_ADMIN_ENDPOINTS", false), + EnableIngestEnrichment: envBool("ENABLE_INGEST_ENRICHMENT", true), + PLCDirectoryURL: envOrDefault("PLC_DIRECTORY_URL", "https://plc.directory"), + IdentityServiceURL: envOrDefault("IDENTITY_SERVICE_URL", "https://public.api.bsky.app"), + XRPCTimeout: envDuration("XRPC_TIMEOUT", 15*time.Second), + ConstellationURL: envOrDefault("CONSTELLATION_URL", "https://constellation.microcosm.blue"), + ConstellationUserAgent: envOrDefault("CONSTELLATION_USER_AGENT", "twister/1.0 (https://tangled.org/desertthunder.dev/twisted; Owais )"), + ConstellationTimeout: envDuration("CONSTELLATION_TIMEOUT", 10*time.Second), + ConstellationCacheTTL: envDuration("CONSTELLATION_CACHE_TTL", 5*time.Minute), OAuthClientID: os.Getenv("OAUTH_CLIENT_ID"), OAuthRedirectURIs: envSlice("OAUTH_REDIRECT_URIS", nil), JetstreamURL: envOrDefault("JETSTREAM_URL", "wss://jetstream2.us-east.bsky.network/subscribe"), diff --git a/packages/api/internal/config/config_test.go b/packages/api/internal/config/config_test.go index 91a60d9..f07774f 100644 --- a/packages/api/internal/config/config_test.go +++ b/packages/api/internal/config/config_test.go @@ -57,3 +57,26 @@ func TestLoadLocalUsesCurrentWorkingDirectoryWhenUnset(t *testing.T) { t.Fatalf("TursoURL: got %q, want %q", cfg.TursoURL, wantURL) } } + +func TestLoadReadThroughDefaults(t *testing.T) { + t.Setenv("TURSO_DATABASE_URL", "file:test.db") + t.Setenv("TURSO_AUTH_TOKEN", "") + t.Setenv("INDEXED_COLLECTIONS", "sh.tangled.repo,sh.tangled.repo.issue") + t.Setenv("READ_THROUGH_MODE", "") + t.Setenv("READ_THROUGH_COLLECTIONS", "") + t.Setenv("READ_THROUGH_MAX_ATTEMPTS", "") + + cfg, err := Load(LoadOptions{}) + if err != nil { + t.Fatalf("load config: %v", err) + } + if cfg.ReadThroughMode != "missing" { + t.Fatalf("ReadThroughMode: got %q", cfg.ReadThroughMode) + } + if cfg.ReadThroughCollections != "sh.tangled.repo,sh.tangled.repo.issue" { + t.Fatalf("ReadThroughCollections: got %q", cfg.ReadThroughCollections) + } + if cfg.ReadThroughMaxAttempts != 5 { + t.Fatalf("ReadThroughMaxAttempts: got %d", cfg.ReadThroughMaxAttempts) + } +} diff --git a/packages/api/internal/index/enrich.go b/packages/api/internal/index/enrich.go new file mode 100644 index 0000000..55be1b5 --- /dev/null +++ b/packages/api/internal/index/enrich.go @@ -0,0 +1,68 @@ +package index + +import ( + "context" + + "tangled.org/desertthunder.dev/twister/internal/normalize" + "tangled.org/desertthunder.dev/twister/internal/store" + "tangled.org/desertthunder.dev/twister/internal/xrpc" +) + +func (p *Processor) enrichDocument( + ctx context.Context, doc *store.Document, record map[string]any, +) error { + handle, err := p.store.GetIdentityHandle(ctx, doc.DID) + if err != nil { + return err + } + if handle != "" { + doc.AuthorHandle = handle + if doc.RecordType == "profile" { + doc.Title = handle + } + } + if p.xrpc == nil { + return nil + } + if doc.RepoDID != "" && doc.RepoName == "" { + if uri := repoURIFromRecord(record); uri != "" { + _, _, repoRKey, err := normalize.ParseATURI(uri) + if err == nil && repoRKey != "" { + if name, err := p.xrpc.ResolveRepoName(ctx, doc.RepoDID, repoRKey); err == nil { + doc.RepoName = name + } + } + } + } + if doc.AuthorHandle == "" && doc.DID != "" { + if info, err := p.xrpc.ResolveIdentity(ctx, doc.DID); err == nil && info.Handle != "" { + doc.AuthorHandle = info.Handle + if doc.RecordType == "profile" { + doc.Title = info.Handle + } + } + } + if doc.WebURL == "" { + ownerHandle := doc.AuthorHandle + if doc.RepoDID != "" && doc.RepoDID != doc.DID { + if handle, err := p.store.GetIdentityHandle(ctx, doc.RepoDID); err == nil && handle != "" { + ownerHandle = handle + } else if info, err := p.xrpc.ResolveIdentity(ctx, doc.RepoDID); err == nil && info.Handle != "" { + ownerHandle = info.Handle + } + } + doc.WebURL = xrpc.BuildWebURL(ownerHandle, doc.RepoName, doc.RecordType, doc.RKey) + } + return nil +} + +func repoURIFromRecord(record map[string]any) string { + if uri, _ := record["repo"].(string); uri != "" { + return uri + } + target, _ := record["target"].(map[string]any) + if uri, _ := target["repo"].(string); uri != "" { + return uri + } + return "" +} diff --git a/packages/api/internal/index/errors.go b/packages/api/internal/index/errors.go new file mode 100644 index 0000000..89d0936 --- /dev/null +++ b/packages/api/internal/index/errors.go @@ -0,0 +1,24 @@ +package index + +import "fmt" + +type PermanentError struct { + Decision string + Err error +} + +func (e *PermanentError) Error() string { + if e.Err == nil { + return e.Decision + } + return fmt.Sprintf("%s: %v", e.Decision, e.Err) +} + +func (e *PermanentError) Unwrap() error { + return e.Err +} + +func IsPermanent(err error) (*PermanentError, bool) { + perr, ok := err.(*PermanentError) + return perr, ok +} diff --git a/packages/api/internal/index/index.go b/packages/api/internal/index/index.go deleted file mode 100644 index 60a0228..0000000 --- a/packages/api/internal/index/index.go +++ /dev/null @@ -1 +0,0 @@ -package index diff --git a/packages/api/internal/index/policy.go b/packages/api/internal/index/policy.go new file mode 100644 index 0000000..5223b20 --- /dev/null +++ b/packages/api/internal/index/policy.go @@ -0,0 +1,85 @@ +package index + +import "strings" + +const ( + ReadThroughOff = "off" + ReadThroughMissing = "missing" + ReadThroughBroad = "broad" +) + +type Policy struct { + indexed allowlist + readThrough allowlist + mode string +} + +func NewPolicy(indexedCollections, readThroughCollections, mode string) Policy { + if strings.TrimSpace(readThroughCollections) == "" { + readThroughCollections = indexedCollections + } + return Policy{ + indexed: parseAllowlist(indexedCollections), + readThrough: parseAllowlist(readThroughCollections), + mode: normalizeMode(mode), + } +} + +func (p Policy) Allows(source, collection string) bool { + if source == "read_through" { + return p.readThrough.match(collection) + } + return p.indexed.match(collection) +} + +func (p Policy) ReadThroughMode() string { + return p.mode +} + +func normalizeMode(mode string) string { + switch strings.TrimSpace(strings.ToLower(mode)) { + case ReadThroughOff, ReadThroughBroad: + return strings.TrimSpace(strings.ToLower(mode)) + default: + return ReadThroughMissing + } +} + +type allowlist struct { + entries []string +} + +func parseAllowlist(raw string) allowlist { + if strings.TrimSpace(raw) == "" { + return allowlist{} + } + parts := strings.FieldsFunc(raw, func(r rune) bool { + return r == ',' || r == ' ' || r == '\n' || r == '\t' + }) + entries := make([]string, 0, len(parts)) + for _, part := range parts { + entry := strings.TrimSpace(part) + if entry != "" { + entries = append(entries, entry) + } + } + return allowlist{entries: entries} +} + +func (a allowlist) match(collection string) bool { + if len(a.entries) == 0 { + return true + } + for _, entry := range a.entries { + if entry == collection { + return true + } + if strings.HasSuffix(entry, "*") { + prefix := strings.TrimSuffix(entry, "*") + if strings.HasPrefix(collection, prefix) { + return true + } + } + } + return false +} diff --git a/packages/api/internal/index/processor.go b/packages/api/internal/index/processor.go new file mode 100644 index 0000000..4a91ada --- /dev/null +++ b/packages/api/internal/index/processor.go @@ -0,0 +1,101 @@ +package index + +import ( + "context" + "fmt" + "log/slog" + + "tangled.org/desertthunder.dev/twister/internal/normalize" + "tangled.org/desertthunder.dev/twister/internal/store" + "tangled.org/desertthunder.dev/twister/internal/xrpc" +) + +type Result struct { + Decision string + DocumentID string + Collection string + CID string +} + +type Processor struct { + store store.IndexingStore + registry *normalize.Registry + xrpc *xrpc.Client + policy Policy + log *slog.Logger +} + +func NewProcessor( + st store.IndexingStore, registry *normalize.Registry, xrpcClient *xrpc.Client, + policy Policy, log *slog.Logger, +) *Processor { + if log == nil { + log = slog.Default() + } + return &Processor{store: st, registry: registry, xrpc: xrpcClient, policy: policy, log: log} +} + +func (p *Processor) ProcessRecord( + ctx context.Context, source string, event normalize.TapRecordEvent, +) (*Result, error) { + if event.Record == nil { + return &Result{Decision: "skip_missing_record"}, nil + } + record := event.Record + result := &Result{ + Decision: "skip_unknown", + DocumentID: normalize.StableID(record.DID, record.Collection, record.RKey), + Collection: record.Collection, + CID: record.CID, + } + if !p.policy.Allows(source, record.Collection) { + result.Decision = "skip_collection" + return result, nil + } + if handler, ok := p.registry.StateHandler(record.Collection); ok { + if record.Action == "delete" { + result.Decision = "skip_state_delete" + return result, nil + } + update, err := handler.HandleState(event) + if err != nil { + return result, &PermanentError{Decision: "invalid_state_record", Err: err} + } + if err := p.store.UpdateRecordState(ctx, update.SubjectURI, update.State); err != nil { + return result, fmt.Errorf("update state: %w", err) + } + result.Decision = "state_updated" + return result, nil + } + adapter, ok := p.registry.Adapter(record.Collection) + if !ok { + result.Decision = "skip_unsupported_collection" + return result, nil + } + if record.Action == "delete" { + if err := p.store.MarkDeleted(ctx, result.DocumentID); err != nil { + return result, fmt.Errorf("mark deleted: %w", err) + } + result.Decision = "document_deleted" + return result, nil + } + if record.Record == nil { + return result, &PermanentError{Decision: "missing_record_payload"} + } + if !adapter.Searchable(record.Record) { + result.Decision = "skip_unsearchable" + return result, nil + } + doc, err := adapter.Normalize(event) + if err != nil { + return result, &PermanentError{Decision: "normalize_failed", Err: err} + } + if err := p.enrichDocument(ctx, doc, record.Record); err != nil { + return result, err + } + if err := p.store.UpsertDocument(ctx, doc); err != nil { + return result, fmt.Errorf("upsert document: %w", err) + } + result.Decision = "document_upserted" + return result, nil +} diff --git a/packages/api/internal/ingest/ingest.go b/packages/api/internal/ingest/ingest.go index a3aca03..d7b521b 100644 --- a/packages/api/internal/ingest/ingest.go +++ b/packages/api/internal/ingest/ingest.go @@ -10,6 +10,7 @@ import ( "sync" "time" + idx "tangled.org/desertthunder.dev/twister/internal/index" "tangled.org/desertthunder.dev/twister/internal/normalize" "tangled.org/desertthunder.dev/twister/internal/store" "tangled.org/desertthunder.dev/twister/internal/xrpc" @@ -30,10 +31,10 @@ type client interface { // Runner ingests Tap events into the store. type Runner struct { store store.Store - registry *normalize.Registry tap client - xrpcClient *xrpc.Client - allowlist allowlist + registry *normalize.Registry + policy idx.Policy + processor *idx.Processor consumerName string log *slog.Logger resumeCursor int64 @@ -47,11 +48,13 @@ func NewRunner(st store.Store, registry *normalize.Registry, tap client, indexed if log == nil { log = slog.Default() } + policy := idx.NewPolicy(indexedCollections, indexedCollections, idx.ReadThroughMissing) return &Runner{ store: st, - registry: registry, tap: tap, - allowlist: parseAllowlist(indexedCollections), + registry: registry, + policy: policy, + processor: idx.NewProcessor(st, registry, nil, policy, log), consumerName: defaultConsumerName, log: log, } @@ -59,7 +62,10 @@ func NewRunner(st store.Store, registry *normalize.Registry, tap client, indexed // SetXRPCClient enables ingest-time enrichment via XRPC lookups. func (r *Runner) SetXRPCClient(c *xrpc.Client) { - r.xrpcClient = c + if c == nil { + return + } + r.processor = idx.NewProcessor(r.store, r.registry, c, r.policy, r.log) } func (r *Runner) Run(ctx context.Context) error { @@ -184,100 +190,48 @@ func (r *Runner) processEvent(ctx context.Context, event normalize.TapRecordEven } func (r *Runner) processRecordEvent(ctx context.Context, event normalize.TapRecordEvent) error { - if event.Record == nil { - return r.advanceCursorAndAck(ctx, event.ID) - } - record := event.Record - if !r.allowlist.match(record.Collection) { - return r.advanceCursorAndAck(ctx, event.ID) - } - - if handler, ok := r.registry.StateHandler(record.Collection); ok { - if record.Action == "delete" { - return r.advanceCursorAndAck(ctx, event.ID) - } - update, err := handler.HandleState(event) - if err != nil { - r.log.Warn("state normalization failed", - slog.Int64("event_id", event.ID), - slog.String("collection", record.Collection), - slog.String("did", record.DID), - slog.String("rkey", record.RKey), - slog.String("error", err.Error()), - ) - return r.advanceCursorAndAck(ctx, event.ID) - } - if err := r.store.UpdateRecordState(ctx, update.SubjectURI, update.State); err != nil { - return err - } - return r.advanceCursorAndAck(ctx, event.ID) - } - - adapter, ok := r.registry.Adapter(record.Collection) - if !ok { - return r.advanceCursorAndAck(ctx, event.ID) - } - - switch record.Action { - case "delete": - docID := normalize.StableID(record.DID, record.Collection, record.RKey) - if err := r.store.MarkDeleted(ctx, docID); err != nil { - return err - } - return r.advanceCursorAndAck(ctx, event.ID) - case "create", "update": - if record.Record == nil { - r.log.Warn("record payload missing", - slog.Int64("event_id", event.ID), - slog.String("collection", record.Collection), - slog.String("did", record.DID), - slog.String("rkey", record.RKey), - ) - return r.advanceCursorAndAck(ctx, event.ID) - } - default: - return r.advanceCursorAndAck(ctx, event.ID) - } - - doc, err := adapter.Normalize(event) - if err != nil { - r.log.Warn("normalization failed", + result, err := r.processor.ProcessRecord(ctx, store.IndexSourceTap, event) + if perr, ok := idx.IsPermanent(err); ok { + r.log.Warn("tap processing skipped", slog.Int64("event_id", event.ID), slog.String("collection", record.Collection), slog.String("did", record.DID), slog.String("rkey", record.RKey), - slog.String("error", err.Error()), + slog.String("decision", perr.Decision), + slog.String("error", perr.Error()), ) + _ = r.store.AppendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: store.IndexSourceTap, + DocumentID: normalize.StableID(record.DID, record.Collection, record.RKey), + Collection: record.Collection, + CID: record.CID, + Decision: perr.Decision, + Error: perr.Error(), + }) return r.advanceCursorAndAck(ctx, event.ID) } - - handle, err := r.store.GetIdentityHandle(ctx, record.DID) if err != nil { return err } - if handle != "" { - doc.AuthorHandle = handle - if doc.RecordType == "profile" { - doc.Title = handle - } - } - - r.enrichDocument(ctx, doc) - - if err := r.store.UpsertDocument(ctx, doc); err != nil { - return err + if result != nil { + _ = r.store.AppendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: store.IndexSourceTap, + DocumentID: result.DocumentID, + Collection: result.Collection, + CID: result.CID, + Decision: result.Decision, + }) } - return r.advanceCursorAndAck(ctx, event.ID) } func (r *Runner) advanceCursorAndAck(ctx context.Context, eventID int64) error { cursor := fmt.Sprintf("%d", eventID) - if err := r.tap.AckEvent(ctx, eventID); err != nil { + if err := r.persistCursorWithRetry(ctx, cursor, eventID); err != nil { return err } - if err := r.persistCursorWithRetry(ctx, cursor, eventID); err != nil { + if err := r.tap.AckEvent(ctx, eventID); err != nil { return err } r.markProcessed(cursor) @@ -295,7 +249,7 @@ func (r *Runner) persistCursorWithRetry(ctx context.Context, cursor string, even } else { attempt++ backoff := retryBackoff(attempt) - r.log.Error("cursor persist failed after ack", + r.log.Error("cursor persist failed before ack", slog.Int64("event_id", eventID), slog.Int("attempt", attempt), slog.Duration("retry_in", backoff), @@ -346,113 +300,6 @@ func (r *Runner) markProcessed(cursor string) { r.statusMu.Unlock() } -type allowlist struct { - entries []string -} - -func parseAllowlist(raw string) allowlist { - if strings.TrimSpace(raw) == "" { - return allowlist{entries: nil} - } - parts := strings.FieldsFunc(raw, func(r rune) bool { - return r == ',' || r == ' ' || r == '\n' || r == '\t' - }) - entries := make([]string, 0, len(parts)) - for _, part := range parts { - entry := strings.TrimSpace(part) - if entry == "" { - continue - } - entries = append(entries, entry) - } - return allowlist{entries: entries} -} - -func (a allowlist) match(collection string) bool { - if len(a.entries) == 0 { - return true - } - for _, entry := range a.entries { - if entry == collection { - return true - } - if strings.HasSuffix(entry, "*") { - prefix := strings.TrimSuffix(entry, "*") - if strings.HasPrefix(collection, prefix) { - return true - } - } - } - return false -} - -// enrichDocument fills RepoName, AuthorHandle, and WebURL via XRPC when possible. -// Failures are logged but never block ingestion. -func (r *Runner) enrichDocument(ctx context.Context, doc *store.Document) { - if r.xrpcClient == nil { - return - } - - if doc.RepoDID != "" && doc.RepoName == "" { - repoRKey := extractRepoRKey(doc.ATURI, doc.Collection) - if repoRKey != "" { - name, err := r.xrpcClient.ResolveRepoName(ctx, doc.RepoDID, repoRKey) - if err != nil { - r.log.Debug("enrich: resolve repo name failed", - slog.String("doc_id", doc.ID), - slog.String("repo_did", doc.RepoDID), - slog.String("error", err.Error()), - ) - } else { - doc.RepoName = name - } - } - } - - if doc.AuthorHandle == "" && doc.DID != "" { - info, err := r.xrpcClient.ResolveIdentity(ctx, doc.DID) - if err != nil { - r.log.Debug("enrich: resolve author handle failed", - slog.String("doc_id", doc.ID), - slog.String("did", doc.DID), - slog.String("error", err.Error()), - ) - } else if info.Handle != "" { - doc.AuthorHandle = info.Handle - } - } - - if doc.WebURL == "" { - ownerHandle := doc.AuthorHandle - if doc.RepoDID != "" && doc.RepoDID != doc.DID { - repoOwnerHandle, err := r.store.GetIdentityHandle(ctx, doc.RepoDID) - if err == nil && repoOwnerHandle != "" { - ownerHandle = repoOwnerHandle - } else if r.xrpcClient != nil { - info, err := r.xrpcClient.ResolveIdentity(ctx, doc.RepoDID) - if err == nil && info.Handle != "" { - ownerHandle = info.Handle - } - } - } - doc.WebURL = xrpc.BuildWebURL(ownerHandle, doc.RepoName, doc.RecordType, doc.RKey) - } -} - -// extractRepoRKey attempts to extract the repo rkey from the document context. -// For repo-scoped collections like sh.tangled.repo.issue, the AT-URI is -// at://did/collection/rkey but the repo is identified by RepoDID. We look -// for a stored repo document, or try common rkey patterns. -func extractRepoRKey(atURI, collection string) string { - if collection == "sh.tangled.repo" { - parts := strings.SplitN(atURI, "/", 5) - if len(parts) >= 5 { - return parts[4] - } - } - return "" -} - func retryBackoff(attempt int) time.Duration { if attempt < 1 { attempt = 1 diff --git a/packages/api/internal/ingest/ingest_test.go b/packages/api/internal/ingest/ingest_test.go index ef9d319..56fba48 100644 --- a/packages/api/internal/ingest/ingest_test.go +++ b/packages/api/internal/ingest/ingest_test.go @@ -6,6 +6,7 @@ import ( "log/slog" "testing" + idx "tangled.org/desertthunder.dev/twister/internal/index" "tangled.org/desertthunder.dev/twister/internal/normalize" "tangled.org/desertthunder.dev/twister/internal/store" ) @@ -36,7 +37,9 @@ type fakeStore struct { initialSync *store.SyncState recordStates map[string]string handles map[string]string - onSetSync func() + jobs map[string]*store.IndexingJob + audits []store.IndexingAuditInput + onSetSync func() } func newFakeStore() *fakeStore { @@ -44,7 +47,8 @@ func newFakeStore() *fakeStore { docs: make(map[string]*store.Document), deleted: make(map[string]bool), recordStates: make(map[string]string), - handles: make(map[string]string), + handles: make(map[string]string), + jobs: make(map[string]*store.IndexingJob), } } @@ -100,7 +104,11 @@ func (f *fakeStore) EnqueueIndexingJob(_ context.Context, _ store.IndexingJobInp return nil } -func (f *fakeStore) ClaimIndexingJob(_ context.Context) (*store.IndexingJob, error) { +func (f *fakeStore) GetIndexingJob(_ context.Context, documentID string) (*store.IndexingJob, error) { + return f.jobs[documentID], nil +} + +func (f *fakeStore) ClaimIndexingJob(_ context.Context, _, _ string) (*store.IndexingJob, error) { return nil, nil } @@ -112,6 +120,27 @@ func (f *fakeStore) RetryIndexingJob(_ context.Context, _, _, _ string) error { return nil } +func (f *fakeStore) FailIndexingJob(_ context.Context, _, _, _ string) error { + return nil +} + +func (f *fakeStore) ListIndexingJobs(_ context.Context, _ store.IndexingJobFilter) ([]*store.IndexingJob, error) { + return nil, nil +} + +func (f *fakeStore) GetIndexingJobStats(_ context.Context) (*store.IndexingJobStats, error) { + return &store.IndexingJobStats{}, nil +} + +func (f *fakeStore) AppendIndexingAudit(_ context.Context, input store.IndexingAuditInput) error { + f.audits = append(f.audits, input) + return nil +} + +func (f *fakeStore) ListIndexingAudit(_ context.Context, _ store.IndexingAuditFilter) ([]*store.IndexingAuditEntry, error) { + return nil, nil +} + func (f *fakeStore) GetFollowSubjects(_ context.Context, _ string) ([]string, error) { return nil, nil } @@ -302,14 +331,14 @@ func TestRunner_NormalizationFailureAdvancesCursor(t *testing.T) { } func TestAllowlistMatching(t *testing.T) { - a := parseAllowlist("sh.tangled.repo, sh.tangled.string sh.tangled.actor.*") - if !a.match("sh.tangled.repo") { + policy := idx.NewPolicy("sh.tangled.repo, sh.tangled.string sh.tangled.actor.*", "", idx.ReadThroughMissing) + if !policy.Allows(store.IndexSourceTap, "sh.tangled.repo") { t.Fatal("expected exact match") } - if !a.match("sh.tangled.actor.profile") { + if !policy.Allows(store.IndexSourceTap, "sh.tangled.actor.profile") { t.Fatal("expected wildcard prefix match") } - if a.match("app.bsky.feed.post") { + if policy.Allows(store.IndexSourceTap, "app.bsky.feed.post") { t.Fatal("unexpected match") } } @@ -337,7 +366,7 @@ func TestRunner_InitializeCursorResume(t *testing.T) { } } -func TestRunner_AckBeforeCursorPersist(t *testing.T) { +func TestRunner_PersistCursorBeforeAck(t *testing.T) { st := newFakeStore() tap := &fakeTapClient{} r := newRunnerForTest(st, tap, "sh.tangled.*") @@ -345,8 +374,8 @@ func TestRunner_AckBeforeCursorPersist(t *testing.T) { acked := false tap.onAck = func(_ int64) { acked = true } st.onSetSync = func() { - if !acked { - t.Fatalf("cursor persisted before ack") + if acked { + t.Fatalf("ack happened before cursor persisted") } } diff --git a/packages/api/internal/store/indexing_audit.go b/packages/api/internal/store/indexing_audit.go new file mode 100644 index 0000000..1e78506 --- /dev/null +++ b/packages/api/internal/store/indexing_audit.go @@ -0,0 +1,81 @@ +package store + +import ( + "context" + "fmt" + "time" +) + +func (s *SQLStore) AppendIndexingAudit(ctx context.Context, input IndexingAuditInput) error { + now := time.Now().UTC().Format(time.RFC3339) + _, err := s.db.ExecContext(ctx, ` + INSERT INTO indexing_audit ( + source, document_id, collection, cid, decision, attempt, error, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + input.Source, input.DocumentID, input.Collection, input.CID, + input.Decision, input.Attempt, nullableStr(input.Error), now, + ) + if err != nil { + return fmt.Errorf("append indexing audit: %w", err) + } + return nil +} + +func (s *SQLStore) ListIndexingAudit( + ctx context.Context, filter IndexingAuditFilter, +) ([]*IndexingAuditEntry, error) { + query := ` + SELECT id, source, document_id, collection, cid, decision, + attempt, COALESCE(error, ''), created_at + FROM indexing_audit + WHERE 1 = 1` + args := []any{} + + if filter.DocumentID != "" { + query += " AND document_id = ?" + args = append(args, filter.DocumentID) + } + if filter.Source != "" { + query += " AND source = ?" + args = append(args, filter.Source) + } + if filter.Decision != "" { + query += " AND decision = ?" + args = append(args, filter.Decision) + } + + query += " ORDER BY created_at DESC" + limit := filter.Limit + if limit <= 0 { + limit = 50 + } + query += " LIMIT ?" + args = append(args, limit) + if filter.Offset > 0 { + query += " OFFSET ?" + args = append(args, filter.Offset) + } + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("list indexing audit: %w", err) + } + defer rows.Close() + + var entries []*IndexingAuditEntry + for rows.Next() { + entry := &IndexingAuditEntry{} + if err := rows.Scan( + &entry.ID, &entry.Source, &entry.DocumentID, &entry.Collection, + &entry.CID, &entry.Decision, &entry.Attempt, &entry.Error, + &entry.CreatedAt, + ); err != nil { + return nil, fmt.Errorf("scan indexing audit: %w", err) + } + entries = append(entries, entry) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate indexing audit: %w", err) + } + return entries, nil +} diff --git a/packages/api/internal/store/indexing_jobs.go b/packages/api/internal/store/indexing_jobs.go new file mode 100644 index 0000000..1b4a58d --- /dev/null +++ b/packages/api/internal/store/indexing_jobs.go @@ -0,0 +1,284 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" + "time" +) + +func (s *SQLStore) GetIndexingJob(ctx context.Context, documentID string) (*IndexingJob, error) { + row := s.db.QueryRowContext(ctx, ` + SELECT document_id, did, collection, rkey, cid, record_json, source, + attempts, status, COALESCE(last_error, ''), scheduled_at, updated_at, + COALESCE(lease_owner, ''), COALESCE(lease_expires_at, ''), + COALESCE(completed_at, '') + FROM indexing_jobs + WHERE document_id = ?`, documentID) + + job, err := scanIndexingJob(row) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("get indexing job: %w", err) + } + return job, nil +} + +func (s *SQLStore) EnqueueIndexingJob(ctx context.Context, input IndexingJobInput) error { + now := time.Now().UTC().Format(time.RFC3339) + source := strings.TrimSpace(input.Source) + if source == "" { + source = IndexSourceReadThrough + } + + _, err := s.db.ExecContext(ctx, ` + INSERT INTO indexing_jobs ( + document_id, did, collection, rkey, cid, record_json, source, + status, attempts, last_error, scheduled_at, updated_at, + lease_owner, lease_expires_at, completed_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, NULL, ?, ?, '', '', '') + ON CONFLICT(document_id) DO UPDATE SET + did = excluded.did, + collection = excluded.collection, + rkey = excluded.rkey, + cid = excluded.cid, + record_json = excluded.record_json, + source = excluded.source, + status = excluded.status, + last_error = NULL, + scheduled_at = excluded.scheduled_at, + updated_at = excluded.updated_at, + lease_owner = '', + lease_expires_at = '', + completed_at = ''`, + input.DocumentID, input.DID, input.Collection, input.RKey, input.CID, + input.RecordJSON, source, IndexingJobPending, now, now, + ) + if err != nil { + return fmt.Errorf("enqueue indexing job: %w", err) + } + return nil +} + +func (s *SQLStore) ClaimIndexingJob( + ctx context.Context, workerID string, leaseUntil string, +) (*IndexingJob, error) { + now := time.Now().UTC().Format(time.RFC3339) + conn, err := s.db.Conn(ctx) + if err != nil { + return nil, fmt.Errorf("claim indexing job conn: %w", err) + } + defer conn.Close() + + if _, err := conn.ExecContext(ctx, `BEGIN IMMEDIATE`); err != nil { + return nil, fmt.Errorf("begin immediate claim tx: %w", err) + } + committed := false + defer func() { + if !committed { + _, _ = conn.ExecContext(context.Background(), `ROLLBACK`) + } + }() + + var documentID string + err = conn.QueryRowContext(ctx, ` + SELECT document_id + FROM indexing_jobs + WHERE (status = ? AND datetime(scheduled_at) <= datetime(?)) + OR ( + status = ? + AND lease_expires_at != '' + AND datetime(lease_expires_at) <= datetime(?) + ) + ORDER BY scheduled_at ASC, updated_at ASC + LIMIT 1`, + IndexingJobPending, now, IndexingJobProcessing, now, + ).Scan(&documentID) + if errors.Is(err, sql.ErrNoRows) { + if _, err := conn.ExecContext(ctx, `COMMIT`); err != nil { + return nil, fmt.Errorf("commit empty claim tx: %w", err) + } + committed = true + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("select claim candidate: %w", err) + } + + row := conn.QueryRowContext(ctx, ` + UPDATE indexing_jobs + SET status = ?, updated_at = ?, lease_owner = ?, lease_expires_at = ? + WHERE document_id = ? + RETURNING document_id, did, collection, rkey, cid, record_json, source, + attempts, status, COALESCE(last_error, ''), scheduled_at, + updated_at, COALESCE(lease_owner, ''), + COALESCE(lease_expires_at, ''), COALESCE(completed_at, '')`, + IndexingJobProcessing, now, workerID, leaseUntil, documentID, + ) + + job, err := scanIndexingJob(row) + if err != nil { + return nil, fmt.Errorf("update claim candidate: %w", err) + } + if _, err := conn.ExecContext(ctx, `COMMIT`); err != nil { + return nil, fmt.Errorf("commit claim tx: %w", err) + } + committed = true + return job, nil +} + +func (s *SQLStore) CompleteIndexingJob(ctx context.Context, documentID string) error { + now := time.Now().UTC().Format(time.RFC3339) + _, err := s.db.ExecContext(ctx, ` + UPDATE indexing_jobs + SET status = ?, updated_at = ?, completed_at = ?, + lease_owner = '', lease_expires_at = '', last_error = NULL + WHERE document_id = ?`, + IndexingJobCompleted, now, now, documentID, + ) + if err != nil { + return fmt.Errorf("complete indexing job: %w", err) + } + return nil +} + +func (s *SQLStore) RetryIndexingJob( + ctx context.Context, documentID string, nextScheduledAt string, lastError string, +) error { + now := time.Now().UTC().Format(time.RFC3339) + _, err := s.db.ExecContext(ctx, ` + UPDATE indexing_jobs + SET status = ?, attempts = attempts + 1, last_error = ?, scheduled_at = ?, + updated_at = ?, lease_owner = '', lease_expires_at = '' + WHERE document_id = ?`, + IndexingJobPending, lastError, nextScheduledAt, now, documentID, + ) + if err != nil { + return fmt.Errorf("retry indexing job: %w", err) + } + return nil +} + +func (s *SQLStore) FailIndexingJob( + ctx context.Context, documentID string, status string, lastError string, +) error { + now := time.Now().UTC().Format(time.RFC3339) + _, err := s.db.ExecContext(ctx, ` + UPDATE indexing_jobs + SET status = ?, attempts = attempts + 1, last_error = ?, updated_at = ?, + lease_owner = '', lease_expires_at = '' + WHERE document_id = ?`, + status, lastError, now, documentID, + ) + if err != nil { + return fmt.Errorf("fail indexing job: %w", err) + } + return nil +} + +func (s *SQLStore) ListIndexingJobs( + ctx context.Context, filter IndexingJobFilter, +) ([]*IndexingJob, error) { + query := ` + SELECT document_id, did, collection, rkey, cid, record_json, source, + attempts, status, COALESCE(last_error, ''), scheduled_at, updated_at, + COALESCE(lease_owner, ''), COALESCE(lease_expires_at, ''), + COALESCE(completed_at, '') + FROM indexing_jobs + WHERE 1 = 1` + args := []any{} + + if filter.DocumentID != "" { + query += " AND document_id = ?" + args = append(args, filter.DocumentID) + } + if filter.Status != "" { + query += " AND status = ?" + args = append(args, filter.Status) + } + if filter.Source != "" { + query += " AND source = ?" + args = append(args, filter.Source) + } + + query += " ORDER BY updated_at DESC" + limit := filter.Limit + if limit <= 0 { + limit = 50 + } + query += " LIMIT ?" + args = append(args, limit) + if filter.Offset > 0 { + query += " OFFSET ?" + args = append(args, filter.Offset) + } + + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("list indexing jobs: %w", err) + } + defer rows.Close() + + var jobs []*IndexingJob + for rows.Next() { + job, err := scanIndexingJob(rows) + if err != nil { + return nil, fmt.Errorf("scan indexing job: %w", err) + } + jobs = append(jobs, job) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate indexing jobs: %w", err) + } + return jobs, nil +} + +func (s *SQLStore) GetIndexingJobStats(ctx context.Context) (*IndexingJobStats, error) { + row := s.db.QueryRowContext(ctx, ` + SELECT + COUNT(*) FILTER (WHERE status = ?), + COUNT(*) FILTER (WHERE status = ?), + COUNT(*) FILTER (WHERE status = ?), + COUNT(*) FILTER (WHERE status = ?), + COUNT(*) FILTER (WHERE status = ?), + COALESCE(MIN(CASE WHEN status = ? THEN scheduled_at END), ''), + COALESCE(MIN(CASE WHEN status = ? THEN updated_at END), ''), + COALESCE(MAX(completed_at), ''), + COALESCE(MAX(updated_at), '') + FROM indexing_jobs`, + IndexingJobPending, IndexingJobProcessing, IndexingJobCompleted, + IndexingJobFailed, IndexingJobDeadLetter, IndexingJobPending, + IndexingJobProcessing, + ) + + stats := &IndexingJobStats{} + err := row.Scan( + &stats.Pending, &stats.Processing, &stats.Completed, &stats.Failed, + &stats.DeadLetter, &stats.OldestPendingAt, &stats.OldestRunningAt, + &stats.LastCompletedAt, &stats.LastProcessedAt, + ) + if err != nil { + return nil, fmt.Errorf("get indexing job stats: %w", err) + } + return stats, nil +} + +func scanIndexingJob(scanner interface { + Scan(dest ...any) error +}) (*IndexingJob, error) { + job := &IndexingJob{} + err := scanner.Scan( + &job.DocumentID, &job.DID, &job.Collection, &job.RKey, &job.CID, + &job.RecordJSON, &job.Source, &job.Attempts, &job.Status, + &job.LastError, &job.ScheduledAt, &job.UpdatedAt, &job.LeaseOwner, + &job.LeaseUntil, &job.CompletedAt, + ) + if err != nil { + return nil, err + } + return job, nil +} diff --git a/packages/api/internal/store/indexing_types.go b/packages/api/internal/store/indexing_types.go new file mode 100644 index 0000000..406b5bb --- /dev/null +++ b/packages/api/internal/store/indexing_types.go @@ -0,0 +1,74 @@ +package store + +import "context" + +const ( + IndexSourceTap = "tap" + IndexSourceReadThrough = "read_through" + IndexSourceBackfill = "backfill" + IndexSourceAdmin = "admin" + + IndexingJobPending = "pending" + IndexingJobProcessing = "processing" + IndexingJobCompleted = "completed" + IndexingJobFailed = "failed" + IndexingJobDeadLetter = "dead_letter" +) + +type IndexingJobFilter struct { + Status string + Source string + DocumentID string + Limit int + Offset int +} + +type IndexingAuditFilter struct { + Source string + Decision string + DocumentID string + Limit int + Offset int +} + +type IndexingJobStats struct { + Pending int64 + Processing int64 + Completed int64 + Failed int64 + DeadLetter int64 + OldestPendingAt string + OldestRunningAt string + LastCompletedAt string + LastProcessedAt string +} + +type IndexingAuditEntry struct { + ID int64 + Source string + DocumentID string + Collection string + CID string + Decision string + Attempt int + Error string + CreatedAt string +} + +type IndexingAuditInput struct { + Source string + DocumentID string + Collection string + CID string + Decision string + Attempt int + Error string +} + +type IndexingStore interface { + GetDocument(ctx context.Context, id string) (*Document, error) + GetIdentityHandle(ctx context.Context, did string) (string, error) + MarkDeleted(ctx context.Context, id string) error + UpdateRecordState(ctx context.Context, subjectURI string, state string) error + UpsertDocument(ctx context.Context, doc *Document) error +} diff --git a/packages/api/internal/store/migrations/008_indexing_observability.sql b/packages/api/internal/store/migrations/008_indexing_observability.sql new file mode 100644 index 0000000..5efe283 --- /dev/null +++ b/packages/api/internal/store/migrations/008_indexing_observability.sql @@ -0,0 +1,32 @@ +ALTER TABLE indexing_jobs + ADD COLUMN source TEXT NOT NULL DEFAULT 'read_through'; + +ALTER TABLE indexing_jobs + ADD COLUMN lease_owner TEXT DEFAULT ''; + +ALTER TABLE indexing_jobs + ADD COLUMN lease_expires_at TEXT DEFAULT ''; + +ALTER TABLE indexing_jobs + ADD COLUMN completed_at TEXT DEFAULT ''; + +CREATE INDEX IF NOT EXISTS idx_indexing_jobs_claim + ON indexing_jobs(status, scheduled_at, lease_expires_at, updated_at); + +CREATE TABLE IF NOT EXISTS indexing_audit ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + document_id TEXT NOT NULL, + collection TEXT NOT NULL, + cid TEXT NOT NULL, + decision TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 0, + error TEXT, + created_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_indexing_audit_created + ON indexing_audit(created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_indexing_audit_document + ON indexing_audit(document_id, created_at DESC); diff --git a/packages/api/internal/store/sql_store.go b/packages/api/internal/store/sql_store.go index 681aeb8..6b558f5 100644 --- a/packages/api/internal/store/sql_store.go +++ b/packages/api/internal/store/sql_store.go @@ -253,122 +253,6 @@ func (s *SQLStore) GetIdentityHandle(ctx context.Context, did string) (string, e return handle.String, nil } -func (s *SQLStore) EnqueueIndexingJob(ctx context.Context, input IndexingJobInput) error { - now := time.Now().UTC().Format(time.RFC3339) - _, err := s.db.ExecContext(ctx, ` - INSERT INTO indexing_jobs ( - document_id, did, collection, rkey, cid, record_json, - status, attempts, last_error, scheduled_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, NULL, ?, ?) - ON CONFLICT(document_id) DO UPDATE SET - did = excluded.did, - collection = excluded.collection, - rkey = excluded.rkey, - cid = excluded.cid, - record_json = excluded.record_json, - status = 'pending', - last_error = NULL, - scheduled_at = excluded.scheduled_at, - updated_at = excluded.updated_at`, - input.DocumentID, input.DID, input.Collection, input.RKey, input.CID, input.RecordJSON, now, now, - ) - if err != nil { - return fmt.Errorf("enqueue indexing job: %w", err) - } - return nil -} - -func (s *SQLStore) ClaimIndexingJob(ctx context.Context) (*IndexingJob, error) { - now := time.Now().UTC().Format(time.RFC3339) - staleCutoff := time.Now().UTC().Add(-5 * time.Minute).Format(time.RFC3339) - - tx, err := s.db.BeginTx(ctx, nil) - if err != nil { - return nil, fmt.Errorf("begin claim indexing job tx: %w", err) - } - defer tx.Rollback() - - var job IndexingJob - row := tx.QueryRowContext(ctx, ` - SELECT document_id, did, collection, rkey, cid, record_json, - attempts, status, COALESCE(last_error, ''), scheduled_at, updated_at - FROM indexing_jobs - WHERE (status = 'pending' AND scheduled_at <= ?) - OR (status = 'processing' AND updated_at <= ?) - ORDER BY scheduled_at ASC, updated_at ASC - LIMIT 1`, now, staleCutoff) - - err = row.Scan( - &job.DocumentID, - &job.DID, - &job.Collection, - &job.RKey, - &job.CID, - &job.RecordJSON, - &job.Attempts, - &job.Status, - &job.LastError, - &job.ScheduledAt, - &job.UpdatedAt, - ) - if errors.Is(err, sql.ErrNoRows) { - return nil, nil - } - if err != nil { - return nil, fmt.Errorf("select indexing job: %w", err) - } - - res, err := tx.ExecContext(ctx, ` - UPDATE indexing_jobs - SET status = 'processing', updated_at = ? - WHERE document_id = ?`, - now, job.DocumentID, - ) - if err != nil { - return nil, fmt.Errorf("mark indexing job processing: %w", err) - } - affected, err := res.RowsAffected() - if err != nil { - return nil, fmt.Errorf("rows affected claim indexing job: %w", err) - } - if affected == 0 { - return nil, nil - } - - if err := tx.Commit(); err != nil { - return nil, fmt.Errorf("commit claim indexing job tx: %w", err) - } - job.Status = "processing" - job.UpdatedAt = now - return &job, nil -} - -func (s *SQLStore) CompleteIndexingJob(ctx context.Context, documentID string) error { - _, err := s.db.ExecContext(ctx, `DELETE FROM indexing_jobs WHERE document_id = ?`, documentID) - if err != nil { - return fmt.Errorf("complete indexing job: %w", err) - } - return nil -} - -func (s *SQLStore) RetryIndexingJob(ctx context.Context, documentID string, nextScheduledAt string, lastError string) error { - now := time.Now().UTC().Format(time.RFC3339) - _, err := s.db.ExecContext(ctx, ` - UPDATE indexing_jobs - SET status = 'pending', - attempts = attempts + 1, - last_error = ?, - scheduled_at = ?, - updated_at = ? - WHERE document_id = ?`, - lastError, nextScheduledAt, now, documentID, - ) - if err != nil { - return fmt.Errorf("retry indexing job: %w", err) - } - return nil -} - func (s *SQLStore) GetFollowSubjects(ctx context.Context, did string) ([]string, error) { rows, err := s.db.QueryContext(ctx, ` SELECT DISTINCT repo_did diff --git a/packages/api/internal/store/store.go b/packages/api/internal/store/store.go index e1d42df..fc6c91d 100644 --- a/packages/api/internal/store/store.go +++ b/packages/api/internal/store/store.go @@ -49,11 +49,15 @@ type IndexingJob struct { RKey string CID string RecordJSON string + Source string Attempts int Status string LastError string ScheduledAt string UpdatedAt string + LeaseOwner string + LeaseUntil string + CompletedAt string } // JetstreamEvent is a cached JetStream activity event. @@ -86,6 +90,7 @@ type IndexingJobInput struct { RKey string CID string RecordJSON string + Source string } // DocumentFilter scopes a ListDocuments query to a subset of documents. @@ -110,10 +115,16 @@ type Store interface { UpdateRecordState(ctx context.Context, subjectURI string, state string) error UpsertIdentityHandle(ctx context.Context, did, handle string, isActive bool, status string) error GetIdentityHandle(ctx context.Context, did string) (string, error) + GetIndexingJob(ctx context.Context, documentID string) (*IndexingJob, error) EnqueueIndexingJob(ctx context.Context, input IndexingJobInput) error - ClaimIndexingJob(ctx context.Context) (*IndexingJob, error) + ClaimIndexingJob(ctx context.Context, workerID string, leaseUntil string) (*IndexingJob, error) CompleteIndexingJob(ctx context.Context, documentID string) error RetryIndexingJob(ctx context.Context, documentID string, nextScheduledAt string, lastError string) error + FailIndexingJob(ctx context.Context, documentID string, status string, lastError string) error + ListIndexingJobs(ctx context.Context, filter IndexingJobFilter) ([]*IndexingJob, error) + GetIndexingJobStats(ctx context.Context) (*IndexingJobStats, error) + AppendIndexingAudit(ctx context.Context, input IndexingAuditInput) error + ListIndexingAudit(ctx context.Context, filter IndexingAuditFilter) ([]*IndexingAuditEntry, error) GetFollowSubjects(ctx context.Context, did string) ([]string, error) GetRepoCollaborators(ctx context.Context, repoOwnerDID string) ([]string, error) CountDocuments(ctx context.Context) (int64, error) diff --git a/packages/api/internal/store/store_test.go b/packages/api/internal/store/store_test.go index b8d4243..473b14c 100644 --- a/packages/api/internal/store/store_test.go +++ b/packages/api/internal/store/store_test.go @@ -4,7 +4,9 @@ import ( "context" "os" "path/filepath" + "sync" "testing" + "time" "tangled.org/desertthunder.dev/twister/internal/store" ) @@ -383,6 +385,7 @@ func TestIntegration(t *testing.T) { RKey: "repo1", CID: "cid-repo1", RecordJSON: `{"name":"repo1"}`, + Source: store.IndexSourceReadThrough, } if err := st.EnqueueIndexingJob(ctx, job); err != nil { t.Fatalf("enqueue indexing job: %v", err) @@ -391,7 +394,7 @@ func TestIntegration(t *testing.T) { t.Fatalf("enqueue indexing job second call: %v", err) } - claimed, err := st.ClaimIndexingJob(ctx) + claimed, err := st.ClaimIndexingJob(ctx, "worker-a", time.Now().Add(time.Minute).Format(time.RFC3339)) if err != nil { t.Fatalf("claim indexing job: %v", err) } @@ -406,7 +409,7 @@ func TestIntegration(t *testing.T) { t.Fatalf("retry indexing job: %v", err) } - none, err := st.ClaimIndexingJob(ctx) + none, err := st.ClaimIndexingJob(ctx, "worker-b", time.Now().Add(time.Minute).Format(time.RFC3339)) if err != nil { t.Fatalf("claim delayed indexing job: %v", err) } @@ -418,7 +421,7 @@ func TestIntegration(t *testing.T) { t.Fatalf("retry indexing job now: %v", err) } - claimed, err = st.ClaimIndexingJob(ctx) + claimed, err = st.ClaimIndexingJob(ctx, "worker-c", time.Now().Add(time.Minute).Format(time.RFC3339)) if err != nil { t.Fatalf("claim retried indexing job: %v", err) } @@ -430,12 +433,94 @@ func TestIntegration(t *testing.T) { t.Fatalf("complete indexing job: %v", err) } - claimed, err = st.ClaimIndexingJob(ctx) + claimed, err = st.ClaimIndexingJob(ctx, "worker-d", time.Now().Add(time.Minute).Format(time.RFC3339)) if err != nil { t.Fatalf("claim after complete: %v", err) } if claimed != nil { t.Fatalf("expected no job after complete, got %#v", claimed) } + + got, err := st.GetIndexingJob(ctx, job.DocumentID) + if err != nil { + t.Fatalf("get completed job: %v", err) + } + if got == nil || got.Status != store.IndexingJobCompleted { + t.Fatalf("expected completed job row, got %#v", got) + } + }) + + t.Run("indexing claim is single winner", func(t *testing.T) { + job := store.IndexingJobInput{ + DocumentID: "did:plc:owner|sh.tangled.repo|repo2", + DID: "did:plc:owner", + Collection: "sh.tangled.repo", + RKey: "repo2", + CID: "cid-repo2", + RecordJSON: `{"name":"repo2"}`, + Source: store.IndexSourceReadThrough, + } + if err := st.EnqueueIndexingJob(ctx, job); err != nil { + t.Fatalf("enqueue job: %v", err) + } + + var wg sync.WaitGroup + results := make(chan *store.IndexingJob, 2) + errs := make(chan error, 2) + for _, worker := range []string{"worker-1", "worker-2"} { + wg.Add(1) + go func(worker string) { + defer wg.Done() + claimed, err := st.ClaimIndexingJob( + ctx, worker, time.Now().Add(time.Minute).Format(time.RFC3339), + ) + errs <- err + results <- claimed + }(worker) + } + wg.Wait() + close(results) + close(errs) + + claimedCount := 0 + for err := range errs { + if err != nil { + t.Fatalf("claim concurrent job: %v", err) + } + } + for claimed := range results { + if claimed != nil { + claimedCount++ + } + } + if claimedCount != 1 { + t.Fatalf("expected exactly one claimant, got %d", claimedCount) + } + }) + + t.Run("indexing audit and stats", func(t *testing.T) { + if err := st.AppendIndexingAudit(ctx, store.IndexingAuditInput{ + Source: store.IndexSourceReadThrough, + DocumentID: "doc-audit", + Collection: "sh.tangled.repo", + CID: "cid-audit", + Decision: "enqueued", + }); err != nil { + t.Fatalf("append indexing audit: %v", err) + } + stats, err := st.GetIndexingJobStats(ctx) + if err != nil { + t.Fatalf("get indexing stats: %v", err) + } + if stats.Completed < 1 { + t.Fatalf("expected completed jobs in stats, got %#v", stats) + } + entries, err := st.ListIndexingAudit(ctx, store.IndexingAuditFilter{DocumentID: "doc-audit"}) + if err != nil { + t.Fatalf("list indexing audit: %v", err) + } + if len(entries) != 1 || entries[0].Decision != "enqueued" { + t.Fatalf("unexpected audit rows: %#v", entries) + } }) } diff --git a/packages/api/overview.md b/packages/api/overview.md deleted file mode 100644 index 3e35ee2..0000000 --- a/packages/api/overview.md +++ /dev/null @@ -1,144 +0,0 @@ -# Twister Backend Overview - -Twister is a Go backend service that indexes Tangled network content and serves -search queries for the Twisted mobile client. It complements Tangled's public -APIs by providing global search and derived data that's hard to compute client-side. - -## Core Responsibilities - -1. **Real-time Indexing**: Consumes record changes from Tap (Tangled's event - stream) and indexes them into SQLite FTS5 for full-text search. - -2. **Search API**: Exposes HTTP endpoints for keyword search over repos, - profiles, issues, pull requests, and follows. - -3. **Graph Augmentation**: Caches follower relationships and provides profile - summaries not easily derivable from the public knot/PDS APIs. - -## Architecture - -```text -┌─────────────┐ ┌─────────────┐ ┌─────────────┐ -│ Tap │────▶│ Indexer │────▶│ Turso │ -│ (event stream) │ (ingest) │ │ (SQLite) │ -└─────────────┘ └─────────────┘ └──────┬──────┘ - │ -┌─────────────┐ ┌─────────────┐ │ -│ Twisted │────▶│ API │◀───────────┘ -│ (mobile) │ │ (search) │ -└─────────────┘ └─────────────┘ -``` - -## Commands - -The `twister` binary provides six subcommands: - -- `api` / `serve`: HTTP server for search and proxy endpoints -- `indexer`: Tap consumer that processes record events into the index -- `backfill`: Discovers users from seeds and registers repos with Tap -- `reindex`: Rebuilds the FTS index from existing documents -- `enrich`: Backfills derived fields (handles, URLs) on existing documents -- `healthcheck`: One-shot health probe - -## Data Flow - -### Indexing Pipeline (internal/ingest) - -1. **Tap Consumer** (`internal/tapclient`): Connects to Tap WebSocket, reads - record events with cursor-based resume. - -2. **Normalizers** (`internal/normalize`): Collection-specific adapters convert - raw records into normalized `Document` structs. Each collection (repos, - issues, PRs, follows, profiles) has a dedicated adapter. - -3. **Store** (`internal/store`): Upserts documents into SQLite with FTS5 index. - Tracks cursor position in `sync_state` table for resume-after-restart. - -### Search (internal/search) - -- **Keyword Search**: SQLite FTS5 full-text search with BM25 ranking -- **Filters**: By collection, author, repo, language, date range, state -- **Results**: Denormalized documents with snippet highlighting - -## Key Packages - -| Package | Purpose | -| ------------------------ | ------------------------------------------ | -| `internal/api` | HTTP router, handlers, middleware | -| `internal/store` | Database access layer (Turso/local SQLite) | -| `internal/ingest` | Tap event processing pipeline | -| `internal/normalize` | Record adapters for each collection type | -| `internal/search` | FTS5 query execution | -| `internal/backfill` | Graph discovery and Tap registration | -| `internal/xrpc` | XRPC client for Tangled PDS/knot APIs | -| `internal/constellation` | External service for graph queries | - -## API Endpoints - -### Search - -- `GET /search` - Unified search across all collections -- `GET /search/keyword` - FTS5 keyword search - -### Actors & Repos - -- `GET /actors/{handle}` - Profile data -- `GET /actors/{handle}/repos` - User's repositories -- `GET /actors/{handle}/repos/{repo}` - Repo details (tree, log, branches) - -### Issues & Pulls - -- `GET /issues/{handle}/{rkey}` - Issue detail -- `GET /pulls/{handle}/{rkey}` - Pull request detail - -### Identity - -- `GET /identity/resolve` - Resolve handle to DID -- `GET /identity/did/{did}` - DID document lookup - -### Proxy - -- `GET /xrpc/knot/{knot}/{nsid}` - Proxy to Tangled knot -- `GET /xrpc/pds/{pds}/{nsid}` - Proxy to PDS -- `GET /xrpc/bsky/{nsid}` - Proxy to Bluesky - -## Database Schema - -Core tables in Turso/libSQL: - -- `documents`: Denormalized search documents (title, body, metadata) -- `documents_fts`: FTS5 virtual table for full-text search -- `identity_handles`: DID → handle mapping with active status -- `record_state`: Issue/PR state (open/closed/merged) -- `sync_state`: Cursor tracking for Tap consumer resume -- `indexing_jobs`: Queue for async read-through indexing - -## Configuration - -Environment variables (loaded by `internal/config`): - -| Variable | Purpose | -| --------------------- | ------------------------------------------ | -| `HTTP_BIND_ADDR` | API server address (default `:8080`) | -| `TURSO_DATABASE_URL` | libsql:// URL for Turso or file: for local | -| `TURSO_AUTH_TOKEN` | Turso auth token (empty for local) | -| `TAP_URL` | Tap WebSocket URL (required for indexer) | -| `TAP_AUTH_PASSWORD` | Tap authentication password | -| `INDEXED_COLLECTIONS` | Comma-separated allowlist (empty = all) | - -## Local Development - -```bash -# API only (uses local SQLite file) -just api-dev - -# Indexer only -just api-run-indexer - -# Both in parallel (three terminals) -pnpm dev # Frontend -pnpm api:run:api # API -pnpm api:run:indexer # Indexer -``` - -For remote Turso, use `--remote` flag or set `TURSO_DATABASE_URL` in `.env`. diff --git a/packages/scripts/api/README.md b/packages/scripts/api/README.md deleted file mode 100644 index c2e561a..0000000 --- a/packages/scripts/api/README.md +++ /dev/null @@ -1,24 +0,0 @@ -# Twister API Smoke Checks - -Python smoke checks for Twister API endpoints, managed with uv. - -## Usage - -From the repo root: - -```sh -# Run all -uv run --project packages/scripts/api twister-api-smoke -# Run specific checks (healthz | readyz | search | documents | indexing | activity) -uv run --project packages/scripts/api twister-api-smoke --check healthz -``` - -## Options - -- `--verbose` for detailed output of API responses (JSON) -- `--base-url` (or env `TWISTER_API_BASE_URL`, default `http://localhost:8080`) -- `--query` for search check (default `twisted`) -- `--document-id` for documents check -- `--actor-handle` for indexing check (default `desertthunder.dev`) -- `--repo-at-uri` for repo fixture indexing/search checks -- `--profile-at-uri` for profile fixture indexing/search checks diff --git a/packages/scripts/api/pyproject.toml b/packages/scripts/api/pyproject.toml index 4f8e45e..584792a 100644 --- a/packages/scripts/api/pyproject.toml +++ b/packages/scripts/api/pyproject.toml @@ -2,7 +2,6 @@ name = "twister-api-smoke" version = "0.1.0" description = "Smoke checks for Twister API endpoints" -readme = "README.md" requires-python = ">=3.11" dependencies = [] diff --git a/packages/scripts/api/src/twister_api_smoke/cli.py b/packages/scripts/api/src/twister_api_smoke/cli.py index 601c3bb..9d9d24f 100644 --- a/packages/scripts/api/src/twister_api_smoke/cli.py +++ b/packages/scripts/api/src/twister_api_smoke/cli.py @@ -1,3 +1,33 @@ +"""Twister API Smoke Checks. + + +Python smoke checks for Twister API endpoints, managed with uv. + +Usage +----- + +From the repo root: + +.. code:: sh + + # Run all + uv run --project packages/scripts/api twister-api-smoke + # Run specific checks (healthz | readyz | search | documents | indexing | admin | activity) + uv run --project packages/scripts/api twister-api-smoke --check healthz + +Options +------- + +- ``--verbose`` for detailed output of API responses (JSON) +- ``--base-url`` (or env ``TWISTER_API_BASE_URL``, default ``http://localhost:8080``) +- ``--query`` for search check (default ``twisted``) +- ``--document-id`` for documents check +- ``--actor-handle`` for indexing check (default ``desertthunder.dev``) +- ``--repo-at-uri`` for repo fixture indexing/search checks +- ``--profile-at-uri`` for profile fixture indexing/search checks +- ``--admin-token`` (or env ``ADMIN_AUTH_TOKEN``) for admin smoke checks +""" + import argparse import enum import json @@ -44,11 +74,20 @@ class Options: actor_handle: str repo_at_uri: str profile_at_uri: str + admin_token: str verbose: bool -def http_get_status(url: str) -> int: +def auth_headers(token: str) -> dict[str, str]: + if not token: + return {} + return {"Authorization": f"Bearer {token}"} + + +def http_get_status(url: str, headers: dict[str, str] | None = None) -> int: req = Request(url, method="GET") + for key, value in (headers or {}).items(): + req.add_header(key, value) try: with urlopen(req, timeout=10) as resp: return resp.status @@ -58,13 +97,19 @@ def http_get_status(url: str) -> int: fail(f"request failed for {url}: {err}") -def http_get_json(url: str, params: dict[str, str] | None = None) -> Any: +def http_get_json( + url: str, + params: dict[str, str] | None = None, + headers: dict[str, str] | None = None, +) -> Any: if params: query = urlencode(params) sep = "&" if "?" in url else "?" url = f"{url}{sep}{query}" req = Request(url, method="GET") + for key, value in (headers or {}).items(): + req.add_header(key, value) try: with urlopen(req, timeout=15) as resp: payload = resp.read().decode("utf-8") @@ -208,6 +253,22 @@ def check_documents(opts: Options) -> None: echo("documents ok") +def resolve_repo_name(opts: Options) -> str: + payload = http_get_json(urljoin(opts.base_url, f"/actors/{opts.actor_handle}/repos")) + records = payload.get("records") if isinstance(payload, dict) else None + if not isinstance(records, list): + fail("repo listing payload is missing records") + for record in records: + if not isinstance(record, dict): + continue + if record.get("uri") != opts.repo_at_uri: + continue + value = record.get("value") + if isinstance(value, dict) and isinstance(value.get("name"), str): + return value["name"] + fail(f"repo fixture uri not found in /actors/{opts.actor_handle}/repos: {opts.repo_at_uri}") + + def check_indexing(opts: Options) -> None: repo_id = at_uri_to_document_id(opts.repo_at_uri) profile_id = at_uri_to_document_id(opts.profile_at_uri) @@ -220,8 +281,9 @@ def check_indexing(opts: Options) -> None: echo(f"triggering read-through fetch via /actors/{opts.actor_handle}") assert_status(urljoin(opts.base_url, f"/actors/{opts.actor_handle}"), 200) - echo(f"triggering read-through fetch via /actors/{opts.actor_handle}/repos") - assert_status(urljoin(opts.base_url, f"/actors/{opts.actor_handle}/repos"), 200) + repo_name = resolve_repo_name(opts) + echo(f"triggering read-through fetch via /actors/{opts.actor_handle}/repos/{repo_name}") + assert_status(urljoin(opts.base_url, f"/actors/{opts.actor_handle}/repos/{repo_name}"), 200) echo(f"waiting for queued indexing of profile fixture {profile_id}") for _ in range(30): @@ -247,6 +309,20 @@ def check_indexing(opts: Options) -> None: fail(f"repo fixture did not become available at /documents/{repo_encoded} within 30s") +def check_admin(opts: Options) -> None: + if not opts.admin_token: + fail("admin check requires --admin-token or ADMIN_AUTH_TOKEN") + echo("checking GET /admin/status") + payload = http_get_json( + urljoin(opts.base_url, "/admin/status"), + headers=auth_headers(opts.admin_token), + ) + if not isinstance(payload, dict) or "read_through" not in payload: + fail("admin status response is missing read_through data") + maybe_log_json(opts, "admin-status", payload) + echo("admin ok") + + def check_activity(opts: Options) -> None: echo("checking websocket handshake on /activity/stream") parsed = urlparse(opts.base_url) @@ -285,6 +361,7 @@ CHECKS: dict[str, Any] = { "search": check_search, "documents": check_documents, "indexing": check_indexing, + "admin": check_admin, "activity": check_activity, } @@ -321,6 +398,11 @@ def parse_args(argv: list[str]) -> Options: default="at://did:plc:xg2vq45muivyy3xwatcehspu/sh.tangled.actor.profile/self", help="Profile AT URI expected to be fetched and indexed by smoke checks", ) + parser.add_argument( + "--admin-token", + default=os.environ.get("ADMIN_AUTH_TOKEN", ""), + help="Bearer token for admin smoke checks", + ) parser.add_argument( "--verbose", action="store_true", help="Print JSON payloads returned by smoke endpoints" ) @@ -335,6 +417,7 @@ def parse_args(argv: list[str]) -> Options: actor_handle=ns.actor_handle, repo_at_uri=ns.repo_at_uri, profile_at_uri=ns.profile_at_uri, + admin_token=ns.admin_token, verbose=ns.verbose, ) @@ -342,14 +425,10 @@ def parse_args(argv: list[str]) -> Options: def main(argv: list[str] | None = None) -> int: opts = parse_args(sys.argv[1:] if argv is None else argv) if opts.check == "all": - for name in ( - "healthz", - "readyz", - "indexing", - "search", - "documents", - "activity", - ): + checks = ["healthz", "readyz", "indexing", "search", "documents", "activity"] + if opts.admin_token: + checks.append("admin") + for name in checks: CHECKS[name](opts) echo("all API smoke checks passed") return 0