diff --git a/README.md b/README.md index 78211f1..21bd59a 100644 --- a/README.md +++ b/README.md @@ -19,17 +19,9 @@ That keeps direct browsing honest while giving the client one place to ask for c ## Development -Use the top-level [`justfile`](justfile) for common workflows: +Use the top-level [`justfile`](justfile) for common workflows (`just --list` to view) -```bash -just dev -just build -just test -just api-run-api -``` - -The committed `apps/twisted/.env` points at production. Use `apps/twisted/.env.local` -for machine-local overrides such as a localhost API or OAuth callback. +Use `apps/twisted/.env.local` for machine-local overrides such as a localhost API or OAuth callback. ## Run Locally @@ -104,94 +96,8 @@ VITE_TWISTER_API_BASE_URL=http://localhost:8080 Dev builds keep the current OAuth flow available. Production builds are read-only and hide auth entry points for now. -### 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 - -Use one Turso database per environment, for example: - -- `twister-dev` -- `twister-prod` - -Do not introduce separate app variable names for dev and prod. Always use the same variables: - -- `TURSO_DATABASE_URL` -- `TURSO_AUTH_TOKEN` - -Only the values change per environment. - -Example: - -```bash -# Development -TURSO_DATABASE_URL=libsql://twister-dev-your-org.turso.io -TURSO_AUTH_TOKEN=... - -# Production -TURSO_DATABASE_URL=libsql://twister-prod-your-org.turso.io -TURSO_AUTH_TOKEN=... -``` - -### Railway - -Create or reuse one Railway project containing: - -- existing `tap` -- `api` running `twister api` -- `indexer` running `twister indexer` - -Set these shared variables on the Railway services: - -- `TURSO_DATABASE_URL` -- `TURSO_AUTH_TOKEN` -- `LOG_LEVEL` -- `LOG_FORMAT` - -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: - -- `TAP_URL` -- `TAP_AUTH_PASSWORD` -- `INDEXED_COLLECTIONS` - -If you use separate Railway environments for dev and prod, keep the same variable names in both and only swap the Turso values. - -### First Bootstrap - -For a brand-new environment: +## Attributions -1. Point `TURSO_DATABASE_URL` and `TURSO_AUTH_TOKEN` at the target database. -2. Deploy `api` and `indexer` on Railway. -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. +This project relies heavily on the work of the [Tangled team](https://tangled.org/tangled.org) (duh) +and the infrastructure made available by [microcosm](https://microcosm.blue), specifically +Lightrail and Constellation. diff --git a/docs/README.md b/docs/README.md index 64878c7..8ff537a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -8,6 +8,7 @@ Completed work — architecture, APIs, and data models as built. - [`reference/app.md`](reference/app.md) — Ionic Vue mobile app - [`reference/deployment-walkthrough.md`](reference/deployment-walkthrough.md) — Railway deployment guide - [`reference/lexicons.md`](reference/lexicons.md) — Tangled AT Protocol record types +- [`reference/metrics.md`](reference/metrics.md) — Railway and Turso usage checks after deploy - [`reference/resync.md`](reference/resync.md) — Backfill and repo-resync recovery playbook ## Specs diff --git a/docs/reference/api.md b/docs/reference/api.md index 89c7664..14359a3 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -159,7 +159,10 @@ The built-in XRPC client provides typed access to AT Protocol endpoints with cac ## Backfill -The backfill command discovers users from a seed file and registers them with Tap for indexing. Discovery fans out via follow graphs and repo collaborators up to a configurable hop depth (default 2). Supports dry-run mode, configurable concurrency and batch sizes, and is idempotent. +The backfill command now defaults to `--source lightrail`: it calls +`com.atproto.sync.listReposByCollection`, dedupes returned DIDs, and batch +submits them to Tap. `--source graph` keeps the older seed-file follow and +collaborator crawl for targeted fallback runs. ## Configuration @@ -185,12 +188,16 @@ All configuration is via environment variables (with `.env` file support): | `PLC_DIRECTORY_URL` | `https://plc.directory` | PLC Directory | | `XRPC_TIMEOUT` | 15s | XRPC HTTP timeout | +Recommended production practice is to use explicit search-relevant collection +lists for `INDEXED_COLLECTIONS` and `READ_THROUGH_COLLECTIONS`, not +`sh.tangled.*`, and to leave `sh.tangled.graph.follow` out of both. + ## Deployment Deployed on Railway with three services: -- **api** — HTTP server (port 8080, health at `/healthz`) -- **indexer** — Tap consumer (health at `:9090/healthz`) +- **api** — HTTP server (port 8080, health at `/readyz`) +- **indexer** — Tap consumer (health at `:9090/health`) - **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. diff --git a/docs/reference/deployment-walkthrough.md b/docs/reference/deployment-walkthrough.md index aee55ee..408716f 100644 --- a/docs/reference/deployment-walkthrough.md +++ b/docs/reference/deployment-walkthrough.md @@ -23,7 +23,6 @@ Before you start, have these ready: - a Railway account and the Railway CLI - a Turso database URL and auth token - a Tap URL and Tap auth password -- a seed list for the first backfill run From this machine: ```sh @@ -68,7 +67,7 @@ Set these on `api`: - `SEARCH_DEFAULT_LIMIT=20` - `SEARCH_MAX_LIMIT=100` - `READ_THROUGH_MODE=missing` -- `READ_THROUGH_COLLECTIONS=sh.tangled.*` +- `READ_THROUGH_COLLECTIONS=` - `READ_THROUGH_MAX_ATTEMPTS=5` - `ENABLE_ADMIN_ENDPOINTS=false` - `ADMIN_AUTH_TOKEN=` @@ -76,8 +75,10 @@ Set these on `indexer`: - `INDEXER_HEALTH_ADDR=0.0.0.0:${{ PORT }}` - `TAP_URL=` - `TAP_AUTH_PASSWORD=` -- `INDEXED_COLLECTIONS=sh.tangled.*` +- `INDEXED_COLLECTIONS=` - `ENABLE_INGEST_ENRICHMENT=true` +Do not use `sh.tangled.*` for those allowlists. Match the Lightrail-backed +search collection set and leave `sh.tangled.graph.follow` out. Optional OAuth variables for a Railway-hosted web client metadata endpoint: - `OAUTH_CLIENT_ID` - `OAUTH_REDIRECT_URIS` @@ -114,12 +115,13 @@ A fresh environment is not search-ready just because the services booted. 3. Confirm the `api` domain returns `200` from `/readyz`. 4. Confirm the `indexer` returns `200` from `/health`. 5. Run the initial backfill against the same Turso and Tap environment. -One simple way to run backfill from this machine is to use the same env values -locally and execute: +Use Railway shell so the command runs inside the live `indexer` environment: ```sh -cd /Users/owais/Projects/Twisted/packages/api -go run ./main.go backfill --seeds /path/to/seeds.txt +cd /Users/owais/Projects/Twisted +railway link # Select indexer service if prompted +railway shell +twister backfill --source lightrail ``` Do not call the environment ready until that first backfill has completed. @@ -139,12 +141,3 @@ pnpm --dir apps/twisted dev pnpm --dir apps/twisted build pnpm --dir apps/twisted exec cap sync ``` - -## Operating Model - -This is the practical split: - -- Railway hosts the always-on backend -- Turso stores indexed data -- this machine, or CI, builds the mobile app and points it at Railway -If you later want a Railway-hosted web frontend, add that as a separate service. diff --git a/docs/reference/metrics.md b/docs/reference/metrics.md new file mode 100644 index 0000000..1ff14a5 --- /dev/null +++ b/docs/reference/metrics.md @@ -0,0 +1,126 @@ +# Metrics To Watch + +Use this after deploying the Lightrail-backed backfill flow and detail-only +read-through changes. + +## Goal + +Confirm that: + +- the API stops creating broad read-through churn during browse traffic +- the indexer still keeps search current through Tap +- bootstrap backfills become cheaper and more predictable + +## Railway + +Watch both `api` and `indexer` for 24 to 48 hours after deploy. + +### API service + +Expected direction: + +- lower average CPU +- fewer latency spikes on browse-heavy endpoints +- lower memory churn from fewer queued background jobs + +Useful checks: + +- CPU usage before and after deploy +- memory usage before and after deploy +- request latency for browse-heavy periods +- restart count + +If this change is helping, the API should look flatter under normal browsing, +especially when clients hit repo lists, issue lists, pull lists, or follows. + +### Indexer service + +Expected direction: + +- similar steady-state load during normal Tap ingest +- shorter, more deliberate spikes only when `twister backfill` is run + +Useful checks: + +- CPU during normal operation +- CPU during `twister backfill --source lightrail` +- memory during backfill +- restart count + +The indexer may still spike during an initial bootstrap. That is expected. The +important change is that the API should stop causing constant incidental work. + +## Turso + +This is where the clearest savings should show up. + +Expected direction: + +- fewer write operations +- fewer row updates in indexing job tables +- lower write amplification from browse traffic + +Useful checks: + +- total row writes +- total queries +- write-heavy windows during normal app usage +- latency on write statements if you have it + +The main reduction should come from no longer enqueueing whole list responses +into `indexing_jobs` during browse requests. + +## Twister Admin Signals + +If admin endpoints are enabled, compare these before and after deploy: + +- `read_through.pending` +- `read_through.processing` +- `read_through.failed` +- `read_through.dead_letter` +- `read_through.last_processed_at` + +Healthy post-change behavior: + +- pending stays near zero most of the time +- processing only bumps when detail pages fetch missing records +- failed and dead-letter counts grow slowly, not continuously + +Relevant endpoint: + +```sh +curl -H "Authorization: Bearer $ADMIN_AUTH_TOKEN" http:///admin/status +``` + +## What To Compare + +Use the same day-of-week and similar traffic windows if possible. + +Good comparisons: + +- 24 hours before deploy vs 24 hours after deploy +- one browse-heavy period before vs after +- one bootstrap backfill run before vs after + +## Success Signals + +Treat the rollout as successful if most of these are true: + +- API CPU is lower or less spiky under normal browsing +- Turso writes drop during browse-heavy traffic +- read-through queue counts stay close to zero most of the time +- backfill runs complete with fewer upstream calls and cleaner batching +- search freshness still tracks Tap ingest without visible regressions + +## Failure Signals + +Investigate if you see any of these: + +- search misses rise after deploy +- detail pages repeatedly enqueue the same records +- `read_through.pending` grows and does not drain +- indexer CPU stays elevated long after a bootstrap run +- Turso writes do not drop despite the handler changes + +If that happens, inspect Tap coverage first, then spot-check whether operators +ran `twister backfill --source lightrail` for the environment. diff --git a/docs/reference/resync.md b/docs/reference/resync.md index 0513e44..e9d6740 100644 --- a/docs/reference/resync.md +++ b/docs/reference/resync.md @@ -1,6 +1,6 @@ --- title: Backfill & Resync Playbook -updated: 2026-03-25 +updated: 2026-03-26 --- Twister's search index has three recovery paths. Choose based on what broke. @@ -39,19 +39,23 @@ Persists cursor to `sync_state` table under consumer name `indexer-tap-v1`. ### `twister backfill` -Discovers users via follow graph from seed DIDs/handles, checks Tap status for -each, and registers untracked repos with Tap `/repos/add`. +Defaults to `--source lightrail`: discovers DIDs from +`com.atproto.sync.listReposByCollection` and submits them to Tap in batches. +Use `--source graph` only for targeted fallback seeding from handles or DIDs. ```sh -# dry-run first -twister backfill --seeds seeds.txt --max-hops 2 --dry-run +# full-network dry-run first +twister backfill --dry-run -# real run -twister backfill --seeds seeds.txt --max-hops 2 \ +# full-network bootstrap +twister backfill + +# targeted fallback +twister backfill --source graph --seeds seeds.txt --max-hops 2 \ --concurrency 5 --batch-size 10 --batch-delay 1s ``` -Safe to re-run. Discovery deduplicates and `repos/add` is idempotent. +Safe to re-run. Discovery deduplicates and `repos/add` is treated as idempotent. ### `twister reindex` @@ -106,7 +110,7 @@ but does not appear in `/search`, the document was never indexed. 1. Check if the DID is tracked by Tap. If not, run `backfill`: ```sh - twister backfill --seeds --max-hops 0 + twister backfill --source graph --seeds --max-hops 0 ``` 2. Once Tap is tracking the DID, the `indexer` will deliver historical events. @@ -135,8 +139,8 @@ full-loss event. 2. Register repos with Tap: ```sh - twister backfill --seeds seeds.txt --max-hops 2 --dry-run - twister backfill --seeds seeds.txt --max-hops 2 + twister backfill --dry-run + twister backfill ``` 3. Start the indexer and let it consume: `twister indexer` diff --git a/docs/todo.md b/docs/todo.md index 9dc2b45..b89a506 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -1,6 +1,6 @@ --- title: Parking Lot -updated: 2026-03-25 +updated: 2026-03-26 --- Search stabilization is active roadmap work now, not parking-lot work. @@ -8,4 +8,3 @@ Search stabilization is active roadmap work now, not parking-lot work. Still parked: - Semantic and hybrid search stay deferred until local-only storage, smoke tests, read-through indexing, and the JetStream cache are stable. -- Revisit `com.atproto.sync.listReposByCollection` as a complementary backfill discovery source after the Tap-driven indexing path is reliable. diff --git a/packages/api/internal/api/actors.go b/packages/api/internal/api/actors.go index fabf99a..fd02650 100644 --- a/packages/api/internal/api/actors.go +++ b/packages/api/internal/api/actors.go @@ -90,7 +90,6 @@ func (s *Server) resolveRepo(r *http.Request, handleOrDID, repoName string) (*re if err != nil { return nil, fmt.Errorf("list repos for %s: %w", actor.DID, err) } - s.enqueueXRPCList(r.Context(), entries) for _, entry := range entries { name, _ := entry.Value["name"].(string) @@ -230,7 +229,6 @@ func (s *Server) handleListActorRepos(w http.ResponseWriter, r *http.Request) { s.actorError(w, err) return } - s.enqueueXRPCList(r.Context(), entries) records := make([]recordEntry, len(entries)) for i, e := range entries { @@ -432,7 +430,6 @@ func (s *Server) handleRepoIssues(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch issues")) return } - s.enqueueXRPCList(r.Context(), issues) var records []issueEntry for _, e := range issues { @@ -470,7 +467,6 @@ func (s *Server) handleRepoPulls(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch pulls")) return } - s.enqueueXRPCList(r.Context(), pulls) var records []pullEntry for _, e := range pulls { @@ -509,7 +505,6 @@ func (s *Server) handleActorIssues(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch issues")) return } - s.enqueueXRPCList(r.Context(), issues) records := make([]issueEntry, len(issues)) for i, e := range issues { @@ -540,7 +535,6 @@ func (s *Server) handleActorPulls(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch pulls")) return } - s.enqueueXRPCList(r.Context(), pulls) records := make([]pullEntry, len(pulls)) for i, e := range pulls { @@ -571,7 +565,6 @@ func (s *Server) handleActorFollowing(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch follows")) return } - s.enqueueXRPCList(r.Context(), entries) records := make([]recordEntry, len(entries)) for i, e := range entries { @@ -599,7 +592,6 @@ func (s *Server) handleActorStrings(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch strings")) return } - s.enqueueXRPCList(r.Context(), entries) records := make([]recordEntry, len(entries)) for i, e := range entries { @@ -669,7 +661,6 @@ func (s *Server) handleIssueComments(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch comments")) return } - s.enqueueXRPCList(r.Context(), entries) var records []recordEntry for _, e := range entries { @@ -746,7 +737,6 @@ func (s *Server) handlePullComments(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch comments")) return } - s.enqueueXRPCList(r.Context(), entries) var records []recordEntry for _, e := range entries { diff --git a/packages/api/internal/api/readthrough_test.go b/packages/api/internal/api/readthrough_test.go index da34a28..3050970 100644 --- a/packages/api/internal/api/readthrough_test.go +++ b/packages/api/internal/api/readthrough_test.go @@ -67,7 +67,7 @@ func TestEnqueueXRPCRecordOnlyQueuesOncePerCID(t *testing.T) { } } -func TestHandleActorFollowingEnqueuesRecords(t *testing.T) { +func TestHandleActorFollowingDoesNotEnqueueRecords(t *testing.T) { var upstream *httptest.Server upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { @@ -112,16 +112,12 @@ func TestHandleActorFollowingEnqueuesRecords(t *testing.T) { if rec.Code != http.StatusOK { t.Fatalf("status: got %d body=%s", rec.Code, rec.Body.String()) } - if len(st.jobs) != 1 { - t.Fatalf("expected one queued follow job, got %#v", st.jobs) - } - job := st.jobs["did:plc:alice|sh.tangled.graph.follow|1"] - if job == nil { - t.Fatalf("expected follow indexing job, got %#v", st.jobs) + if len(st.jobs) != 0 { + t.Fatalf("expected no queued follow jobs, got %#v", st.jobs) } } -func TestHandleActorReposEnqueuesRecords(t *testing.T) { +func TestHandleActorReposDoesNotEnqueueRecords(t *testing.T) { var upstream *httptest.Server upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { @@ -167,15 +163,65 @@ func TestHandleActorReposEnqueuesRecords(t *testing.T) { 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 repo jobs, got %#v", st.jobs) + } +} + +func TestHandleGetActorEnqueuesRecord(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.getRecord": + _ = json.NewEncoder(w).Encode(map[string]any{ + "uri": "at://did:plc:alice/sh.tangled.actor.profile/self", + "cid": "cid-1", + "value": map[string]any{ + "$type": "sh.tangled.actor.profile", + "description": "hello", + }, + }) + 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}", srv.handleGetActor) + + req := httptest.NewRequest(http.MethodGet, "/actors/alice.tangled.org", 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) != 1 { - t.Fatalf("expected one queued repo job, got %#v", st.jobs) + t.Fatalf("expected one queued profile job, got %#v", st.jobs) } - job := st.jobs["did:plc:alice|sh.tangled.repo|repo1"] - if job == nil { - t.Fatalf("expected repo indexing job, got %#v", st.jobs) + if st.jobs["did:plc:alice|sh.tangled.actor.profile|self"] == nil { + t.Fatalf("expected profile indexing job, got %#v", st.jobs) } } diff --git a/packages/api/internal/backfill/backfill.go b/packages/api/internal/backfill/backfill.go index abc6a55..144f0b4 100644 --- a/packages/api/internal/backfill/backfill.go +++ b/packages/api/internal/backfill/backfill.go @@ -5,6 +5,7 @@ import ( "fmt" "log/slog" "sort" + "strings" "sync" "time" @@ -20,12 +21,13 @@ type discoveryStore interface { // Runner executes seed resolution, graph discovery, and Tap registration. type Runner struct { - store discoveryStore - tap tapAdmin - resolver handleResolver - follows followFetcher - profiles profileFetcher - log *slog.Logger + store discoveryStore + tap tapAdmin + resolver handleResolver + follows followFetcher + profiles profileFetcher + lightrail lightrailRepoLister + log *slog.Logger } func NewRunner(store discoveryStore, tap tapAdmin, xrpcClient *xrpc.Client, log *slog.Logger) *Runner { @@ -34,24 +36,29 @@ func NewRunner(store discoveryStore, tap tapAdmin, xrpcClient *xrpc.Client, log NewXRPCHandleResolver(xrpcClient), NewXRPCFollowFetcher(xrpcClient), NewXRPCProfileFetcher(xrpcClient), + NewHTTPLightrailClient(), log, ) } -func NewRunnerWithDeps(store discoveryStore, tap tapAdmin, resolver handleResolver, follows followFetcher, profiles profileFetcher, log *slog.Logger) *Runner { +func NewRunnerWithDeps( + store discoveryStore, tap tapAdmin, resolver handleResolver, + follows followFetcher, profiles profileFetcher, lightrail lightrailRepoLister, + log *slog.Logger, +) *Runner { if log == nil { log = slog.Default() } - return &Runner{store: store, tap: tap, resolver: resolver, follows: follows, profiles: profiles, log: log} + if lightrail == nil { + lightrail = NewHTTPLightrailClient() + } + return &Runner{ + store: store, tap: tap, resolver: resolver, follows: follows, + profiles: profiles, lightrail: lightrail, log: log, + } } func (r *Runner) Run(ctx context.Context, opts Options) error { - if opts.SeedsPath == "" { - return fmt.Errorf("--seeds is required") - } - if opts.MaxHops < 0 { - return fmt.Errorf("--max-hops must be >= 0") - } if opts.Concurrency <= 0 { opts.Concurrency = 5 } @@ -61,6 +68,30 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { if opts.BatchDelay < 0 { return fmt.Errorf("--batch-delay must be >= 0") } + if opts.PageLimit <= 0 { + opts.PageLimit = DefaultPageLimit + } + if strings.TrimSpace(opts.LightrailURL) == "" { + opts.LightrailURL = DefaultLightrailURL + } + + switch normalizeSource(opts.Source) { + case SourceLightrail: + return r.runLightrail(ctx, opts) + case SourceGraph: + return r.runGraph(ctx, opts) + default: + return fmt.Errorf("unsupported --source %q", opts.Source) + } +} + +func (r *Runner) runGraph(ctx context.Context, opts Options) error { + if opts.SeedsPath == "" { + return fmt.Errorf("--seeds is required for --source graph") + } + if opts.MaxHops < 0 { + return fmt.Errorf("--max-hops must be >= 0") + } seedEntries, err := parseSeedInput(opts.SeedsPath) if err != nil { @@ -75,6 +106,7 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { } r.log.Info("starting backfill discovery", + slog.String("source", SourceGraph), slog.Int("seed_count", len(seeds)), slog.Int("max_hops", opts.MaxHops), slog.Int("concurrency", opts.Concurrency), @@ -123,52 +155,10 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { slog.Int("to_submit", len(toSubmit)), ) - submitted := 0 - submitFailures := 0 - for i := 0; i < len(toSubmit); i += opts.BatchSize { - end := i + opts.BatchSize - if end > len(toSubmit) { - end = len(toSubmit) - } - batch := toSubmit[i:end] - if err := r.tap.AddRepos(ctx, batch); err != nil { - r.log.Warn("tap batch submission failed", - slog.Int("batch_start", i), - slog.Int("batch_end", end), - slog.Int("batch_size", len(batch)), - slog.String("error", err.Error()), - ) - for _, did := range batch { - if err := r.tap.AddRepos(ctx, []string{did}); err != nil { - submitFailures++ - r.log.Warn("tap repo submission failed", - slog.String("did", did), - slog.String("error", err.Error()), - ) - continue - } - submitted++ - r.log.Info("submitted Tap repo", slog.String("did", did), slog.Int("submitted_total", submitted)) - } - } else { - submitted += len(batch) - r.log.Info("submitted Tap batch", - slog.Int("batch_start", i), - slog.Int("batch_end", end), - slog.Int("batch_size", len(batch)), - slog.Int("submitted_total", submitted), - ) - } - if end < len(toSubmit) && opts.BatchDelay > 0 { - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(opts.BatchDelay): - } - } - } + submitted, submitFailures := r.submitDIDs(ctx, toSubmit, opts.BatchSize, opts.BatchDelay) r.log.Info("backfill complete", + slog.String("source", SourceGraph), slog.Int("discovered_total", len(discovered)), slog.Int("already_tracked", alreadyTracked), slog.Int("backfill_in_progress", inProgress), @@ -184,6 +174,53 @@ func (r *Runner) Run(ctx context.Context, opts Options) error { return nil } +func (r *Runner) runLightrail(ctx context.Context, opts Options) error { + collections := normalizeCollections(opts.Collections) + if len(collections) == 0 { + collections = append([]string(nil), DefaultCollections...) + } + + r.log.Info("starting backfill discovery", + slog.String("source", SourceLightrail), + slog.String("lightrail_url", opts.LightrailURL), + slog.Int("collection_count", len(collections)), + slog.Int("page_limit", opts.PageLimit), + ) + + dids, err := r.lightrail.ListReposByCollection( + ctx, opts.LightrailURL, collections, opts.PageLimit, + ) + if err != nil { + return fmt.Errorf("discover repos from lightrail: %w", err) + } + dids = normalizeDIDs(dids) + sort.Strings(dids) + discovered := make([]DiscoveredUser, 0, len(dids)) + for _, did := range dids { + discovered = append(discovered, DiscoveredUser{ + DID: did, Hop: 0, Source: strings.Join(collections, ","), Reason: "collection", + }) + } + + r.log.Info("discovery complete", + slog.String("source", SourceLightrail), + slog.Int("discovered_total", len(discovered)), + ) + if opts.DryRun { + r.log.Info("dry-run mode enabled; skipping Tap mutations") + return nil + } + + submitted, submitFailures := r.submitDIDs(ctx, dids, opts.BatchSize, opts.BatchDelay) + r.log.Info("backfill complete", + slog.String("source", SourceLightrail), + slog.Int("discovered_total", len(discovered)), + slog.Int("submitted", submitted), + slog.Int("submit_failures", submitFailures), + ) + return nil +} + // resolveSeeds returns (dids, did→handle map, error). The handle map contains // entries for seeds that were specified as handles rather than DIDs. func (r *Runner) resolveSeeds(ctx context.Context, entries []seedEntry) ([]string, map[string]string, error) { @@ -319,6 +356,98 @@ func (r *Runner) discover(ctx context.Context, seeds []string, maxHops int, conc return ordered, nil } +func (r *Runner) submitDIDs( + ctx context.Context, dids []string, batchSize int, batchDelay time.Duration, +) (int, int) { + submitted := 0 + submitFailures := 0 + for i := 0; i < len(dids); i += batchSize { + end := i + batchSize + if end > len(dids) { + end = len(dids) + } + batch := dids[i:end] + if err := r.tap.AddRepos(ctx, batch); err != nil { + r.log.Warn("tap batch submission failed", + slog.Int("batch_start", i), + slog.Int("batch_end", end), + slog.Int("batch_size", len(batch)), + slog.String("error", err.Error()), + ) + for _, did := range batch { + if err := r.tap.AddRepos(ctx, []string{did}); err != nil { + submitFailures++ + r.log.Warn("tap repo submission failed", + slog.String("did", did), + slog.String("error", err.Error()), + ) + continue + } + submitted++ + r.log.Info("submitted Tap repo", + slog.String("did", did), + slog.Int("submitted_total", submitted), + ) + } + } else { + submitted += len(batch) + r.log.Info("submitted Tap batch", + slog.Int("batch_start", i), + slog.Int("batch_end", end), + slog.Int("batch_size", len(batch)), + slog.Int("submitted_total", submitted), + ) + } + if end < len(dids) && batchDelay > 0 { + select { + case <-ctx.Done(): + return submitted, submitFailures + case <-time.After(batchDelay): + } + } + } + return submitted, submitFailures +} + +func normalizeSource(source string) string { + switch strings.ToLower(strings.TrimSpace(source)) { + case "", SourceLightrail: + return SourceLightrail + case SourceGraph: + return SourceGraph + default: + return strings.ToLower(strings.TrimSpace(source)) + } +} + +func normalizeCollections(collections []string) []string { + seen := make(map[string]bool) + normalized := make([]string, 0, len(collections)) + for _, collection := range collections { + collection = strings.TrimSpace(collection) + if collection == "" || seen[collection] { + continue + } + seen[collection] = true + normalized = append(normalized, collection) + } + return normalized +} + +func normalizeDIDs(dids []string) []string { + seen := make(map[string]bool) + normalized := make([]string, 0, len(dids)) + for _, did := range dids { + did = strings.TrimSpace(did) + if did == "" || seen[did] { + continue + } + seen[did] = true + normalized = append(normalized, did) + } + return normalized +} + // indexProfiles fetches sh.tangled.actor.profile records via XRPC for each // discovered user, persists the DID→handle mapping, and upserts a searchable // profile document. diff --git a/packages/api/internal/backfill/backfill_test.go b/packages/api/internal/backfill/backfill_test.go index 7eeb2a2..82fade9 100644 --- a/packages/api/internal/backfill/backfill_test.go +++ b/packages/api/internal/backfill/backfill_test.go @@ -95,6 +95,28 @@ func (f *fakeProfileFetcher) FetchProfile(_ context.Context, did string) (*Profi return &ProfileRecord{}, nil } +type fakeLightrailRepoLister struct { + dids []string + err error + calls int + baseURL string + collections []string + limit int +} + +func (f *fakeLightrailRepoLister) ListReposByCollection( + _ context.Context, baseURL string, collections []string, limit int, +) ([]string, error) { + f.calls++ + f.baseURL = baseURL + f.collections = append([]string(nil), collections...) + f.limit = limit + if f.err != nil { + return nil, f.err + } + return append([]string(nil), f.dids...), nil +} + func TestRunner_DiscoveryAndSubmit(t *testing.T) { st := &fakeStore{ collaborators: map[string][]string{ @@ -105,7 +127,11 @@ func TestRunner_DiscoveryAndSubmit(t *testing.T) { tap := &fakeTapAdmin{statuses: map[string]RepoStatus{"did:plc:f1": {Found: true, Tracked: true, Backfilled: true}}} resolver := &fakeResolver{mapping: map[string]string{"alice.tangled.sh": "did:plc:seed"}} log := slog.New(slog.NewTextHandler(io.Discard, nil)) - r := NewRunnerWithDeps(st, tap, resolver, follows, &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, log) + r := NewRunnerWithDeps( + st, tap, resolver, follows, + &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, + &fakeLightrailRepoLister{}, log, + ) dir := t.TempDir() seedsPath := filepath.Join(dir, "seeds.txt") @@ -118,6 +144,7 @@ func TestRunner_DiscoveryAndSubmit(t *testing.T) { MaxHops: 1, Concurrency: 2, BatchSize: 2, + Source: SourceGraph, }) if err != nil { t.Fatalf("run backfill: %v", err) @@ -137,7 +164,11 @@ func TestRunner_DryRunSkipsMutations(t *testing.T) { tap := &fakeTapAdmin{statuses: map[string]RepoStatus{}} resolver := &fakeResolver{mapping: map[string]string{"alice.tangled.sh": "did:plc:seed"}} log := slog.New(slog.NewTextHandler(io.Discard, nil)) - r := NewRunnerWithDeps(st, tap, resolver, follows, &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, log) + r := NewRunnerWithDeps( + st, tap, resolver, follows, + &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, + &fakeLightrailRepoLister{}, log, + ) dir := t.TempDir() seedsPath := filepath.Join(dir, "seeds.txt") @@ -151,6 +182,7 @@ func TestRunner_DryRunSkipsMutations(t *testing.T) { DryRun: true, Concurrency: 1, BatchSize: 10, + Source: SourceGraph, }) if err != nil { t.Fatalf("run dry-run backfill: %v", err) @@ -168,7 +200,11 @@ func TestRunner_SkipsInProgressBackfills(t *testing.T) { }} resolver := &fakeResolver{mapping: map[string]string{"alice.tangled.sh": "did:plc:seed"}} log := slog.New(slog.NewTextHandler(io.Discard, nil)) - r := NewRunnerWithDeps(st, tap, resolver, follows, &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, log) + r := NewRunnerWithDeps( + st, tap, resolver, follows, + &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, + &fakeLightrailRepoLister{}, log, + ) dir := t.TempDir() seedsPath := filepath.Join(dir, "seeds.txt") @@ -176,7 +212,9 @@ func TestRunner_SkipsInProgressBackfills(t *testing.T) { t.Fatalf("write seeds: %v", err) } - err := r.Run(context.Background(), Options{SeedsPath: seedsPath, MaxHops: 0}) + err := r.Run(context.Background(), Options{ + SeedsPath: seedsPath, MaxHops: 0, Source: SourceGraph, + }) if err != nil { t.Fatalf("run backfill: %v", err) } @@ -198,7 +236,11 @@ func TestRunner_ContinuesWhenRepoStatusFails(t *testing.T) { } resolver := &fakeResolver{mapping: map[string]string{"alice.tangled.sh": "did:plc:seed"}} log := slog.New(slog.NewTextHandler(io.Discard, nil)) - r := NewRunnerWithDeps(st, tap, resolver, follows, &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, log) + r := NewRunnerWithDeps( + st, tap, resolver, follows, + &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, + &fakeLightrailRepoLister{}, log, + ) dir := t.TempDir() seedsPath := filepath.Join(dir, "seeds.txt") @@ -211,6 +253,7 @@ func TestRunner_ContinuesWhenRepoStatusFails(t *testing.T) { MaxHops: 1, Concurrency: 1, BatchSize: 10, + Source: SourceGraph, }) if err != nil { t.Fatalf("run backfill: %v", err) @@ -250,7 +293,11 @@ func TestRunner_FallsBackToSingleRepoSubmissionOnBatchFailure(t *testing.T) { } resolver := &fakeResolver{mapping: map[string]string{"alice.tangled.sh": "did:plc:seed"}} log := slog.New(slog.NewTextHandler(io.Discard, nil)) - r := NewRunnerWithDeps(st, tap, resolver, follows, &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, log) + r := NewRunnerWithDeps( + st, tap, resolver, follows, + &fakeProfileFetcher{profiles: map[string]*ProfileRecord{}}, + &fakeLightrailRepoLister{}, log, + ) dir := t.TempDir() seedsPath := filepath.Join(dir, "seeds.txt") @@ -263,6 +310,7 @@ func TestRunner_FallsBackToSingleRepoSubmissionOnBatchFailure(t *testing.T) { MaxHops: 1, Concurrency: 1, BatchSize: 10, + Source: SourceGraph, }) if err != nil { t.Fatalf("run backfill: %v", err) @@ -297,7 +345,9 @@ func TestRunner_IndexesProfilesAndHandles(t *testing.T) { }, }} log := slog.New(slog.NewTextHandler(io.Discard, nil)) - r := NewRunnerWithDeps(st, tap, resolver, follows, profiles, log) + r := NewRunnerWithDeps( + st, tap, resolver, follows, profiles, &fakeLightrailRepoLister{}, log, + ) dir := t.TempDir() seedsPath := filepath.Join(dir, "seeds.txt") @@ -305,7 +355,9 @@ func TestRunner_IndexesProfilesAndHandles(t *testing.T) { t.Fatalf("write seeds: %v", err) } - err := r.Run(context.Background(), Options{SeedsPath: seedsPath, MaxHops: 0}) + err := r.Run(context.Background(), Options{ + SeedsPath: seedsPath, MaxHops: 0, Source: SourceGraph, + }) if err != nil { t.Fatalf("run backfill: %v", err) } @@ -334,3 +386,67 @@ func TestRunner_IndexesProfilesAndHandles(t *testing.T) { t.Errorf("expected summary to contain location, got %q", doc.Summary) } } + +func TestRunner_LightrailDryRunSkipsTapAndProfileIndexing(t *testing.T) { + st := &fakeStore{collaborators: map[string][]string{}} + tap := &fakeTapAdmin{ + statusErrs: map[string]error{"did:plc:a": errors.New("should not be called")}, + } + lightrail := &fakeLightrailRepoLister{dids: []string{"did:plc:b", "did:plc:a"}} + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + r := NewRunnerWithDeps( + st, tap, &fakeResolver{}, &fakeFollowFetcher{}, &fakeProfileFetcher{}, + lightrail, log, + ) + + err := r.Run(context.Background(), Options{ + Source: SourceLightrail, DryRun: true, + LightrailURL: "https://example.test", PageLimit: 500, + }) + if err != nil { + t.Fatalf("run lightrail dry-run: %v", err) + } + if len(tap.added) != 0 { + t.Fatalf("expected no Tap submissions, got %#v", tap.added) + } + if len(st.documents) != 0 { + t.Fatalf("expected no profile indexing, got %#v", st.documents) + } + if lightrail.calls != 1 { + t.Fatalf("expected one Lightrail call, got %d", lightrail.calls) + } + if len(lightrail.collections) != len(DefaultCollections) { + t.Fatalf("expected default collections, got %#v", lightrail.collections) + } +} + +func TestRunner_LightrailSubmitsWithoutRepoStatusChecks(t *testing.T) { + st := &fakeStore{collaborators: map[string][]string{}} + tap := &fakeTapAdmin{ + statusErrs: map[string]error{ + "did:plc:a": errors.New("RepoStatus should not be called in lightrail mode"), + }, + } + lightrail := &fakeLightrailRepoLister{ + dids: []string{"did:plc:b", "did:plc:a", "did:plc:b"}, + } + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + r := NewRunnerWithDeps( + st, tap, &fakeResolver{}, &fakeFollowFetcher{}, &fakeProfileFetcher{}, + lightrail, log, + ) + + err := r.Run(context.Background(), Options{ + Source: SourceLightrail, BatchSize: 10, + Collections: []string{"sh.tangled.repo"}, + }) + if err != nil { + t.Fatalf("run lightrail backfill: %v", err) + } + if len(tap.added) != 1 { + t.Fatalf("expected one Tap batch, got %#v", tap.added) + } + if len(tap.added[0]) != 2 { + t.Fatalf("expected deduped DIDs, got %#v", tap.added) + } +} diff --git a/packages/api/internal/backfill/doc.go b/packages/api/internal/backfill/doc.go index 2f925cf..b2bc192 100644 --- a/packages/api/internal/backfill/doc.go +++ b/packages/api/internal/backfill/doc.go @@ -1,14 +1,19 @@ -// Package backfill provides graph bootstrap tooling for Twister. +// Package backfill provides Tap bootstrap tooling for Twister. // // # Backfill Runbook // -// This runbook covers initial graph bootstrap and repeat runs using: +// This runbook covers initial bootstrap and repeat runs using: // // twister backfill // -// # Seeds Input +// `--source lightrail` is the default and discovers DIDs from +// com.atproto.sync.listReposByCollection. `--source graph` keeps the older +// handle/DID seed crawl for targeted fallback runs. // -// The `--seeds` flag supports either of these forms: +// # Graph Seeds Input +// +// The `--seeds` flag applies only to `--source graph` and supports either of +// these forms: // // 1. File path: // @@ -39,26 +44,24 @@ // // # First Bootstrap // -// 1. Copy and customize seeds: +// 1. Run full-network dry-run: // -// cp docs/api/seeds.txt /tmp/twister-seeds.txt +// twister backfill --dry-run // -// 2. Run dry-run first: +// 2. Run real bootstrap: // -// twister backfill --seeds /tmp/twister-seeds.txt --max-hops 2 --dry-run +// twister backfill // -// 3. Run real backfill: +// 3. Use graph mode only for targeted fallback: // -// twister backfill --seeds /tmp/twister-seeds.txt --max-hops 2 --concurrency 5 --batch-size 10 --batch-delay 1s +// twister backfill --source graph --seeds /tmp/twister-seeds.txt --max-hops 2 // -// Watch logs for seed count, hop-level discoveries, already-tracked vs submitted -// users, and batch progress totals. +// Watch logs for discovery totals and Tap submission progress. // // # Repeat Run // -// Append new candidate users to the seed source, run dry-run, then run the real -// command again. Reruns are safe because discovery deduplicates in-memory and -// Tap /repos/add is treated as idempotent. +// Re-run `twister backfill` whenever you need to reseed the authoritative Tap +// corpus. Append graph seeds only when using `--source graph`. // // # Dry-Run Safety // diff --git a/packages/api/internal/backfill/lightrail.go b/packages/api/internal/backfill/lightrail.go new file mode 100644 index 0000000..c8a93a0 --- /dev/null +++ b/packages/api/internal/backfill/lightrail.go @@ -0,0 +1,124 @@ +package backfill + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" +) + +const listReposByCollectionNSID = "com.atproto.sync.listReposByCollection" + +type lightrailRepoLister interface { + ListReposByCollection( + ctx context.Context, baseURL string, collections []string, limit int, + ) ([]string, error) +} + +type listReposByCollectionResponse struct { + Cursor string `json:"cursor"` + Repos []listReposByCollection `json:"repos"` +} + +type listReposByCollection struct { + DID string `json:"did"` +} + +type HTTPLightrailClient struct { + client *http.Client +} + +func NewHTTPLightrailClient() *HTTPLightrailClient { + return &HTTPLightrailClient{ + client: &http.Client{Timeout: 15 * time.Second}, + } +} + +func (c *HTTPLightrailClient) ListReposByCollection( + ctx context.Context, baseURL string, collections []string, limit int, +) ([]string, error) { + baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/") + if baseURL == "" { + return nil, fmt.Errorf("lightrail url is required") + } + if limit <= 0 { + limit = DefaultPageLimit + } + + seen := make(map[string]bool) + dids := make([]string, 0) + cursor := "" + for { + resp, err := c.listReposByCollectionPage(ctx, baseURL, collections, limit, cursor) + if err != nil { + return nil, err + } + for _, repo := range resp.Repos { + did := strings.TrimSpace(repo.DID) + if did == "" || seen[did] { + continue + } + seen[did] = true + dids = append(dids, did) + } + if resp.Cursor == "" { + return dids, nil + } + if resp.Cursor == cursor { + return nil, fmt.Errorf("listReposByCollection repeated cursor %q", cursor) + } + cursor = resp.Cursor + } +} + +func (c *HTTPLightrailClient) listReposByCollectionPage( + ctx context.Context, baseURL string, collections []string, limit int, cursor string, +) (*listReposByCollectionResponse, error) { + params := url.Values{} + for _, collection := range collections { + collection = strings.TrimSpace(collection) + if collection != "" { + params.Add("collection", collection) + } + } + if limit > 0 { + params.Set("limit", fmt.Sprintf("%d", limit)) + } + if cursor != "" { + params.Set("cursor", cursor) + } + + endpoint := baseURL + "/xrpc/" + listReposByCollectionNSID + if encoded := params.Encode(); encoded != "" { + endpoint += "?" + encoded + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, fmt.Errorf("build listReposByCollection request: %w", err) + } + + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("listReposByCollection request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return nil, fmt.Errorf( + "listReposByCollection failed: status %d: %s", + resp.StatusCode, + strings.TrimSpace(string(body)), + ) + } + + var payload listReposByCollectionResponse + if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil { + return nil, fmt.Errorf("decode listReposByCollection response: %w", err) + } + return &payload, nil +} diff --git a/packages/api/internal/backfill/lightrail_test.go b/packages/api/internal/backfill/lightrail_test.go new file mode 100644 index 0000000..101a6da --- /dev/null +++ b/packages/api/internal/backfill/lightrail_test.go @@ -0,0 +1,123 @@ +package backfill + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "testing" +) + +func TestHTTPLightrailClientListReposByCollectionSinglePage(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/xrpc/"+listReposByCollectionNSID { + http.NotFound(w, r) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "repos": []map[string]string{ + {"did": "did:plc:a"}, + {"did": "did:plc:b"}, + }, + }) + })) + defer srv.Close() + + client := NewHTTPLightrailClient() + dids, err := client.ListReposByCollection( + context.Background(), srv.URL, []string{"sh.tangled.repo"}, 100, + ) + if err != nil { + t.Fatalf("list repos: %v", err) + } + + want := []string{"did:plc:a", "did:plc:b"} + if !reflect.DeepEqual(dids, want) { + t.Fatalf("dids: got %#v want %#v", dids, want) + } +} + +func TestHTTPLightrailClientListReposByCollectionPaginatesAndDedupes(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + query := r.URL.Query() + cursor := query.Get("cursor") + switch cursor { + case "": + _ = json.NewEncoder(w).Encode(map[string]any{ + "cursor": "page-2", + "repos": []map[string]string{ + {"did": "did:plc:a"}, + {"did": "did:plc:b"}, + }, + }) + case "page-2": + _ = json.NewEncoder(w).Encode(map[string]any{ + "repos": []map[string]string{ + {"did": "did:plc:b"}, + {"did": "did:plc:c"}, + }, + }) + default: + t.Fatalf("unexpected cursor %q", cursor) + } + })) + defer srv.Close() + + client := NewHTTPLightrailClient() + dids, err := client.ListReposByCollection( + context.Background(), srv.URL, []string{"sh.tangled.repo"}, 2, + ) + if err != nil { + t.Fatalf("list repos: %v", err) + } + + want := []string{"did:plc:a", "did:plc:b", "did:plc:c"} + if !reflect.DeepEqual(dids, want) { + t.Fatalf("dids: got %#v want %#v", dids, want) + } +} + +func TestHTTPLightrailClientListReposByCollectionAddsRepeatedCollections(t *testing.T) { + var got url.Values + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got = r.URL.Query() + _ = json.NewEncoder(w).Encode(map[string]any{"repos": []map[string]string{}}) + })) + defer srv.Close() + + client := NewHTTPLightrailClient() + if _, err := client.ListReposByCollection( + context.Background(), + srv.URL, + []string{"sh.tangled.actor.profile", "sh.tangled.repo"}, + 50, + ); err != nil { + t.Fatalf("list repos: %v", err) + } + + if got.Get("limit") != "50" { + t.Fatalf("limit: got %q", got.Get("limit")) + } + if values := got["collection"]; !reflect.DeepEqual(values, []string{ + "sh.tangled.actor.profile", + "sh.tangled.repo", + }) { + t.Fatalf("collections: got %#v", values) + } +} + +func TestHTTPLightrailClientListReposByCollectionErrorsOnNonSuccess(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, `{"message":"boom"}`, http.StatusBadGateway) + })) + defer srv.Close() + + client := NewHTTPLightrailClient() + if _, err := client.ListReposByCollection( + context.Background(), srv.URL, []string{"sh.tangled.repo"}, 100, + ); err == nil { + t.Fatal("expected error") + } +} diff --git a/packages/api/internal/backfill/types.go b/packages/api/internal/backfill/types.go index 4d56350..f45143f 100644 --- a/packages/api/internal/backfill/types.go +++ b/packages/api/internal/backfill/types.go @@ -2,14 +2,37 @@ package backfill import "time" +const ( + SourceLightrail = "lightrail" + SourceGraph = "graph" + DefaultLightrailURL = "https://lightrail.microcosm.blue" + DefaultPageLimit = 10000 +) + +var DefaultCollections = []string{ + "sh.tangled.repo", + "sh.tangled.repo.issue", + "sh.tangled.repo.issue.state", + "sh.tangled.repo.issue.comment", + "sh.tangled.repo.pull", + "sh.tangled.repo.pull.status", + "sh.tangled.repo.pull.comment", + "sh.tangled.string", + "sh.tangled.actor.profile", +} + // Options configures a backfill run. type Options struct { - SeedsPath string - MaxHops int - DryRun bool - Concurrency int - BatchSize int - BatchDelay time.Duration + SeedsPath string + MaxHops int + DryRun bool + Concurrency int + BatchSize int + BatchDelay time.Duration + Source string + LightrailURL string + Collections []string + PageLimit int } // DiscoveredUser contains crawl metadata for an included DID. diff --git a/packages/api/internal/view/static/search.js b/packages/api/internal/view/static/search.js index 687ee92..6ec13e9 100644 --- a/packages/api/internal/view/static/search.js +++ b/packages/api/internal/view/static/search.js @@ -1,5 +1,7 @@ function searchApp() { const TANGLED_BASE = "https://tangled.org"; + const PDS_BASE = "https://pds.ls"; + const DOCUMENTS_BASE = "/documents"; return { query: "", @@ -89,10 +91,6 @@ function searchApp() { this.doSearch(false); }, - resultMode(r) { - return this.resolveResult(r).mode; - }, - resultURL(r) { return this.resolveResult(r).url; }, @@ -101,43 +99,37 @@ function searchApp() { return this.resolveResult(r).warning; }, + jsonURL(r) { + return DOCUMENTS_BASE + "/" + encodeURIComponent(r.id); + }, + + pdsURL(r) { + return r.at_uri ? PDS_BASE + "/" + r.at_uri : ""; + }, + resolveResult(r) { const parsed = this.parseATURI(r.at_uri); const author = this.normalizeOwner(r.author_handle) || this.normalizeSegment(r.did) || parsed.did; const repoOwner = this.normalizeOwner(r.repo_owner_handle) || author; const repoName = this.normalizeSegment(r.repo_name); - if (r.record_type === "issue") { - if (!r.at_uri) { - return { - mode: "none", - url: "", - warning: "This issue is missing its AT URI, so Twister cannot copy or link it yet.", - }; - } - return { mode: "copy", url: "", warning: "" }; - } - if (r.record_type === "string") { const owner = author || parsed.did; const rkey = parsed.rkey; const url = r.web_url || (owner && rkey ? this.buildTangledURL("strings", owner, rkey) : ""); const warning = url ? "" : "This string is indexed from AT Protocol, but Tangled no longer has a page for it."; - return { mode: url ? "link" : "none", url, warning }; - } - - if (r.web_url) { - return { mode: "link", url: r.web_url, warning: "" }; + return { url, warning }; } let url = ""; switch (r.record_type) { case "profile": - url = author ? this.buildTangledURL(author) : ""; + url = r.web_url || (author ? this.buildTangledURL(author) : ""); break; case "repo": - url = repoOwner && repoName ? this.buildTangledURL(repoOwner, repoName) : ""; + url = r.web_url || (repoOwner && repoName ? this.buildTangledURL(repoOwner, repoName) : ""); break; + case "issue": case "issue_comment": url = repoOwner && repoName ? this.buildTangledURL(repoOwner, repoName, "issues") : ""; break; @@ -145,28 +137,47 @@ function searchApp() { case "pull_comment": url = repoOwner && repoName ? this.buildTangledURL(repoOwner, repoName, "pulls") : ""; break; + default: + url = r.web_url || ""; } return url - ? { mode: "link", url, warning: "" } + ? { url, warning: "" } : { - mode: "none", url: "", warning: "This record is indexed from AT Protocol, but Tangled does not currently expose a page for it.", }; }, - async copyIssueATURI(r) { + async copyATURI(r) { + const label = this.recordLabel(r); if (!r.at_uri) { - this.showToast("Issue AT URI is unavailable."); + this.showToast(label + " AT URI is unavailable."); return; } try { await this.writeClipboard(r.at_uri); - this.showToast("Issue AT URI copied."); + this.showToast(label + " AT URI copied."); } catch (_) { - this.showToast("Could not copy the issue AT URI."); + this.showToast("Could not copy the " + label.toLowerCase() + " AT URI."); + } + }, + + recordLabel(r) { + switch (r.record_type) { + case "profile": + return "User"; + case "repo": + return "Repo"; + case "issue": + return "Issue"; + case "pull": + return "Pull"; + default: { + const label = (r.record_type || "record").replace(/_/g, " "); + return label.charAt(0).toUpperCase() + label.slice(1); + } } }, diff --git a/packages/api/internal/view/static/style.css b/packages/api/internal/view/static/style.css index 1c2bd4a..4e55801 100644 --- a/packages/api/internal/view/static/style.css +++ b/packages/api/internal/view/static/style.css @@ -110,7 +110,7 @@ a:hover { text-decoration: underline; } font-size: .9rem; cursor: pointer; } -.btn:hover { border-color: var(--accent); color: var(--accent); } +.btn:hover { border-color: var(--accent); color: var(--accent); text-decoration: none; } .btn:disabled { opacity: .5; cursor: default; } .btn-primary { background: var(--accent); color: var(--bg); border-color: var(--accent); font-weight: 500; } .btn-primary:hover { background: #6b93e8; color: var(--bg); } @@ -137,6 +137,12 @@ a:hover { text-decoration: underline; } /* Messages */ .msg { padding: 1rem; color: var(--text-dim); text-align: center; } .msg-error { color: #f7768e; } +.msg-empty { + background: var(--surface); + border: 1px solid var(--border); + border-radius: var(--radius); +} +.msg-empty p:last-child { margin-bottom: 0; } /* Result cards */ .card { @@ -151,13 +157,6 @@ a:hover { text-decoration: underline; } min-width: 0; } .card:hover { border-color: var(--accent); text-decoration: none; } -.card-button { - width: 100%; - text-align: left; - font: inherit; - cursor: pointer; -} -.card-disabled:hover { border-color: var(--border); } .card-head { display: flex; align-items: flex-start; @@ -210,6 +209,16 @@ a:hover { text-decoration: underline; } overflow-wrap: anywhere; word-break: break-word; } +.card-actions { + display: flex; + gap: .5rem; + flex-wrap: wrap; + margin-top: .75rem; +} +.btn-card { + padding: .45rem .7rem; + font-size: .82rem; +} .meta-sep::before { content: "\00b7"; margin-right: .5rem; } .card-warning { margin-top: .6rem; diff --git a/packages/api/internal/view/templates/index.html b/packages/api/internal/view/templates/index.html index 84b5211..ec35bee 100644 --- a/packages/api/internal/view/templates/index.html +++ b/packages/api/internal/view/templates/index.html @@ -35,65 +35,78 @@ +