From fcd96d3fc1c59851baa12b1104bf388cbf0a86aa Mon Sep 17 00:00:00 2001 From: Owais Jamil Date: Tue, 24 Mar 2026 22:56:55 -0500 Subject: [PATCH] feat: migrate api to twister/xrpc/ --- apps/twisted/README.md | 48 +- apps/twisted/src/core/config/project.ts | 9 + .../src/features/activity/ActivityPage.vue | 26 +- apps/twisted/src/features/home/HomePage.vue | 6 +- .../src/features/profile/UserProfilePage.vue | 13 +- .../src/features/repo/IssueDetailPage.vue | 34 +- .../features/repo/PullRequestDetailPage.vue | 32 +- .../src/features/repo/RepoDetailPage.vue | 38 +- apps/twisted/src/features/repo/RepoFiles.vue | 12 +- .../src/services/constellation/queries.ts | 26 +- apps/twisted/src/services/jetstream/client.ts | 14 +- apps/twisted/src/services/tangled/.gitkeep | 0 .../twisted/src/services/tangled/endpoints.ts | 572 +++++------ apps/twisted/src/services/tangled/queries.ts | 605 +++++------- .../src/services/tangled/repo-assets.ts | 25 +- .../tests/unit/tangled-normalizers.spec.ts | 60 +- apps/twisted/tsconfig.json | 1 + packages/api/README.md | 62 +- packages/api/internal/api/actors.go | 900 ++++++++++++++++++ packages/api/internal/api/api.go | 36 +- packages/api/internal/api/proxy.go | 182 ++++ packages/api/internal/constellation/client.go | 64 +- packages/api/internal/store/db.go | 7 - packages/api/main.go | 8 +- 24 files changed, 1837 insertions(+), 943 deletions(-) delete mode 100644 apps/twisted/src/services/tangled/.gitkeep create mode 100644 packages/api/internal/api/actors.go create mode 100644 packages/api/internal/api/proxy.go diff --git a/apps/twisted/README.md b/apps/twisted/README.md index 826be52..ef253c8 100644 --- a/apps/twisted/README.md +++ b/apps/twisted/README.md @@ -1,3 +1,49 @@ # Twisted App -Ionic Vue client for the Twisted monorepo. +Ionic Vue client for Twisted — a Tangled browser and search app for Android & iOS. + +## Requirements + +- Node.js 20+ +- pnpm +- The Twister API running locally (see `packages/api/README.md`) + +## Running locally + +```sh +# From the repo root, install dependencies +pnpm install + +# Copy env file and point it at your local Twister API +cp apps/twisted/.env.example apps/twisted/.env + +# Start the Vite dev server +cd apps/twisted +pnpm dev +``` + +The dev server runs at `http://localhost:5173` by default. + +## Environment variables + +| Variable | Default | Description | +| --------------------------- | ----------------------- | --------------------------------------------------------------------------------------------------------------- | +| `VITE_TWISTER_API_BASE_URL` | `http://localhost:8080` | Base URL of the Twister API. All app requests (AT Protocol, Jetstream, Constellation) are proxied through this. | + +> All upstream requests — knot XRPC, PDS records, handle resolution, DID documents, +> Constellation backlink counts, and the Jetstream activity stream — are routed +> through the Twister API. The app makes no direct calls to external services. + +## Building for mobile + +```sh +# Build the web assets +pnpm build + +# Sync to native projects +pnpm cap sync + +# Open in Xcode / Android Studio +pnpm cap open ios +pnpm cap open android +``` diff --git a/apps/twisted/src/core/config/project.ts b/apps/twisted/src/core/config/project.ts index c69dd0c..aaf34d5 100644 --- a/apps/twisted/src/core/config/project.ts +++ b/apps/twisted/src/core/config/project.ts @@ -10,3 +10,12 @@ export function getTwisterApiUrl(path: string): string { return new URL(path.replace(/^\/+/, ""), `${twisterApiBaseUrl}/`).toString(); } + +export function getTwisterWsUrl(path: string): string { + if (!hasTwisterApi) { + throw new Error("Twister API base URL is not configured."); + } + + const wsBase = twisterApiBaseUrl.replace(/^http/, (m) => (m === "https" ? "wss" : "ws")); + return new URL(path.replace(/^\/+/, ""), `${wsBase}/`).toString(); +} diff --git a/apps/twisted/src/features/activity/ActivityPage.vue b/apps/twisted/src/features/activity/ActivityPage.vue index c2df23b..c1844be 100644 --- a/apps/twisted/src/features/activity/ActivityPage.vue +++ b/apps/twisted/src/features/activity/ActivityPage.vue @@ -68,7 +68,6 @@ @click="handleItemClick(item)" @actor-click="handleActorClick(item)" /> - @@ -97,7 +96,7 @@ import EmptyState from "@/components/common/EmptyState.vue"; import SkeletonLoader from "@/components/common/SkeletonLoader.vue"; import { JetstreamClient } from "@/services/jetstream/client.js"; - import { resolveHandleFromDid } from "@/services/tangled/endpoints.js"; + import { fetchActor } from "@/services/tangled/endpoints.js"; import type { ActivityItem } from "@/domain/models/activity.js"; type ConnectionStatus = "connecting" | "connected" | "disconnected"; @@ -154,17 +153,15 @@ function resolveHandle(did: string): void { if (handleCache.value.has(did)) return; - // Optimistically mark as in-progress by setting to DID to avoid re-entrancy handleCache.value.set(did, did); - resolveHandleFromDid(did) - .then((handle) => { + fetchActor(did) + .then((actor) => { const next = new Map(handleCache.value); - next.set(did, handle); + next.set(did, actor.handle); handleCache.value = next; }) .catch(() => { - // Leave the placeholder handle from the item handleCache.value.delete(did); }); } @@ -230,29 +227,24 @@ if (handle && handle !== item.actorDid) { router.push(`/tabs/activity/user/${handle}`); } else { - // Resolve then navigate - resolveHandleFromDid(item.actorDid) - .then((h) => { + fetchActor(item.actorDid) + .then((actor) => { + const h = actor.handle; const next = new Map(handleCache.value); next.set(item.actorDid, h); handleCache.value = next; router.push(`/tabs/activity/user/${h}`); }) - .catch(() => { - // Cannot navigate without a handle - }); + .catch(console.warn); } } function handleItemClick(item: ActivityItem) { - // Navigate to repo if we can determine owner handle and repo name if (item.targetName && item.targetOwnerDid) { const ownerHandle = handleCache.value.get(item.targetOwnerDid); if (ownerHandle && ownerHandle !== item.targetOwnerDid) { router.push(`/tabs/activity/repo/${ownerHandle}/${item.targetName}`); - } - // If handle not yet resolved, resolve it in background for future clicks - else { + } else { resolveHandle(item.targetOwnerDid); } } diff --git a/apps/twisted/src/features/home/HomePage.vue b/apps/twisted/src/features/home/HomePage.vue index ee10949..5e2e397 100644 --- a/apps/twisted/src/features/home/HomePage.vue +++ b/apps/twisted/src/features/home/HomePage.vue @@ -182,12 +182,10 @@ const hasRecentItems = computed(() => recentRepos.value.length > 0 || recentProfiles.value.length > 0); const identity = useIdentity(activeHandle, { enabled: computed(() => !!activeHandle.value) }); - const did = computed(() => identity.data.value?.did ?? ""); - const pds = computed(() => identity.data.value?.pds ?? ""); const hasResolvedIdentity = computed(() => !!identity.data.value); - const profileQuery = useActorProfile(pds, did, activeHandle, undefined, { enabled: hasResolvedIdentity }); - const reposQuery = useUserRepos(pds, did, activeHandle, { enabled: hasResolvedIdentity }); + const profileQuery = useActorProfile(activeHandle, undefined, { enabled: hasResolvedIdentity }); + const reposQuery = useUserRepos(activeHandle, { enabled: hasResolvedIdentity }); const repos = computed(() => reposQuery.data.value ?? []); const displayName = computed(() => profileQuery.data.value?.displayName ?? "Public Tangled account"); diff --git a/apps/twisted/src/features/profile/UserProfilePage.vue b/apps/twisted/src/features/profile/UserProfilePage.vue index 9dc6361..8d1a9dd 100644 --- a/apps/twisted/src/features/profile/UserProfilePage.vue +++ b/apps/twisted/src/features/profile/UserProfilePage.vue @@ -195,7 +195,6 @@ const section = ref<"repos" | "strings" | "issues" | "prs" | "following">("repos const identity = useIdentity(handle); const did = computed(() => identity.data.value?.did ?? ""); -const pds = computed(() => identity.data.value?.pds ?? ""); const hasIdentity = computed(() => !!identity.data.value); const tabPrefix = computed(() => { if (route.path.startsWith("/tabs/explore")) return "/tabs/explore"; @@ -203,12 +202,12 @@ const tabPrefix = computed(() => { return "/tabs/home"; }); -const profileQuery = useActorProfile(pds, did, handle, undefined, { enabled: hasIdentity }); -const reposQuery = useUserRepos(pds, did, handle, { enabled: hasIdentity }); -const stringsQuery = useUserStrings(pds, did, { enabled: hasIdentity }); -const issuesQuery = useUserIssues(pds, did, handle, { enabled: hasIdentity }); -const pullRequestsQuery = useUserPullRequests(pds, did, handle, { enabled: hasIdentity }); -const followingQuery = useUserFollowing(pds, did, { enabled: hasIdentity }); +const profileQuery = useActorProfile(handle, undefined, { enabled: hasIdentity }); +const reposQuery = useUserRepos(handle, { enabled: hasIdentity }); +const stringsQuery = useUserStrings(handle, { enabled: hasIdentity }); +const issuesQuery = useUserIssues(handle, { enabled: hasIdentity }); +const pullRequestsQuery = useUserPullRequests(handle, { enabled: hasIdentity }); +const followingQuery = useUserFollowing(handle, { enabled: hasIdentity }); const indexedProfileSummaryQuery = useIndexedProfileSummary(did, { enabled: hasIdentity }); const profile = computed(() => profileQuery.data.value); diff --git a/apps/twisted/src/features/repo/IssueDetailPage.vue b/apps/twisted/src/features/repo/IssueDetailPage.vue index a20e5c7..3077631 100644 --- a/apps/twisted/src/features/repo/IssueDetailPage.vue +++ b/apps/twisted/src/features/repo/IssueDetailPage.vue @@ -92,7 +92,7 @@ import SkeletonLoader from "@/components/common/SkeletonLoader.vue"; import EmptyState from "@/components/common/EmptyState.vue"; import MarkdownRenderer from "@/components/repo/MarkdownRenderer.vue"; import CommentThread from "@/components/repo/CommentThread.vue"; -import { useIdentity, useRepoRecord, useIssueDetail, useIssueComments, useDefaultBranch } from "@/services/tangled/queries.js"; +import { useRepoRecord, useIssueDetail, useIssueComments, useDefaultBranch } from "@/services/tangled/queries.js"; import type { RepoAssetContext } from "@/services/tangled/repo-assets.js"; const route = useRoute(); @@ -100,33 +100,23 @@ const owner = computed(() => String(route.params.owner ?? "")); const repoName = computed(() => String(route.params.repo ?? "")); const issueId = computed(() => String(route.params.issueId ?? "")); -const identity = useIdentity(owner, { enabled: computed(() => !!owner.value) }); -const did = computed(() => identity.data.value?.did ?? ""); -const pds = computed(() => identity.data.value?.pds ?? ""); -const hasIdentity = computed(() => !!identity.data.value); - -const repoQuery = useRepoRecord(pds, did, repoName, owner, { enabled: hasIdentity }); -const knotHost = computed(() => repoQuery.data.value?.knot ?? ""); -const knotRepo = computed(() => (did.value && repoName.value ? `${did.value}/${repoName.value}` : "")); -const branchQuery = useDefaultBranch(knotHost, knotRepo, { - enabled: computed(() => !!knotHost.value && !!knotRepo.value), +const repoQuery = useRepoRecord(owner, repoName, { enabled: computed(() => !!owner.value && !!repoName.value) }); +const branchQuery = useDefaultBranch(owner, repoName, { + enabled: computed(() => !!owner.value && !!repoName.value), }); const repoAtUri = computed(() => repoQuery.data.value?.atUri ?? ""); const markdownContext = computed(() => { - if (!knotHost.value || !knotRepo.value || !branchQuery.data.value?.name) return undefined; + if (!owner.value || !repoName.value || !branchQuery.data.value?.name) return undefined; return { owner: owner.value, repo: repoName.value, branch: branchQuery.data.value.name, - knotHost: knotHost.value, - knotRepo: knotRepo.value, }; }); -const issueQuery = useIssueDetail(pds, did, owner, issueId, { enabled: hasIdentity }); -const issueAtUri = computed(() => issueQuery.data.value?.atUri ?? ""); -const commentsQuery = useIssueComments(pds, did, owner, issueAtUri, { enabled: computed(() => !!issueAtUri.value) }); +const issueQuery = useIssueDetail(owner, issueId, { enabled: computed(() => !!owner.value && !!issueId.value) }); +const commentsQuery = useIssueComments(owner, issueId, { enabled: computed(() => !!owner.value && !!issueId.value) }); const issue = computed(() => { const value = issueQuery.data.value; @@ -137,17 +127,13 @@ const issue = computed(() => { const comments = computed(() => commentsQuery.data.value ?? []); const isLoading = computed( - () => - identity.isPending.value || - repoQuery.isPending.value || - issueQuery.isPending.value || - commentsQuery.isPending.value, + () => repoQuery.isPending.value || issueQuery.isPending.value || commentsQuery.isPending.value, ); const isError = computed( - () => identity.isError.value || repoQuery.isError.value || issueQuery.isError.value || commentsQuery.isError.value, + () => repoQuery.isError.value || issueQuery.isError.value || commentsQuery.isError.value, ); const errorMessage = computed(() => { - const error = identity.error.value ?? repoQuery.error.value ?? issueQuery.error.value ?? commentsQuery.error.value; + const error = repoQuery.error.value ?? issueQuery.error.value ?? commentsQuery.error.value; return error instanceof Error ? error.message : "An unexpected error occurred."; }); diff --git a/apps/twisted/src/features/repo/PullRequestDetailPage.vue b/apps/twisted/src/features/repo/PullRequestDetailPage.vue index 8f484d4..1be7900 100644 --- a/apps/twisted/src/features/repo/PullRequestDetailPage.vue +++ b/apps/twisted/src/features/repo/PullRequestDetailPage.vue @@ -102,7 +102,6 @@ import EmptyState from "@/components/common/EmptyState.vue"; import MarkdownRenderer from "@/components/repo/MarkdownRenderer.vue"; import CommentThread from "@/components/repo/CommentThread.vue"; import { - useIdentity, useRepoRecord, useDefaultBranch, usePullRequestDetail, @@ -115,34 +114,24 @@ const owner = computed(() => String(route.params.owner ?? "")); const repoName = computed(() => String(route.params.repo ?? "")); const pullId = computed(() => String(route.params.pullId ?? "")); -const identity = useIdentity(owner, { enabled: computed(() => !!owner.value) }); -const did = computed(() => identity.data.value?.did ?? ""); -const pds = computed(() => identity.data.value?.pds ?? ""); -const hasIdentity = computed(() => !!identity.data.value); - -const repoQuery = useRepoRecord(pds, did, repoName, owner, { enabled: hasIdentity }); -const knotHost = computed(() => repoQuery.data.value?.knot ?? ""); -const knotRepo = computed(() => (did.value && repoName.value ? `${did.value}/${repoName.value}` : "")); -const branchQuery = useDefaultBranch(knotHost, knotRepo, { - enabled: computed(() => !!knotHost.value && !!knotRepo.value), +const repoQuery = useRepoRecord(owner, repoName, { enabled: computed(() => !!owner.value && !!repoName.value) }); +const branchQuery = useDefaultBranch(owner, repoName, { + enabled: computed(() => !!owner.value && !!repoName.value), }); const repoAtUri = computed(() => repoQuery.data.value?.atUri ?? ""); const markdownContext = computed(() => { - if (!knotHost.value || !knotRepo.value || !branchQuery.data.value?.name) return undefined; + if (!owner.value || !repoName.value || !branchQuery.data.value?.name) return undefined; return { owner: owner.value, repo: repoName.value, branch: branchQuery.data.value.name, - knotHost: knotHost.value, - knotRepo: knotRepo.value, }; }); -const pullQuery = usePullRequestDetail(pds, did, owner, pullId, { enabled: hasIdentity }); -const pullAtUri = computed(() => pullQuery.data.value?.atUri ?? ""); -const commentsQuery = usePullRequestComments(pds, did, owner, pullAtUri, { - enabled: computed(() => !!pullAtUri.value), +const pullQuery = usePullRequestDetail(owner, pullId, { enabled: computed(() => !!owner.value && !!pullId.value) }); +const commentsQuery = usePullRequestComments(owner, pullId, { + enabled: computed(() => !!owner.value && !!pullId.value), }); const pullRequest = computed(() => { @@ -154,14 +143,13 @@ const pullRequest = computed(() => { const comments = computed(() => commentsQuery.data.value ?? []); const isLoading = computed( - () => - identity.isPending.value || repoQuery.isPending.value || pullQuery.isPending.value || commentsQuery.isPending.value, + () => repoQuery.isPending.value || pullQuery.isPending.value || commentsQuery.isPending.value, ); const isError = computed( - () => identity.isError.value || repoQuery.isError.value || pullQuery.isError.value || commentsQuery.isError.value, + () => repoQuery.isError.value || pullQuery.isError.value || commentsQuery.isError.value, ); const errorMessage = computed(() => { - const error = identity.error.value ?? repoQuery.error.value ?? pullQuery.error.value ?? commentsQuery.error.value; + const error = repoQuery.error.value ?? pullQuery.error.value ?? commentsQuery.error.value; return error instanceof Error ? error.message : "An unexpected error occurred."; }); diff --git a/apps/twisted/src/features/repo/RepoDetailPage.vue b/apps/twisted/src/features/repo/RepoDetailPage.vue index 83fab64..af3795d 100644 --- a/apps/twisted/src/features/repo/RepoDetailPage.vue +++ b/apps/twisted/src/features/repo/RepoDetailPage.vue @@ -45,8 +45,8 @@ :markdown-context="markdownContext" /> { router.replace({ path: route.path, query: nextQuery }); }); -const identity = useIdentity(owner, { enabled: computed(() => !!owner.value) }); -const did = computed(() => identity.data.value?.did ?? ""); -const pds = computed(() => identity.data.value?.pds ?? ""); -const hasIdentity = computed(() => !!identity.data.value); +const recordQuery = useRepoRecord(owner, repoName, { enabled: computed(() => !!owner.value && !!repoName.value) }); +const hasRecord = computed(() => !!recordQuery.data.value); -const recordQuery = useRepoRecord(pds, did, repoName, owner, { enabled: hasIdentity }); -const knotHost = computed(() => recordQuery.data.value?.knot ?? ""); -const knotRepo = computed(() => (did.value && repoName.value ? `${did.value}/${repoName.value}` : "")); -const hasRecord = computed(() => !!recordQuery.data.value?.knot && !!did.value); - -const branchQuery = useDefaultBranch(knotHost, knotRepo, { enabled: hasRecord }); +const branchQuery = useDefaultBranch(owner, repoName, { enabled: hasRecord }); const defaultBranch = computed(() => branchQuery.data.value?.name ?? ""); const hasBranch = computed(() => !!branchQuery.data.value?.name); const markdownContext = computed(() => { - if (!knotHost.value || !knotRepo.value || !defaultBranch.value) return undefined; + if (!owner.value || !repoName.value || !defaultBranch.value) return undefined; return { owner: owner.value, repo: repoName.value, branch: defaultBranch.value, - knotHost: knotHost.value, - knotRepo: knotRepo.value, sourcePath: "README.md", }; }); -const languagesQuery = useRepoLanguages(knotHost, knotRepo, undefined, { enabled: hasBranch }); -const readmeQuery = useRepoBlob(knotHost, knotRepo, defaultBranch, "README.md", { readme: true, enabled: hasBranch }); -const logQuery = useRepoLog(knotHost, knotRepo, defaultBranch, { limit: 20, enabled: hasBranch }); +const languagesQuery = useRepoLanguages(owner, repoName, undefined, { enabled: hasBranch }); +const readmeQuery = useRepoBlob(owner, repoName, defaultBranch, "README.md", { readme: true, enabled: hasBranch }); +const logQuery = useRepoLog(owner, repoName, defaultBranch, { limit: 20, enabled: hasBranch }); const repo = computed((): RepoDetail | undefined => { const rec = recordQuery.data.value; @@ -176,8 +166,8 @@ const hasAtUri = computed(() => !!repoAtUri.value); const starCountQuery = useRepoStarCount(repoAtUri, { enabled: hasAtUri }); -const issuesQuery = useRepoIssues(pds, did, owner, repoAtUri, { enabled: hasAtUri }); -const prsQuery = useRepoPRs(pds, did, owner, repoAtUri, { enabled: hasAtUri }); +const issuesQuery = useRepoIssues(owner, repoName, { enabled: hasAtUri }); +const prsQuery = useRepoPRs(owner, repoName, { enabled: hasAtUri }); const commits = computed(() => logQuery.data.value ?? []); const issues = computed(() => issuesQuery.data.value ?? []); @@ -189,10 +179,10 @@ const tabPrefix = computed(() => { return "/tabs/home"; }); -const isLoading = computed(() => identity.isPending.value || recordQuery.isPending.value); -const isError = computed(() => identity.isError.value || recordQuery.isError.value); +const isLoading = computed(() => recordQuery.isPending.value); +const isError = computed(() => recordQuery.isError.value); const errorMessage = computed(() => { - const err = identity.error.value ?? recordQuery.error.value; + const err = recordQuery.error.value; return err instanceof Error ? err.message : "An unexpected error occurred."; }); diff --git a/apps/twisted/src/features/repo/RepoFiles.vue b/apps/twisted/src/features/repo/RepoFiles.vue index 14b60d1..9ded590 100644 --- a/apps/twisted/src/features/repo/RepoFiles.vue +++ b/apps/twisted/src/features/repo/RepoFiles.vue @@ -78,17 +78,17 @@ import type { RepoFile } from "@/domain/models/repo.js"; import { highlightCode } from "@/lib/syntax.js"; import { createObjectUrlFromBlobContent } from "@/services/tangled/repo-assets.js"; -const props = defineProps<{ knotHost: string; knotRepo: string; branch: string }>(); +const props = defineProps<{ owner: string; repo: string; branch: string }>(); const selectedFile = ref(null); const currentPath = ref(""); const treeQuery = useRepoTree( - computed(() => props.knotHost), - computed(() => props.knotRepo), + computed(() => props.owner), + computed(() => props.repo), computed(() => props.branch), currentPath, - { enabled: computed(() => !!props.knotHost && !!props.knotRepo && !!props.branch) }, + { enabled: computed(() => !!props.owner && !!props.repo && !!props.branch) }, ); const sortedFiles = computed(() => { @@ -107,8 +107,8 @@ const filePath = computed(() => selectedFile.value?.path ?? ""); const isFileSelected = computed(() => !!selectedFile.value && selectedFile.value.type === "file"); const blobQuery = useRepoBlob( - computed(() => props.knotHost), - computed(() => props.knotRepo), + computed(() => props.owner), + computed(() => props.repo), computed(() => props.branch), filePath, { enabled: isFileSelected }, diff --git a/apps/twisted/src/services/constellation/queries.ts b/apps/twisted/src/services/constellation/queries.ts index 52aa43e..f5334b2 100644 --- a/apps/twisted/src/services/constellation/queries.ts +++ b/apps/twisted/src/services/constellation/queries.ts @@ -1,37 +1,25 @@ /** - * TanStack Query hooks for the Constellation backlink API. - * https://constellation.microcosm.blue - * - * Constellation is a public AT Protocol backlink index. It answers - * "how many records link to this subject?" — star counts, follower - * counts, reaction counts — without requiring authentication. - * - * Calling it directly from the app avoids adding per-resource endpoints - * to the Twister API for every social signal we need. + * TanStack Query hooks for Constellation backlink counts, proxied through the Twister API. */ import { useQuery } from "@tanstack/vue-query"; import { computed, toValue } from "vue"; import type { MaybeRef } from "vue"; +import { getTwisterApiUrl } from "@/core/config/project.js"; -const CONSTELLATION_BASE = "https://constellation.microcosm.blue"; - -// AT Protocol collection + field paths used as Constellation "sources". -const SOURCE_STAR = "sh.tangled.feed.star:subject.uri"; -const SOURCE_FOLLOW = "sh.tangled.graph.follow:subject"; +const SOURCE_STAR = "sh.tangled.feed.star:.subject"; +const SOURCE_FOLLOW = "sh.tangled.graph.follow:.subject"; const MIN = 60_000; async function fetchBacklinksCount(subject: string, source: string): Promise { - const url = new URL(`${CONSTELLATION_BASE}/xrpc/blue.microcosm.links.getBacklinksCount`); + const url = new URL(getTwisterApiUrl("/backlinks/count")); url.searchParams.set("subject", subject); url.searchParams.set("source", source); - const res = await fetch(url.toString(), { - headers: { Accept: "application/json" }, - }); + const res = await fetch(url.toString(), { headers: { Accept: "application/json" } }); if (!res.ok) { - throw new Error(`Constellation request failed: ${res.status}`); + throw new Error(`Backlinks request failed: ${res.status}`); } const data = (await res.json()) as { count: number }; diff --git a/apps/twisted/src/services/jetstream/client.ts b/apps/twisted/src/services/jetstream/client.ts index a09d445..89bc29e 100644 --- a/apps/twisted/src/services/jetstream/client.ts +++ b/apps/twisted/src/services/jetstream/client.ts @@ -1,18 +1,14 @@ /** - * JetstreamClient — subscribes to the AT Protocol Jetstream WebSocket firehose - * and filters for sh.tangled.* collection events, emitting ActivityItems. + * JetstreamClient — subscribes to the Twister activity stream WebSocket, which + * proxies the AT Protocol Jetstream firehose and filters for sh.tangled.* events. * * Connects on demand, auto-reconnects after disconnection, and tracks the last * event cursor so gap-free resume is possible on reconnect. - * - * Data source decision: Jetstream is chosen over PDS polling because it provides - * a public, real-time stream of all network events without requiring authentication - * or prior knowledge of specific user DIDs. PDS polling would require a known list - * of accounts to follow, and the Twister API does not yet expose an activity feed. */ import type { ActivityItem } from "@/domain/models/activity.js"; +import { getTwisterWsUrl } from "@/core/config/project.js"; -const JETSTREAM_URL = "wss://jetstream2.us-east.bsky.network/subscribe"; +const ACTIVITY_STREAM_PATH = "/activity/stream"; const MAX_ITEMS = 200; const RECONNECT_DELAY_MS = 3_000; @@ -160,7 +156,7 @@ export class JetstreamClient { if (this.cursor !== null) { params.set("cursor", String(this.cursor)); } - return `${JETSTREAM_URL}?${params.toString()}`; + return `${getTwisterWsUrl(ACTIVITY_STREAM_PATH)}?${params.toString()}`; } private _open(): void { diff --git a/apps/twisted/src/services/tangled/.gitkeep b/apps/twisted/src/services/tangled/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/apps/twisted/src/services/tangled/endpoints.ts b/apps/twisted/src/services/tangled/endpoints.ts index a77cf19..d9c17d3 100644 --- a/apps/twisted/src/services/tangled/endpoints.ts +++ b/apps/twisted/src/services/tangled/endpoints.ts @@ -1,21 +1,9 @@ /** - * Typed wrappers around XRPC queries to Tangled knots and the AT Protocol PDS. + * Typed wrappers around Twister API endpoints. * - * Knot endpoints use raw fetch so we can control query serialization for the - * `repo=did:.../repoName` parameter. PDS endpoints also use raw fetch because - * some `com.atproto.repo.*` calls are not typed in the installed packages. - * - * --- API Validation Notes (to verify against live endpoints) --- - * Knot XRPC base: https:///xrpc/ (e.g. us-west.tangled.sh) - * PDS XRPC base: https://bsky.social/xrpc/ (or user's own PDS) - * - * CORS: knot endpoints need to be confirmed CORS-safe from a browser context. - * Appview (tangled.org) serves HTML/HTMX — not a JSON API. Profile & repo - * metadata must come from PDS records via com.atproto.repo.getRecord. - * - * Data routing: - * - Git data (tree, blob, log, branches, languages) → knot XRPC - * - Repo metadata & profile → PDS com.atproto.repo.getRecord/listRecords + * The app calls Twister for everything — no direct XRPC calls to PDSes or + * knots. Twister resolves handles, routes to the right knot/PDS, and returns + * the raw Lexicon records so the existing normalizers can still apply. */ import { @@ -23,429 +11,295 @@ import { ShTangledRepoBlob, ShTangledRepoGetDefaultBranch, ShTangledRepoLanguages, - ShTangledRepoTags, ShTangledRepoDiff, ShTangledRepoCompare, ShTangledRepo, ShTangledActorProfile, ShTangledRepoIssue, ShTangledRepoIssueComment, - ShTangledRepoIssueState, ShTangledRepoPull, ShTangledRepoPullComment, - ShTangledRepoPullStatus, ShTangledGraphFollow, ShTangledString, } from "@atcute/tangled"; import { throwOnXrpcError } from "@/services/atproto/client.js"; -import { MalformedResponseError, NotFoundError } from "@/core/errors/tangled.js"; +import { getTwisterApiUrl } from "@/core/config/project.js"; -type KnotParams = Record>; -type BlueskyProfileResponse = { did: string; handle: string; displayName?: string; avatar?: string }; +export type RecordEntry = { uri: string; cid: string; value: T }; +export type IssueEntry = RecordEntry & { state: "open" | "closed" }; +export type PullEntry = RecordEntry & { status: "open" | "merged" | "closed" }; -function encodeKnotQueryParam(key: string, value: string | number | boolean): string { - const encodedValue = encodeURIComponent(String(value)); - return `${encodeURIComponent(key)}=${key === "repo" ? encodedValue.replaceAll("%2F", "/") : encodedValue}`; -} +export type ActorResponse = { + did: string; + handle: string; + pds: string; + profile: RecordEntry; + bsky?: { displayName?: string; avatar?: string } | null; +}; -function buildKnotQuery(params: KnotParams): string { - const pairs: string[] = []; +export type ActorReposResponse = { did: string; handle: string; records: RecordEntry[] }; - for (const [key, rawValue] of Object.entries(params)) { - if (rawValue === undefined) continue; +export type ActorRepoResponse = { + did: string; + handle: string; + knot_host: string; + record: RecordEntry; +}; - if (Array.isArray(rawValue)) { - for (const value of rawValue) { - pairs.push(encodeKnotQueryParam(key, value)); - } - continue; - } +export type ActorIssuesResponse = { did: string; handle: string; records: IssueEntry[] }; - pairs.push(encodeKnotQueryParam(key, rawValue)); - } +export type ActorPullsResponse = { did: string; handle: string; records: PullEntry[] }; - return pairs.length > 0 ? `?${pairs.join("&")}` : ""; -} +export type ActorFollowingResponse = { did: string; handle: string; records: RecordEntry[] }; -export function buildKnotUrl(knotHost: string, nsid: string, params: KnotParams): string { - return `https://${knotHost}/xrpc/${nsid}${buildKnotQuery(params)}`; -} +export type ActorStringsResponse = { did: string; handle: string; records: RecordEntry[] }; + +export type IssueDetailResponse = IssueEntry; -async function readKnotError(res: Response): Promise { - const contentType = res.headers.get("content-type") ?? ""; +export type IssueCommentsResponse = { + did: string; + handle: string; + issueUri: string; + records: RecordEntry[]; +}; - if (contentType.includes("application/json")) { +export type PullDetailResponse = PullEntry; + +export type PullCommentsResponse = { + did: string; + handle: string; + pullUri: string; + records: RecordEntry[]; +}; + +async function get(path: string, params?: Record): Promise { + const url = new URL(getTwisterApiUrl(path)); + if (params) { + for (const [key, value] of Object.entries(params)) { + if (value !== undefined) url.searchParams.set(key, value); + } + } + const res = await fetch(url.toString()); + if (!res.ok) { const body = (await res.json().catch(() => ({}))) as { error?: string; message?: string }; throwOnXrpcError(res.status, body.error ?? "Unknown", body.message); } - - const text = await res.text().catch(() => ""); - throwOnXrpcError(res.status, "Unknown", text || undefined); + return res.json() as Promise; } -async function fetchKnotJson(knotHost: string, nsid: string, params: KnotParams): Promise { - const res = await fetch(buildKnotUrl(knotHost, nsid, params)); - if (!res.ok) return readKnotError(res); - return res.json() as Promise; +async function getProxy(path: string, params?: Record): Promise { + const normalized: Record = {}; + if (params) { + for (const [key, value] of Object.entries(params)) { + normalized[key] = value === undefined ? undefined : String(value); + } + } + return get(path, normalized); } -async function fetchKnotBytes(knotHost: string, nsid: string, params: KnotParams): Promise { - const res = await fetch(buildKnotUrl(knotHost, nsid, params)); - if (!res.ok) return readKnotError(res); +async function getBytes(path: string, params?: Record): Promise { + const url = new URL(getTwisterApiUrl(path)); + if (params) { + for (const [key, value] of Object.entries(params)) { + if (value !== undefined) url.searchParams.set(key, value); + } + } + const res = await fetch(url.toString()); + if (!res.ok) { + const body = (await res.json().catch(() => ({}))) as { error?: string; message?: string }; + throwOnXrpcError(res.status, body.error ?? "Unknown", body.message); + } return new Uint8Array(await res.arrayBuffer()); } -export async function fetchRepoTree( - knotHost: string, - params: ShTangledRepoTree.$params, -): Promise { - return fetchKnotJson(knotHost, "sh.tangled.repo.tree", params); +export async function fetchActor(handle: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}`); } -export async function fetchRepoBlob( - knotHost: string, - params: ShTangledRepoBlob.$params, -): Promise { - return fetchKnotJson(knotHost, "sh.tangled.repo.blob", params); +export async function fetchActorRepos(handle: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/repos`); } -export async function fetchDefaultBranch( - knotHost: string, - params: ShTangledRepoGetDefaultBranch.$params, -): Promise { - return fetchKnotJson(knotHost, "sh.tangled.repo.getDefaultBranch", params); +export async function fetchActorRepo(handle: string, repo: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}`); } -export async function fetchLanguages( - knotHost: string, - params: ShTangledRepoLanguages.$params, -): Promise { - return fetchKnotJson(knotHost, "sh.tangled.repo.languages", params); +export async function fetchActorFollowing(handle: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/following`); } -/** - * Fetch commit log. The wire format is a raw blob; the decoded text is returned - * as-is so the normalizer can handle it once the format is confirmed against - * the live API. Expected: newline-delimited JSON or git log text. - */ -export async function fetchRepoLog( - knotHost: string, - params: { repo: string; ref: string; path?: string; limit?: number; cursor?: string }, -): Promise { - return new TextDecoder().decode(await fetchKnotBytes(knotHost, "sh.tangled.repo.log", params)); +export async function fetchActorStrings(handle: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/strings`); } -/** - * Fetch branch list. The wire format is a raw blob; decoded text is returned - * for the normalizer to parse once the live format is confirmed. - */ -export async function fetchRepoBranches( - knotHost: string, - params: { repo: string; limit?: number; cursor?: string }, -): Promise { - return new TextDecoder().decode(await fetchKnotBytes(knotHost, "sh.tangled.repo.branches", params)); +export async function fetchActorIssues(handle: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/issues`); } -/** Tag list. Wire format is a raw blob — decoded text returned for normalizer. */ -export async function fetchRepoTags(knotHost: string, params: ShTangledRepoTags.$params): Promise { - return new TextDecoder().decode(await fetchKnotBytes(knotHost, "sh.tangled.repo.tags", params)); +export async function fetchActorPulls(handle: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/pulls`); } -/** Diff for a ref. Wire format is a raw blob — patch text. */ -export async function fetchRepoDiff(knotHost: string, params: ShTangledRepoDiff.$params): Promise { - return new TextDecoder().decode(await fetchKnotBytes(knotHost, "sh.tangled.repo.diff", params)); -} +type ResolveHandleResponse = { did: string }; -/** Comparison between two revisions. Wire format is a raw blob — patch text. */ -export async function fetchRepoCompare(knotHost: string, params: ShTangledRepoCompare.$params): Promise { - return new TextDecoder().decode(await fetchKnotBytes(knotHost, "sh.tangled.repo.compare", params)); -} +type DidDocument = { + service?: Array<{ id?: string; type?: string; serviceEndpoint?: string }>; + alsoKnownAs?: string[]; +}; -type GetRecordResponse = { uri: string; cid: string; value: T }; +export async function resolveHandle(handle: string): Promise { + const data = await getProxy("/identity/resolve", { handle }); + return data.did; +} -/** - * Fetch a single record from the AT Protocol PDS. - * Uses raw fetch against /xrpc/com.atproto.repo.getRecord since this NSID - * is not currently typed in the installed @atcute packages. - */ -async function getRecord( - pds: string, - repo: string, - collection: string, - rkey: string, -): Promise> { - const url = new URL(`https://${pds}/xrpc/com.atproto.repo.getRecord`); - url.searchParams.set("repo", repo); - url.searchParams.set("collection", collection); - url.searchParams.set("rkey", rkey); +export async function fetchDidDocument(did: string): Promise { + return get(`/identity/did/${encodeURIComponent(did)}`); +} - const res = await fetch(url.toString()); - if (!res.ok) { - const body = (await res.json().catch(() => ({}))) as { error?: string; message?: string }; - throwOnXrpcError(res.status, body.error ?? "Unknown", body.message); +export async function resolvePds(did: string): Promise { + const doc = await fetchDidDocument(did); + const endpoint = doc.service?.find((entry) => entry.id === "#atproto_pds")?.serviceEndpoint; + if (!endpoint) { + throw new Error(`No PDS endpoint found in DID document for ${did}`); } - return res.json() as Promise>; + return new URL(endpoint).hostname; } -export async function fetchActorProfile( - pds: string, - did: string, -): Promise> { - return getRecord(pds, did, "sh.tangled.actor.profile", "self"); +export async function resolveDidIdentity(did: string): Promise<{ did: string; handle: string; pds: string }> { + const actor = await fetchActor(did); + return { did: actor.did, handle: actor.handle, pds: new URL(actor.pds).hostname }; } -export async function fetchBlueskyProfile(actor: string): Promise { - const url = new URL("https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile"); - url.searchParams.set("actor", actor); - - const res = await fetch(url.toString()); - if (!res.ok) { - const body = (await res.json().catch(() => ({}))) as { error?: string; message?: string }; - throwOnXrpcError(res.status, body.error ?? "Unknown", body.message); - } +export async function fetchBskyXrpc(nsid: string, params?: Record): Promise { + return getProxy(`/xrpc/bsky/${encodeURIComponent(nsid)}`, params); +} - return res.json() as Promise; +export async function fetchPdsXrpc( + pdsHost: string, + nsid: string, + params?: Record, +): Promise { + return getProxy(`/xrpc/pds/${encodeURIComponent(pdsHost)}/${encodeURIComponent(nsid)}`, params); } -/** - * Fetch a repo record by its PDS record key. - * This is distinct from the repo's `name`, which is the identifier used by - * knot endpoints in the `did:.../repoName` format. - */ -export async function fetchRepoRecord( - pds: string, - did: string, - rkey: string, -): Promise> { - return getRecord(pds, did, "sh.tangled.repo", rkey); +export async function fetchKnotXrpc( + knotHost: string, + nsid: string, + params?: Record, +): Promise { + return getProxy(`/xrpc/knot/${encodeURIComponent(knotHost)}/${encodeURIComponent(nsid)}`, params); } -/** - * Fetch a repo record by matching on the record's `name` field. - * Use this when the UI route or knot API identifies a repo by repo name rather - * than by the underlying AT Protocol record key. - */ -export async function fetchRepoRecordByName( - pds: string, - did: string, - repoName: string, -): Promise> { - let cursor: string | undefined; - - for (;;) { - const response = await listRepoRecords(pds, did, 100, cursor); - const record = response.records.find((entry) => entry.value.name === repoName); - if (record) return record; - if (!response.cursor) break; - cursor = response.cursor; - } +export async function fetchRepoTree( + handle: string, + repo: string, + params: ShTangledRepoTree.$params, +): Promise { + return get( + `/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/tree`, + { ref: params.ref, path: params.path }, + ); +} - throw new NotFoundError(`Repository ${repoName}`); +export async function fetchRepoBlob( + handle: string, + repo: string, + params: ShTangledRepoBlob.$params, +): Promise { + return get( + `/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/blob`, + { ref: params.ref, path: params.path }, + ); } -export async function fetchIssueRecord( - pds: string, - did: string, - rkey: string, -): Promise> { - return getRecord(pds, did, "sh.tangled.repo.issue", rkey); +export async function fetchDefaultBranch(handle: string, repo: string): Promise { + return get( + `/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/default-branch`, + ); } -export async function fetchPullRecord( - pds: string, - did: string, - rkey: string, -): Promise> { - return getRecord(pds, did, "sh.tangled.repo.pull", rkey); +export async function fetchLanguages( + handle: string, + repo: string, + ref?: string, +): Promise { + return get( + `/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/languages`, + ref ? { ref } : undefined, + ); } -/** - * Resolve an AT Protocol handle to a DID via bsky.social. - * Returns the DID string (e.g. "did:plc:xxx"). - */ -export async function resolveHandle(handle: string): Promise { - const url = new URL("https://bsky.social/xrpc/com.atproto.identity.resolveHandle"); - url.searchParams.set("handle", handle); - const res = await fetch(url.toString()); - if (!res.ok) { - const body = (await res.json().catch(() => ({}))) as { error?: string; message?: string }; - throwOnXrpcError(res.status, body.error ?? "Unknown", body.message); - } - const data = (await res.json()) as { did: string }; - return data.did; +export async function fetchRepoLog( + handle: string, + repo: string, + params: { ref: string; path?: string; limit?: number; cursor?: string }, +): Promise { + const p: Record = { ref: params.ref, path: params.path }; + if (params.limit !== undefined) p.limit = String(params.limit); + if (params.cursor !== undefined) p.cursor = params.cursor; + return new TextDecoder().decode( + await getBytes(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/log`, p), + ); } -type DidDocument = { alsoKnownAs?: string[]; service?: Array<{ id: string; type: string; serviceEndpoint: string }> }; +export async function fetchRepoBranches( + handle: string, + repo: string, + params?: { limit?: number; cursor?: string }, +): Promise { + const p: Record = {}; + if (params?.limit !== undefined) p.limit = String(params.limit); + if (params?.cursor !== undefined) p.cursor = params.cursor; + return new TextDecoder().decode( + await getBytes(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/branches`, p), + ); +} -async function fetchDidDocument(did: string): Promise { - let docUrl: string; - if (did.startsWith("did:plc:")) { - docUrl = `https://plc.directory/${did}`; - } else if (did.startsWith("did:web:")) { - const host = did.slice("did:web:".length); - docUrl = `https://${host}/.well-known/did.json`; - } else { - throw new MalformedResponseError("resolveHandle", `Unsupported DID method: ${did}`); - } +export async function fetchRepoTags(handle: string, repo: string): Promise { + return new TextDecoder().decode( + await getBytes(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/tags`), + ); +} - const res = await fetch(docUrl); - if (!res.ok) throwOnXrpcError(res.status, "ResolveFailed", `Could not fetch DID document: ${did}`); - return (await res.json()) as DidDocument; +export async function fetchRepoDiff(handle: string, repo: string, params: ShTangledRepoDiff.$params): Promise { + return new TextDecoder().decode( + await getBytes(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/diff`, { ref: params.ref }), + ); } -/** - * Fetch the DID document for a DID and extract the PDS service endpoint hostname. - * Supports did:plc (via plc.directory) and did:web. - */ -export async function resolvePds(did: string): Promise { - const doc = await fetchDidDocument(did); - const svc = doc.service?.find((s) => s.id === "#atproto_pds"); - if (!svc?.serviceEndpoint) { - throw new MalformedResponseError("resolvePds", `No PDS endpoint in DID document: ${did}`); - } - return new URL(svc.serviceEndpoint).hostname; +export async function fetchRepoCompare( + handle: string, + repo: string, + params: ShTangledRepoCompare.$params, +): Promise { + return new TextDecoder().decode( + await getBytes(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/compare`, { + from: params.rev1, + to: params.rev2, + }), + ); } -export async function resolveHandleFromDid(did: string): Promise { - if (did.startsWith("did:web:")) return did.slice("did:web:".length); +export async function fetchRepoIssues(handle: string, repo: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/issues`); +} - const doc = await fetchDidDocument(did); - const alias = doc.alsoKnownAs?.find((entry) => entry.startsWith("at://")); - if (!alias) { - throw new MalformedResponseError("resolveHandleFromDid", `No handle alias in DID document: ${did}`); - } +export async function fetchRepoPulls(handle: string, repo: string): Promise { + return get(`/actors/${encodeURIComponent(handle)}/repos/${encodeURIComponent(repo)}/pulls`); +} - return alias.slice("at://".length); +export async function fetchIssueDetail(handle: string, rkey: string): Promise { + return get(`/issues/${encodeURIComponent(handle)}/${encodeURIComponent(rkey)}`); } -export async function resolveDidIdentity(did: string): Promise<{ did: string; handle: string; pds: string }> { - const [handle, pds] = await Promise.all([resolveHandleFromDid(did), resolvePds(did)]); - return { did, handle, pds }; -} - -type ListRecordsResponse = { records: Array<{ uri: string; cid: string; value: T }>; cursor?: string }; - -async function listRecords( - pds: string, - did: string, - collection: string, - limit = 50, - cursor?: string, -): Promise> { - const url = new URL(`https://${pds}/xrpc/com.atproto.repo.listRecords`); - url.searchParams.set("repo", did); - url.searchParams.set("collection", collection); - url.searchParams.set("limit", String(limit)); - if (cursor) url.searchParams.set("cursor", cursor); - const res = await fetch(url.toString()); - if (!res.ok) { - const body = (await res.json().catch(() => ({}))) as { error?: string; message?: string }; - throwOnXrpcError(res.status, body.error ?? "Unknown", body.message); - } - return res.json() as Promise>; -} - -/** List sh.tangled.repo.issue records from a user's PDS. */ -export async function listIssueRecords( - pds: string, - did: string, - limit = 50, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.repo.issue", limit, cursor); -} - -/** List sh.tangled.repo.issue.state records from a user's PDS. */ -export async function listIssueStateRecords( - pds: string, - did: string, - limit = 100, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.repo.issue.state", limit, cursor); -} - -/** List sh.tangled.repo.issue.comment records from a user's PDS. */ -export async function listIssueCommentRecords( - pds: string, - did: string, - limit = 100, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.repo.issue.comment", limit, cursor); -} - -/** List sh.tangled.repo.pull records from a user's PDS. */ -export async function listPullRecords( - pds: string, - did: string, - limit = 50, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.repo.pull", limit, cursor); -} - -/** List sh.tangled.repo.pull.status records from a user's PDS. */ -export async function listPullStatusRecords( - pds: string, - did: string, - limit = 100, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.repo.pull.status", limit, cursor); -} - -/** List sh.tangled.repo.pull.comment records from a user's PDS. */ -export async function listPullCommentRecords( - pds: string, - did: string, - limit = 100, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.repo.pull.comment", limit, cursor); -} - -export async function listFollowRecords( - pds: string, - did: string, - limit = 100, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.graph.follow", limit, cursor); -} - -export async function listStringRecords( - pds: string, - did: string, - limit = 100, - cursor?: string, -): Promise> { - return listRecords(pds, did, "sh.tangled.string", limit, cursor); +export async function fetchIssueComments(handle: string, rkey: string): Promise { + return get(`/issues/${encodeURIComponent(handle)}/${encodeURIComponent(rkey)}/comments`); } -/** - * List all sh.tangled.repo records from a user's PDS. - * Uses com.atproto.repo.listRecords since it's not in the installed lexicons. - */ -export async function listRepoRecords( - pds: string, - did: string, - limit = 50, - cursor?: string, -): Promise<{ records: Array<{ uri: string; cid: string; value: ShTangledRepo.Main }>; cursor?: string }> { - const url = new URL(`https://${pds}/xrpc/com.atproto.repo.listRecords`); - url.searchParams.set("repo", did); - url.searchParams.set("collection", "sh.tangled.repo"); - url.searchParams.set("limit", String(limit)); - if (cursor) url.searchParams.set("cursor", cursor); +export async function fetchPullDetail(handle: string, rkey: string): Promise { + return get(`/pulls/${encodeURIComponent(handle)}/${encodeURIComponent(rkey)}`); +} - const res = await fetch(url.toString()); - if (!res.ok) { - const body = (await res.json().catch(() => ({}))) as { error?: string; message?: string }; - throwOnXrpcError(res.status, body.error ?? "Unknown", body.message); - } - return res.json() as Promise<{ - records: Array<{ uri: string; cid: string; value: ShTangledRepo.Main }>; - cursor?: string; - }>; +export async function fetchPullComments(handle: string, rkey: string): Promise { + return get(`/pulls/${encodeURIComponent(handle)}/${encodeURIComponent(rkey)}/comments`); } diff --git a/apps/twisted/src/services/tangled/queries.ts b/apps/twisted/src/services/tangled/queries.ts index de7fd24..856da57 100644 --- a/apps/twisted/src/services/tangled/queries.ts +++ b/apps/twisted/src/services/tangled/queries.ts @@ -1,16 +1,6 @@ /** - * TanStack Query hooks for Tangled data. - * These are the only entry points Vue components should use — no direct - * imports of @atcute/* or service/endpoint functions in components. - * - * Cache strategy: - * Repo metadata stale: 5m gc: 30m - * File tree stale: 2m gc: 10m - * File content stale: 5m gc: 30m - * Commit log stale: 2m gc: 10m - * Branches stale: 2m gc: 10m - * Profile stale: 10m gc: 60m - * README stale: 5m gc: 30m + * TanStack Query hooks for Tangled data via Twister API. + * Components should only use these hooks rather than endpoint wrappers directly. */ import { useQuery } from "@tanstack/vue-query"; @@ -19,6 +9,13 @@ import type { MaybeRef } from "vue"; import type { FollowedUserSummary } from "@/domain/models/follow.js"; import type { StringSummary } from "@/domain/models/string.js"; import { + fetchActor, + fetchActorRepos, + fetchActorRepo, + fetchActorFollowing, + fetchActorStrings, + fetchActorIssues, + fetchActorPulls, fetchRepoTree, fetchRepoBlob, fetchDefaultBranch, @@ -28,23 +25,12 @@ import { fetchRepoTags, fetchRepoDiff, fetchRepoCompare, - fetchActorProfile, - fetchBlueskyProfile, - fetchRepoRecordByName, - fetchIssueRecord, - fetchPullRecord, - listRepoRecords, - listIssueRecords, - listIssueCommentRecords, - listIssueStateRecords, - listPullRecords, - listPullCommentRecords, - listPullStatusRecords, - listFollowRecords, - listStringRecords, - resolveHandle, - resolveDidIdentity, - resolvePds, + fetchRepoIssues, + fetchRepoPulls, + fetchIssueDetail, + fetchIssueComments, + fetchPullDetail, + fetchPullComments, } from "./endpoints.js"; import { normalizeTree, @@ -79,36 +65,18 @@ function isEnabled(required: boolean, enabled?: MaybeRef): boolean { return required && (enabled === undefined || !!toValue(enabled)); } -async function resolveBlueskyProfile( - record: Awaited>["value"], - did: string, -): Promise<{ displayName?: string; avatar?: string }> { - if (!record.bluesky) return {}; - - try { - const profile = await fetchBlueskyProfile(did); - return { displayName: profile.displayName, avatar: profile.avatar }; - } catch { - return {}; - } -} - /** Resolved identity: DID + PDS hostname for an AT Protocol handle. */ export type Identity = { did: string; pds: string }; -/** - * Resolve an AT Protocol handle to its DID and PDS hostname. - * Result is cached for 10 minutes (handles rarely change). - */ +/** Resolve a handle through Twister actor endpoint. */ export function useIdentity(handle: MaybeRef, options: { enabled?: MaybeRef } = {}) { const normalizedHandle = computed(() => toValue(handle).trim()); return useQuery({ queryKey: computed(() => ["identity", normalizedHandle.value]), queryFn: async (): Promise => { - const did = await resolveHandle(normalizedHandle.value); - const pds = await resolvePds(did); - return { did, pds }; + const actor = await fetchActor(normalizedHandle.value); + return { did: actor.did, pds: new URL(actor.pds).hostname }; }, enabled: computed(() => isEnabled(hasText(normalizedHandle), options.enabled)), staleTime: 10 * MIN, @@ -118,21 +86,22 @@ export function useIdentity(handle: MaybeRef, options: { enabled?: Maybe /** File tree for a path within a repo. */ export function useRepoTree( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, ref: MaybeRef, path: MaybeRef = undefined, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["tree", toValue(knotHost), toValue(repo), toValue(ref), toValue(path)]), + queryKey: computed(() => ["tree", h.value, r.value, toValue(ref), toValue(path)]), queryFn: () => - fetchRepoTree(toValue(knotHost), { - repo: toValue(repo), - ref: toValue(ref), - path: toValue(path), - }).then((out) => normalizeTree(out, toValue(path) ?? "")), - enabled: options.enabled, + fetchRepoTree(h.value, r.value, { repo: `${h.value}/${r.value}`, ref: toValue(ref), path: toValue(path) }).then( + (out) => normalizeTree(out, toValue(path) ?? ""), + ), + enabled: computed(() => isEnabled(hasText(h) && hasText(r) && hasText(ref), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); @@ -140,21 +109,22 @@ export function useRepoTree( /** Raw file content (blob) for a specific path and ref. */ export function useRepoBlob( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, ref: MaybeRef, path: MaybeRef, options: { readme?: boolean; enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["blob", toValue(knotHost), toValue(repo), toValue(ref), toValue(path)]), + queryKey: computed(() => ["blob", h.value, r.value, toValue(ref), toValue(path)]), queryFn: () => - fetchRepoBlob(toValue(knotHost), { - repo: toValue(repo), - ref: toValue(ref), - path: toValue(path), - }).then(normalizeBlob), - enabled: options.enabled, + fetchRepoBlob(h.value, r.value, { repo: `${h.value}/${r.value}`, ref: toValue(ref), path: toValue(path) }).then( + normalizeBlob, + ), + enabled: computed(() => isEnabled(hasText(h) && hasText(r) && hasText(ref) && hasText(path), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); @@ -162,15 +132,17 @@ export function useRepoBlob( /** Default branch name + latest commit for a repo. */ export function useDefaultBranch( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["defaultBranch", toValue(knotHost), toValue(repo)]), - queryFn: () => - fetchDefaultBranch(toValue(knotHost), { repo: toValue(repo) }).then(normalizeDefaultBranch), - enabled: options.enabled, + queryKey: computed(() => ["defaultBranch", h.value, r.value]), + queryFn: () => fetchDefaultBranch(h.value, r.value).then(normalizeDefaultBranch), + enabled: computed(() => isEnabled(hasText(h) && hasText(r), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); @@ -178,16 +150,18 @@ export function useDefaultBranch( /** Language breakdown for a repo (percentages). */ export function useRepoLanguages( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, ref?: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["languages", toValue(knotHost), toValue(repo), toValue(ref)]), - queryFn: () => - fetchLanguages(toValue(knotHost), { repo: toValue(repo), ref: toValue(ref) }).then(normalizeLanguages), - enabled: options.enabled, + queryKey: computed(() => ["languages", h.value, r.value, toValue(ref)]), + queryFn: () => fetchLanguages(h.value, r.value, toValue(ref)).then(normalizeLanguages), + enabled: computed(() => isEnabled(hasText(h) && hasText(r), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); @@ -195,7 +169,7 @@ export function useRepoLanguages( /** Paginated commit log for a repo/ref. */ export function useRepoLog( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, ref: MaybeRef, options: { @@ -205,24 +179,19 @@ export function useRepoLog( enabled?: MaybeRef; } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => [ - "log", - toValue(knotHost), - toValue(repo), - toValue(ref), - toValue(options.path), - toValue(options.cursor), - ]), + queryKey: computed(() => ["log", h.value, r.value, toValue(ref), toValue(options.path), toValue(options.cursor)]), queryFn: () => - fetchRepoLog(toValue(knotHost), { - repo: toValue(repo), + fetchRepoLog(h.value, r.value, { ref: toValue(ref), path: toValue(options.path), limit: options.limit, cursor: toValue(options.cursor), }).then(normalizeLogText), - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h) && hasText(r) && hasText(ref), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); @@ -230,106 +199,82 @@ export function useRepoLog( /** Branch list for a repo. */ export function useRepoBranches( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, defaultBranch?: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["branches", toValue(knotHost), toValue(repo)]), + queryKey: computed(() => ["branches", h.value, r.value]), queryFn: () => - fetchRepoBranches(toValue(knotHost), { repo: toValue(repo) }).then((raw) => - normalizeBranchesText(raw, toValue(defaultBranch)), - ), - enabled: options.enabled, + fetchRepoBranches(h.value, r.value).then((raw) => normalizeBranchesText(raw, toValue(defaultBranch))), + enabled: computed(() => isEnabled(hasText(h) && hasText(r), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } -/** - * Fetch a repo's PDS record (metadata: description, topics, knot, etc.). - * `pds` is the PDS hostname, e.g. "bsky.social". - */ +/** Fetch a repo record (metadata: description, topics, knot, etc.). */ export function useRepoRecord( - pds: MaybeRef, - did: MaybeRef, - repoName: MaybeRef, handle: MaybeRef, + repoName: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { - const normalizedPds = computed(() => toValue(pds).trim()); - const normalizedDid = computed(() => toValue(did).trim()); - const normalizedRepoName = computed(() => toValue(repoName).trim()); + const h = computed(() => toValue(handle).trim()); + const repo = computed(() => toValue(repoName).trim()); return useQuery({ - queryKey: computed(() => ["repoRecord", normalizedPds.value, normalizedDid.value, normalizedRepoName.value]), + queryKey: computed(() => ["repoRecord", h.value, repo.value]), queryFn: async () => { - const { value: record, uri } = await fetchRepoRecordByName( - normalizedPds.value, - normalizedDid.value, - normalizedRepoName.value, - ).then((r) => ({ - value: r.value, - uri: r.uri, - })); - return normalizeRepoRecord(record, toValue(did), toValue(handle), uri); + const response = await fetchActorRepo(h.value, repo.value); + return normalizeRepoRecord(response.record.value, response.did, response.handle, response.record.uri); }, - enabled: computed(() => - isEnabled(hasText(normalizedPds) && hasText(normalizedDid) && hasText(normalizedRepoName), options.enabled), - ), + enabled: computed(() => isEnabled(hasText(h) && hasText(repo), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); } -/** List all repos for a user from their PDS. */ -export function useUserRepos( - pds: MaybeRef, - did: MaybeRef, - handle: MaybeRef, - options: { enabled?: MaybeRef } = {}, -) { - const normalizedPds = computed(() => toValue(pds).trim()); - const normalizedDid = computed(() => toValue(did).trim()); +/** List all repos for a user. */ +export function useUserRepos(handle: MaybeRef, options: { enabled?: MaybeRef } = {}) { + const h = computed(() => toValue(handle).trim()); return useQuery({ - queryKey: computed(() => ["userRepos", normalizedPds.value, normalizedDid.value]), + queryKey: computed(() => ["userRepos", h.value]), queryFn: async () => { - const { records } = await listRepoRecords(normalizedPds.value, normalizedDid.value); - return records.map((r) => normalizeRepoRecord(r.value, toValue(did), toValue(handle), r.uri)); + const response = await fetchActorRepos(h.value); + return response.records.map((r) => normalizeRepoRecord(r.value, response.did, response.handle, r.uri)); }, - enabled: computed(() => isEnabled(hasText(normalizedPds) && hasText(normalizedDid), options.enabled)), + enabled: computed(() => isEnabled(hasText(h), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); } -/** Fetch a user's Tangled actor profile from their PDS. */ +/** Fetch a user's Tangled actor profile. */ export function useActorProfile( - pds: MaybeRef, - did: MaybeRef, handle: MaybeRef, displayName?: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { - const normalizedPds = computed(() => toValue(pds).trim()); - const normalizedDid = computed(() => toValue(did).trim()); + const h = computed(() => toValue(handle).trim()); return useQuery({ - queryKey: computed(() => ["actorProfile", normalizedPds.value, normalizedDid.value]), + queryKey: computed(() => ["actorProfile", h.value]), queryFn: async () => { - const { value } = await fetchActorProfile(normalizedPds.value, normalizedDid.value); - const bluesky = await resolveBlueskyProfile(value, normalizedDid.value); + const response = await fetchActor(h.value); return normalizeActorProfile( - value, - toValue(did), - toValue(handle), - bluesky.displayName ?? toValue(displayName), - bluesky.avatar, + response.profile.value, + response.did, + response.handle, + response.bsky?.displayName ?? toValue(displayName), + response.bsky?.avatar, ); }, - enabled: computed(() => isEnabled(hasText(normalizedPds) && hasText(normalizedDid), options.enabled)), + enabled: computed(() => isEnabled(hasText(h), options.enabled)), staleTime: 10 * MIN, gcTime: 60 * MIN, }); @@ -337,15 +282,17 @@ export function useActorProfile( /** Tag list for a repo. Wire format is a raw blob; parsed as lines. */ export function useRepoTags( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["tags", toValue(knotHost), toValue(repo)]), - queryFn: () => - fetchRepoTags(toValue(knotHost), { repo: toValue(repo) }).then((raw) => raw.trim().split("\n").filter(Boolean)), - enabled: options.enabled, + queryKey: computed(() => ["tags", h.value, r.value]), + queryFn: () => fetchRepoTags(h.value, r.value).then((raw) => raw.trim().split("\n").filter(Boolean)), + enabled: computed(() => isEnabled(hasText(h) && hasText(r), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); @@ -353,15 +300,18 @@ export function useRepoTags( /** Unified diff for a ref (patch text). */ export function useRepoDiff( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, ref: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["diff", toValue(knotHost), toValue(repo), toValue(ref)]), - queryFn: () => fetchRepoDiff(toValue(knotHost), { repo: toValue(repo), ref: toValue(ref) }), - enabled: options.enabled, + queryKey: computed(() => ["diff", h.value, r.value, toValue(ref)]), + queryFn: () => fetchRepoDiff(h.value, r.value, { repo: `${h.value}/${r.value}`, ref: toValue(ref) }), + enabled: computed(() => isEnabled(hasText(h) && hasText(r) && hasText(ref), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); @@ -369,152 +319,118 @@ export function useRepoDiff( /** Comparison diff between two revisions (patch text). */ export function useRepoCompare( - knotHost: MaybeRef, + handle: MaybeRef, repo: MaybeRef, rev1: MaybeRef, rev2: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["compare", toValue(knotHost), toValue(repo), toValue(rev1), toValue(rev2)]), + queryKey: computed(() => ["compare", h.value, r.value, toValue(rev1), toValue(rev2)]), queryFn: () => - fetchRepoCompare(toValue(knotHost), { - repo: toValue(repo), - rev1: toValue(rev1), - rev2: toValue(rev2), - }), - enabled: options.enabled, + fetchRepoCompare(h.value, r.value, { repo: `${h.value}/${r.value}`, rev1: toValue(rev1), rev2: toValue(rev2) }), + enabled: computed(() => isEnabled(hasText(h) && hasText(r) && hasText(rev1) && hasText(rev2), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); } -/** - * Issues for a repo. Lists sh.tangled.repo.issue records from the owner's PDS, - * filtered by repo AT URI, joined with state from sh.tangled.repo.issue.state. - */ +/** Issues for a repo, returned with current state by the backend. */ export function useRepoIssues( - pds: MaybeRef, - did: MaybeRef, handle: MaybeRef, - repoAtUri: MaybeRef, + repo: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["issues", toValue(pds), toValue(did), toValue(repoAtUri)]), + queryKey: computed(() => ["issues", h.value, r.value]), queryFn: async () => { - const [issuesRes, statesRes] = await Promise.all([ - listIssueRecords(toValue(pds), toValue(did)), - listIssueStateRecords(toValue(pds), toValue(did)), - ]); - - const stateMap = new Map(); - for (const s of statesRes.records) { - const closed = s.value.state === "sh.tangled.repo.issue.state.closed"; - stateMap.set(s.value.issue, closed ? "closed" : "open"); - } - - const targetRepo = toValue(repoAtUri); - return issuesRes.records - .filter((r) => !targetRepo || r.value.repo === targetRepo) - .map((r) => normalizeIssueRecord(r.value, r.uri, toValue(did), toValue(handle), stateMap.get(r.uri) ?? "open")); + const response = await fetchRepoIssues(h.value, r.value); + return response.records.map((entry) => + normalizeIssueRecord(entry.value, entry.uri, response.did, response.handle, entry.state), + ); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h) && hasText(r), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } -/** - * Pull requests for a repo. Lists sh.tangled.repo.pull records from the owner's - * PDS filtered by target repo AT URI, joined with status records. - */ +/** Pull requests for a repo, returned with current status by the backend. */ export function useRepoPRs( - pds: MaybeRef, - did: MaybeRef, handle: MaybeRef, - repoAtUri: MaybeRef, + repo: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const r = computed(() => toValue(repo).trim()); + return useQuery({ - queryKey: computed(() => ["prs", toValue(pds), toValue(did), toValue(repoAtUri)]), + queryKey: computed(() => ["prs", h.value, r.value]), queryFn: async () => { - const [pullsRes, statusesRes] = await Promise.all([ - listPullRecords(toValue(pds), toValue(did)), - listPullStatusRecords(toValue(pds), toValue(did)), - ]); - - const statusMap = new Map(); - for (const s of statusesRes.records) { - const raw = s.value.status ?? "sh.tangled.repo.pull.status.open"; - const status = - raw === "sh.tangled.repo.pull.status.merged" - ? "merged" - : raw === "sh.tangled.repo.pull.status.closed" - ? "closed" - : "open"; - statusMap.set(s.value.pull, status); - } - - const targetRepo = toValue(repoAtUri); - return pullsRes.records - .filter((r) => !targetRepo || r.value.target.repo === targetRepo) - .map((r) => normalizePullRecord(r.value, r.uri, toValue(did), toValue(handle), statusMap.get(r.uri) ?? "open")); + const response = await fetchRepoPulls(h.value, r.value); + return response.records.map((entry) => + normalizePullRecord(entry.value, entry.uri, response.did, response.handle, entry.status), + ); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h) && hasText(r), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } -export function useUserStrings( - pds: MaybeRef, - did: MaybeRef, - options: { enabled?: MaybeRef } = {}, -) { +export function useUserStrings(handle: MaybeRef, options: { enabled?: MaybeRef } = {}) { + const h = computed(() => toValue(handle).trim()); + return useQuery({ - queryKey: computed(() => ["userStrings", toValue(pds), toValue(did)]), + queryKey: computed(() => ["userStrings", h.value]), queryFn: async (): Promise => { - const { records } = await listStringRecords(toValue(pds), toValue(did)); - return records + const response = await fetchActorStrings(h.value); + return response.records .map((record) => normalizeStringRecord(record.value, record.uri)) .sort((left, right) => Date.parse(right.createdAt) - Date.parse(left.createdAt)); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } -export function useUserFollowing( - pds: MaybeRef, - did: MaybeRef, - options: { enabled?: MaybeRef } = {}, -) { +export function useUserFollowing(handle: MaybeRef, options: { enabled?: MaybeRef } = {}) { + const h = computed(() => toValue(handle).trim()); + return useQuery({ - queryKey: computed(() => ["userFollowing", toValue(pds), toValue(did)]), + queryKey: computed(() => ["userFollowing", h.value]), queryFn: async (): Promise => { - const { records } = await listFollowRecords(toValue(pds), toValue(did)); - const follows = records + const response = await fetchActorFollowing(h.value); + const follows = response.records .map((record) => normalizeFollowRecord(record.value, record.uri)) .sort((left, right) => Date.parse(right.createdAt) - Date.parse(left.createdAt)); return Promise.all( follows.map(async (follow) => { - const subject = await resolveDidIdentity(follow.subjectDid); - try { - const { value } = await fetchActorProfile(subject.pds, subject.did); - const bluesky = await resolveBlueskyProfile(value, subject.did); + const subject = await fetchActor(follow.subjectDid); return { - ...normalizeActorProfile(value, subject.did, subject.handle, bluesky.displayName, bluesky.avatar), + ...normalizeActorProfile( + subject.profile.value, + subject.did, + subject.handle, + subject.bsky?.displayName, + subject.bsky?.avatar, + ), followAtUri: follow.atUri, followedAt: follow.createdAt, }; } catch { return { - did: subject.did, - handle: subject.handle, + did: follow.subjectDid, + handle: follow.subjectDid, followAtUri: follow.atUri, followedAt: follow.createdAt, }; @@ -522,238 +438,173 @@ export function useUserFollowing( }), ); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h), options.enabled)), staleTime: 5 * MIN, gcTime: 30 * MIN, }); } -export function useUserIssues( - pds: MaybeRef, - did: MaybeRef, - handle: MaybeRef, - options: { enabled?: MaybeRef } = {}, -) { +export function useUserIssues(handle: MaybeRef, options: { enabled?: MaybeRef } = {}) { + const h = computed(() => toValue(handle).trim()); + return useQuery({ - queryKey: computed(() => ["userIssues", toValue(pds), toValue(did)]), + queryKey: computed(() => ["userIssues", h.value]), queryFn: async () => { - const [issuesRes, statesRes] = await Promise.all([ - listIssueRecords(toValue(pds), toValue(did)), - listIssueStateRecords(toValue(pds), toValue(did)), - ]); - - const stateMap = new Map(); - for (const stateRecord of statesRes.records) { - const closed = stateRecord.value.state === "sh.tangled.repo.issue.state.closed"; - stateMap.set(stateRecord.value.issue, closed ? "closed" : "open"); - } - - return issuesRes.records - .map((record) => - normalizeIssueRecord( - record.value, - record.uri, - toValue(did), - toValue(handle), - stateMap.get(record.uri) ?? "open", - ), - ) + const response = await fetchActorIssues(h.value); + return response.records + .map((entry) => normalizeIssueRecord(entry.value, entry.uri, response.did, response.handle, entry.state)) .sort((left, right) => Date.parse(right.createdAt) - Date.parse(left.createdAt)); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } -export function useUserPullRequests( - pds: MaybeRef, - did: MaybeRef, - handle: MaybeRef, - options: { enabled?: MaybeRef } = {}, -) { +export function useUserPullRequests(handle: MaybeRef, options: { enabled?: MaybeRef } = {}) { + const h = computed(() => toValue(handle).trim()); + return useQuery({ - queryKey: computed(() => ["userPullRequests", toValue(pds), toValue(did)]), + queryKey: computed(() => ["userPullRequests", h.value]), queryFn: async () => { - const [pullsRes, statusesRes] = await Promise.all([ - listPullRecords(toValue(pds), toValue(did)), - listPullStatusRecords(toValue(pds), toValue(did)), - ]); - - const statusMap = new Map(); - for (const statusRecord of statusesRes.records) { - const raw = statusRecord.value.status ?? "sh.tangled.repo.pull.status.open"; - const status = - raw === "sh.tangled.repo.pull.status.merged" - ? "merged" - : raw === "sh.tangled.repo.pull.status.closed" - ? "closed" - : "open"; - statusMap.set(statusRecord.value.pull, status); - } - - return pullsRes.records - .map((record) => - normalizePullRecord( - record.value, - record.uri, - toValue(did), - toValue(handle), - statusMap.get(record.uri) ?? "open", - ), - ) + const response = await fetchActorPulls(h.value); + return response.records + .map((entry) => normalizePullRecord(entry.value, entry.uri, response.did, response.handle, entry.status)) .sort((left, right) => Date.parse(right.createdAt) - Date.parse(left.createdAt)); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } export function useIssueDetail( - pds: MaybeRef, - did: MaybeRef, handle: MaybeRef, issueRkey: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const rkey = computed(() => toValue(issueRkey).trim()); + return useQuery({ - queryKey: computed(() => ["issueDetail", toValue(pds), toValue(did), toValue(issueRkey)]), + queryKey: computed(() => ["issueDetail", h.value, rkey.value]), queryFn: async () => { - const [issueRes, statesRes, commentsRes] = await Promise.all([ - fetchIssueRecord(toValue(pds), toValue(did), toValue(issueRkey)), - listIssueStateRecords(toValue(pds), toValue(did)), - listIssueCommentRecords(toValue(pds), toValue(did)), + const [issueRes, commentsRes, actor] = await Promise.all([ + fetchIssueDetail(h.value, rkey.value), + fetchIssueComments(h.value, rkey.value), + fetchActor(h.value), ]); - const currentState = statesRes.records.reduce<"open" | "closed">((state, record) => { - if (record.value.issue !== issueRes.uri) return state; - return record.value.state === "sh.tangled.repo.issue.state.closed" ? "closed" : "open"; - }, "open"); - - const commentCount = commentsRes.records.filter((record) => record.value.issue === issueRes.uri).length; - return normalizeIssueDetail( issueRes.value, issueRes.uri, - toValue(did), - toValue(handle), - currentState, - commentCount, + actor.did, + actor.handle, + issueRes.state, + commentsRes.records.length, ); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h) && hasText(rkey), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } export function useIssueComments( - pds: MaybeRef, - did: MaybeRef, handle: MaybeRef, - issueAtUri: MaybeRef, + issueRkey: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const rkey = computed(() => toValue(issueRkey).trim()); + return useQuery({ - queryKey: computed(() => ["issueComments", toValue(pds), toValue(did), toValue(issueAtUri)]), + queryKey: computed(() => ["issueComments", h.value, rkey.value]), queryFn: async () => { - const response = await listIssueCommentRecords(toValue(pds), toValue(did)); - const comments = response.records - .filter((record) => record.value.issue === toValue(issueAtUri)) - .map((record) => normalizeIssueComment(record.value, record.uri, toValue(did), toValue(handle))); + const [response, actor] = await Promise.all([fetchIssueComments(h.value, rkey.value), fetchActor(h.value)]); + const comments = response.records.map((record) => + normalizeIssueComment(record.value, record.uri, actor.did, actor.handle), + ); return buildIssueCommentThread(comments); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h) && hasText(rkey), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } export function usePullRequestDetail( - pds: MaybeRef, - did: MaybeRef, handle: MaybeRef, pullRkey: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const rkey = computed(() => toValue(pullRkey).trim()); + return useQuery({ - queryKey: computed(() => ["pullDetail", toValue(pds), toValue(did), toValue(pullRkey)]), + queryKey: computed(() => ["pullDetail", h.value, rkey.value]), queryFn: async () => { - const [pullRes, statusesRes, commentsRes] = await Promise.all([ - fetchPullRecord(toValue(pds), toValue(did), toValue(pullRkey)), - listPullStatusRecords(toValue(pds), toValue(did)), - listPullCommentRecords(toValue(pds), toValue(did)), + const [pullRes, commentsRes, actor] = await Promise.all([ + fetchPullDetail(h.value, rkey.value), + fetchPullComments(h.value, rkey.value), + fetchActor(h.value), ]); - const currentStatus = statusesRes.records.reduce<"open" | "merged" | "closed">((status, record) => { - if (record.value.pull !== pullRes.uri) return status; - if (record.value.status === "sh.tangled.repo.pull.status.merged") return "merged"; - if (record.value.status === "sh.tangled.repo.pull.status.closed") return "closed"; - return "open"; - }, "open"); - - const roundCount = commentsRes.records.filter((record) => record.value.pull === pullRes.uri).length; - - return normalizePullDetail(pullRes.value, pullRes.uri, toValue(did), toValue(handle), currentStatus, roundCount); + return normalizePullDetail( + pullRes.value, + pullRes.uri, + actor.did, + actor.handle, + pullRes.status, + commentsRes.records.length, + ); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h) && hasText(rkey), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } export function usePullRequestComments( - pds: MaybeRef, - did: MaybeRef, handle: MaybeRef, - pullAtUri: MaybeRef, + pullRkey: MaybeRef, options: { enabled?: MaybeRef } = {}, ) { + const h = computed(() => toValue(handle).trim()); + const rkey = computed(() => toValue(pullRkey).trim()); + return useQuery({ - queryKey: computed(() => ["pullComments", toValue(pds), toValue(did), toValue(pullAtUri)]), + queryKey: computed(() => ["pullComments", h.value, rkey.value]), queryFn: async () => { - const response = await listPullCommentRecords(toValue(pds), toValue(did)); + const [response, actor] = await Promise.all([fetchPullComments(h.value, rkey.value), fetchActor(h.value)]); return response.records - .filter((record) => record.value.pull === toValue(pullAtUri)) - .map((record) => normalizePullComment(record.value, record.uri, toValue(did), toValue(handle))) + .map((record) => normalizePullComment(record.value, record.uri, actor.did, actor.handle)) .sort((left, right) => Date.parse(left.createdAt) - Date.parse(right.createdAt)); }, - enabled: options.enabled, + enabled: computed(() => isEnabled(hasText(h) && hasText(rkey), options.enabled)), staleTime: 2 * MIN, gcTime: 10 * MIN, }); } -/** - * Composite hook: fetch repo PDS record + default branch + languages in - * parallel, returning a merged RepoDetail. - */ -export function useRepoDetail( - pds: MaybeRef, - did: MaybeRef, - repoName: MaybeRef, - knotHost: MaybeRef, - handle: MaybeRef, -) { - const knotRepo = computed(() => `${toValue(did)}/${toValue(repoName)}`); - - const record = useRepoRecord(pds, did, repoName, handle); - const branch = useDefaultBranch(knotHost, knotRepo); - const languages = useRepoLanguages(knotHost, knotRepo); +/** Composite hook: repo record + default branch + languages in parallel. */ +export function useRepoDetail(handle: MaybeRef, repoName: MaybeRef) { + const record = useRepoRecord(handle, repoName); + const branch = useDefaultBranch(handle, repoName); + const languages = useRepoLanguages(handle, repoName); const data = computed(() => { if (!record.data.value) return undefined; return normalizeRepoRecordToDetail( { name: toValue(repoName), - knot: toValue(knotHost), + knot: record.data.value.knot, createdAt: record.data.value.updatedAt ?? "", $type: "sh.tangled.repo", }, - toValue(did), - toValue(handle), + record.data.value.ownerDid, + record.data.value.ownerHandle, record.data.value.atUri, { defaultBranch: branch.data.value?.name, languages: languages.data.value }, ); diff --git a/apps/twisted/src/services/tangled/repo-assets.ts b/apps/twisted/src/services/tangled/repo-assets.ts index a54bf38..b5f07d9 100644 --- a/apps/twisted/src/services/tangled/repo-assets.ts +++ b/apps/twisted/src/services/tangled/repo-assets.ts @@ -1,14 +1,7 @@ import { fetchRepoBlob } from "./endpoints.js"; import { normalizeBlob, type BlobContent } from "./normalizers.js"; -export type RepoAssetContext = { - owner: string; - repo: string; - branch: string; - knotHost: string; - knotRepo: string; - sourcePath?: string; -}; +export type RepoAssetContext = { owner: string; repo: string; branch: string; sourcePath?: string }; const EXTERNAL_URL_RE = /^(?:[a-z][a-z0-9+.-]*:)?\/\//i; const SCHEME_RE = /^[a-z][a-z0-9+.-]*:/i; @@ -47,9 +40,7 @@ export function resolveRepoRelativePath(sourcePath: string | undefined, src: str const [pathOnly] = value.split(/[?#]/, 1); if (!pathOnly) return null; - const baseSegments = value.startsWith("/") - ? [] - : (sourcePath ? dirname(sourcePath).split("/").filter(Boolean) : []); + const baseSegments = value.startsWith("/") ? [] : sourcePath ? dirname(sourcePath).split("/").filter(Boolean) : []; const segments = pathOnly.replace(/^\/+/, "").split("/"); const resolved = [...baseSegments]; @@ -77,11 +68,13 @@ export async function resolveRepoImageUrl( if (!repoPath) return null; try { - const blob = normalizeBlob(await fetchRepoBlob(context.knotHost, { - repo: context.knotRepo, - ref: context.branch, - path: repoPath, - })); + const blob = normalizeBlob( + await fetchRepoBlob(context.owner, context.repo, { + repo: `${context.owner}/${context.repo}`, + ref: context.branch, + path: repoPath, + }), + ); const objectUrl = createObjectUrlFromBlobContent(blob); if (objectUrl) return { url: objectUrl, revoke: true }; diff --git a/apps/twisted/tests/unit/tangled-normalizers.spec.ts b/apps/twisted/tests/unit/tangled-normalizers.spec.ts index f7df23d..5ddd88d 100644 --- a/apps/twisted/tests/unit/tangled-normalizers.spec.ts +++ b/apps/twisted/tests/unit/tangled-normalizers.spec.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from "vitest"; -import { buildKnotUrl } from "@/services/tangled/endpoints.js"; -import { buildIssueCommentThread, normalizeLogText, normalizeRepoRecord, normalizeTree } from "@/services/tangled/normalizers.js"; +import { + buildIssueCommentThread, + normalizeLogText, + normalizeRepoRecord, + normalizeTree, +} from "@/services/tangled/normalizers.js"; import { buildPublicRawUrl, resolveRepoRelativePath } from "@/services/tangled/repo-assets.js"; import { getAtUriRkey, parseAtUri } from "@/services/tangled/uris.js"; import type { IssueComment } from "@/domain/models/comment.js"; @@ -44,23 +48,34 @@ describe("AT URI helpers", () => { expect(repo.rkey).toBe("writer-app"); }); - it("preserves the repo slash in knot XRPC query strings", () => { - const url = buildKnotUrl("knot1.tangled.sh", "sh.tangled.repo.getDefaultBranch", { - repo: "did:plc:xg2vq45muivyy3xwatcehspu/writer", - }); - - expect(url).toContain("repo=did%3Aplc%3Axg2vq45muivyy3xwatcehspu/writer"); - expect(url).not.toContain("%2Fwriter"); - }); - it("derives file kinds from zero-padded git modes", () => { const files = normalizeTree({ files: [ - { mode: "0040000", name: ".github", size: 75, last_commit: { hash: "a", message: "dir", when: "2026-03-23T00:00:00Z" } }, - { mode: "0100644", name: "README.md", size: 3126, last_commit: { hash: "b", message: "file", when: "2026-03-23T00:00:00Z" } }, - { mode: "0160000", name: "vendor/lib", size: 0, last_commit: { hash: "c", message: "submodule", when: "2026-03-23T00:00:00Z" } }, + { + mode: "0040000", + name: ".github", + size: 75, + last_commit: { hash: "a", message: "dir", when: "2026-03-23T00:00:00Z" }, + }, + { + mode: "0100644", + name: "README.md", + size: 3126, + last_commit: { hash: "b", message: "file", when: "2026-03-23T00:00:00Z" }, + }, + { + mode: "0160000", + name: "vendor/lib", + size: 0, + last_commit: { hash: "c", message: "submodule", when: "2026-03-23T00:00:00Z" }, + }, ], - lastCommit: { hash: "a", message: "dir", when: "2026-03-23T00:00:00Z", author: { name: "Test", email: "test@example.com", when: "" } }, + lastCommit: { + hash: "a", + message: "dir", + when: "2026-03-23T00:00:00Z", + author: { name: "Test", email: "test@example.com", when: "" }, + }, ref: "main", }); @@ -128,15 +143,12 @@ describe("AT URI helpers", () => { }); it("builds public raw URLs for repo assets", () => { - expect(buildPublicRawUrl({ - owner: "desertthunder.dev", - repo: "writer", - branch: "main", - knotHost: "unused", - knotRepo: "unused", - }, "www/src/static/images/context-menu-in-sidebar.png")).toBe( - "https://tangled.org/desertthunder.dev/writer/raw/main/www/src/static/images/context-menu-in-sidebar.png", - ); + expect( + buildPublicRawUrl( + { owner: "desertthunder.dev", repo: "writer", branch: "main", knotHost: "unused", knotRepo: "unused" }, + "www/src/static/images/context-menu-in-sidebar.png", + ), + ).toBe("https://tangled.org/desertthunder.dev/writer/raw/main/www/src/static/images/context-menu-in-sidebar.png"); }); }); diff --git a/apps/twisted/tsconfig.json b/apps/twisted/tsconfig.json index c3287dc..87e79a9 100644 --- a/apps/twisted/tsconfig.json +++ b/apps/twisted/tsconfig.json @@ -9,6 +9,7 @@ "resolveJsonModule": true, "isolatedModules": true, "esModuleInterop": true, + "allowImportingTsExtensions": true, "lib": ["ESNext", "DOM"], "skipLibCheck": true, "noEmit": true, diff --git a/packages/api/README.md b/packages/api/README.md index b3dd019..6efefee 100644 --- a/packages/api/README.md +++ b/packages/api/README.md @@ -1,3 +1,63 @@ # Twister -Tap-based indexing and search service for Tangled. +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. + +## 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 +``` + +## 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` | diff --git a/packages/api/internal/api/actors.go b/packages/api/internal/api/actors.go new file mode 100644 index 0000000..f44146b --- /dev/null +++ b/packages/api/internal/api/actors.go @@ -0,0 +1,900 @@ +package api + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "strings" + + "tangled.org/desertthunder.dev/twister/internal/xrpc" +) + +// recordEntry is the common shape for a single PDS record returned to the app. +type recordEntry struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value map[string]any `json:"value"` +} + +// issueEntry extends recordEntry with pre-joined issue state. +type issueEntry struct { + recordEntry + State string `json:"state"` // "open" or "closed" +} + +// pullEntry extends recordEntry with pre-joined pull status. +type pullEntry struct { + recordEntry + Status string `json:"status"` // "open", "merged", or "closed" +} + +// 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" +} + +// repoContext extends actorContext with the repo's knot host and AT URI. +type repoContext struct { + actorContext + KnotHost string `json:"knot_host"` + AtURI string `json:"at_uri"` + RepoName string `json:"repo_name"` +} + +// resolveActor resolves a handle (or DID) to its DID, PDS, and canonical handle. +func (s *Server) resolveActor(r *http.Request, handleOrDID string) (*actorContext, error) { + ctx := r.Context() + + var did string + if strings.HasPrefix(handleOrDID, "did:") { + did = handleOrDID + } else { + var err error + did, err = s.xrpc.ResolveHandle(ctx, handleOrDID) + if err != nil { + return nil, fmt.Errorf("resolve handle %q: %w", handleOrDID, err) + } + } + + identity, err := s.xrpc.ResolveIdentity(ctx, did) + if err != nil { + return nil, fmt.Errorf("resolve identity %q: %w", did, err) + } + + return &actorContext{ + DID: identity.DID, + Handle: identity.Handle, + PDS: identity.PDS, + }, nil +} + +// resolveRepo resolves a handle + repo name to actor + knot host + AT URI. +func (s *Server) resolveRepo(r *http.Request, handleOrDID, repoName string) (*repoContext, error) { + actor, err := s.resolveActor(r, handleOrDID) + if err != nil { + return nil, err + } + + entries, err := s.xrpc.ListAllRecords(r.Context(), actor.PDS, actor.DID, "sh.tangled.repo") + if err != nil { + return nil, fmt.Errorf("list repos for %s: %w", actor.DID, err) + } + + for _, entry := range entries { + name, _ := entry.Value["name"].(string) + if name == repoName { + knot, _ := entry.Value["knot"].(string) + return &repoContext{ + actorContext: *actor, + KnotHost: knot, + AtURI: entry.URI, + RepoName: repoName, + }, nil + } + } + + return nil, &xrpc.NotFoundError{Message: fmt.Sprintf("repo %q not found for %s", repoName, handleOrDID)} +} + +// knotCall makes a GET request to a knot's XRPC endpoint and streams the response body. +// The caller is responsible for closing the returned ReadCloser. +func (s *Server) knotCall(r *http.Request, knotHost, nsid string, params url.Values) (io.ReadCloser, string, error) { + if knotHost == "" { + return nil, "", fmt.Errorf("repo has no knot host") + } + knotURL := "https://" + knotHost + body, err := s.xrpc.Call(r.Context(), knotURL, nsid, params) + if err != nil { + return nil, "", err + } + return body, knotURL, nil +} + +// proxyKnotJSON calls a knot endpoint and writes the JSON response verbatim. +func (s *Server) proxyKnotJSON(w http.ResponseWriter, r *http.Request, repo *repoContext, nsid string, params url.Values) { + params.Set("repo", repo.DID+"/"+repo.RepoName) + body, _, err := s.knotCall(r, repo.KnotHost, nsid, params) + if err != nil { + s.knotError(w, err) + return + } + defer body.Close() + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = io.Copy(w, body) +} + +// proxyKnotBytes calls a knot endpoint and writes the raw bytes verbatim. +func (s *Server) proxyKnotBytes(w http.ResponseWriter, r *http.Request, repo *repoContext, nsid string, params url.Values) { + params.Set("repo", repo.DID+"/"+repo.RepoName) + body, _, err := s.knotCall(r, repo.KnotHost, nsid, params) + if err != nil { + s.knotError(w, err) + return + } + defer body.Close() + w.Header().Set("Content-Type", "application/octet-stream") + w.WriteHeader(http.StatusOK) + _, _ = io.Copy(w, body) +} + +func (s *Server) knotError(w http.ResponseWriter, err error) { + var nfe *xrpc.NotFoundError + var xe *xrpc.XRPCError + switch { + case isError(err, &nfe): + writeJSON(w, http.StatusNotFound, errorBody("not_found", nfe.Message)) + case isError(err, &xe): + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", xe.Message)) + default: + s.log.Debug("knot call failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "upstream request failed")) + } +} + +func (s *Server) actorError(w http.ResponseWriter, err error) { + var nfe *xrpc.NotFoundError + var xe *xrpc.XRPCError + switch { + case isError(err, &nfe): + writeJSON(w, http.StatusNotFound, errorBody("not_found", nfe.Message)) + case isError(err, &xe): + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", xe.Message)) + default: + s.log.Debug("actor resolve failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusBadGateway, errorBody("resolve_error", "failed to resolve actor")) + } +} + +// isError is a type-safe errors.As replacement for pointer receiver targets. +func isError[T error](err error, target *T) bool { + if err == nil { + return false + } + + type unwrapper interface{ Unwrap() error } + for e := err; e != nil; { + if t, ok := e.(T); ok { + *target = t + return true + } + if u, ok := e.(unwrapper); ok { + e = u.Unwrap() + } else { + break + } + } + return false +} + +// handleGetActor returns the actor's Tangled profile + optional Bluesky info. +// GET /actors/{handle} +func (s *Server) handleGetActor(w http.ResponseWriter, r *http.Request) { + handle := r.PathValue("handle") + + actor, err := s.resolveActor(r, handle) + if err != nil { + s.actorError(w, err) + return + } + + rec, err := s.xrpc.GetRecord(r.Context(), actor.PDS, actor.DID, "sh.tangled.actor.profile", "self") + if err != nil { + s.actorError(w, err) + return + } + + var bsky *bskyProfileResponse + if linked, _ := rec.Value["bluesky"].(bool); linked { + bsky = s.fetchBskyProfile(r, actor.DID) + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "pds": actor.PDS, + "profile": recordEntry{ + URI: rec.URI, + CID: rec.CID, + Value: rec.Value, + }, + "bsky": bsky, + }) +} + +// handleListActorRepos returns all sh.tangled.repo records for an actor. +// GET /actors/{handle}/repos +func (s *Server) handleListActorRepos(w http.ResponseWriter, r *http.Request) { + handle := r.PathValue("handle") + + actor, err := s.resolveActor(r, handle) + if err != nil { + s.actorError(w, err) + return + } + + entries, err := s.xrpc.ListAllRecords(r.Context(), actor.PDS, actor.DID, "sh.tangled.repo") + if err != nil { + s.actorError(w, err) + return + } + + records := make([]recordEntry, len(entries)) + for i, e := range entries { + records[i] = recordEntry{URI: e.URI, CID: e.CID, Value: e.Value} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "records": records, + }) +} + +// handleGetActorRepo returns the repo record for a specific repo by name. +// GET /actors/{handle}/repos/{repo} +func (s *Server) handleGetActorRepo(w http.ResponseWriter, r *http.Request) { + handle := r.PathValue("handle") + repoName := r.PathValue("repo") + + repo, err := s.resolveRepo(r, handle, repoName) + if err != nil { + s.actorError(w, err) + return + } + + did, _, rkey, parseErr := parseATURI(repo.AtURI) + if parseErr != nil { + writeJSON(w, http.StatusInternalServerError, errorBody("internal_error", "invalid AT URI")) + return + } + + rec, err := s.xrpc.GetRecord(r.Context(), repo.PDS, did, "sh.tangled.repo", rkey) + if err != nil { + s.actorError(w, err) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": repo.DID, + "handle": repo.Handle, + "knot_host": repo.KnotHost, + "record": recordEntry{ + URI: rec.URI, + CID: rec.CID, + Value: rec.Value, + }, + }) +} + +// handleRepoTree proxies sh.tangled.repo.tree to the knot. +// GET /actors/{handle}/repos/{repo}/tree +func (s *Server) handleRepoTree(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + params := url.Values{} + for _, k := range []string{"ref", "path"} { + if v := r.URL.Query().Get(k); v != "" { + params.Set(k, v) + } + } + s.proxyKnotJSON(w, r, repo, "sh.tangled.repo.tree", params) +} + +// handleRepoBlob proxies sh.tangled.repo.blob to the knot. +// GET /actors/{handle}/repos/{repo}/blob +func (s *Server) handleRepoBlob(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + params := url.Values{} + for _, k := range []string{"ref", "path"} { + if v := r.URL.Query().Get(k); v != "" { + params.Set(k, v) + } + } + s.proxyKnotJSON(w, r, repo, "sh.tangled.repo.blob", params) +} + +// handleRepoLog proxies sh.tangled.repo.log (raw bytes) to the knot. +// GET /actors/{handle}/repos/{repo}/log +func (s *Server) handleRepoLog(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + params := url.Values{} + for _, k := range []string{"ref", "path", "limit", "cursor"} { + if v := r.URL.Query().Get(k); v != "" { + params.Set(k, v) + } + } + s.proxyKnotBytes(w, r, repo, "sh.tangled.repo.log", params) +} + +// handleRepoBranches proxies sh.tangled.repo.branches (raw bytes) to the knot. +// GET /actors/{handle}/repos/{repo}/branches +func (s *Server) handleRepoBranches(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + params := url.Values{} + for _, k := range []string{"limit", "cursor"} { + if v := r.URL.Query().Get(k); v != "" { + params.Set(k, v) + } + } + s.proxyKnotBytes(w, r, repo, "sh.tangled.repo.branches", params) +} + +// handleRepoDefaultBranch proxies sh.tangled.repo.getDefaultBranch (JSON) to the knot. +// GET /actors/{handle}/repos/{repo}/default-branch +func (s *Server) handleRepoDefaultBranch(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + s.proxyKnotJSON(w, r, repo, "sh.tangled.repo.getDefaultBranch", url.Values{}) +} + +// handleRepoLanguages proxies sh.tangled.repo.languages (JSON) to the knot. +// GET /actors/{handle}/repos/{repo}/languages +func (s *Server) handleRepoLanguages(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + params := url.Values{} + if v := r.URL.Query().Get("ref"); v != "" { + params.Set("ref", v) + } + s.proxyKnotJSON(w, r, repo, "sh.tangled.repo.languages", params) +} + +// handleRepoTags proxies sh.tangled.repo.tags (raw bytes) to the knot. +// GET /actors/{handle}/repos/{repo}/tags +func (s *Server) handleRepoTags(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + s.proxyKnotBytes(w, r, repo, "sh.tangled.repo.tags", url.Values{}) +} + +// handleRepoDiff proxies sh.tangled.repo.diff (raw bytes) to the knot. +// GET /actors/{handle}/repos/{repo}/diff +func (s *Server) handleRepoDiff(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + params := url.Values{} + if v := r.URL.Query().Get("ref"); v != "" { + params.Set("ref", v) + } + s.proxyKnotBytes(w, r, repo, "sh.tangled.repo.diff", params) +} + +// handleRepoCompare proxies sh.tangled.repo.compare (raw bytes) to the knot. +// GET /actors/{handle}/repos/{repo}/compare +func (s *Server) handleRepoCompare(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + params := url.Values{} + for _, k := range []string{"from", "to"} { + if v := r.URL.Query().Get(k); v != "" { + params.Set(k, v) + } + } + s.proxyKnotBytes(w, r, repo, "sh.tangled.repo.compare", params) +} + +// handleRepoIssues returns issues for a repo, pre-joined with state. +// GET /actors/{handle}/repos/{repo}/issues +func (s *Server) handleRepoIssues(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + + 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 + } + + var records []issueEntry + for _, e := range issues { + repoURI, _ := e.Value["repo"].(string) + if repo.AtURI != "" && repoURI != repo.AtURI { + continue + } + records = append(records, issueEntry{ + recordEntry: recordEntry{URI: e.URI, CID: e.CID, Value: e.Value}, + State: resolveIssueState(stateMap, e.URI), + }) + } + if records == nil { + records = []issueEntry{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": repo.DID, + "handle": repo.Handle, + "records": records, + }) +} + +// handleRepoPulls returns pull requests for a repo, pre-joined with status. +// GET /actors/{handle}/repos/{repo}/pulls +func (s *Server) handleRepoPulls(w http.ResponseWriter, r *http.Request) { + repo, err := s.resolveRepo(r, r.PathValue("handle"), r.PathValue("repo")) + if err != nil { + s.actorError(w, err) + return + } + + 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 + } + + var records []pullEntry + for _, e := range pulls { + target, _ := e.Value["target"].(map[string]any) + targetRepo, _ := target["repo"].(string) + if repo.AtURI != "" && targetRepo != repo.AtURI { + continue + } + records = append(records, pullEntry{ + recordEntry: recordEntry{URI: e.URI, CID: e.CID, Value: e.Value}, + Status: resolvePullStatus(statusMap, e.URI), + }) + } + if records == nil { + records = []pullEntry{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": repo.DID, + "handle": repo.Handle, + "records": records, + }) +} + +// handleActorIssues returns all issues authored by an actor, pre-joined with state. +// GET /actors/{handle}/issues +func (s *Server) handleActorIssues(w http.ResponseWriter, r *http.Request) { + actor, err := s.resolveActor(r, r.PathValue("handle")) + if err != nil { + s.actorError(w, err) + return + } + + 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 + } + + records := make([]issueEntry, len(issues)) + for i, e := range issues { + records[i] = issueEntry{ + recordEntry: recordEntry{URI: e.URI, CID: e.CID, Value: e.Value}, + State: resolveIssueState(stateMap, e.URI), + } + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "records": records, + }) +} + +// handleActorPulls returns all pull requests authored by an actor, pre-joined with status. +// GET /actors/{handle}/pulls +func (s *Server) handleActorPulls(w http.ResponseWriter, r *http.Request) { + actor, err := s.resolveActor(r, r.PathValue("handle")) + if err != nil { + s.actorError(w, err) + return + } + + 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 + } + + records := make([]pullEntry, len(pulls)) + for i, e := range pulls { + records[i] = pullEntry{ + recordEntry: recordEntry{URI: e.URI, CID: e.CID, Value: e.Value}, + Status: resolvePullStatus(statusMap, e.URI), + } + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "records": records, + }) +} + +// handleActorFollowing returns sh.tangled.graph.follow records for an actor. +// GET /actors/{handle}/following +func (s *Server) handleActorFollowing(w http.ResponseWriter, r *http.Request) { + actor, err := s.resolveActor(r, r.PathValue("handle")) + if err != nil { + s.actorError(w, err) + return + } + + entries, err := s.xrpc.ListAllRecords(r.Context(), actor.PDS, actor.DID, "sh.tangled.graph.follow") + if err != nil { + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch follows")) + return + } + + records := make([]recordEntry, len(entries)) + for i, e := range entries { + records[i] = recordEntry{URI: e.URI, CID: e.CID, Value: e.Value} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "records": records, + }) +} + +// handleActorStrings returns sh.tangled.string records for an actor. +// GET /actors/{handle}/strings +func (s *Server) handleActorStrings(w http.ResponseWriter, r *http.Request) { + actor, err := s.resolveActor(r, r.PathValue("handle")) + if err != nil { + s.actorError(w, err) + return + } + + entries, err := s.xrpc.ListAllRecords(r.Context(), actor.PDS, actor.DID, "sh.tangled.string") + if err != nil { + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch strings")) + return + } + + records := make([]recordEntry, len(entries)) + for i, e := range entries { + records[i] = recordEntry{URI: e.URI, CID: e.CID, Value: e.Value} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "records": records, + }) +} + +// handleIssueDetail returns a single issue with its state. +// GET /issues/{handle}/{rkey} +func (s *Server) handleIssueDetail(w http.ResponseWriter, r *http.Request) { + handle := r.PathValue("handle") + rkey := r.PathValue("rkey") + + actor, err := s.resolveActor(r, handle) + if err != nil { + s.actorError(w, err) + return + } + + rec, err := s.xrpc.GetRecord(r.Context(), actor.PDS, actor.DID, "sh.tangled.repo.issue", rkey) + if err != nil { + s.actorError(w, err) + return + } + + _, stateMap, err := s.fetchIssuesAndStates(r, actor.PDS, actor.DID) + if err != nil { + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch issue states")) + return + } + + writeJSON(w, http.StatusOK, issueEntry{ + recordEntry: recordEntry{URI: rec.URI, CID: rec.CID, Value: rec.Value}, + State: resolveIssueState(stateMap, rec.URI), + }) +} + +// handleIssueComments returns all comments for a specific issue. +// GET /issues/{handle}/{rkey}/comments +func (s *Server) handleIssueComments(w http.ResponseWriter, r *http.Request) { + handle := r.PathValue("handle") + rkey := r.PathValue("rkey") + + actor, err := s.resolveActor(r, handle) + if err != nil { + s.actorError(w, err) + return + } + + issueURI := fmt.Sprintf("at://%s/sh.tangled.repo.issue/%s", actor.DID, rkey) + + entries, err := s.xrpc.ListAllRecords(r.Context(), actor.PDS, actor.DID, "sh.tangled.repo.issue.comment") + if err != nil { + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch comments")) + return + } + + var records []recordEntry + for _, e := range entries { + issue, _ := e.Value["issue"].(string) + if issue == issueURI { + records = append(records, recordEntry{URI: e.URI, CID: e.CID, Value: e.Value}) + } + } + if records == nil { + records = []recordEntry{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "issueUri": issueURI, + "records": records, + }) +} + +// handlePullDetail returns a single pull request with its status. +// GET /pulls/{handle}/{rkey} +func (s *Server) handlePullDetail(w http.ResponseWriter, r *http.Request) { + handle := r.PathValue("handle") + rkey := r.PathValue("rkey") + + actor, err := s.resolveActor(r, handle) + if err != nil { + s.actorError(w, err) + return + } + + rec, err := s.xrpc.GetRecord(r.Context(), actor.PDS, actor.DID, "sh.tangled.repo.pull", rkey) + if err != nil { + s.actorError(w, err) + return + } + + _, statusMap, err := s.fetchPullsAndStatuses(r, actor.PDS, actor.DID) + if err != nil { + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch pull statuses")) + return + } + + writeJSON(w, http.StatusOK, pullEntry{ + recordEntry: recordEntry{URI: rec.URI, CID: rec.CID, Value: rec.Value}, + Status: resolvePullStatus(statusMap, rec.URI), + }) +} + +// handlePullComments returns all comments for a specific pull request. +// GET /pulls/{handle}/{rkey}/comments +func (s *Server) handlePullComments(w http.ResponseWriter, r *http.Request) { + handle := r.PathValue("handle") + rkey := r.PathValue("rkey") + + actor, err := s.resolveActor(r, handle) + if err != nil { + s.actorError(w, err) + return + } + + pullURI := fmt.Sprintf("at://%s/sh.tangled.repo.pull/%s", actor.DID, rkey) + + entries, err := s.xrpc.ListAllRecords(r.Context(), actor.PDS, actor.DID, "sh.tangled.repo.pull.comment") + if err != nil { + writeJSON(w, http.StatusBadGateway, errorBody("upstream_error", "failed to fetch comments")) + return + } + + var records []recordEntry + for _, e := range entries { + pull, _ := e.Value["pull"].(string) + if pull == pullURI { + records = append(records, recordEntry{URI: e.URI, CID: e.CID, Value: e.Value}) + } + } + if records == nil { + records = []recordEntry{} + } + + writeJSON(w, http.StatusOK, map[string]any{ + "did": actor.DID, + "handle": actor.Handle, + "pullUri": pullURI, + "records": records, + }) +} + +func (s *Server) fetchIssuesAndStates(r *http.Request, pds, did string) ([]xrpc.ListRecordEntry, map[string]string, error) { + issueCh := make(chan []xrpc.ListRecordEntry, 1) + stateCh := make(chan []xrpc.ListRecordEntry, 1) + errCh := make(chan error, 2) + + go func() { + entries, err := s.xrpc.ListAllRecords(r.Context(), pds, did, "sh.tangled.repo.issue") + if err != nil { + errCh <- err + return + } + issueCh <- entries + }() + go func() { + entries, err := s.xrpc.ListAllRecords(r.Context(), pds, did, "sh.tangled.repo.issue.state") + if err != nil { + errCh <- err + return + } + stateCh <- entries + }() + + var issues []xrpc.ListRecordEntry + var states []xrpc.ListRecordEntry + for i := 0; i < 2; i++ { + select { + case e := <-issueCh: + issues = e + case e := <-stateCh: + states = e + case err := <-errCh: + return nil, nil, err + } + } + + stateMap := make(map[string]string, len(states)) + for _, e := range states { + issueURI, _ := e.Value["issue"].(string) + state, _ := e.Value["state"].(string) + if issueURI != "" { + stateMap[issueURI] = state + } + } + + return issues, stateMap, nil +} + +func (s *Server) fetchPullsAndStatuses(r *http.Request, pds, did string) ([]xrpc.ListRecordEntry, map[string]string, error) { + pullCh := make(chan []xrpc.ListRecordEntry, 1) + statusCh := make(chan []xrpc.ListRecordEntry, 1) + errCh := make(chan error, 2) + + go func() { + entries, err := s.xrpc.ListAllRecords(r.Context(), pds, did, "sh.tangled.repo.pull") + if err != nil { + errCh <- err + return + } + pullCh <- entries + }() + go func() { + entries, err := s.xrpc.ListAllRecords(r.Context(), pds, did, "sh.tangled.repo.pull.status") + if err != nil { + errCh <- err + return + } + statusCh <- entries + }() + + var pulls []xrpc.ListRecordEntry + var statuses []xrpc.ListRecordEntry + for i := 0; i < 2; i++ { + select { + case e := <-pullCh: + pulls = e + case e := <-statusCh: + statuses = e + case err := <-errCh: + return nil, nil, err + } + } + + statusMap := make(map[string]string, len(statuses)) + for _, e := range statuses { + pullURI, _ := e.Value["pull"].(string) + status, _ := e.Value["status"].(string) + if pullURI != "" { + statusMap[pullURI] = status + } + } + + return pulls, statusMap, nil +} + +func resolveIssueState(stateMap map[string]string, issueURI string) string { + raw := stateMap[issueURI] + if strings.HasSuffix(raw, ".closed") { + return "closed" + } + return "open" +} + +func resolvePullStatus(statusMap map[string]string, pullURI string) string { + raw := statusMap[pullURI] + switch { + case strings.HasSuffix(raw, ".merged"): + return "merged" + case strings.HasSuffix(raw, ".closed"): + return "closed" + default: + return "open" + } +} + +type bskyProfileResponse struct { + DisplayName string `json:"displayName,omitempty"` + Avatar string `json:"avatar,omitempty"` +} + +func (s *Server) fetchBskyProfile(r *http.Request, did string) *bskyProfileResponse { + body, err := s.xrpc.Call(r.Context(), "https://public.api.bsky.app", "app.bsky.actor.getProfile", url.Values{"actor": {did}}) + if err != nil { + return nil + } + defer body.Close() + + var p bskyProfileResponse + if err := json.NewDecoder(body).Decode(&p); err != nil { + return nil + } + return &p +} + +// parseATURI splits an AT URI (at://did/collection/rkey) into its components. +func parseATURI(uri string) (did, collection, rkey string, err error) { + trimmed := strings.TrimPrefix(uri, "at://") + parts := strings.SplitN(trimmed, "/", 3) + if len(parts) != 3 { + return "", "", "", fmt.Errorf("invalid AT URI: %q", uri) + } + return parts[0], parts[1], parts[2], nil +} diff --git a/packages/api/internal/api/api.go b/packages/api/internal/api/api.go index dcfe399..9328588 100644 --- a/packages/api/internal/api/api.go +++ b/packages/api/internal/api/api.go @@ -17,6 +17,7 @@ import ( "tangled.org/desertthunder.dev/twister/internal/search" "tangled.org/desertthunder.dev/twister/internal/store" "tangled.org/desertthunder.dev/twister/internal/view" + "tangled.org/desertthunder.dev/twister/internal/xrpc" ) // Server is the HTTP search API server. @@ -26,16 +27,18 @@ type Server struct { cfg *config.Config log *slog.Logger constellation *constellation.Client + xrpc *xrpc.Client } // New creates a new API server. -func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log *slog.Logger, constellation *constellation.Client) *Server { +func New(searchRepo *search.Repository, st store.Store, cfg *config.Config, log *slog.Logger, constellation *constellation.Client, xrpcClient *xrpc.Client) *Server { return &Server{ search: searchRepo, store: st, cfg: cfg, log: log, constellation: constellation, + xrpc: xrpcClient, } } @@ -53,6 +56,37 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("GET /documents/{id}", s.handleGetDocument) mux.HandleFunc("GET /profiles/{did}/summary", s.handleProfileSummary) + mux.HandleFunc("GET /backlinks/count", s.handleBacklinksCount) + mux.HandleFunc("GET /activity/stream", s.handleActivityStream) + mux.HandleFunc("GET /identity/resolve", s.handleResolveHandle) + mux.HandleFunc("GET /identity/did/{did}", s.handleDidDocument) + mux.HandleFunc("GET /xrpc/knot/{knotHost}/{nsid}", s.handleKnotProxy) + mux.HandleFunc("GET /xrpc/pds/{pds}/{nsid}", s.handlePdsProxy) + mux.HandleFunc("GET /xrpc/bsky/{nsid}", s.handleBskyProxy) + + mux.HandleFunc("GET /actors/{handle}", s.handleGetActor) + mux.HandleFunc("GET /actors/{handle}/repos", s.handleListActorRepos) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}", s.handleGetActorRepo) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/tree", s.handleRepoTree) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/blob", s.handleRepoBlob) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/log", s.handleRepoLog) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/branches", s.handleRepoBranches) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/default-branch", s.handleRepoDefaultBranch) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/languages", s.handleRepoLanguages) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/tags", s.handleRepoTags) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/diff", s.handleRepoDiff) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/compare", s.handleRepoCompare) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/issues", s.handleRepoIssues) + mux.HandleFunc("GET /actors/{handle}/repos/{repo}/pulls", s.handleRepoPulls) + mux.HandleFunc("GET /actors/{handle}/issues", s.handleActorIssues) + mux.HandleFunc("GET /actors/{handle}/pulls", s.handleActorPulls) + mux.HandleFunc("GET /actors/{handle}/following", s.handleActorFollowing) + mux.HandleFunc("GET /actors/{handle}/strings", s.handleActorStrings) + mux.HandleFunc("GET /issues/{handle}/{rkey}", s.handleIssueDetail) + mux.HandleFunc("GET /issues/{handle}/{rkey}/comments", s.handleIssueComments) + mux.HandleFunc("GET /pulls/{handle}/{rkey}", s.handlePullDetail) + mux.HandleFunc("GET /pulls/{handle}/{rkey}/comments", s.handlePullComments) + if s.cfg.EnableAdminEndpoints { mux.HandleFunc("POST /admin/reindex", s.handleAdminReindex) mux.HandleFunc("POST /admin/reembed", s.handleNotImplemented) diff --git a/packages/api/internal/api/proxy.go b/packages/api/internal/api/proxy.go new file mode 100644 index 0000000..de80a30 --- /dev/null +++ b/packages/api/internal/api/proxy.go @@ -0,0 +1,182 @@ +package api + +import ( + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "time" + + "github.com/coder/websocket" + "tangled.org/desertthunder.dev/twister/internal/constellation" +) + +var proxyHTTPClient = &http.Client{Timeout: 30 * time.Second} + +// isValidHost checks that a host string is safe to use in an upstream URL. +// Rejects empty strings, anything with path separators or whitespace. +func isValidHost(host string) bool { + return len(host) > 0 && len(host) < 256 && + !strings.ContainsAny(host, "/ \t\r\n") +} + +// proxyHTTP fetches upstreamURL and writes the response (headers + body) to w verbatim. +func (s *Server) proxyHTTP(w http.ResponseWriter, r *http.Request, upstreamURL string) { + req, err := http.NewRequestWithContext(r.Context(), http.MethodGet, upstreamURL, nil) + if err != nil { + writeJSON(w, http.StatusInternalServerError, errorBody("proxy_error", "failed to build upstream request")) + return + } + if accept := r.Header.Get("Accept"); accept != "" { + req.Header.Set("Accept", accept) + } + + resp, err := proxyHTTPClient.Do(req) + if err != nil { + writeJSON(w, http.StatusBadGateway, errorBody("proxy_error", "upstream request failed")) + return + } + defer resp.Body.Close() + + if ct := resp.Header.Get("Content-Type"); ct != "" { + w.Header().Set("Content-Type", ct) + } + w.WriteHeader(resp.StatusCode) + _, _ = io.Copy(w, resp.Body) +} + +// handleKnotProxy proxies GET requests to a Tangled knot's XRPC endpoint. +// Route: GET /xrpc/knot/{knotHost}/{nsid} +func (s *Server) handleKnotProxy(w http.ResponseWriter, r *http.Request) { + knotHost := r.PathValue("knotHost") + nsid := r.PathValue("nsid") + if !isValidHost(knotHost) { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "invalid knot host")) + return + } + upstream := fmt.Sprintf("https://%s/xrpc/%s", knotHost, nsid) + if r.URL.RawQuery != "" { + upstream += "?" + r.URL.RawQuery + } + s.proxyHTTP(w, r, upstream) +} + +// handlePdsProxy proxies GET requests to an AT Protocol PDS XRPC endpoint. +// Route: GET /xrpc/pds/{pds}/{nsid} +func (s *Server) handlePdsProxy(w http.ResponseWriter, r *http.Request) { + pds := r.PathValue("pds") + nsid := r.PathValue("nsid") + if !isValidHost(pds) { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "invalid PDS host")) + return + } + upstream := fmt.Sprintf("https://%s/xrpc/%s", pds, nsid) + if r.URL.RawQuery != "" { + upstream += "?" + r.URL.RawQuery + } + s.proxyHTTP(w, r, upstream) +} + +// handleBskyProxy proxies GET requests to the Bluesky public API. +// Route: GET /xrpc/bsky/{nsid} +func (s *Server) handleBskyProxy(w http.ResponseWriter, r *http.Request) { + nsid := r.PathValue("nsid") + upstream := fmt.Sprintf("https://public.api.bsky.app/xrpc/%s", nsid) + if r.URL.RawQuery != "" { + upstream += "?" + r.URL.RawQuery + } + s.proxyHTTP(w, r, upstream) +} + +// handleResolveHandle proxies handle → DID resolution through bsky.social. +// Route: GET /identity/resolve?handle=... +func (s *Server) handleResolveHandle(w http.ResponseWriter, r *http.Request) { + upstream := "https://bsky.social/xrpc/com.atproto.identity.resolveHandle" + if r.URL.RawQuery != "" { + upstream += "?" + r.URL.RawQuery + } + s.proxyHTTP(w, r, upstream) +} + +// handleDidDocument fetches a DID document from plc.directory (did:plc) or +// the well-known endpoint (did:web) and proxies the response. +// Route: GET /identity/did/{did} +func (s *Server) handleDidDocument(w http.ResponseWriter, r *http.Request) { + did := r.PathValue("did") + var docURL string + switch { + case strings.HasPrefix(did, "did:plc:"): + docURL = fmt.Sprintf("https://plc.directory/%s", did) + case strings.HasPrefix(did, "did:web:"): + host := strings.TrimPrefix(did, "did:web:") + docURL = fmt.Sprintf("https://%s/.well-known/did.json", host) + default: + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "unsupported DID method")) + return + } + s.proxyHTTP(w, r, docURL) +} + +// handleBacklinksCount returns the Constellation backlinks count for a subject/source pair. +// Route: GET /backlinks/count?subject=...&source=... +func (s *Server) handleBacklinksCount(w http.ResponseWriter, r *http.Request) { + if s.constellation == nil { + writeJSON(w, http.StatusServiceUnavailable, errorBody("not_configured", "constellation is not configured")) + return + } + subject := r.URL.Query().Get("subject") + source := r.URL.Query().Get("source") + if subject == "" || source == "" { + writeJSON(w, http.StatusBadRequest, errorBody("invalid_parameter", "subject and source are required")) + return + } + n, err := s.constellation.GetBacklinksCount(r.Context(), constellation.BacklinksParams{ + Subject: subject, + Source: source, + }) + if err != nil { + s.log.Debug("backlinks count failed", slog.String("error", err.Error())) + writeJSON(w, http.StatusBadGateway, errorBody("constellation_error", "failed to fetch backlinks count")) + return + } + writeJSON(w, http.StatusOK, map[string]int{"count": n}) +} + +// handleActivityStream accepts a WebSocket connection and proxies it to the +// Jetstream firehose, forwarding all query parameters (wantedCollections, cursor, etc.) +// Route: GET /activity/stream +func (s *Server) handleActivityStream(w http.ResponseWriter, r *http.Request) { + clientConn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ + OriginPatterns: []string{"*"}, + }) + if err != nil { + s.log.Debug("activity stream: accept failed", slog.String("error", err.Error())) + return + } + defer clientConn.CloseNow() + + jetstreamURL := "wss://jetstream2.us-east.bsky.network/subscribe" + if r.URL.RawQuery != "" { + jetstreamURL += "?" + r.URL.RawQuery + } + + ctx := r.Context() + jetstreamConn, _, err := websocket.Dial(ctx, jetstreamURL, nil) + if err != nil { + s.log.Debug("activity stream: jetstream dial failed", slog.String("error", err.Error())) + _ = clientConn.Close(websocket.StatusBadGateway, "failed to connect to upstream") + return + } + defer jetstreamConn.CloseNow() + + for { + msgType, msg, err := jetstreamConn.Read(ctx) + if err != nil { + return + } + if err := clientConn.Write(ctx, msgType, msg); err != nil { + return + } + } +} diff --git a/packages/api/internal/constellation/client.go b/packages/api/internal/constellation/client.go index e0d120b..915472d 100644 --- a/packages/api/internal/constellation/client.go +++ b/packages/api/internal/constellation/client.go @@ -11,6 +11,7 @@ import ( "io" "net/http" "net/url" + "strings" "time" ) @@ -21,10 +22,11 @@ const ( ) // Source constants for common Tangled collections. +// Format: "collection:path" where path uses Constellation dot-notation (e.g. ".subject"). const ( - SourceStarURI string = "sh.tangled.feed.star:subject.uri" - SourceFollowDID string = "sh.tangled.graph.follow:subject" - SourceReactionURI string = "sh.tangled.feed.reaction:subject.uri" + SourceStarURI string = "sh.tangled.feed.star:.subject" + SourceFollowDID string = "sh.tangled.graph.follow:.subject" + SourceReactionURI string = "sh.tangled.feed.reaction:.subject" ) // Client is an HTTP client for the Constellation backlink API. @@ -89,47 +91,61 @@ type BacklinksParams struct { // BacklinkRecord is one entry returned by GetBacklinks. type BacklinkRecord struct { - URI string `json:"uri"` - CID string `json:"cid"` - ActorDID string `json:"actorDid"` - CreatedAt string `json:"createdAt"` + DID string `json:"did"` + Collection string `json:"collection"` + RKey string `json:"rkey"` } // BacklinksResponse is returned by GetBacklinks. type BacklinksResponse struct { - Records []BacklinkRecord `json:"records"` - Cursor string `json:"cursor,omitempty"` + Total int `json:"total"` + LinkingRecords []BacklinkRecord `json:"linking_records"` + Cursor *string `json:"cursor,omitempty"` } // GetBacklinksCount returns the count of records linking to the given subject. // Results are cached with the configured TTL. Errors are returned without caching. +// p.Source must be in "collection:path" format, e.g. "sh.tangled.feed.star:.subject". func (c *Client) GetBacklinksCount(ctx context.Context, p BacklinksParams) (int, error) { cacheKey := "count\x00" + p.Subject + "\x00" + p.Source if n, ok := c.countCache.Get(cacheKey); ok { return n, nil } + collection, path, ok := strings.Cut(p.Source, ":") + if !ok { + return 0, fmt.Errorf("constellation: invalid source %q: expected collection:path", p.Source) + } + params := url.Values{} - params.Set("subject", p.Subject) - params.Set("source", p.Source) + params.Set("target", p.Subject) + params.Set("collection", collection) + params.Set("path", path) var resp struct { - Count int `json:"count"` + Total int `json:"total"` } - if err := c.get(ctx, "blue.microcosm.links.getBacklinksCount", params, &resp); err != nil { + if err := c.getLinks(ctx, params, &resp); err != nil { return 0, err } - c.countCache.Set(cacheKey, resp.Count) - return resp.Count, nil + c.countCache.Set(cacheKey, resp.Total) + return resp.Total, nil } // GetBacklinks returns records linking to the given subject. // Results are not cached because paginated lists change frequently. +// p.Source must be in "collection:path" format, e.g. "sh.tangled.feed.star:.subject". func (c *Client) GetBacklinks(ctx context.Context, p BacklinksParams) (*BacklinksResponse, error) { + collection, path, ok := strings.Cut(p.Source, ":") + if !ok { + return nil, fmt.Errorf("constellation: invalid source %q: expected collection:path", p.Source) + } + params := url.Values{} - params.Set("subject", p.Subject) - params.Set("source", p.Source) + params.Set("target", p.Subject) + params.Set("collection", collection) + params.Set("path", path) if p.DID != "" { params.Set("did", p.DID) } @@ -141,14 +157,14 @@ func (c *Client) GetBacklinks(ctx context.Context, p BacklinksParams) (*Backlink } var resp BacklinksResponse - if err := c.get(ctx, "blue.microcosm.links.getBacklinks", params, &resp); err != nil { + if err := c.getLinks(ctx, params, &resp); err != nil { return nil, err } return &resp, nil } -func (c *Client) get(ctx context.Context, method string, params url.Values, out any) error { - u := c.baseURL + "/xrpc/" + method +func (c *Client) getLinks(ctx context.Context, params url.Values, out any) error { + u := c.baseURL + "/links" if len(params) > 0 { u += "?" + params.Encode() } @@ -163,7 +179,7 @@ func (c *Client) get(ctx context.Context, method string, params url.Values, out resp, err := c.http.Do(req) if err != nil { - return fmt.Errorf("constellation: request %s: %w", method, err) + return fmt.Errorf("constellation: request /links: %w", err) } defer resp.Body.Close() @@ -173,13 +189,13 @@ func (c *Client) get(ctx context.Context, method string, params url.Values, out Message string `json:"message"` } if json.Unmarshal(body, &errResp) == nil && errResp.Message != "" { - return fmt.Errorf("constellation: %s: status %d: %s", method, resp.StatusCode, errResp.Message) + return fmt.Errorf("constellation: /links: status %d: %s", resp.StatusCode, errResp.Message) } - return fmt.Errorf("constellation: %s: status %d", method, resp.StatusCode) + return fmt.Errorf("constellation: /links: status %d", resp.StatusCode) } if err := json.NewDecoder(resp.Body).Decode(out); err != nil { - return fmt.Errorf("constellation: %s: decode response: %w", method, err) + return fmt.Errorf("constellation: /links: decode response: %w", err) } return nil } diff --git a/packages/api/internal/store/db.go b/packages/api/internal/store/db.go index 9d16c3c..b59c414 100644 --- a/packages/api/internal/store/db.go +++ b/packages/api/internal/store/db.go @@ -61,8 +61,6 @@ func Migrate(db *sql.DB, url string) error { return fmt.Errorf("create schema_migrations table: %w", err) } - // For databases that were created before migration tracking was added, - // backfill schema_migrations by introspecting which tables/columns exist. if err := backfillMigrationHistory(db); err != nil { return fmt.Errorf("backfill migration history: %w", err) } @@ -115,7 +113,6 @@ func backfillMigrationHistory(db *sql.DB) error { return nil } - // If the documents table does not exist yet this is a fresh database — nothing to backfill. if !sqliteTableExists(db, "documents") { return nil } @@ -127,20 +124,16 @@ func backfillMigrationHistory(db *sql.DB) error { ) } - // 001 — documents table is present. mark("001_initial.sql") - // 002 — identity_handles table. if sqliteTableExists(db, "identity_handles") { mark("002_identity_handles.sql") } - // 003 — documents_fts virtual table. if sqliteTableExists(db, "documents_fts") { mark("003_documents_fts.sql") } - // 004 — web_url column on documents. if sqliteColumnExists(db, "documents", "web_url") { mark("004_web_url.sql") } diff --git a/packages/api/main.go b/packages/api/main.go index 8141c8e..ff2f871 100644 --- a/packages/api/main.go +++ b/packages/api/main.go @@ -106,7 +106,13 @@ func newAPICmd(local *bool) *cobra.Command { ) log.Info("constellation client configured", slog.String("url", cfg.ConstellationURL)) - srv := api.New(searchRepo, st, cfg, log, constellationClient) + xrpcClient := xrpc.NewClient( + xrpc.WithPLCDirectory(cfg.PLCDirectoryURL), + xrpc.WithIdentityService(cfg.IdentityServiceURL), + xrpc.WithTimeout(cfg.XRPCTimeout), + ) + + srv := api.New(searchRepo, st, cfg, log, constellationClient, xrpcClient) ctx, cancel := baseContext() defer cancel() -- knotmirror 0.0.1