From 7b3dc7448f34383a96369c269e23507d0f3ffd3f Mon Sep 17 00:00:00 2001 From: Owais Jamil Date: Wed, 25 Mar 2026 16:03:10 +0000 Subject: [PATCH] fix: tangled links & pds + handle <-> DID resolution --- docs/roadmap.md | 8 ++++---- packages/api/README.md | 18 +++++++++++++++++- apps/twisted/src/mocks/repos.ts | 2 +- packages/api/internal/api/actors.go | 12 +++++++++--- packages/api/internal/api/readthrough.go | 122 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- packages/api/internal/backfill/backfill.go | 2 +- packages/api/internal/config/config.go | 2 +- packages/api/internal/ingest/ingest_test.go | 4 ++++ packages/api/internal/reindex/reindex.go | 12 ++++++++---- packages/api/internal/store/sql_store.go | 9 +++++++++ packages/api/internal/store/store.go | 1 + packages/api/internal/xrpc/did.go | 4 +--- packages/api/internal/xrpc/did_test.go | 25 +++++++++++++++++++++++++ packages/api/internal/xrpc/records.go | 3 +++ packages/api/internal/xrpc/repo.go | 14 +++++++------- packages/api/internal/xrpc/repo_test.go | 14 +++++++------- apps/twisted/src/features/activity/ActivityPage.vue | 3 +-- apps/twisted/src/services/tangled/repo-assets.ts | 4 +++- packages/api/internal/view/templates/layout.html | 2 +- packages/api/internal/view/templates/docs/index.html | 2 +- 20 file(s) changed, 225 insertion(s)(+), 38 deletion(s)(-) diff --git a/docs/roadmap.md b/docs/roadmap.md --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -20,14 +20,14 @@ - Follow AT URI (desertthunder.dev follows npmx): `at://did:plc:xg2vq45muivyy3xwatcehspu/sh.tangled.graph.follow/3mhofstanru22` - Star AT URI (desertthunder.dev stars microcosm-rs): `at://did:plc:lulmyldiq4sb2ikags5sfb25/sh.tangled.repo/3lvsxzinfz222` - ~~Add `just` targets for smoke-test runs locally and against a remote base URL~~ directly invoking the scripts is fine. -- [ ] Reuse the existing normalization and upsert path for on-demand indexing jobs -- [ ] Trigger indexing jobs from repo, issue, PR, profile, and similar fetch handlers -- [ ] Add dedupe, retries, and observability for indexing jobs +- [x] Reuse the existing normalization and upsert path for on-demand indexing jobs +- [x] Trigger indexing jobs from repo, issue, PR, profile, and similar fetch handlers +- [x] Add dedupe, retries, and observability for indexing jobs - [ ] Add a JetStream cache consumer with a persisted timestamp cursor - [ ] Seed the JetStream cursor to `now - 24h` on first boot and rewind slightly on reconnect - [ ] Store and serve bounded recent activity from the local cache - [ ] Keep Tap as the authoritative indexing and bulk backfill path -- [ ] Define a controlled backfill and repo-resync playbook for recovery +- [ ] Define a controlled backfill and repo-resync playbook for recovery (`docs/references/resync.md`) ## API: Constellation Integration diff --git a/packages/api/README.md b/packages/api/README.md --- a/packages/api/README.md +++ b/packages/api/README.md @@ -142,8 +142,24 @@ 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 +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 diff --git a/apps/twisted/src/mocks/repos.ts b/apps/twisted/src/mocks/repos.ts --- a/apps/twisted/src/mocks/repos.ts +++ b/apps/twisted/src/mocks/repos.ts @@ -120,7 +120,7 @@ const README_CONTENT = `# twisted -A mobile companion reader for [Tangled](https://tangled.sh), built with Ionic Vue and Capacitor. +A mobile companion reader for [Tangled](https://tangled.org), built with Ionic Vue and Capacitor. ## Features diff --git a/packages/api/internal/api/actors.go b/packages/api/internal/api/actors.go --- a/packages/api/internal/api/actors.go +++ b/packages/api/internal/api/actors.go @@ -22,20 +22,23 @@ // issueEntry extends recordEntry with pre-joined issue state. type issueEntry struct { recordEntry - State string `json:"state"` // "open" or "closed" + // "open" or "closed" + State string `json:"state"` } // pullEntry extends recordEntry with pre-joined pull status. type pullEntry struct { recordEntry - Status string `json:"status"` // "open", "merged", or "closed" + // "open", "merged", or "closed" + Status string `json:"status"` } // actorContext holds resolved identity for a request. type actorContext struct { DID string `json:"did"` Handle string `json:"handle"` - PDS string `json:"pds"` // full URL, e.g. "https://bsky.social" + // full URL, e.g. "https://bsky.social" + PDS string `json:"pds"` } // repoContext extends actorContext with the repo's knot host and AT URI. @@ -64,6 +67,9 @@ identity, err := s.xrpc.ResolveIdentity(ctx, did) if err != nil { return nil, fmt.Errorf("resolve identity %q: %w", did, err) + } + if identity.PDS == "" { + return nil, fmt.Errorf("no atproto pds in did document for %q", did) } return &actorContext{ diff --git a/packages/api/internal/api/readthrough.go b/packages/api/internal/api/readthrough.go --- a/packages/api/internal/api/readthrough.go +++ b/packages/api/internal/api/readthrough.go @@ -5,6 +5,7 @@ "encoding/json" "fmt" "log/slog" + "sync" "time" "tangled.org/desertthunder.dev/twister/internal/normalize" @@ -12,11 +13,19 @@ "tangled.org/desertthunder.dev/twister/internal/xrpc" ) -const readThroughIdlePoll = 1 * time.Second +const ( + readThroughIdlePoll = 1 * time.Second + readThroughStatusInterval = 30 * time.Second + maxIndexingAttempts = 10 +) func (s *Server) runReadThroughIndexer(ctx context.Context) { ticker := time.NewTicker(readThroughIdlePoll) defer ticker.Stop() + + var mu sync.Mutex + var processedTick int64 + go s.runIndexerStatusLogger(ctx, &mu, &processedTick) s.log.Info("read-through indexer worker started") for { @@ -45,6 +54,15 @@ } 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)) @@ -70,6 +88,37 @@ slog.String("error", err.Error()), ) continue + } + + s.log.Debug("read-through job completed", slog.String("document_id", job.DocumentID)) + mu.Lock() + processedTick++ + mu.Unlock() + } +} + +func (s *Server) runIndexerStatusLogger(ctx context.Context, mu *sync.Mutex, processedTick *int64) { + ticker := time.NewTicker(readThroughStatusInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + mu.Lock() + n := *processedTick + *processedTick = 0 + mu.Unlock() + + pending, err := s.store.CountPendingIndexingJobs(ctx) + if err != nil { + s.log.Warn("read-through status: count failed", slog.String("error", err.Error())) + continue + } + s.log.Info("read-through indexer status", + slog.Int64("jobs_processed", n), + slog.Int64("jobs_pending", pending), + ) } } } @@ -124,6 +173,8 @@ } } + s.enrichDocument(ctx, doc, record) + if err := s.store.UpsertDocument(ctx, doc); err != nil { return fmt.Errorf("upsert document: %w", err) } @@ -137,6 +188,75 @@ } } return nil +} + +// 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 { + 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) + } +} + +// 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 target, _ := record["target"].(map[string]any); target != nil { + if uri, _ := target["repo"].(string); uri != "" { + return uri + } + } + return "" } func (s *Server) enqueueXRPCRecord(ctx context.Context, uri, cid string, value map[string]any) { diff --git a/packages/api/internal/backfill/backfill.go b/packages/api/internal/backfill/backfill.go --- a/packages/api/internal/backfill/backfill.go +++ b/packages/api/internal/backfill/backfill.go @@ -189,7 +189,7 @@ func (r *Runner) resolveSeeds(ctx context.Context, entries []seedEntry) ([]string, map[string]string, error) { seen := map[string]bool{} seeds := make([]string, 0, len(entries)) - handles := make(map[string]string) // did → handle + handles := make(map[string]string) for _, entry := range entries { if entry.isDID { seen[entry.raw] = true diff --git a/packages/api/internal/config/config.go b/packages/api/internal/config/config.go --- a/packages/api/internal/config/config.go +++ b/packages/api/internal/config/config.go @@ -82,7 +82,7 @@ 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.sh; Owais )"), + 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"), diff --git a/packages/api/internal/ingest/ingest_test.go b/packages/api/internal/ingest/ingest_test.go --- a/packages/api/internal/ingest/ingest_test.go +++ b/packages/api/internal/ingest/ingest_test.go @@ -143,6 +143,10 @@ return int64(len(f.docs)), nil } +func (f *fakeStore) CountPendingIndexingJobs(_ context.Context) (int64, error) { + return 0, nil +} + func (f *fakeStore) Ping(_ context.Context) error { return nil } diff --git a/packages/api/internal/reindex/reindex.go b/packages/api/internal/reindex/reindex.go --- a/packages/api/internal/reindex/reindex.go +++ b/packages/api/internal/reindex/reindex.go @@ -12,10 +12,14 @@ // Options controls which documents are reindexed. type Options struct { - Collection string // reindex documents in this collection only - DID string // reindex documents authored by this DID only - DocumentID string // reindex a single document by stable ID - DryRun bool // log intended work without writing + // reindex documents in this collection only + Collection string + // reindex documents authored by this DID only + DID string + // reindex a single document by stable ID + DocumentID string + // log intended work without writing + DryRun bool } // Result summarises the outcome of a reindex run. diff --git a/packages/api/internal/store/sql_store.go b/packages/api/internal/store/sql_store.go --- a/packages/api/internal/store/sql_store.go +++ b/packages/api/internal/store/sql_store.go @@ -460,6 +460,15 @@ return n, nil } +func (s *SQLStore) CountPendingIndexingJobs(ctx context.Context) (int64, error) { + var n int64 + err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM indexing_jobs WHERE status = 'pending'`).Scan(&n) + if err != nil { + return 0, fmt.Errorf("count pending indexing jobs: %w", err) + } + return n, nil +} + func (s *SQLStore) Ping(ctx context.Context) error { return s.db.PingContext(ctx) } diff --git a/packages/api/internal/store/store.go b/packages/api/internal/store/store.go --- a/packages/api/internal/store/store.go +++ b/packages/api/internal/store/store.go @@ -96,5 +96,6 @@ GetFollowSubjects(ctx context.Context, did string) ([]string, error) GetRepoCollaborators(ctx context.Context, repoOwnerDID string) ([]string, error) CountDocuments(ctx context.Context) (int64, error) + CountPendingIndexingJobs(ctx context.Context) (int64, error) Ping(ctx context.Context) error } diff --git a/packages/api/internal/xrpc/did.go b/packages/api/internal/xrpc/did.go --- a/packages/api/internal/xrpc/did.go +++ b/packages/api/internal/xrpc/did.go @@ -77,6 +77,7 @@ } // ResolveIdentity resolves a DID to its PDS endpoint and handle. +// Both fields are best-effort: callers that require PDS should check info.PDS != "". func (c *Client) ResolveIdentity(ctx context.Context, did string) (*IdentityInfo, error) { doc, err := c.ResolveDIDDoc(ctx, did) if err != nil { @@ -90,9 +91,6 @@ info.PDS = strings.TrimSpace(svc.ServiceEndpoint) break } - } - if info.PDS == "" { - return nil, fmt.Errorf("no atproto pds endpoint in did document for %s", did) } for _, aka := range doc.AlsoKnownAs { diff --git a/packages/api/internal/xrpc/did_test.go b/packages/api/internal/xrpc/did_test.go --- a/packages/api/internal/xrpc/did_test.go +++ b/packages/api/internal/xrpc/did_test.go @@ -85,6 +85,31 @@ } } +func TestResolveIdentity_NoPDS(t *testing.T) { + doc := DIDDocument{ + ID: "did:plc:nopds", + AlsoKnownAs: []string{"at://alice.test"}, + Service: []DIDService{}, + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(doc) + })) + defer srv.Close() + + c := NewClient(WithPLCDirectory(srv.URL)) + info, err := c.ResolveIdentity(context.Background(), "did:plc:nopds") + if err != nil { + t.Fatal("ResolveIdentity should succeed even with no PDS service:", err) + } + if info.Handle != "alice.test" { + t.Errorf("expected handle alice.test, got %q", info.Handle) + } + if info.PDS != "" { + t.Errorf("expected empty PDS, got %q", info.PDS) + } +} + func TestResolveDIDDoc_UnsupportedMethod(t *testing.T) { c := NewClient() _, err := c.ResolveDIDDoc(context.Background(), "did:key:z123") diff --git a/packages/api/internal/xrpc/records.go b/packages/api/internal/xrpc/records.go --- a/packages/api/internal/xrpc/records.go +++ b/packages/api/internal/xrpc/records.go @@ -120,5 +120,8 @@ if err != nil { return "", fmt.Errorf("resolve pds for %s: %w", repo, err) } + if info.PDS == "" { + return "", fmt.Errorf("no atproto pds in did document for %s", repo) + } return info.PDS, nil } diff --git a/packages/api/internal/xrpc/repo.go b/packages/api/internal/xrpc/repo.go --- a/packages/api/internal/xrpc/repo.go +++ b/packages/api/internal/xrpc/repo.go @@ -31,7 +31,7 @@ return name, nil } -// BuildWebURL builds a canonical tangled.sh URL for a record. +// BuildWebURL builds a canonical tangled.org URL for a record. // recordType should be one of: "repo", "issue", "pull", "issue_comment", "pull_comment", "profile". func BuildWebURL(ownerHandle, repoName, recordType, rkey string) string { if ownerHandle == "" { @@ -41,32 +41,32 @@ switch recordType { case "profile": - return fmt.Sprintf("https://tangled.sh/%s", owner) + return fmt.Sprintf("https://tangled.org/%s", owner) case "repo": if repoName == "" { return "" } - return fmt.Sprintf("https://tangled.sh/%s/%s", owner, repoName) + return fmt.Sprintf("https://tangled.org/%s/%s", owner, repoName) case "issue": if repoName == "" || rkey == "" { return "" } - return fmt.Sprintf("https://tangled.sh/%s/%s/issues/%s", owner, repoName, rkey) + return fmt.Sprintf("https://tangled.org/%s/%s/issues/%s", owner, repoName, rkey) case "pull": if repoName == "" || rkey == "" { return "" } - return fmt.Sprintf("https://tangled.sh/%s/%s/pulls/%s", owner, repoName, rkey) + return fmt.Sprintf("https://tangled.org/%s/%s/pulls/%s", owner, repoName, rkey) case "issue_comment": if repoName == "" { return "" } - return fmt.Sprintf("https://tangled.sh/%s/%s/issues", owner, repoName) + return fmt.Sprintf("https://tangled.org/%s/%s/issues", owner, repoName) case "pull_comment": if repoName == "" { return "" } - return fmt.Sprintf("https://tangled.sh/%s/%s/pulls", owner, repoName) + return fmt.Sprintf("https://tangled.org/%s/%s/pulls", owner, repoName) default: return "" } diff --git a/packages/api/internal/xrpc/repo_test.go b/packages/api/internal/xrpc/repo_test.go --- a/packages/api/internal/xrpc/repo_test.go +++ b/packages/api/internal/xrpc/repo_test.go @@ -7,13 +7,13 @@ owner, repo, recordType, rkey string want string }{ - {"alice.test", "myrepo", "repo", "", "https://tangled.sh/alice.test/myrepo"}, - {"alice.test", "myrepo", "issue", "123", "https://tangled.sh/alice.test/myrepo/issues/123"}, - {"alice.test", "myrepo", "pull", "456", "https://tangled.sh/alice.test/myrepo/pulls/456"}, - {"alice.test", "myrepo", "issue_comment", "789", "https://tangled.sh/alice.test/myrepo/issues"}, - {"alice.test", "myrepo", "pull_comment", "789", "https://tangled.sh/alice.test/myrepo/pulls"}, - {"alice.test", "", "profile", "", "https://tangled.sh/alice.test"}, - {"@alice.test", "myrepo", "repo", "", "https://tangled.sh/alice.test/myrepo"}, + {"alice.test", "myrepo", "repo", "", "https://tangled.org/alice.test/myrepo"}, + {"alice.test", "myrepo", "issue", "123", "https://tangled.org/alice.test/myrepo/issues/123"}, + {"alice.test", "myrepo", "pull", "456", "https://tangled.org/alice.test/myrepo/pulls/456"}, + {"alice.test", "myrepo", "issue_comment", "789", "https://tangled.org/alice.test/myrepo/issues"}, + {"alice.test", "myrepo", "pull_comment", "789", "https://tangled.org/alice.test/myrepo/pulls"}, + {"alice.test", "", "profile", "", "https://tangled.org/alice.test"}, + {"@alice.test", "myrepo", "repo", "", "https://tangled.org/alice.test/myrepo"}, {"", "myrepo", "repo", "", ""}, {"alice.test", "", "repo", "", ""}, {"alice.test", "myrepo", "unknown", "", ""}, diff --git a/apps/twisted/src/features/activity/ActivityPage.vue b/apps/twisted/src/features/activity/ActivityPage.vue --- a/apps/twisted/src/features/activity/ActivityPage.vue +++ b/apps/twisted/src/features/activity/ActivityPage.vue @@ -193,7 +193,7 @@ onIonViewWillLeave(() => { client.disconnect(); - status.value = "connecting"; // Reset so next enter shows "connecting" + status.value = "connecting"; }); onUnmounted(() => { @@ -217,7 +217,6 @@ client.disconnect(); status.value = "connecting"; client.connect(); - // Complete the refresher after a short delay await new Promise((resolve) => setTimeout(resolve, 1000)); (event.target as HTMLIonRefresherElement).complete(); } diff --git a/apps/twisted/src/services/tangled/repo-assets.ts b/apps/twisted/src/services/tangled/repo-assets.ts --- a/apps/twisted/src/services/tangled/repo-assets.ts +++ b/apps/twisted/src/services/tangled/repo-assets.ts @@ -79,7 +79,9 @@ const objectUrl = createObjectUrlFromBlobContent(blob); if (objectUrl) return { url: objectUrl, revoke: true }; } catch { - // Fall back to the public raw URL if the XRPC lookup fails. + console.warn( + `Failed to fetch blob for ${context.owner}/${context.repo} at ${repoPath}, falling back to public URL.`, + ); } return { url: buildPublicRawUrl(context, repoPath), revoke: false }; diff --git a/packages/api/internal/view/templates/layout.html b/packages/api/internal/view/templates/layout.html --- a/packages/api/internal/view/templates/layout.html +++ b/packages/api/internal/view/templates/layout.html @@ -25,7 +25,7 @@ {{block "scripts" .}}{{end}} diff --git a/packages/api/internal/view/templates/docs/index.html b/packages/api/internal/view/templates/docs/index.html --- a/packages/api/internal/view/templates/docs/index.html +++ b/packages/api/internal/view/templates/docs/index.html @@ -1,7 +1,7 @@ {{define "title"}}API Docs — Twister{{end}} {{define "content"}}

API Documentation

-

Twister exposes a public JSON API for searching indexed Tangled content. No authentication is required for read endpoints.

+

Twister exposes a public JSON API for searching indexed Tangled content. No authentication is required for read endpoints.

Base URL

https://<your-twister-domain>
-- tangled.sh