From 1bb78a953be4b64957904997b17561019c710376 Mon Sep 17 00:00:00 2001 From: Florian <45694132+flo-bit@users.noreply.github.com> Date: Sat, 25 Apr 2026 15:21:37 +0200 Subject: [PATCH] add labels --- README.md | 1 + docs/01-indexing.md | 1 + docs/08-labels.md | 148 ++++++++ packages/contrail/package.json | 1 + packages/contrail/src/cli.ts | 66 +++- packages/contrail/src/contrail.ts | 29 ++ packages/contrail/src/core/db/schema.ts | 8 + packages/contrail/src/core/labels/apply.ts | 64 ++++ packages/contrail/src/core/labels/hydrate.ts | Bin 0 -> 3547 bytes packages/contrail/src/core/labels/resolve.ts | 134 ++++++++ packages/contrail/src/core/labels/schema.ts | 30 ++ packages/contrail/src/core/labels/select.ts | 75 +++++ .../contrail/src/core/labels/subscribe.ts | 315 ++++++++++++++++++ packages/contrail/src/core/labels/types.ts | 63 ++++ .../contrail/src/core/router/collection.ts | 97 +++++- packages/contrail/src/core/router/index.ts | 21 ++ packages/contrail/src/core/router/profiles.ts | 3 + packages/contrail/src/core/types.ts | 5 + packages/contrail/src/index.ts | 19 ++ packages/contrail/src/worker/index.ts | 6 + packages/contrail/tests/labels-router.test.ts | 141 ++++++++ packages/contrail/tests/labels.test.ts | 188 +++++++++++ pnpm-lock.yaml | 3 + 23 files changed, 1404 insertions(+), 14 deletions(-) create mode 100644 docs/08-labels.md create mode 100644 packages/contrail/src/core/labels/apply.ts create mode 100644 packages/contrail/src/core/labels/hydrate.ts create mode 100644 packages/contrail/src/core/labels/resolve.ts create mode 100644 packages/contrail/src/core/labels/schema.ts create mode 100644 packages/contrail/src/core/labels/select.ts create mode 100644 packages/contrail/src/core/labels/subscribe.ts create mode 100644 packages/contrail/src/core/labels/types.ts create mode 100644 packages/contrail/tests/labels-router.test.ts create mode 100644 packages/contrail/tests/labels.test.ts diff --git a/README.md b/README.md index 64d09f0..d9605f8 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,7 @@ returns every `community.lexicon.calendar.event` record published anywhere on at - [Spaces](./docs/05-spaces.md) — permissioned records stored by the appview - [Communities](./docs/06-communities.md) — group-controlled atproto DIDs - [Sync](./docs/07-sync.md) — reactive client-side store over `watchRecords` +- [Labels](./docs/08-labels.md) — atproto-native moderation hydration from external labelers - Frameworks: [SvelteKit + Cloudflare](./docs/frameworks/sveltekit-cloudflare.md) ## Packages diff --git a/docs/01-indexing.md b/docs/01-indexing.md index 0b8d70d..81e44f1 100644 --- a/docs/01-indexing.md +++ b/docs/01-indexing.md @@ -203,3 +203,4 @@ const db = createPostgresDatabase(pool); | `spaces` | — | See [Spaces](./05-spaces.md) | | `community` | — | See [Communities](./06-communities.md) | | `realtime` | — | See [Sync](./07-sync.md) | +| `labels` | — | See [Labels](./08-labels.md) | diff --git a/docs/08-labels.md b/docs/08-labels.md new file mode 100644 index 0000000..f6c92dd --- /dev/null +++ b/docs/08-labels.md @@ -0,0 +1,148 @@ +# Labels + +Atproto-native moderation hydration. Subscribe to one or more labelers, index their labels, and attach them to records and profiles in your XRPC responses. Opt-in; zero cost if you don't enable it. + +## Mental model + +> A **label** is a `(src, uri, val)` triple authored by a labeler DID. A **labeler** is a regular atproto account that publishes signed annotations about other accounts and records via `com.atproto.label.subscribeLabels`. + +- One contrail deployment can subscribe to many labelers. +- The caller of your XRPC picks which subset to honor per request via the `atproto-accept-labelers` header (or `?labelers=` query param when headers are awkward — SSE/WS). +- Labels hydrate onto every `listRecords`, `getRecord`, `getProfile`, and `?profiles=true` response without changing your collection config. +- This module only consumes labels. Producing them — your appview emitting its own labels — is a separate question. See *Future work* below. + +## Enable + +```ts +import type { ContrailConfig } from "@atmo-dev/contrail"; + +const config: ContrailConfig = { + namespace: "com.example", + collections: { /* ... */ }, + labels: { + sources: [ + { did: "did:plc:ar7c4by46qjdydhdevvrndac" }, // bsky moderation + { did: "did:plc:newsmast" }, + ], + }, +}; +``` + +`initSchema` creates a `labels` table and a `labeler_cursors` table. Both live on the main DB; nothing per-collection. + +## Caller selection + +Per request, contrail picks accepted labelers in this order: + +1. `atproto-accept-labelers: did:plc:a, did:plc:b` — the spec's HTTP header. +2. `?labelers=did:plc:a,did:plc:b` — fallback for transports that can't set headers easily. +3. `config.labels.defaults` — operator policy. +4. Every entry in `config.labels.sources`. + +The list is intersected with what's actually configured (unknowns dropped — see [`allowUserSupplied`](#allowusersupplied) below) and capped at `maxPerRequest` (default 20). Contrail echoes the applied set back via `atproto-content-labelers`. + +``` +GET /xrpc/com.example.event.listRecords + atproto-accept-labelers: did:plc:ar7c4by46qjdydhdevvrndac +``` + +→ + +```jsonc +// Response: atproto-content-labelers: did:plc:ar7c4by46qjdydhdevvrndac +{ + "records": [ + { + "uri": "at://did:plc:.../com.example.event/...", + "value": { /* ... */ }, + "labels": [ + { + "src": "did:plc:ar7c4by46qjdydhdevvrndac", + "uri": "at://did:plc:.../com.example.event/...", + "val": "spam", + "cts": "2026-04-25T00:00:00.000Z" + } + ] + } + ] +} +``` + +`labels` matches `com.atproto.label.defs#label` field-for-field — pass it straight to atproto SDK moderation helpers. + +### `allowUserSupplied` + +Default: `false` — caller-supplied DIDs that aren't in `sources` are silently dropped. Set `true` to honor them anyway. The current request still only returns labels for already-indexed sources; lazy registration of new labelers is future work. + +### `defaults: []` + +Set defaults to an empty array if you want strict opt-in: callers that send no header / param see no labels at all. + +## Hydration semantics + +For each `(src, uri, val)` tuple visible to the caller, hydration picks the row with the highest `cts`. If that row has `neg=true`, the label is treated as retracted and dropped. Expired rows (`exp` past `now`) are filtered at the SQL level. CID-pinned labels apply only when the indexed record's CID matches. + +Account-level labels (subject = bare DID) hydrate onto profiles. They appear inside each `ProfileEntry.labels` of the `profiles` array on `?profiles=true` responses, and on `getProfile`. + +## Ingestion + +`com.atproto.label.subscribeLabels` is a per-labeler WebSocket firehose with a CBOR frame envelope. Contrail mirrors its existing Jetstream pipeline: + +| Mode | Function | When | +|---|---|---| +| Cron-driven | `contrail.ingestLabels()` | Cloudflare Workers — one drain per cron tick | +| Persistent | `contrail.runPersistentLabels()` | Node / long-lived servers — one socket per labeler, auto-reconnect | +| One-shot backfill | `pnpm contrail labels-backfill [--remote]` | Local script, drains until each labeler reports caught up | + +When `config.labels` is set, the bundled `createWorker` already calls `ingestLabels()` from `scheduled()` alongside `ingest()` — no boilerplate. + +```ts +// node / long-lived +const ac = new AbortController(); +await Promise.all([ + contrail.runPersistent({ signal: ac.signal }), + contrail.runPersistentLabels({ signal: ac.signal }), +]); +``` + +Per-labeler cursors live in `labeler_cursors` (`{did, cursor, endpoint, resolved_at}`). Endpoints are resolved from the DID doc's `service[id="#atproto_labeler"]` and cached for 6h. On `#info { name: "OutdatedCursor" }` frames, contrail resets the cursor to `0` so the next cycle re-backfills. + +### `backfill: false` + +Per source. Default: backfill from `cursor=0` on first sight. Set `false` to start at "now" — useful for very chatty labelers where you don't need history. + +```ts +labels: { + sources: [{ did: "did:plc:somenoisylabeler", backfill: false }], +} +``` + +## Storage + +```sql +CREATE TABLE labels ( + src TEXT NOT NULL, -- labeler DID + uri TEXT NOT NULL, -- subject: at://... or did:... + val TEXT NOT NULL, -- label value + cid TEXT, -- optional record-version pin + neg INTEGER NOT NULL DEFAULT 0, + exp INTEGER, -- expiry, unix sec + cts INTEGER NOT NULL, -- creation time, unix sec + sig BLOB, -- signature bytes (stored, not verified in v1) + PRIMARY KEY (src, uri, val, cts) +); +``` + +The PK includes `cts`, so a `neg=true` retraction is a *new row* that replaces the previous decision via the read-time collapse rule above — never an in-place mutation. This matches the spec, tolerates out-of-order delivery, and survives a labeler that flip-flops. + +## What's not here + +- **Signature verification.** `sig` is stored if the labeler supplies it, but contrail does not verify it in v1. Document as TODO; most appviews skip it. +- **Live label updates on `watchRecords`.** The realtime stream snapshots labels with the initial query but does not push label deltas. Adding this means publishing `labels:` topic events from the ingest worker and merging them in `runQueryStream` — future work. +- **Spaces / community labels.** Hydration already runs on the spaces read paths, so labels emitted by a labeler-DID member of a space (or under a community DID) will show up if you write them to the `labels` table. The auth surface for "make this DID a labeler in this space" is not yet exposed as XRPCs. +- **Outbound `subscribeLabels`.** Contrail does not republish labels. Communities-as-labelers (using the community DID as `src`) is a natural extension once you want to act as a labeler instead of just consume. +- **Label definitions / preferences UX.** Custom label names, blur behaviors, severity, and per-user preference state belong on the *client*, fetched directly from each labeler. Contrail intentionally stays out of this. + +## Design + +Follows the [atproto label spec](https://atproto.com/specs/label) literally. The wire format on responses matches `com.atproto.label.defs#label` so existing atproto SDKs can consume it directly. Storage is the data model normalized into rows; ingestion mirrors Jetstream both in code shape and in operator UX. diff --git a/packages/contrail/package.json b/packages/contrail/package.json index b111360..4b6db28 100644 --- a/packages/contrail/package.json +++ b/packages/contrail/package.json @@ -65,6 +65,7 @@ }, "dependencies": { "@atcute/atproto": "^3.1.10", + "@atcute/cbor": "^2.3.2", "@atcute/cid": "^2.4.1", "@atcute/client": "^4.2.1", "@atcute/identity": "^1.1.4", diff --git a/packages/contrail/src/cli.ts b/packages/contrail/src/cli.ts index 9246dae..505b5f4 100644 --- a/packages/contrail/src/cli.ts +++ b/packages/contrail/src/cli.ts @@ -19,15 +19,17 @@ import { Contrail } from "./contrail.js"; import type { CollectionStats, RefreshResult } from "./core/refresh.js"; import type { Database } from "./core/types.js"; -type Subcommand = "backfill" | "refresh" | "dev" | "help"; +type Subcommand = "backfill" | "refresh" | "labels-backfill" | "dev" | "help"; const USAGE = `contrail [options] Subcommands: - backfill One-time bulk load from each known DID's PDS (resumable) - refresh Fresh sweep: reconcile PDS vs DB, report missing + stale - dev Local wrangler dev + auto-trigger cron + backfill/refresh prompts - help Print this message + backfill One-time bulk load from each known DID's PDS (resumable) + refresh Fresh sweep: reconcile PDS vs DB, report missing + stale + labels-backfill One-shot drain per configured labeler — runs catch-up cycles + until each labeler has no more pending events (resumable) + dev Local wrangler dev + auto-trigger cron + backfill/refresh prompts + help Print this message Options (backfill): --config Path to Contrail config file (TS or JS). @@ -287,6 +289,60 @@ async function main(): Promise { return 0; } + if (opts.cmd === "labels-backfill") { + const configPath = resolveConfigPath(opts); + if (!configPath) return 1; + const config = await loadConfig(configPath); + if (!config.labels || config.labels.sources.length === 0) { + console.error("No labels configured (config.labels.sources is empty)."); + return 1; + } + const { getPlatformProxy } = await import("wrangler"); + const { env, dispose } = await getPlatformProxy({ + environment: opts.remote ? "production" : undefined, + }); + try { + const db = (env as Record)[opts.binding] as Database | undefined; + if (!db) { + console.error(`No binding named "${opts.binding}" in wrangler env.`); + return 1; + } + const contrail = new Contrail(config); + await contrail.init(db); + + // Run cycles until each cycle drains nothing new — measured by the + // labeler_cursors not advancing across two consecutive cycles. + const before = new Map(); + let stable = 0; + while (stable < 2) { + const rows = ( + await db + .prepare("SELECT did, cursor FROM labeler_cursors") + .all<{ did: string; cursor: number }>() + ).results ?? []; + for (const r of rows) before.set(r.did, r.cursor); + await contrail.ingestLabels({ timeoutMs: 60_000 }, db); + const after = ( + await db + .prepare("SELECT did, cursor FROM labeler_cursors") + .all<{ did: string; cursor: number }>() + ).results ?? []; + let advanced = false; + for (const r of after) { + if ((before.get(r.did) ?? -1) !== r.cursor) { + advanced = true; + break; + } + } + stable = advanced ? 0 : stable + 1; + } + console.log("labels-backfill: caught up"); + return 0; + } finally { + await dispose(); + } + } + if (opts.cmd === "dev") return cmdDev(opts); console.error(USAGE); diff --git a/packages/contrail/src/contrail.ts b/packages/contrail/src/contrail.ts index 5f0fb72..630fbc9 100644 --- a/packages/contrail/src/contrail.ts +++ b/packages/contrail/src/contrail.ts @@ -13,6 +13,11 @@ import { processNotifyUris } from "./core/router/notify"; import type { NotifyResult } from "./core/router/notify"; import { runPersistent as runPersistentIngestion } from "./core/persistent"; import type { PersistentIngestOptions } from "./core/persistent"; +import { + runLabelIngestCycle, + runPersistentLabels, + type PersistentLabelsOptions, +} from "./core/labels/subscribe"; import type { PubSub } from "./core/realtime/types"; import { InMemoryPubSub } from "./core/realtime/in-memory"; import { createApp, type CreateAppOptions } from "./core/router"; @@ -99,6 +104,30 @@ export class Contrail { }); } + /** Run one labeler ingestion cycle — for every labeler in `config.labels.sources`, + * drains pending `subscribeLabels` frames and persists them to the `labels` + * table. No-op when `config.labels` is unset. Mirrors `ingest()`. */ + async ingestLabels( + options?: { timeoutMs?: number }, + db?: Database, + ): Promise { + if (!this.config.labels) return; + await runLabelIngestCycle(this.getDb(db), this.config, options?.timeoutMs); + } + + /** Long-lived label ingestion — one socket per labeler, auto-reconnect on drop. + * No-op when `config.labels` is unset. Mirrors `runPersistent()`. */ + async runPersistentLabels( + options?: Omit, + db?: Database, + ): Promise { + if (!this.config.labels) return; + await runPersistentLabels(this.getDb(db), this.config, { + ...options, + logger: this.config.logger, + }); + } + /** Discover users from relays. Returns discovered DIDs. */ async discover(db?: Database): Promise { const d = this.getDb(db); diff --git a/packages/contrail/src/core/db/schema.ts b/packages/contrail/src/core/db/schema.ts index 7bf06d7..3f3c676 100644 --- a/packages/contrail/src/core/db/schema.ts +++ b/packages/contrail/src/core/db/schema.ts @@ -12,6 +12,7 @@ import { import { getSearchableFields } from "../search"; import { buildSpacesBaseSchema } from "../spaces/schema"; import { buildCommunitySchema } from "../community/schema"; +import { buildLabelsSchema } from "../labels/schema"; function getResolved(config: ContrailConfig): ResolvedMaps { return (config as ResolvedContrailConfig)._resolved ?? resolveConfig(config)._resolved; @@ -317,6 +318,13 @@ export async function initSchema( await target.batch(communityStmts.map((s) => target.prepare(s))); } + if (config.labels) { + // Labels tables live on the main DB — they're keyed by at-URI / DID and + // are read alongside public records during hydration. + const labelsStmts = buildLabelsSchema(dialect); + await db.batch(labelsStmts.map((s) => db.prepare(s))); + } + // FTS5 may not be available (e.g. node:sqlite) — skip gracefully for (const stmt of ftsStatements) { try { diff --git a/packages/contrail/src/core/labels/apply.ts b/packages/contrail/src/core/labels/apply.ts new file mode 100644 index 0000000..cbf5558 --- /dev/null +++ b/packages/contrail/src/core/labels/apply.ts @@ -0,0 +1,64 @@ +import type { Database, Statement } from "../types"; + +/** Wire shape of a single `com.atproto.label.defs#label` entry. Field names + * match the spec exactly. We accept the spec's ISO-8601 strings and + * convert to unix seconds at the storage boundary. */ +export interface IncomingLabel { + src: string; + uri: string; + val: string; + cid?: string; + neg?: boolean; + exp?: string; + cts: string; + sig?: Uint8Array; +} + +/** Upsert a batch of labels. Idempotent on `(src, uri, val, cts)`. Bad rows + * (missing required fields, unparseable timestamps) are dropped silently; + * we don't want one malformed label to abort an entire labeler frame. */ +export async function applyLabels( + db: Database, + labels: IncomingLabel[], +): Promise { + if (labels.length === 0) return 0; + const stmts: Statement[] = []; + let kept = 0; + for (const l of labels) { + if (!l.src || !l.uri || !l.val || !l.cts) continue; + const cts = isoToUnixSec(l.cts); + if (cts == null) continue; + const exp = l.exp ? isoToUnixSec(l.exp) : null; + stmts.push( + db + .prepare( + `INSERT INTO labels (src, uri, val, cid, neg, exp, cts, sig) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(src, uri, val, cts) DO UPDATE SET + cid = excluded.cid, + neg = excluded.neg, + exp = excluded.exp, + sig = excluded.sig`, + ) + .bind( + l.src, + l.uri, + l.val, + l.cid ?? null, + l.neg ? 1 : 0, + exp, + cts, + l.sig ?? null, + ), + ); + kept++; + } + if (stmts.length > 0) await db.batch(stmts); + return kept; +} + +function isoToUnixSec(iso: string): number | null { + const ms = Date.parse(iso); + if (!Number.isFinite(ms)) return null; + return Math.floor(ms / 1000); +} diff --git a/packages/contrail/src/core/labels/hydrate.ts b/packages/contrail/src/core/labels/hydrate.ts new file mode 100644 index 0000000000000000000000000000000000000000..02b947a44ef764f5c31f64fefd468727f76fd7d1 GIT binary patch literal 3547 zcmd1IEyyn_Q7EY_NL8p-a7iplOiCERx2bWmLzAS zrg-KBmZla}A}cM*FD*$e($7fEDM&2>>EY7X(ozV|EK1ca&PXgsRY=P(Qb<(Ds7xtJ zEJ;mK$Vp5}%~5#NFhwC3Y-X`SLUMktUSdf>QGQ9j9#~p0B{i*B8O%)JQqWRRNXtyk zNzqNqFVY3G6pHf|lJoP5OLJ3;iWQO*^Arjai;ER9OB9MriV`!^GfEUnGK%s`(=!xG z@)e*)D+IfE7wai#>2sx4fc>ADSCU$kmYAHX-~n}<57=$hTnY-sMafnm!!q;It+^Bw zN{cd)xn+qt$lT=26nkVoC;C%ADCFm6mVm+utS2vDA+@3)v#3%bvsj@xwFIg^wW0vSLYN5($wjG& zC7Jno3MHAjsl_FUxdl*N$tA@ISJrZIA^abdSPu4jP=2`rIlhM(t5BnmSDKRp@>E`G zx|Kp+X>L+#5r~^wQ2^s2DFV9)rl^(+oO0Y!OOi7nsiYX1c8XI=6!Oy)ic6ESQj<%H zbrjMvb4pT+KyjO#n3I!Qq+YC$n4Fwi017v7s!1&>)&VD+Bf(?VJ z%Pdw%R7l86O}7O_X@WvgYDrOIGAPLP6oOJqN{jM93JOw-bfGR*$W1KJ<5JK98xjol z1vuFzr79$r=!OP)7AxczDI_Hpr7F01x)g)6M@dGiLUCelszOO(QckKuMt)98u|iUQ zNd`DNQz22PUUDFhVd=VlhC+6001!BwJU zuIN}hJ8%-qOjFQ+22ZhGPHJ9yNrr;0t*wHALQM@kGEpQnK>=M_l&4T#Yt6-_prEg> z;F6kBT967)7^uPK;_0H0n^>ukSWu9fSfr4dr;w16nNpkpj^M;%1yJ}x0v}>nVrfZ6 zevvg;931WtNw68o`MIf((lIBqxI{r47SMWJ3gCjMM4>b(vO$B`gLjwZ? zO<1TpW#*-1=A{=alw~HO7yOAusa9am6%?hWWmf1Y=;`StXO!k;$Lc8rl!A)gRE31p zih=}4m_jqER#IwOeo?9xIBXN*<30W2;}aB+`~)h{LGh6Uv05R&C?&NBRzX0F0!ar4 z`k?3nD?`LVQGR)`f~`Vgd17V>n!|07ia0w^#sFmvP|{aWNC75RwrK}R7u zGerkfQ0st7U>$|zl41q7Ab($I$;71q0pT95L9QUJ3Z8xn8Y^@C)_vQSc7}nX6!@V2`G%v?vp779^B36CjQRr6*7lgGIWI0whks*;fIa z9$={tqyZk60h!6!;82H_T}Aoj3I(Y}n8l%U;I@lF4hNFwFVWu&?+xA z1>_EJQS9vLqEL{TXQhw`$sr2h6q1*in+mE7LD8C-mYJ*osZ$g{suK$eav(`Ar&6IL zADmb8^&$GoQj3Z+^Yio+Jk!953R2C3TY0G|3W-o-bdxhvKy@p)L07{EsmC!~u$Q)493@Qe0X@kT{i!wna8OUAW<~B&IEHMWr25vf*6o=${7Uye# z^6ORTwSxwzm>HJG`j wd6^Z#smZX`u_mky1}=u6nn48_sJ_;KDpt^f)Yy6@`JTc4!QdvDh9<~-0O=5WfdBvi literal 0 HcmV?d00001 diff --git a/packages/contrail/src/core/labels/resolve.ts b/packages/contrail/src/core/labels/resolve.ts new file mode 100644 index 0000000..694bfec --- /dev/null +++ b/packages/contrail/src/core/labels/resolve.ts @@ -0,0 +1,134 @@ +import { + CompositeDidDocumentResolver, + PlcDidDocumentResolver, + WebDidDocumentResolver, +} from "@atcute/identity-resolver"; +import type { Did } from "@atcute/lexicons"; +import type { Database } from "../types"; + +/** Reject endpoint URLs that point to private/internal addresses or non-HTTPS. + * Mirrors the validator in core/client.ts — labeler endpoints should be + * publicly reachable for the same reasons PDS endpoints should. */ +function validateEndpointUrl(url: string): boolean { + try { + const parsed = new URL(url); + if (parsed.protocol !== "https:") return false; + const host = parsed.hostname; + if (host === "localhost" || host === "127.0.0.1" || host === "[::1]") return false; + if (host.startsWith("10.")) return false; + if (host.startsWith("192.168.")) return false; + if (host.startsWith("169.254.")) return false; + if (/^172\.(1[6-9]|2\d|3[01])\./.test(host)) return false; + return true; + } catch { + return false; + } +} + +const didResolver = new CompositeDidDocumentResolver({ + methods: { + plc: new PlcDidDocumentResolver(), + web: new WebDidDocumentResolver(), + }, +}); + +/** Look up the labeler service endpoint from a DID. + * Reads the DID doc's `service[id="#atproto_labeler"].serviceEndpoint`. */ +export async function resolveLabelerEndpoint(did: string): Promise { + if (!did.startsWith("did:plc:") && !did.startsWith("did:web:")) return null; + try { + const doc = await didResolver.resolve(did as Did<"plc"> | Did<"web">); + const endpoint = doc.service + ?.find((s) => s.id === "#atproto_labeler") + ?.serviceEndpoint?.toString(); + if (!endpoint) return null; + if (!validateEndpointUrl(endpoint)) return null; + return endpoint; + } catch { + return null; + } +} + +/** State row for a labeler — the per-DID equivalent of the singleton + * jetstream `cursor` table, with cached endpoint to avoid repeated DID-doc + * fetches. */ +export interface LabelerState { + did: string; + cursor: number; + endpoint: string | null; + resolved_at: number | null; +} + +const ENDPOINT_TTL_MS = 6 * 60 * 60 * 1000; // 6h, matches the recommended client cache for label-defs + +/** Get cached `(endpoint, cursor)` for a labeler. Resolves endpoint on + * cache miss or staleness; persists endpoint + resolved_at back to the DB + * so subsequent ingest cycles avoid the network round-trip. */ +export async function getLabelerState( + db: Database, + did: string, + endpointOverride: string | undefined, +): Promise { + const row = await db + .prepare( + "SELECT did, cursor, endpoint, resolved_at FROM labeler_cursors WHERE did = ?", + ) + .bind(did) + .first(); + + let endpoint = endpointOverride ?? row?.endpoint ?? null; + const stale = + !row?.resolved_at || Date.now() - row.resolved_at > ENDPOINT_TTL_MS; + + if (!endpoint || (!endpointOverride && stale)) { + endpoint = await resolveLabelerEndpoint(did); + if (!endpoint) return null; + const now = Date.now(); + await db + .prepare( + `INSERT INTO labeler_cursors (did, cursor, endpoint, resolved_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(did) DO UPDATE SET endpoint = excluded.endpoint, resolved_at = excluded.resolved_at`, + ) + .bind(did, row?.cursor ?? 0, endpoint, now) + .run(); + return { + did, + cursor: row?.cursor ?? 0, + endpoint, + resolved_at: now, + }; + } + + return row ?? { did, cursor: 0, endpoint, resolved_at: null }; +} + +/** Persist the highest seen seq number for a labeler. Idempotent; + * the next ingest cycle resumes from `cursor + 1` via the `?cursor=` param. */ +export async function saveLabelerCursor( + db: Database, + did: string, + cursor: number, +): Promise { + await db + .prepare( + `INSERT INTO labeler_cursors (did, cursor) + VALUES (?, ?) + ON CONFLICT(did) DO UPDATE SET cursor = excluded.cursor`, + ) + .bind(did, cursor) + .run(); +} + +/** Reset cursor to 0 — used in response to `#info { name: "OutdatedCursor" }` + * frames, which signal that the labeler's seq history was rewound. */ +export async function resetLabelerCursor(db: Database, did: string): Promise { + await db + .prepare( + `INSERT INTO labeler_cursors (did, cursor) + VALUES (?, 0) + ON CONFLICT(did) DO UPDATE SET cursor = 0`, + ) + .bind(did) + .run(); +} diff --git a/packages/contrail/src/core/labels/schema.ts b/packages/contrail/src/core/labels/schema.ts new file mode 100644 index 0000000..486ce27 --- /dev/null +++ b/packages/contrail/src/core/labels/schema.ts @@ -0,0 +1,30 @@ +import type { SqlDialect } from "../dialect"; + +/** DDL for the labels module. Single `labels` table covers record-level + * (uri starts with `at://`) and account-level (uri is a bare DID) entries — + * the spec collapses both into the same row shape. `labeler_cursors` + * mirrors the role of the singleton `cursor` table for jetstream, but + * per-labeler. */ +export function buildLabelsSchema(dialect: SqlDialect): string[] { + return [ + `CREATE TABLE IF NOT EXISTS labels ( + src TEXT NOT NULL, + uri TEXT NOT NULL, + val TEXT NOT NULL, + cid TEXT, + neg INTEGER NOT NULL DEFAULT 0, + exp ${dialect.bigintType}, + cts ${dialect.bigintType} NOT NULL, + sig BLOB, + PRIMARY KEY (src, uri, val, cts) + )`, + `CREATE INDEX IF NOT EXISTS idx_labels_uri ON labels(uri)`, + `CREATE INDEX IF NOT EXISTS idx_labels_src_cts ON labels(src, cts DESC)`, + `CREATE TABLE IF NOT EXISTS labeler_cursors ( + did TEXT PRIMARY KEY, + cursor ${dialect.bigintType} NOT NULL DEFAULT 0, + endpoint TEXT, + resolved_at ${dialect.bigintType} + )`, + ]; +} diff --git a/packages/contrail/src/core/labels/select.ts b/packages/contrail/src/core/labels/select.ts new file mode 100644 index 0000000..4293109 --- /dev/null +++ b/packages/contrail/src/core/labels/select.ts @@ -0,0 +1,75 @@ +import type { LabelsConfig } from "./types"; +import { DEFAULT_LABELS_MAX_PER_REQUEST } from "./types"; + +/** Pick which labelers to honor for this request. + * + * Order of precedence: + * 1. `atproto-accept-labelers` header (atproto spec) + * 2. `?labelers=` query param (fallback for SSE/WS where headers are awkward) + * 3. `config.defaults` (operator policy) + * 4. every entry in `config.sources` + * + * Each candidate DID is checked against `config.sources`. Unknown DIDs are + * dropped unless `allowUserSupplied: true`, in which case they're returned + * in `lazyAdd` for the caller to schedule a registration. The active + * request gets results only from already-indexed sources. + * + * Header values can carry `;param` modifiers (e.g. `did:plc:...;redact`); + * v1 strips and ignores those — only the bare DID is honored. */ +export interface SelectedLabelers { + /** DIDs to use for hydration this request. */ + accepted: string[]; + /** DIDs the caller asked for that aren't configured (only populated when + * `allowUserSupplied: true`). The hydrator ignores these for the current + * request — the caller should enqueue registration as a follow-up. */ + lazyAdd: string[]; +} + +export function selectAcceptedLabelers( + headerValue: string | null | undefined, + paramValue: string | null | undefined, + cfg: LabelsConfig, +): SelectedLabelers { + const cap = cfg.maxPerRequest ?? DEFAULT_LABELS_MAX_PER_REQUEST; + const known = new Set(cfg.sources.map((s) => s.did)); + + const fromCaller = parseLabelerList(headerValue) ?? parseLabelerList(paramValue); + + let candidates: string[]; + if (fromCaller && fromCaller.length > 0) { + candidates = fromCaller; + } else { + candidates = (cfg.defaults ?? cfg.sources.map((s) => s.did)).slice(); + } + + const accepted: string[] = []; + const lazyAdd: string[] = []; + const seen = new Set(); + for (const did of candidates) { + if (seen.has(did)) continue; + seen.add(did); + if (known.has(did)) { + accepted.push(did); + } else if (cfg.allowUserSupplied) { + lazyAdd.push(did); + } + if (accepted.length >= cap) break; + } + + return { accepted, lazyAdd }; +} + +/** Parse a comma-separated DID list. Returns null when the input is empty + * or undefined so callers can distinguish "absent" from "empty list" (the + * latter — `atproto-accept-labelers: ` — is technically valid and means + * "no labelers"; we treat it the same as absent for ergonomics). */ +function parseLabelerList(value: string | null | undefined): string[] | null { + if (!value) return null; + const out: string[] = []; + for (const raw of value.split(",")) { + // Drop `;param` modifiers from the spec (e.g. `;redact`). v1 ignores them. + const head = raw.split(";")[0]!.trim(); + if (head.startsWith("did:")) out.push(head); + } + return out.length > 0 ? out : null; +} diff --git a/packages/contrail/src/core/labels/subscribe.ts b/packages/contrail/src/core/labels/subscribe.ts new file mode 100644 index 0000000..c80ed2b --- /dev/null +++ b/packages/contrail/src/core/labels/subscribe.ts @@ -0,0 +1,315 @@ +import { decodeFirst } from "@atcute/cbor"; +import type { ContrailConfig, Database, Logger } from "../types"; +import type { LabelerSource } from "./types"; +import { applyLabels, type IncomingLabel } from "./apply"; +import { + getLabelerState, + resetLabelerCursor, + saveLabelerCursor, +} from "./resolve"; + +const DEFAULT_CYCLE_TIMEOUT_MS = 25_000; +const DEFAULT_BATCH_SIZE = 100; +const DEFAULT_FLUSH_INTERVAL_MS = 5_000; + +function getLogger(config: ContrailConfig): Logger { + return config.logger ?? console; +} + +/** One catch-up cycle for every configured labeler. Designed to fit inside a + * Cloudflare Workers cron tick — we drain frames until the labeler has no + * more buffered events for us, or `timeoutMs` is reached, then save cursor + * and disconnect. Mirrors the shape of `runIngestCycle` for jetstream. */ +export async function runLabelIngestCycle( + db: Database, + config: ContrailConfig, + timeoutMs = DEFAULT_CYCLE_TIMEOUT_MS, +): Promise { + if (!config.labels) return; + const log = getLogger(config); + const deadline = Date.now() + timeoutMs; + + for (const source of config.labels.sources) { + if (Date.now() >= deadline) { + log.log(`[labels] cycle deadline hit before processing ${source.did}`); + break; + } + const remaining = Math.max(2_000, deadline - Date.now()); + try { + await pumpOneLabeler(db, source, log, remaining, /* persistent */ false); + } catch (err) { + log.warn(`[labels] cycle for ${source.did} failed: ${err}`); + } + } +} + +export interface PersistentLabelsOptions { + signal?: AbortSignal; + batchSize?: number; + flushIntervalMs?: number; + logger?: Logger; +} + +/** Long-lived equivalent — keeps one socket per labeler open forever, with + * exponential backoff reconnect. Mirrors `runPersistent` for jetstream. */ +export async function runPersistentLabels( + db: Database, + config: ContrailConfig, + options: PersistentLabelsOptions = {}, +): Promise { + if (!config.labels) return; + const log = options.logger ?? config.logger ?? console; + const signal = options.signal; + + const tasks = config.labels.sources.map((source) => + runOneLabelerForever(db, source, log, signal, options), + ); + await Promise.all(tasks); +} + +async function runOneLabelerForever( + db: Database, + source: LabelerSource, + log: Logger, + signal: AbortSignal | undefined, + options: PersistentLabelsOptions, +): Promise { + let attempts = 0; + while (!signal?.aborted) { + try { + await pumpOneLabeler(db, source, log, /* timeoutMs */ Infinity, true, { + signal, + batchSize: options.batchSize ?? DEFAULT_BATCH_SIZE, + flushIntervalMs: options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS, + }); + attempts = 0; + } catch (err) { + if (signal?.aborted) break; + log.error(`[labels] ${source.did} stream error: ${err}`); + const delay = Math.min(1_000 * 2 ** attempts, 30_000); + attempts++; + log.log(`[labels] ${source.did} reconnecting in ${delay}ms (attempt ${attempts})`); + await new Promise((r) => setTimeout(r, delay)); + } + } +} + +interface PumpOptions { + signal?: AbortSignal; + batchSize?: number; + flushIntervalMs?: number; +} + +/** Open a `subscribeLabels` WebSocket, drain frames into a buffer, flush + * the buffer to `labels` in batches, and persist the seq cursor. Returns + * when: + * - the labeler closes the socket cleanly (caught up + no more events) + * - `timeoutMs` is reached (cron mode) + * - `signal` is aborted (persistent mode) + * - an error tears the socket down (caller may retry) */ +async function pumpOneLabeler( + db: Database, + source: LabelerSource, + log: Logger, + timeoutMs: number, + persistent: boolean, + pumpOpts: PumpOptions = {}, +): Promise { + const state = await getLabelerState(db, source.did, source.endpoint); + if (!state) { + log.warn(`[labels] could not resolve labeler endpoint for ${source.did}; skipping`); + return; + } + + // First-time policy: cursor 0 = "from the beginning" if backfill is on + // (default), null = "from now" otherwise. After the first cycle we always + // resume from the saved cursor — `backfill` only flips the start point. + const isFirstRun = state.cursor === 0 && state.resolved_at === null; + const backfill = source.backfill !== false; + const startCursor = isFirstRun && !backfill ? null : state.cursor; + + const url = buildWsUrl(state.endpoint!, startCursor); + log.log(`[labels] connecting to ${source.did} (cursor=${startCursor ?? "now"})`); + + const ws = new WebSocket(url); + ws.binaryType = "arraybuffer"; + + const buffer: IncomingLabel[] = []; + let highestSeq = state.cursor; + let flushing = false; + let resolveDone!: () => void; + let rejectDone!: (err: unknown) => void; + const done = new Promise((res, rej) => { + resolveDone = res; + rejectDone = rej; + }); + + const flush = async () => { + if (buffer.length === 0 || flushing) return; + flushing = true; + const batch = buffer.splice(0); + try { + const kept = await applyLabels(db, batch); + if (highestSeq > state.cursor) { + await saveLabelerCursor(db, source.did, highestSeq); + state.cursor = highestSeq; + } + log.log( + `[labels] ${source.did} flushed ${kept}/${batch.length} labels, cursor=${highestSeq}`, + ); + } catch (err) { + log.error(`[labels] ${source.did} flush failed: ${err}`); + } finally { + flushing = false; + } + }; + + const batchSize = pumpOpts.batchSize ?? DEFAULT_BATCH_SIZE; + const flushInterval = pumpOpts.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS; + const flushTimer = setInterval(() => { + flush().catch(() => {}); + }, flushInterval); + + const cleanup = () => { + clearInterval(flushTimer); + try { + ws.close(); + } catch { + /* already closed */ + } + }; + + // External abort (persistent mode) — close socket gracefully. + const abortHandler = () => { + cleanup(); + flush().finally(() => resolveDone()); + }; + pumpOpts.signal?.addEventListener("abort", abortHandler, { once: true }); + + // Cron-mode time budget — close socket gracefully when reached. + let deadlineTimer: ReturnType | undefined; + if (Number.isFinite(timeoutMs)) { + deadlineTimer = setTimeout(() => { + log.log(`[labels] ${source.did} cycle deadline reached, closing`); + cleanup(); + flush().finally(() => resolveDone()); + }, timeoutMs); + } + + ws.addEventListener("error", (ev) => { + cleanup(); + if (deadlineTimer) clearTimeout(deadlineTimer); + pumpOpts.signal?.removeEventListener("abort", abortHandler); + rejectDone(new Error(`WebSocket error: ${(ev as ErrorEvent)?.message ?? "unknown"}`)); + }); + + ws.addEventListener("close", () => { + if (deadlineTimer) clearTimeout(deadlineTimer); + pumpOpts.signal?.removeEventListener("abort", abortHandler); + flush().finally(() => { + clearInterval(flushTimer); + resolveDone(); + }); + }); + + ws.addEventListener("message", async (ev) => { + let bytes: Uint8Array; + if (ev.data instanceof ArrayBuffer) { + bytes = new Uint8Array(ev.data); + } else if (ev.data instanceof Uint8Array) { + bytes = ev.data; + } else { + // Binary-only protocol — text frames shouldn't arrive. + return; + } + const frame = decodeFrame(bytes); + if (!frame) return; + + if (frame.t === "#labels") { + const seq = Number(frame.payload?.seq ?? 0); + const rawLabels = Array.isArray(frame.payload?.labels) ? frame.payload.labels : []; + for (const raw of rawLabels) { + const lab = normalizeLabel(raw, source.did); + if (lab) buffer.push(lab); + } + if (Number.isFinite(seq) && seq > highestSeq) highestSeq = seq; + if (buffer.length >= batchSize) { + flush().catch(() => {}); + } + } else if (frame.t === "#info") { + const name = String(frame.payload?.name ?? ""); + log.log(`[labels] ${source.did} info: ${name}`); + if (name === "OutdatedCursor") { + // Labeler rewound its log — discard our cursor and let the next + // run start from the beginning. We don't reconnect here; the + // caller (or the persistent loop) will pick up the reset on retry. + await resetLabelerCursor(db, source.did); + cleanup(); + } + } else if (frame.op === -1) { + log.warn(`[labels] ${source.did} error frame: ${JSON.stringify(frame.payload)}`); + cleanup(); + } + }); + + // Workers WebSocket doesn't always emit `open`; just await `done` directly. + await done; +} + +function buildWsUrl(httpEndpoint: string, cursor: number | null): string { + const u = new URL("/xrpc/com.atproto.label.subscribeLabels", httpEndpoint); + // wss:// for HTTPS endpoints — the protocol on the labeler service is + // expected to be HTTPS already (validated at resolution time). + u.protocol = u.protocol === "https:" ? "wss:" : "ws:"; + if (cursor !== null) u.searchParams.set("cursor", String(cursor)); + return u.toString(); +} + +interface DecodedFrame { + op: number; + t: string | undefined; + payload: Record; +} + +/** Decode an atproto subscription frame: two consecutive CBOR objects. + * Header `{ op, t? }`, payload — shape depends on `t`. Returns null on + * decode failure or non-object frames. */ +function decodeFrame(bytes: Uint8Array): DecodedFrame | null { + try { + const [header, rest] = decodeFirst(bytes); + if (!header || typeof header !== "object") return null; + const op = typeof (header as { op?: number }).op === "number" ? (header as { op: number }).op : 1; + const t = typeof (header as { t?: string }).t === "string" ? (header as { t: string }).t : undefined; + const [payload] = decodeFirst(rest); + if (!payload || typeof payload !== "object") return null; + return { op, t, payload: payload as Record }; + } catch { + return null; + } +} + +/** Coerce a wire `Label` object into our `IncomingLabel` shape. Returns + * null when required fields are missing — we'd rather skip a row than + * insert one with placeholder values. */ +function normalizeLabel(raw: unknown, expectedSrc: string): IncomingLabel | null { + if (!raw || typeof raw !== "object") return null; + const r = raw as Record; + const src = typeof r.src === "string" ? r.src : null; + const uri = typeof r.uri === "string" ? r.uri : null; + const val = typeof r.val === "string" ? r.val : null; + const cts = typeof r.cts === "string" ? r.cts : null; + if (!src || !uri || !val || !cts) return null; + // A labeler shouldn't emit labels under a different `src` than its own + // DID — drop them rather than poison our table with cross-issuer rows. + if (src !== expectedSrc) return null; + return { + src, + uri, + val, + cts, + cid: typeof r.cid === "string" ? r.cid : undefined, + neg: r.neg === true, + exp: typeof r.exp === "string" ? r.exp : undefined, + sig: r.sig instanceof Uint8Array ? r.sig : undefined, + }; +} diff --git a/packages/contrail/src/core/labels/types.ts b/packages/contrail/src/core/labels/types.ts new file mode 100644 index 0000000..c5833cf --- /dev/null +++ b/packages/contrail/src/core/labels/types.ts @@ -0,0 +1,63 @@ +import type { Database } from "../types"; + +/** A labeler the operator wants contrail to track. */ +export interface LabelerSource { + /** Labeler DID — `did:plc:...` or `did:web:...`. */ + did: string; + /** Override the service endpoint resolution. Otherwise resolved from the + * DID doc's `service[id="#atproto_labeler"].serviceEndpoint`. */ + endpoint?: string; + /** Backfill from `cursor=0` on first sight. Defaults to true. Set false + * for "start from now" — useful for very chatty labelers. */ + backfill?: boolean; +} + +export interface LabelsConfig { + /** Labelers to subscribe to and index. */ + sources: LabelerSource[]; + /** DIDs honored when the caller sends no `atproto-accept-labelers` / + * `?labelers=`. Defaults to every entry in `sources`. Set `[]` for + * opt-in-only — clients see no labels unless they ask. */ + defaults?: string[]; + /** Honor caller-supplied DIDs that aren't in `sources`. Default: false. + * When true, unknown DIDs in the request are accepted (and the labeler + * may be lazily registered for ingest in a later release). */ + allowUserSupplied?: boolean; + /** Per-request cap. Default: 20 (matches Bluesky). */ + maxPerRequest?: number; +} + +export const DEFAULT_LABELS_MAX_PER_REQUEST = 20; + +/** A single label as stored. Matches `com.atproto.label.defs#label`. */ +export interface LabelRow { + /** Issuing labeler DID. */ + src: string; + /** Subject — at-URI for record labels, plain DID for account labels. */ + uri: string; + /** Label value — kebab-case, ≤128 bytes per spec. */ + val: string; + /** Optional CID pin to a specific record version. */ + cid: string | null; + /** When true, retracts a previously-emitted label for the same (src, uri, val). */ + neg: boolean; + /** Expiry, unix seconds. Past this, hydration drops the row. */ + exp: number | null; + /** Creation timestamp, unix seconds — what we collapse on. */ + cts: number; + /** Raw signature bytes. Stored when present so we can re-emit later; + * not verified in v1. */ + sig: Uint8Array | null; +} + +/** Per-labeler state row — endpoint cache and last-seen seq cursor. */ +export interface LabelerCursorRow { + did: string; + cursor: number; + endpoint: string | null; + resolved_at: number | null; +} + +export interface AdapterContext { + db: Database; +} diff --git a/packages/contrail/src/core/router/collection.ts b/packages/contrail/src/core/router/collection.ts index 2e07719..e440522 100644 --- a/packages/contrail/src/core/router/collection.ts +++ b/packages/contrail/src/core/router/collection.ts @@ -16,6 +16,8 @@ import { resolveProfiles, collectDids } from "./profiles"; import { resolveActor } from "../identity"; import type { FormattedRecord } from "./helpers"; import { formatRecord, parseIntParam, fieldToParam } from "./helpers"; +import { selectAcceptedLabelers } from "../labels/select"; +import { hydrateLabels } from "../labels/hydrate"; import { verifyServiceAuthRequest, extractInviteToken, checkInviteReadGrant } from "../spaces/auth"; import { checkAccess } from "../spaces/acl"; import { hashInviteToken } from "../invite/token"; @@ -278,8 +280,12 @@ export async function runPipeline( collection: string, params: URLSearchParams, source?: RecordSource, - spaceUris?: string[] -): Promise<{ records: FormattedRecord[]; cursor?: string; profiles?: any[] }> { + spaceUris?: string[], + /** Optional headers from the originating request — used for label + * hydration (`atproto-accept-labelers`). Other entry points pass nothing + * and labels are gated by `?labelers=` / config defaults. */ + headers?: Headers +): Promise<{ records: FormattedRecord[]; cursor?: string; profiles?: any[]; labelersApplied?: string[] }> { const colConfig = config.collections[collection]; if (!colConfig) throw new Error(`Unknown collection: ${collection}`); @@ -433,13 +439,57 @@ export async function runPipeline( ? await resolveProfiles(db, config, allDids) : undefined; + let labelersApplied: string[] | undefined; + if (config.labels) { + const sel = selectAcceptedLabelers( + headers?.get("atproto-accept-labelers") ?? null, + params.get("labelers"), + config.labels, + ); + if (sel.accepted.length > 0) { + const subjects: string[] = [ + ...formattedRecords.map((r) => r.uri), + ...allDids, + ]; + const cidByUri = new Map(); + for (const r of formattedRecords) cidByUri.set(r.uri, r.cid); + const labelsByUri = await hydrateLabels(db, subjects, sel.accepted, cidByUri); + for (const fr of formattedRecords) { + const ls = labelsByUri[fr.uri]; + if (ls && ls.length > 0) fr.labels = ls; + } + if (profileMap) { + for (const entries of Object.values(profileMap)) { + for (const entry of entries) { + const ls = labelsByUri[entry.did]; + if (ls && ls.length > 0) entry.labels = ls; + } + } + } + labelersApplied = sel.accepted; + } + } + return { records: formattedRecords, cursor: result.cursor, ...(profileMap ? { profiles: Object.values(profileMap).flat() } : {}), + ...(labelersApplied ? { labelersApplied } : {}), }; } +/** Serialize a runPipeline result as JSON, echoing + * `atproto-content-labelers` when labels were applied. The result's + * `labelersApplied` field never appears in the response body — it's a + * side channel for the route to read and turn into a header. */ +function jsonWithLabelers(c: Context, result: { labelersApplied?: string[] } & Record) { + const { labelersApplied, ...body } = result; + if (labelersApplied && labelersApplied.length > 0) { + c.header("atproto-content-labelers", labelersApplied.join(",")); + } + return c.json(body); +} + export function registerCollectionRoutes( app: Hono, db: Database, @@ -537,8 +587,8 @@ export function registerCollectionRoutes( // full filter / sort / hydrate / reference surface works on per-space // queries too, not just on the cross-space union path. try { - const result = await runPipeline(db, config, collection, params, undefined, [spaceUri]); - return c.json(result); + const result = await runPipeline(db, config, collection, params, undefined, [spaceUri], c.req.raw.headers); + return jsonWithLabelers(c, result); } catch (e: any) { if (e.message === "Could not resolve actor") { return c.json({ error: e.message }, 400); @@ -571,8 +621,8 @@ export function registerCollectionRoutes( } try { - const result = await runPipeline(db, config, collection, params, undefined, spaceUris); - return c.json(result); + const result = await runPipeline(db, config, collection, params, undefined, spaceUris, c.req.raw.headers); + return jsonWithLabelers(c, result); } catch (e: any) { if (e.message === "Could not resolve actor") { return c.json({ error: e.message }, 400); @@ -759,7 +809,8 @@ export function registerCollectionRoutes( collection, params, undefined, - snapshotSpaces + snapshotSpaces, + c.req.raw.headers ); let ticket: string | undefined; if (ticketSigner && callerDid) { @@ -1040,6 +1091,34 @@ export function registerCollectionRoutes( ? await resolveProfiles(db, config, allDids) : undefined; + let labelersApplied: string[] | undefined; + if (config.labels) { + const sel = selectAcceptedLabelers( + c.req.raw.headers.get("atproto-accept-labelers"), + params.get("labelers"), + config.labels, + ); + if (sel.accepted.length > 0) { + const subjects: string[] = [row.uri, ...allDids]; + const cidByUri = new Map([[row.uri, row.cid]]); + const labelsByUri = await hydrateLabels(db, subjects, sel.accepted, cidByUri); + const ls = labelsByUri[row.uri]; + if (ls && ls.length > 0) (formatted as Record).labels = ls; + if (profileMap) { + for (const entries of Object.values(profileMap)) { + for (const entry of entries) { + const els = labelsByUri[entry.did]; + if (els && els.length > 0) entry.labels = els; + } + } + } + labelersApplied = sel.accepted; + } + } + if (labelersApplied) { + c.header("atproto-content-labelers", labelersApplied.join(",")); + } + return c.json({ ...formatted, ...(profileMap ? { profiles: Object.values(profileMap).flat() } : {}), @@ -1062,8 +1141,8 @@ export function registerCollectionRoutes( const params = new URL(c.req.url).searchParams; try { const source = await handler(db, params, config); - const result = await runPipeline(db, config, collection, params, source); - return c.json(result); + const result = await runPipeline(db, config, collection, params, source, undefined, c.req.raw.headers); + return jsonWithLabelers(c, result); } catch (e: any) { if (e.message === "Could not resolve actor") { return c.json({ error: e.message }, 400); diff --git a/packages/contrail/src/core/router/index.ts b/packages/contrail/src/core/router/index.ts index 413161c..f5e6808 100644 --- a/packages/contrail/src/core/router/index.ts +++ b/packages/contrail/src/core/router/index.ts @@ -24,6 +24,8 @@ import type { PubSub } from "../realtime/types"; import { resolveActor } from "../identity"; import { resolveProfiles } from "./profiles"; import { backfillUser } from "../backfill"; +import { selectAcceptedLabelers } from "../labels/select"; +import { hydrateLabels } from "../labels/hydrate"; export interface SpacesContext { adapter: StorageAdapter; @@ -89,6 +91,25 @@ export function createApp( const profiles = profileMap[did]; if (!profiles || profiles.length === 0) return c.json({ error: "Profile not found" }, 404); + if (config.labels) { + const params = new URL(c.req.url).searchParams; + const sel = selectAcceptedLabelers( + c.req.raw.headers.get("atproto-accept-labelers"), + params.get("labelers"), + config.labels, + ); + if (sel.accepted.length > 0) { + const labelsByUri = await hydrateLabels(db, [did], sel.accepted); + const ls = labelsByUri[did]; + if (ls && ls.length > 0) { + for (const entry of profiles) { + entry.labels = ls; + } + } + c.header("atproto-content-labelers", sel.accepted.join(",")); + } + } + return c.json({ profiles }); }); diff --git a/packages/contrail/src/core/router/profiles.ts b/packages/contrail/src/core/router/profiles.ts index 3e9ec8d..5552357 100644 --- a/packages/contrail/src/core/router/profiles.ts +++ b/packages/contrail/src/core/router/profiles.ts @@ -13,6 +13,9 @@ export interface ProfileEntry { value?: unknown; collection?: string; rkey?: string; + /** Hydrated by the labels module when the caller has accepted-labelers + * active and there are matching labels on this DID. */ + labels?: unknown; } export function collectDids( diff --git a/packages/contrail/src/core/types.ts b/packages/contrail/src/core/types.ts index f9b9279..953d604 100644 --- a/packages/contrail/src/core/types.ts +++ b/packages/contrail/src/core/types.ts @@ -161,6 +161,11 @@ export interface ContrailConfig { /** Realtime module configuration. When set, the service exposes ticket + SSE/WS * subscribe XRPCs, and wraps the spaces adapter to publish events after writes. */ realtime?: import("./realtime/types").RealtimeConfig; + /** Labels module configuration. When set, contrail subscribes to the + * configured labelers, indexes their labels into a single `labels` table, + * and hydrates `record.labels` onto `listRecords` / `getRecord` / profile + * responses gated by the caller's `atproto-accept-labelers` header. */ + labels?: import("./labels/types").LabelsConfig; /** Customize the auto-generated `.permissionSet` lexicon. */ permissionSet?: PermissionSetConfig; } diff --git a/packages/contrail/src/index.ts b/packages/contrail/src/index.ts index 6665862..1e0c0cf 100644 --- a/packages/contrail/src/index.ts +++ b/packages/contrail/src/index.ts @@ -94,6 +94,25 @@ export type { DurableObjectState, } from "./core/realtime/durable-object"; +// Labels +export type { + LabelsConfig, + LabelerSource, + LabelRow, + LabelerCursorRow, +} from "./core/labels/types"; +export type { HydratedLabel } from "./core/labels/hydrate"; +export { hydrateLabels } from "./core/labels/hydrate"; +export { selectAcceptedLabelers } from "./core/labels/select"; +export { applyLabels } from "./core/labels/apply"; +export type { IncomingLabel } from "./core/labels/apply"; +export { + runLabelIngestCycle, + runPersistentLabels, +} from "./core/labels/subscribe"; +export type { PersistentLabelsOptions } from "./core/labels/subscribe"; +export { resolveLabelerEndpoint } from "./core/labels/resolve"; + // Community export { CommunityAdapter, diff --git a/packages/contrail/src/worker/index.ts b/packages/contrail/src/worker/index.ts index b0d2ed1..c9213d7 100644 --- a/packages/contrail/src/worker/index.ts +++ b/packages/contrail/src/worker/index.ts @@ -64,6 +64,12 @@ export function createWorker( const db = env[binding] as Database; await ensureReady(env, db); ctx.waitUntil(contrail.ingest({}, db)); + // Run label ingest alongside the jetstream catch-up so a single cron + // tick keeps both data streams fresh. Only scheduled when configured — + // skipping the no-op promise keeps the worker task list tight. + if (config.labels) { + ctx.waitUntil(contrail.ingestLabels({}, db)); + } }, }; } diff --git a/packages/contrail/tests/labels-router.test.ts b/packages/contrail/tests/labels-router.test.ts new file mode 100644 index 0000000..ff1cea6 --- /dev/null +++ b/packages/contrail/tests/labels-router.test.ts @@ -0,0 +1,141 @@ +/** End-to-end router checks for label hydration: + * - `listRecords` attaches `record.labels` and echoes `atproto-content-labelers` + * - `getRecord` does the same for a single record + * - caller-provided `?labelers=` overrides config defaults + * - unaccepted labelers are dropped */ +import { describe, it, expect } from "vitest"; +import { Contrail } from "../src/contrail"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; +import { applyEvents } from "../src/core/db/records"; +import { applyLabels } from "../src/core/labels/apply"; +import type { IngestEvent } from "../src/core/types"; + +const COLL = "com.example.event"; +const SRC_A = "did:plc:labelerA"; +const SRC_B = "did:plc:labelerB"; +const AUTHOR = "did:plc:author"; +const RKEY = "abc"; +const URI = `at://${AUTHOR}/${COLL}/${RKEY}`; + +function ev(): IngestEvent { + const now = Date.now() * 1000; + return { + uri: URI, + did: AUTHOR, + collection: COLL, + rkey: RKEY, + operation: "create", + cid: "bafy-1", + record: JSON.stringify({ name: "test" }), + time_us: now, + indexed_at: now, + }; +} + +async function setup() { + const db = createSqliteDatabase(":memory:"); + const contrail = new Contrail({ + namespace: "ex", + collections: { event: { collection: COLL } }, + labels: { + sources: [{ did: SRC_A }, { did: SRC_B }], + defaults: [SRC_A], // SRC_B is opt-in via caller + }, + db, + }); + await contrail.init(); + await applyEvents(db, [ev()], contrail.config); + await applyLabels(db, [ + { src: SRC_A, uri: URI, val: "spam", cts: new Date().toISOString() }, + { src: SRC_B, uri: URI, val: "porn", cts: new Date().toISOString() }, + ]); + return { db, contrail }; +} + +describe("labels router integration", () => { + it("listRecords attaches labels and echoes atproto-content-labelers", async () => { + const { contrail } = await setup(); + const app = contrail.app(); + + const res = await app.fetch(new Request(`http://localhost/xrpc/ex.event.listRecords`)); + expect(res.status).toBe(200); + + expect(res.headers.get("atproto-content-labelers")).toBe(SRC_A); + + const body = (await res.json()) as { records: Array<{ uri: string; labels?: any[] }> }; + expect(body.records).toHaveLength(1); + expect(body.records[0]!.labels).toBeDefined(); + expect(body.records[0]!.labels!.map((l: any) => l.val)).toEqual(["spam"]); + }); + + it("getRecord attaches labels for a single record", async () => { + const { contrail } = await setup(); + const app = contrail.app(); + + const res = await app.fetch( + new Request(`http://localhost/xrpc/ex.event.getRecord?uri=${encodeURIComponent(URI)}`), + ); + expect(res.status).toBe(200); + expect(res.headers.get("atproto-content-labelers")).toBe(SRC_A); + + const body = (await res.json()) as { uri: string; labels?: any[] }; + expect(body.labels?.map((l: any) => l.val)).toEqual(["spam"]); + }); + + it("?labelers= override pulls from a non-default source", async () => { + const { contrail } = await setup(); + const app = contrail.app(); + + const res = await app.fetch( + new Request(`http://localhost/xrpc/ex.event.listRecords?labelers=${SRC_B}`), + ); + expect(res.headers.get("atproto-content-labelers")).toBe(SRC_B); + const body = (await res.json()) as { records: Array<{ labels?: any[] }> }; + expect(body.records[0]!.labels?.map((l: any) => l.val)).toEqual(["porn"]); + }); + + it("atproto-accept-labelers header takes precedence over query param", async () => { + const { contrail } = await setup(); + const app = contrail.app(); + + const res = await app.fetch( + new Request(`http://localhost/xrpc/ex.event.listRecords?labelers=${SRC_B}`, { + headers: { "atproto-accept-labelers": SRC_A }, + }), + ); + expect(res.headers.get("atproto-content-labelers")).toBe(SRC_A); + const body = (await res.json()) as { records: Array<{ labels?: any[] }> }; + expect(body.records[0]!.labels?.map((l: any) => l.val)).toEqual(["spam"]); + }); + + it("when no labels are configured, no header and no field", async () => { + const db = createSqliteDatabase(":memory:"); + const contrail = new Contrail({ + namespace: "ex", + collections: { event: { collection: COLL } }, + // no labels: {} block + db, + }); + await contrail.init(); + await applyEvents(db, [ev()], contrail.config); + const app = contrail.app(); + + const res = await app.fetch(new Request(`http://localhost/xrpc/ex.event.listRecords`)); + expect(res.status).toBe(200); + expect(res.headers.get("atproto-content-labelers")).toBeNull(); + const body = (await res.json()) as { records: Array<{ labels?: any[] }> }; + expect(body.records[0]!.labels).toBeUndefined(); + }); + + it("unknown caller-supplied DIDs are dropped (allowUserSupplied off)", async () => { + const { contrail } = await setup(); + const app = contrail.app(); + + const res = await app.fetch( + new Request(`http://localhost/xrpc/ex.event.listRecords?labelers=did:plc:bogus`), + ); + expect(res.headers.get("atproto-content-labelers")).toBeNull(); + const body = (await res.json()) as { records: Array<{ labels?: any[] }> }; + expect(body.records[0]!.labels).toBeUndefined(); + }); +}); diff --git a/packages/contrail/tests/labels.test.ts b/packages/contrail/tests/labels.test.ts new file mode 100644 index 0000000..317064c --- /dev/null +++ b/packages/contrail/tests/labels.test.ts @@ -0,0 +1,188 @@ +import { describe, expect, it } from "vitest"; +import { createSqliteDatabase } from "../src/adapters/sqlite"; +import { initSchema } from "../src/core/db/schema"; +import { resolveConfig, type ContrailConfig } from "../src/core/types"; +import { applyLabels } from "../src/core/labels/apply"; +import { hydrateLabels } from "../src/core/labels/hydrate"; +import { selectAcceptedLabelers } from "../src/core/labels/select"; +import type { LabelsConfig } from "../src/core/labels/types"; + +const SRC_A = "did:plc:labelerA"; +const SRC_B = "did:plc:labelerB"; +const URI_X = "at://did:plc:author/com.example.event/1"; +const URI_Y = "at://did:plc:author/com.example.event/2"; +const ACC_DID = "did:plc:author"; + +function ctsIso(deltaSec = 0): string { + return new Date(Date.now() + deltaSec * 1000).toISOString(); +} + +async function setup() { + const db = createSqliteDatabase(":memory:"); + const config: ContrailConfig = { + namespace: "com.example", + collections: { + event: { collection: "com.example.event" }, + }, + labels: { + sources: [{ did: SRC_A }, { did: SRC_B }], + }, + }; + const resolved = resolveConfig(config); + await initSchema(db, resolved); + return { db, config: resolved }; +} + +describe("labels: applyLabels + hydrate", () => { + it("upserts and hydrates a basic label", async () => { + const { db } = await setup(); + await applyLabels(db, [ + { src: SRC_A, uri: URI_X, val: "spam", cts: ctsIso() }, + ]); + const out = await hydrateLabels(db, [URI_X], [SRC_A]); + expect(out[URI_X]).toHaveLength(1); + expect(out[URI_X][0]!.val).toBe("spam"); + expect(out[URI_X][0]!.src).toBe(SRC_A); + }); + + it("filters by accepted labelers — unaccepted source drops out", async () => { + const { db } = await setup(); + await applyLabels(db, [ + { src: SRC_A, uri: URI_X, val: "a", cts: ctsIso() }, + { src: SRC_B, uri: URI_X, val: "b", cts: ctsIso() }, + ]); + const out = await hydrateLabels(db, [URI_X], [SRC_A]); + expect(out[URI_X]).toHaveLength(1); + expect(out[URI_X][0]!.val).toBe("a"); + }); + + it("collapses (src, uri, val) by latest cts and drops neg=true winners", async () => { + const { db } = await setup(); + await applyLabels(db, [ + { src: SRC_A, uri: URI_X, val: "spam", cts: ctsIso(-10) }, + // Negation arrives later — should retract. + { src: SRC_A, uri: URI_X, val: "spam", cts: ctsIso(0), neg: true }, + ]); + const out = await hydrateLabels(db, [URI_X], [SRC_A]); + expect(out[URI_X] ?? []).toHaveLength(0); + }); + + it("re-positive after a negation wins again", async () => { + const { db } = await setup(); + await applyLabels(db, [ + { src: SRC_A, uri: URI_X, val: "spam", cts: ctsIso(-20) }, + { src: SRC_A, uri: URI_X, val: "spam", cts: ctsIso(-10), neg: true }, + { src: SRC_A, uri: URI_X, val: "spam", cts: ctsIso(0) }, + ]); + const out = await hydrateLabels(db, [URI_X], [SRC_A]); + expect(out[URI_X]).toHaveLength(1); + }); + + it("expired labels are dropped", async () => { + const { db } = await setup(); + const yesterday = new Date(Date.now() - 86400_000).toISOString(); + await applyLabels(db, [ + { src: SRC_A, uri: URI_X, val: "x", cts: ctsIso(-3600), exp: yesterday }, + ]); + const out = await hydrateLabels(db, [URI_X], [SRC_A]); + expect(out[URI_X] ?? []).toHaveLength(0); + }); + + it("CID pin filters to matching record version", async () => { + const { db } = await setup(); + await applyLabels(db, [ + { src: SRC_A, uri: URI_X, val: "v1-only", cts: ctsIso(), cid: "bafy-old" }, + { src: SRC_A, uri: URI_X, val: "any-version", cts: ctsIso() }, + ]); + const cidByUri = new Map([[URI_X, "bafy-new"]]); + const out = await hydrateLabels(db, [URI_X], [SRC_A], cidByUri); + expect(out[URI_X]?.map((l) => l.val).sort()).toEqual(["any-version"]); + }); + + it("account-level label keyed by bare DID hydrates fine", async () => { + const { db } = await setup(); + await applyLabels(db, [ + { src: SRC_A, uri: ACC_DID, val: "!hide", cts: ctsIso() }, + ]); + const out = await hydrateLabels(db, [ACC_DID], [SRC_A]); + expect(out[ACC_DID]).toHaveLength(1); + expect(out[ACC_DID][0]!.val).toBe("!hide"); + }); + + it("cross-subject hydration: many uris in one query", async () => { + const { db } = await setup(); + await applyLabels(db, [ + { src: SRC_A, uri: URI_X, val: "a", cts: ctsIso() }, + { src: SRC_A, uri: URI_Y, val: "b", cts: ctsIso() }, + ]); + const out = await hydrateLabels(db, [URI_X, URI_Y], [SRC_A]); + expect(out[URI_X]).toHaveLength(1); + expect(out[URI_Y]).toHaveLength(1); + }); +}); + +describe("labels: selectAcceptedLabelers", () => { + const cfg: LabelsConfig = { + sources: [{ did: SRC_A }, { did: SRC_B }], + }; + + it("falls back to defaults (= sources) when caller sends nothing", () => { + const sel = selectAcceptedLabelers(null, null, cfg); + expect(sel.accepted).toEqual([SRC_A, SRC_B]); + expect(sel.lazyAdd).toEqual([]); + }); + + it("honors header before query param", () => { + const sel = selectAcceptedLabelers(SRC_A, SRC_B, cfg); + expect(sel.accepted).toEqual([SRC_A]); + }); + + it("falls through to query param when header is empty", () => { + const sel = selectAcceptedLabelers("", SRC_B, cfg); + expect(sel.accepted).toEqual([SRC_B]); + }); + + it("drops unknown DIDs by default", () => { + const sel = selectAcceptedLabelers("did:plc:strangerlabeler", null, cfg); + expect(sel.accepted).toEqual([]); + expect(sel.lazyAdd).toEqual([]); + }); + + it("collects unknowns as lazyAdd when allowUserSupplied is true", () => { + const sel = selectAcceptedLabelers( + `did:plc:strangerlabeler,${SRC_A}`, + null, + { ...cfg, allowUserSupplied: true }, + ); + expect(sel.accepted).toEqual([SRC_A]); + expect(sel.lazyAdd).toEqual(["did:plc:strangerlabeler"]); + }); + + it("strips ;param modifiers from header values", () => { + const sel = selectAcceptedLabelers( + `${SRC_A};redact, ${SRC_B} ; foo`, + null, + cfg, + ); + expect(sel.accepted).toEqual([SRC_A, SRC_B]); + }); + + it("caps at maxPerRequest", () => { + const sel = selectAcceptedLabelers( + `${SRC_A},${SRC_B}`, + null, + { ...cfg, maxPerRequest: 1 }, + ); + expect(sel.accepted).toEqual([SRC_A]); + }); + + it("empty defaults => no labelers when caller is silent (opt-in policy)", () => { + const sel = selectAcceptedLabelers(null, null, { ...cfg, defaults: [] }); + expect(sel.accepted).toEqual([]); + }); + + it("dedupes repeated DIDs in caller list", () => { + const sel = selectAcceptedLabelers(`${SRC_A},${SRC_A},${SRC_A}`, null, cfg); + expect(sel.accepted).toEqual([SRC_A]); + }); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e6c5090..5c07e90 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -349,6 +349,9 @@ importers: '@atcute/atproto': specifier: ^3.1.10 version: 3.1.11 + '@atcute/cbor': + specifier: ^2.3.2 + version: 2.3.2 '@atcute/cid': specifier: ^2.4.1 version: 2.4.1 -- 2.51.2