feat(registry): add labeler service core

.changeset/lucky-labelers-contract.md

@@ -0,0 +1,5 @@
+---
+"@emdash-cms/registry-lexicons": patch
+---
+
+Adds experimental labeler assessment and policy query contracts.

.github/workflows/ci.yml

@@ -124,6 +124,9 @@ jobs:
       - run: pnpm test:unit
         env:
           EMDASH_TEST_PG: postgres://postgres:test@localhost:5432/emdash_test
+      - run: pnpm --filter @emdash-cms/labeler build
+      - run: pnpm --filter @emdash-cms/labeler typecheck
+      - run: pnpm --filter @emdash-cms/labeler test
       # Render tests use the Astro Vite plugin (vitest.repro.config.ts);
       # they can't run under the plain-node config in test:unit.
       - run: pnpm --filter emdash exec vitest run --config vitest.repro.config.ts

apps/aggregator/migrations/0001_init.sql

@@ -178,14 +178,14 @@ CREATE TABLE mirrored_artifacts (
 );
 
 ------------------------------------------------------------------------------
--- Labels (populated when the labeller integration lands)
+-- Labels (populated when the labeler integration lands)
 ------------------------------------------------------------------------------
 
 -- Append-only label history. Every label received is written here, including
 -- negations. Current state is derived from latest cts per (src, uri, val) and
 -- projected into label_state below for hot-path lookups.
 CREATE TABLE labels (
-	src TEXT NOT NULL,                          -- labeller DID
+	src TEXT NOT NULL,                          -- labeler DID
 	uri TEXT NOT NULL,                          -- AT URI of subject
 	cid TEXT,                                   -- optional version-specific CID
 	val TEXT NOT NULL,                          -- e.g. 'security:yanked', '!takedown'
@@ -229,8 +229,8 @@ CREATE TABLE label_state (
 CREATE INDEX idx_label_state_enforce ON label_state(uri, val, trusted)
 	WHERE neg = 0 AND trusted = 1;
 
--- Trusted/known labellers (operator config, edited via deployment).
-CREATE TABLE labellers (
+-- Trusted/known labelers (operator config, edited via deployment).
+CREATE TABLE labelers (
 	did TEXT PRIMARY KEY,
 	endpoint TEXT NOT NULL,                     -- subscribeLabels URL
 	signing_key TEXT NOT NULL,                  -- cached #atproto_label key
@@ -278,9 +278,9 @@ END;
 ------------------------------------------------------------------------------
 
 -- Cursor state for ingest sources (Jetstream microsecond timestamp,
--- subscribeLabels seq cursors per labeller, etc.).
+-- subscribeLabels seq cursors per labeler, etc.).
 CREATE TABLE ingest_state (
-	source TEXT PRIMARY KEY,                    -- 'jetstream', 'labeller:did:web:labels.example.com', etc.
+	source TEXT PRIMARY KEY,                    -- 'jetstream', 'labeler:did:web:labels.example.com', etc.
 	cursor TEXT NOT NULL,
 	updated_at TEXT NOT NULL
 );

apps/aggregator/src/routes/xrpc/getPackage.ts

@@ -19,7 +19,7 @@ export async function getPackage(
 	params: AggregatorGetPackage.$params,
 ): Promise<Response> {
 	// `first-primary` because the same row could become subject to a takedown
-	// label between two reads; once the labeller (Slice 2) writes, the next
+	// label between two reads; once the labeler (Slice 2) writes, the next
 	// read everywhere should reflect it. Per plan §XRPC endpoints.
 	const session = env.DB.withSession("first-primary");
 	const row = await session

apps/aggregator/src/routes/xrpc/router.ts

@@ -52,7 +52,7 @@ const SYNC_GET_RECORD_PATH = "/xrpc/com.atproto.sync.getRecord";
  * caller's origin or credentials -- there are no cookies, no auth, no
  * per-origin policy. We allow `atproto-accept-labelers` and
  * `content-type` as request headers (the only two clients send), echo
- * back the labellers header for symmetry with atproto's labeller-aware
+ * back the labelers header for symmetry with atproto's labeler-aware
  * clients, and cap preflight cache at 24h.
  */
 const CORS_HEADERS: Record<string, string> = {

apps/aggregator/src/routes/xrpc/searchPackages.ts

@@ -10,7 +10,7 @@
  * a non-issue.
  *
  * Takedown filter joins `label_state` even though that table is empty in
- * Slice 1 — keeps the contract honest for Slice 2 (when the labeller starts
+ * Slice 1 — keeps the contract honest for Slice 2 (when the labeler starts
  * writing), and the optimiser short-circuits the NOT EXISTS subquery
  * cheaply when no rows match. See plan §Search.
  *
@@ -145,7 +145,7 @@ function buildBrowseBindings(
 	return out;
 }
 
-/** Hard-enforcement label filter. The Slice 2 labeller writes to
+/** Hard-enforcement label filter. The Slice 2 labeler writes to
  * `label_state`; until then the table is empty so this clause is a no-op
  * (the NOT EXISTS short-circuits). The plan calls for shipping the filter
  * now to lock the contract — adding it later would be a behaviour change

apps/labeler/migrations/0001_init.sql

@@ -0,0 +1,80 @@
+CREATE TABLE issuance_actions (
+	id INTEGER PRIMARY KEY,
+	actor TEXT NOT NULL,
+	type TEXT NOT NULL,
+	reason TEXT NOT NULL,
+	idempotency_key TEXT NOT NULL UNIQUE,
+	created_at TEXT NOT NULL
+);
+
+CREATE TRIGGER issuance_actions_immutable_update
+BEFORE UPDATE ON issuance_actions
+BEGIN
+	SELECT RAISE(ABORT, 'issuance actions are immutable');
+END;
+
+CREATE TRIGGER issuance_actions_immutable_delete
+BEFORE DELETE ON issuance_actions
+BEGIN
+	SELECT RAISE(ABORT, 'issuance actions are immutable');
+END;
+
+CREATE TABLE label_sequence (
+	name TEXT PRIMARY KEY CHECK (name = 'issued_labels'),
+	next_sequence INTEGER NOT NULL CHECK (next_sequence > 0)
+);
+
+INSERT INTO label_sequence (name, next_sequence) VALUES ('issued_labels', 1);
+
+CREATE TABLE issued_labels (
+	id INTEGER PRIMARY KEY,
+	action_id INTEGER NOT NULL UNIQUE REFERENCES issuance_actions(id),
+	sequence INTEGER UNIQUE,
+	ver INTEGER NOT NULL CHECK (ver = 1),
+	src TEXT NOT NULL,
+	uri TEXT NOT NULL,
+	cid TEXT,
+	val TEXT NOT NULL,
+	neg INTEGER NOT NULL DEFAULT 0 CHECK (neg IN (0, 1)),
+	cts TEXT NOT NULL,
+	exp TEXT,
+	sig BLOB NOT NULL,
+	signing_key_id TEXT NOT NULL
+);
+
+CREATE TRIGGER issued_labels_allocate_sequence
+AFTER INSERT ON issued_labels
+BEGIN
+	UPDATE issued_labels
+	SET sequence = (SELECT next_sequence FROM label_sequence WHERE name = 'issued_labels')
+	WHERE id = NEW.id;
+	UPDATE label_sequence SET next_sequence = next_sequence + 1 WHERE name = 'issued_labels';
+END;
+
+CREATE TRIGGER issued_labels_immutable_update
+BEFORE UPDATE ON issued_labels
+WHEN NEW.sequence IS NULL
+  OR (OLD.sequence IS NOT NULL AND OLD.sequence IS NOT NEW.sequence)
+  OR OLD.id IS NOT NEW.id
+  OR OLD.action_id IS NOT NEW.action_id
+  OR OLD.ver IS NOT NEW.ver
+  OR OLD.src IS NOT NEW.src
+  OR OLD.uri IS NOT NEW.uri
+  OR OLD.cid IS NOT NEW.cid
+  OR OLD.val IS NOT NEW.val
+  OR OLD.neg IS NOT NEW.neg
+  OR OLD.cts IS NOT NEW.cts
+  OR OLD.exp IS NOT NEW.exp
+BEGIN
+	SELECT RAISE(ABORT, 'issued labels are immutable');
+END;
+
+CREATE TRIGGER issued_labels_immutable_delete
+BEFORE DELETE ON issued_labels
+BEGIN
+	SELECT RAISE(ABORT, 'issued labels are immutable');
+END;
+
+CREATE INDEX issued_labels_query_order ON issued_labels(sequence);
+CREATE INDEX issued_labels_uri_sequence ON issued_labels(uri, sequence);
+CREATE INDEX issued_labels_source_sequence ON issued_labels(src, sequence);

apps/labeler/package.json

@@ -0,0 +1,29 @@
+{
+	"name": "@emdash-cms/labeler",
+	"version": "0.0.0",
+	"private": true,
+	"description": "EmDash registry moderation label service.",
+	"type": "module",
+	"scripts": {
+		"dev": "vite dev",
+		"build": "vite build",
+		"preview": "vite preview",
+		"deploy": "vite build && wrangler deploy",
+		"typecheck": "tsgo --noEmit",
+		"test": "vitest run",
+		"db:migrate:local": "wrangler d1 migrations apply emdash-labeler --local",
+		"db:migrate": "wrangler d1 migrations apply emdash-labeler --remote"
+	},
+	"dependencies": {
+		"@emdash-cms/registry-moderation": "workspace:*"
+	},
+	"devDependencies": {
+		"@cloudflare/vite-plugin": "catalog:",
+		"@cloudflare/vitest-pool-workers": "catalog:",
+		"@types/node": "catalog:",
+		"typescript": "catalog:",
+		"vite": "catalog:",
+		"vitest": "catalog:",
+		"wrangler": "catalog:"
+	}
+}

apps/labeler/src/config.ts

@@ -0,0 +1,10 @@
+const DID = /^did:[a-z0-9]+:[A-Za-z0-9._:%-]+(?:[:][A-Za-z0-9._:%-]+)*$/;
+
+export interface LabelerConfig {
+	labelerDid: string;
+}
+
+export function getLabelerConfig(env: Pick<Env, "LABELER_DID">): LabelerConfig {
+	if (!DID.test(env.LABELER_DID)) throw new TypeError("LABELER_DID must be a DID");
+	return { labelerDid: env.LABELER_DID };
+}

apps/labeler/src/index.ts

@@ -0,0 +1,23 @@
+import { getLabelerConfig } from "./config.js";
+import { queryLabels } from "./query-labels.js";
+import { xrpcError } from "./xrpc.js";
+
+const QUERY_LABELS_PATH = "/xrpc/com.atproto.label.queryLabels";
+
+export default {
+	async fetch(request: Request, env: Env): Promise<Response> {
+		const pathname = new URL(request.url).pathname;
+		if (pathname.startsWith("/xrpc/")) {
+			// Require the deployment identity even though this first public route
+			// is read-only, so issuance cannot be configured against another DID.
+			try {
+				getLabelerConfig(env);
+			} catch {
+				return xrpcError("InternalServerError", "labeler is not configured", 500);
+			}
+			if (pathname === QUERY_LABELS_PATH) return queryLabels(env.DB, request);
+			return xrpcError("MethodNotSupported", "XRPC method not found", 404);
+		}
+		return new Response("emdash-labeler: not found", { status: 404 });
+	},
+};

apps/labeler/src/query-labels.ts

@@ -0,0 +1,109 @@
+import { xrpcError } from "./xrpc.js";
+
+const DID = /^did:[a-z0-9]+:[A-Za-z0-9._:%-]+(?:[:][A-Za-z0-9._:%-]+)*$/;
+const DIGITS = /^\d+$/;
+const POSITIVE_INTEGER = /^[1-9]\d*$/;
+const DEFAULT_LIMIT = 50;
+const MAX_LIMIT = 250;
+
+interface LabelRow {
+	sequence: number;
+	ver: number;
+	src: string;
+	uri: string;
+	cid: string | null;
+	val: string;
+	neg: number;
+	cts: string;
+	exp: string | null;
+	sig: ArrayBuffer;
+}
+
+export async function queryLabels(db: D1Database, request: Request): Promise<Response> {
+	if (request.method !== "GET") {
+		return xrpcError("MethodNotSupported", "queryLabels only supports GET", 405, { allow: "GET" });
+	}
+	const params = new URL(request.url).searchParams;
+	const uriPatterns = params.getAll("uriPatterns");
+	if (uriPatterns.length === 0) return badRequest("uriPatterns is required");
+	const patterns = uriPatterns.map(parseUriPattern);
+	if (patterns.some((pattern) => pattern === null)) return badRequest("invalid uriPatterns");
+	const sources = params.getAll("sources");
+	if (sources.some((source) => !DID.test(source))) return badRequest("sources must contain DIDs");
+	const limit = parseLimit(params.get("limit"));
+	if (limit === null) return badRequest("limit must be an integer between 1 and 250");
+	const cursor = parseCursor(params.get("cursor"));
+	if (cursor === null) return badRequest("cursor must be a positive integer");
+
+	const patternClauses: string[] = [];
+	const values: (string | number)[] = [];
+	for (const pattern of patterns) {
+		if (!pattern) continue;
+		if (pattern.endsWith("*")) {
+			patternClauses.push("substr(uri, 1, ?) = ?");
+			const prefix = pattern.slice(0, -1);
+			values.push(prefix.length, prefix);
+		} else {
+			patternClauses.push("uri = ?");
+			values.push(pattern);
+		}
+	}
+	const sourceClause =
+		sources.length > 0 ? ` AND src IN (${sources.map(() => "?").join(", ")})` : "";
+	values.push(...sources, cursor ?? 0, limit + 1);
+	const rows = await db
+		.prepare(
+			`SELECT sequence, ver, src, uri, cid, val, neg, cts, exp, sig
+			 FROM issued_labels
+			 WHERE (${patternClauses.join(" OR ")})${sourceClause} AND sequence > ?
+			 ORDER BY sequence ASC LIMIT ?`,
+		)
+		.bind(...values)
+		.all<LabelRow>();
+	const labels = rows.results ?? [];
+	const page = labels.slice(0, limit);
+	const last = page.at(-1);
+	return Response.json({
+		labels: page.map((label) => ({
+			ver: label.ver,
+			src: label.src,
+			uri: label.uri,
+			...(label.cid === null ? {} : { cid: label.cid }),
+			val: label.val,
+			...(label.neg === 1 ? { neg: true } : {}),
+			cts: label.cts,
+			...(label.exp === null ? {} : { exp: label.exp }),
+			sig: { $bytes: toBase64(new Uint8Array(label.sig)) },
+		})),
+		...(labels.length > limit && last ? { cursor: `${last.sequence}` } : {}),
+	});
+}
+
+function parseUriPattern(value: string): string | null {
+	if (value.length === 0 || value.length > 2_000) return null;
+	const firstStar = value.indexOf("*");
+	if (firstStar !== -1 && firstStar !== value.length - 1) return null;
+	return value;
+}
+
+function parseLimit(value: string | null): number | null {
+	if (value === null) return DEFAULT_LIMIT;
+	if (!DIGITS.test(value)) return null;
+	const limit = Number(value);
+	return Number.isSafeInteger(limit) && limit >= 1 && limit <= MAX_LIMIT ? limit : null;
+}
+
+function parseCursor(value: string | null): number | null {
+	if (value === null) return 0;
+	if (!POSITIVE_INTEGER.test(value)) return null;
+	const cursor = Number(value);
+	return Number.isSafeInteger(cursor) ? cursor : null;
+}
+
+function badRequest(message: string): Response {
+	return xrpcError("InvalidRequest", message, 400);
+}
+
+function toBase64(value: Uint8Array): string {
+	return btoa(String.fromCharCode(...value));
+}