feat(labeler): aggregator service binding + read-only client (W8.4 slice-3 foundation)

apps/labeler/src/aggregator-client.ts

@@ -0,0 +1,139 @@
+/**
+ * Read-only client over the aggregator Worker's XRPC surface, reached via
+ * the `AGGREGATOR` service binding (fetch-over-binding; the aggregator is
+ * HTTP/XRPC-only with no RPC entrypoint).
+ *
+ * Each method performs exactly one `fetch` per call with no memoization,
+ * caching, or retry: internal consumers require a fresh read at call time
+ * (W10.4). Callers layer their own retry/backoff if they want it.
+ *
+ * Every request sends a BLANK `atproto-accept-labelers` header to obtain the
+ * genuinely unfiltered view. Counter-intuitively, OMITTING the header does the
+ * opposite: the aggregator resolves an absent header to its default policy
+ * (every trusted labeler, `redact: true`), so a subject redacted by any trusted
+ * labeler — plausibly including this labeler itself — presents as `NotFound`
+ * and is indistinguishable from an unindexed subject. A blank value parses to
+ * an empty accepted-labelers set, which the aggregator enforces as a no-op.
+ * The labeler is the moderation authority performing analysis; it must not be
+ * blinded to redacted subjects when resolving a takedown contact, reading
+ * history context, or re-assessing a flagged subject.
+ *
+ * Boundary: this unfiltered view is for INTERNAL ANALYSIS ONLY. Its results
+ * must never be surfaced to a public serving path without re-applying takedown
+ * enforcement at that layer (see the ratified sync.getRecord / artifact-mirror
+ * serving decisions) — an unfiltered read that reached a public surface would
+ * re-expose taken-down content.
+ *
+ * Views are returned as their `@emdash-cms/registry-lexicons` types without
+ * runtime validation: the aggregator is a first-party in-process binding, and
+ * the acquire stage independently re-verifies artifact checksums downstream.
+ *
+ * Error contract: a missing subject (XRPC `NotFound`, HTTP 404) resolves to
+ * `null`. Any other non-2xx status, or a transport failure, throws — callers
+ * may treat a throw as transient and retry the whole read.
+ */
+
+import type { AggregatorDefs, AggregatorListReleases } from "@emdash-cms/registry-lexicons";
+
+/**
+ * XRPC endpoint host. Irrelevant to routing — a service binding dispatches by
+ * binding, not DNS — but `fetch` needs a syntactically valid absolute URL.
+ */
+const XRPC_BASE = "https://aggregator/xrpc";
+
+/**
+ * ATProto label-negotiation header. Sent with a blank value on every read to
+ * opt out of the aggregator's default trusted-labeler redaction (see the
+ * module doc). Mirrors the aggregator's own local constant; this is a stable
+ * protocol string, not currently exported from a shared package.
+ */
+const ACCEPT_LABELERS_HEADER = "atproto-accept-labelers";
+
+/** NSIDs are fixed per method and never derived from caller data. Only query
+ * param *values* carry caller data, and every value is `encodeURIComponent`-
+ * encoded in {@link buildUrl}; the param *keys* are hard-coded literals. There
+ * is therefore no path for caller data to alter the target host or path. */
+const NSID = {
+	getPackage: "com.emdashcms.experimental.aggregator.getPackage",
+	getLatestRelease: "com.emdashcms.experimental.aggregator.getLatestRelease",
+	listReleases: "com.emdashcms.experimental.aggregator.listReleases",
+} as const;
+
+/** Build an XRPC GET URL from a constant NSID and caller-supplied param
+ * values. Keys are the caller's fixed literals; each value is URL-encoded. */
+function buildUrl(nsid: string, params: Record<string, string>): string {
+	const query = Object.entries(params)
+		.map(([key, value]) => `${key}=${encodeURIComponent(value)}`)
+		.join("&");
+	return `${XRPC_BASE}/${nsid}?${query}`;
+}
+
+/** The `error` field of an XRPC error envelope, or undefined if the body
+ * isn't a recognisable `{ error }` object. */
+function parseXrpcErrorName(body: string): string | undefined {
+	try {
+		const parsed: unknown = JSON.parse(body);
+		if (parsed && typeof parsed === "object" && "error" in parsed) {
+			const { error } = parsed;
+			if (typeof error === "string") return error;
+		}
+	} catch {
+		// Non-JSON error body (e.g. an infra-level 502); fall through to throw.
+	}
+	return undefined;
+}
+
+export class AggregatorClient {
+	readonly #fetcher: Fetcher;
+
+	constructor(fetcher: Fetcher) {
+		this.#fetcher = fetcher;
+	}
+
+	/** Fetch the package view for `(did, slug)`, or `null` if not indexed. */
+	async getPackage(did: string, slug: string): Promise<AggregatorDefs.PackageView | null> {
+		const url = buildUrl(NSID.getPackage, { did, slug });
+		return this.#query<AggregatorDefs.PackageView>(NSID.getPackage, url);
+	}
+
+	/** Fetch the highest-precedence live release for `(did, pkg)`, or `null`
+	 * if the package has no eligible release. */
+	async getLatestRelease(did: string, pkg: string): Promise<AggregatorDefs.ReleaseView | null> {
+		const url = buildUrl(NSID.getLatestRelease, { did, package: pkg });
+		return this.#query<AggregatorDefs.ReleaseView>(NSID.getLatestRelease, url);
+	}
+
+	/** List releases for `(did, pkg)`, newest first, one page per call.
+	 * Returns `null` if the parent package isn't indexed. */
+	async listReleases(
+		did: string,
+		pkg: string,
+		cursor?: string,
+	): Promise<AggregatorListReleases.$output | null> {
+		const params: Record<string, string> = { did, package: pkg };
+		// Only a non-empty cursor is a page token; an empty string would decode as
+		// a malformed cursor (the aggregator 400s), so treat it as "first page".
+		if (cursor) params["cursor"] = cursor;
+		const url = buildUrl(NSID.listReleases, params);
+		return this.#query<AggregatorListReleases.$output>(NSID.listReleases, url);
+	}
+
+	/** One fetch, no retry. `NotFound` → `null`; any other non-2xx or a
+	 * transport failure throws. */
+	async #query<T>(nsid: string, url: string): Promise<T | null> {
+		const response = await this.#fetcher.fetch(url, {
+			headers: { [ACCEPT_LABELERS_HEADER]: "" },
+		});
+		if (response.ok) {
+			return response.json<T>();
+		}
+		const body = await response.text();
+		const errorName = parseXrpcErrorName(body);
+		if (response.status === 404 && errorName === "NotFound") {
+			return null;
+		}
+		throw new Error(
+			`Aggregator ${nsid} failed: ${response.status}${errorName ? ` ${errorName}` : ""}`,
+		);
+	}
+}

apps/labeler/test/aggregator-client.test.ts

@@ -0,0 +1,198 @@
+import { env } from "cloudflare:test";
+import { describe, expect, it } from "vitest";
+
+import { AggregatorClient } from "../src/aggregator-client.js";
+
+/** Records every URL and request init the client fetches and returns a
+ * caller-supplied Response (or throws, to simulate a transport failure). */
+function mockFetcher(handler: (url: string) => Response) {
+	const urls: string[] = [];
+	const inits: (RequestInit | undefined)[] = [];
+	const fetcher = {
+		fetch: (input: string, init?: RequestInit) => {
+			urls.push(input);
+			inits.push(init);
+			return Promise.resolve(handler(input));
+		},
+	} as unknown as Fetcher;
+	return { fetcher, urls, inits };
+}
+
+function jsonResponse(body: unknown, status = 200): Response {
+	return new Response(JSON.stringify(body), {
+		status,
+		headers: { "content-type": "application/json" },
+	});
+}
+
+function notFoundResponse(): Response {
+	return jsonResponse({ error: "NotFound", message: "No package indexed." }, 404);
+}
+
+const BASE = "https://aggregator/xrpc";
+
+const PACKAGE_VIEW = {
+	uri: "at://did:plc:abc123/com.emdashcms.experimental.package.profile/my-plugin",
+	cid: "bafyreib2rxk3rybk3examplecidvalue",
+	did: "did:plc:abc123",
+	slug: "my-plugin",
+	profile: {
+		$type: "com.emdashcms.experimental.package.profile",
+		id: "at://did:plc:abc123/com.emdashcms.experimental.package.profile/my-plugin",
+		type: "plugin",
+		license: "MIT",
+		authors: [{ name: "Ada", handle: "ada.example" }],
+		security: [{ contact: "security@example.test" }],
+		slug: "my-plugin",
+	},
+	indexedAt: "2026-07-01T00:00:00.000Z",
+	labels: [],
+	latestVersion: "1.2.0",
+};
+
+const RELEASE_VIEW = {
+	uri: "at://did:plc:abc123/com.emdashcms.experimental.package.release/1.2.0",
+	cid: "bafyreirelease123examplecidvalue",
+	did: "did:plc:abc123",
+	package: "my-plugin",
+	version: "1.2.0",
+	release: {
+		$type: "com.emdashcms.experimental.package.release",
+		package: "my-plugin",
+		version: "1.2.0",
+		artifacts: {
+			tarball: {
+				url: "https://cdn.example/my-plugin-1.2.0.tgz",
+				checksum: "sha256-0123456789abcdef",
+			},
+		},
+	},
+	mirrors: [],
+	indexedAt: "2026-07-01T00:00:00.000Z",
+	labels: [],
+};
+
+describe("AggregatorClient.getPackage", () => {
+	it("builds the getPackage URL with URL-encoded params and parses the view", async () => {
+		const { fetcher, urls } = mockFetcher(() => jsonResponse(PACKAGE_VIEW));
+		const view = await new AggregatorClient(fetcher).getPackage("did:plc:abc123", "my-plugin");
+
+		expect(urls).toEqual([
+			`${BASE}/com.emdashcms.experimental.aggregator.getPackage?did=did%3Aplc%3Aabc123&slug=my-plugin`,
+		]);
+		expect(view).toEqual(PACKAGE_VIEW);
+	});
+
+	it("returns null on a NotFound (404) error", async () => {
+		const { fetcher } = mockFetcher(() => notFoundResponse());
+		const view = await new AggregatorClient(fetcher).getPackage("did:plc:abc123", "missing");
+		expect(view).toBeNull();
+	});
+
+	it("sends a blank atproto-accept-labelers header to opt out of default redaction", async () => {
+		const { fetcher, inits } = mockFetcher(() => jsonResponse(PACKAGE_VIEW));
+		await new AggregatorClient(fetcher).getPackage("did:plc:abc123", "my-plugin");
+
+		const headers = new Headers(inits[0]?.headers);
+		expect(headers.get("atproto-accept-labelers")).toBe("");
+	});
+
+	it("throws on a 5xx response", async () => {
+		const { fetcher } = mockFetcher(() => new Response("upstream boom", { status: 500 }));
+		await expect(
+			new AggregatorClient(fetcher).getPackage("did:plc:abc123", "my-plugin"),
+		).rejects.toThrow(/getPackage failed: 500/);
+	});
+
+	it("throws when the binding fetch rejects (transport failure)", async () => {
+		const fetcher = {
+			fetch: () => Promise.reject(new Error("binding unreachable")),
+		} as unknown as Fetcher;
+		await expect(
+			new AggregatorClient(fetcher).getPackage("did:plc:abc123", "my-plugin"),
+		).rejects.toThrow("binding unreachable");
+	});
+});
+
+describe("AggregatorClient.getLatestRelease", () => {
+	it("builds the getLatestRelease URL with a `package` param and parses the view", async () => {
+		const { fetcher, urls } = mockFetcher(() => jsonResponse(RELEASE_VIEW));
+		const view = await new AggregatorClient(fetcher).getLatestRelease(
+			"did:plc:abc123",
+			"my-plugin",
+		);
+
+		expect(urls).toEqual([
+			`${BASE}/com.emdashcms.experimental.aggregator.getLatestRelease?did=did%3Aplc%3Aabc123&package=my-plugin`,
+		]);
+		expect(view).toEqual(RELEASE_VIEW);
+	});
+
+	it("returns null on a NotFound (404) error", async () => {
+		const { fetcher } = mockFetcher(() => notFoundResponse());
+		const view = await new AggregatorClient(fetcher).getLatestRelease("did:plc:abc123", "missing");
+		expect(view).toBeNull();
+	});
+});
+
+describe("AggregatorClient.listReleases", () => {
+	it("omits the cursor param when none is given and parses the page", async () => {
+		const page = { releases: [RELEASE_VIEW], cursor: "next-page-cursor" };
+		const { fetcher, urls } = mockFetcher(() => jsonResponse(page));
+		const result = await new AggregatorClient(fetcher).listReleases("did:plc:abc123", "my-plugin");
+
+		expect(urls).toEqual([
+			`${BASE}/com.emdashcms.experimental.aggregator.listReleases?did=did%3Aplc%3Aabc123&package=my-plugin`,
+		]);
+		expect(result).toEqual(page);
+	});
+
+	it("appends a URL-encoded cursor param when given", async () => {
+		const { fetcher, urls } = mockFetcher(() => jsonResponse({ releases: [] }));
+		await new AggregatorClient(fetcher).listReleases("did:plc:abc123", "my-plugin", "abc==");
+
+		expect(urls).toEqual([
+			`${BASE}/com.emdashcms.experimental.aggregator.listReleases?did=did%3Aplc%3Aabc123&package=my-plugin&cursor=abc%3D%3D`,
+		]);
+	});
+
+	it("omits the cursor param for an empty-string cursor (first page, not a 400)", async () => {
+		const { fetcher, urls } = mockFetcher(() => jsonResponse({ releases: [] }));
+		await new AggregatorClient(fetcher).listReleases("did:plc:abc123", "my-plugin", "");
+
+		expect(urls).toEqual([
+			`${BASE}/com.emdashcms.experimental.aggregator.listReleases?did=did%3Aplc%3Aabc123&package=my-plugin`,
+		]);
+	});
+
+	it("returns null when the parent package is NotFound", async () => {
+		const { fetcher } = mockFetcher(() => notFoundResponse());
+		const result = await new AggregatorClient(fetcher).listReleases("did:plc:abc123", "missing");
+		expect(result).toBeNull();
+	});
+});
+
+describe("AggregatorClient param encoding", () => {
+	it("percent-encodes reserved characters so they cannot alter the query", async () => {
+		const { fetcher, urls } = mockFetcher(() => jsonResponse(PACKAGE_VIEW));
+		await new AggregatorClient(fetcher).getPackage("did:plc:a&b c", "s/l&ug");
+
+		expect(urls).toEqual([
+			`${BASE}/com.emdashcms.experimental.aggregator.getPackage?did=did%3Aplc%3Aa%26b%20c&slug=s%2Fl%26ug`,
+		]);
+	});
+});
+
+describe("AGGREGATOR binding transport", () => {
+	// The unfiltered-read contract depends on the blank atproto-accept-labelers
+	// header surviving the service-binding hop: if workerd dropped an
+	// empty-valued header, the aggregator would fall back to default redaction
+	// and return NotFound for redacted subjects, silently blinding analysis.
+	// The vitest AGGREGATOR stub echoes the inbound header as a marker.
+	it("delivers a blank accept-labelers header across the service binding", async () => {
+		const response = await env.AGGREGATOR.fetch("https://aggregator/xrpc/probe", {
+			headers: { "atproto-accept-labelers": "" },
+		});
+		expect(response.headers.get("x-test-accept-labelers")).toBe("empty");
+	});
+});

apps/labeler/vitest.config.ts

@@ -22,6 +22,24 @@ export default defineConfig({
 					LABEL_SIGNING_PRIVATE_KEY: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAE",
 					NOTIFICATION_HASH_PEPPER: "test-notification-hash-pepper",
 				},
+				// wrangler.jsonc declares an AGGREGATOR service binding to the
+				// aggregator Worker, which doesn't exist in the test runtime.
+				// Stub it so miniflare can start; most tests inject their own mock
+				// Fetcher rather than using this binding. The stub still fails loud
+				// (501) for an accidental call, and echoes the inbound
+				// atproto-accept-labelers header as a marker so a binding-transport
+				// test can prove the client's blank value survives the hop (a
+				// dropped empty header would read `absent`, not `empty`).
+				serviceBindings: {
+					AGGREGATOR: (request) => {
+						const raw = request.headers.get("atproto-accept-labelers");
+						const marker = raw === null ? "absent" : raw === "" ? "empty" : `value:${raw}`;
+						return new Response("AGGREGATOR is stubbed in tests", {
+							status: 501,
+							headers: { "x-test-accept-labelers": marker },
+						});
+					},
+				},
 			},
 		}),
 	],