feat(pds): implement com.atproto.identity.submitPlcOperation

.changeset/submit-plc-operation.md

@@ -0,0 +1,5 @@
+---
+"@getcirrus/pds": minor
+---
+
+Implement `com.atproto.identity.submitPlcOperation`. The endpoint forwards an already-signed PLC operation to `plc.directory` on the user's behalf, so migration clients can complete an outbound move without talking to the PLC directory themselves. Pairs with the existing `com.atproto.identity.signPlcOperation` to match the reference PDS migration flow.

packages/pds/src/index.ts

@@ -306,6 +306,11 @@ app.post(
 	requireAuth,
 	identity.signPlcOperation,
 );
+app.post(
+	"/xrpc/com.atproto.identity.submitPlcOperation",
+	requireAuth,
+	identity.submitPlcOperation,
+);
 app.get(
 	"/xrpc/gg.mk.experimental.getMigrationToken",
 	requireAuth,

packages/pds/src/xrpc/identity.ts

@@ -248,6 +248,55 @@ async function signOperation(
 	};
 }
 
+/**
+ * Submit a signed PLC operation to the PLC directory.
+ *
+ * Forwards an already-signed operation (e.g. one produced by
+ * signPlcOperation) to plc.directory on the user's behalf, so
+ * migration clients don't have to talk to the directory themselves.
+ *
+ * Endpoint: POST com.atproto.identity.submitPlcOperation
+ */
+export async function submitPlcOperation(
+	c: Context<AuthedAppEnv>,
+): Promise<Response> {
+	const body = await c.req.json<{ operation?: SignedPlcOperation }>();
+
+	const { operation } = body;
+
+	if (!operation) {
+		return c.json(
+			{
+				error: "InvalidRequest",
+				message: "Missing required parameter: operation",
+			},
+			400,
+		);
+	}
+
+	const res = await fetch(`${PLC_DIRECTORY}/${c.env.DID}`, {
+		method: "POST",
+		headers: { "Content-Type": "application/json" },
+		body: JSON.stringify(operation),
+	});
+
+	if (!res.ok) {
+		const message = await res.text();
+		return new Response(
+			JSON.stringify({
+				error: "PlcDirectoryError",
+				message: message || `PLC directory responded with ${res.status}`,
+			}),
+			{
+				status: res.status,
+				headers: { "Content-Type": "application/json" },
+			},
+		);
+	}
+
+	return new Response(null, { status: 200 });
+}
+
 /**
  * Generate a migration token for the CLI.
  *

packages/pds/test/identity.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from "vitest";
+import { afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { Secp256k1Keypair } from "@atproto/crypto";
 import { env, worker } from "./helpers";
 
@@ -51,4 +51,132 @@ describe("Identity Endpoints", () => {
 			expect(expectedSigningKey.startsWith("did:key:")).toBe(true);
 		});
 	});
+
+	describe("com.atproto.identity.submitPlcOperation", () => {
+		let originalFetch: typeof fetch;
+
+		beforeAll(() => {
+			originalFetch = globalThis.fetch;
+		});
+
+		afterEach(() => {
+			globalThis.fetch = originalFetch;
+			vi.unstubAllGlobals();
+		});
+
+		it("requires authentication", async () => {
+			const response = await worker.fetch(
+				new Request(
+					"http://pds.test/xrpc/com.atproto.identity.submitPlcOperation",
+					{
+						method: "POST",
+						headers: { "Content-Type": "application/json" },
+						body: JSON.stringify({ operation: { type: "plc_operation" } }),
+					},
+				),
+				env,
+			);
+			expect(response.status).toBe(401);
+		});
+
+		it("rejects request without operation", async () => {
+			const response = await worker.fetch(
+				new Request(
+					"http://pds.test/xrpc/com.atproto.identity.submitPlcOperation",
+					{
+						method: "POST",
+						headers: {
+							"Content-Type": "application/json",
+							Authorization: `Bearer ${env.AUTH_TOKEN}`,
+						},
+						body: JSON.stringify({}),
+					},
+				),
+				env,
+			);
+			expect(response.status).toBe(400);
+			const body = (await response.json()) as { error: string };
+			expect(body.error).toBe("InvalidRequest");
+		});
+
+		it("forwards the operation to plc.directory for this DID", async () => {
+			const operation = {
+				type: "plc_operation",
+				prev: "bafyreid",
+				rotationKeys: ["did:key:zRotation"],
+				verificationMethods: { atproto: "did:key:zVerify" },
+				alsoKnownAs: ["at://example.test"],
+				services: {
+					atproto_pds: {
+						type: "AtprotoPersonalDataServer",
+						endpoint: "https://new.pds.example",
+					},
+				},
+				sig: "AAAA",
+			};
+
+			const fetchMock = vi.fn(async (url: string | URL, init?: RequestInit) => {
+				const href = typeof url === "string" ? url : url.toString();
+				expect(href).toBe(`https://plc.directory/${env.DID}`);
+				expect(init?.method).toBe("POST");
+				expect(JSON.parse(init?.body as string)).toEqual(operation);
+				return new Response(null, { status: 200 });
+			});
+			vi.stubGlobal("fetch", fetchMock);
+
+			const response = await worker.fetch(
+				new Request(
+					"http://pds.test/xrpc/com.atproto.identity.submitPlcOperation",
+					{
+						method: "POST",
+						headers: {
+							"Content-Type": "application/json",
+							Authorization: `Bearer ${env.AUTH_TOKEN}`,
+						},
+						body: JSON.stringify({ operation }),
+					},
+				),
+				env,
+			);
+
+			expect(fetchMock).toHaveBeenCalledTimes(1);
+			expect(response.status).toBe(200);
+		});
+
+		it("surfaces PLC directory errors to the caller", async () => {
+			const fetchMock = vi.fn(
+				async () =>
+					new Response("invalid signature", {
+						status: 400,
+						headers: { "Content-Type": "text/plain" },
+					}),
+			);
+			vi.stubGlobal("fetch", fetchMock);
+
+			const response = await worker.fetch(
+				new Request(
+					"http://pds.test/xrpc/com.atproto.identity.submitPlcOperation",
+					{
+						method: "POST",
+						headers: {
+							"Content-Type": "application/json",
+							Authorization: `Bearer ${env.AUTH_TOKEN}`,
+						},
+						body: JSON.stringify({
+							operation: { type: "plc_operation", sig: "bad" },
+						}),
+					},
+				),
+				env,
+			);
+
+			expect(response.status).toBe(400);
+			const body = (await response.json()) as {
+				error: string;
+				message: string;
+			};
+			expect(body.error).toBe("PlcDirectoryError");
+			expect(body.message).toContain("invalid signature");
+		});
+	});
 });