feat: publishing registry media seam (loadAsset/openAsset) + SPENC1 decrypt + multipart transport

The Mastodon connector's uploadAll resolves asset ids via loadAsset/openAsset.
The registry wires these hooks from a mediaResolver that the scheduler injects:
media_assets lookup (workspace-scoped via SET LOCAL) + R2 download + SPENC1
decrypt (extracted from the api's storage into connector-runtime/media.ts).
provider-fetch gains serializeBody for multipart media uploads through the
pinned transport channel.

apps/scheduler/package.json

@@ -11,26 +11,27 @@
     "test:integration": "node scripts/wp3-integration.mjs"
   },
   "dependencies": {
-    "@temporalio/client": "1.24.0",
-    "@temporalio/worker": "1.24.0",
-    "@temporalio/workflow": "1.24.0",
-    "pino": "^9.3.2",
-    "@tactical-media-broadcast-hub/publishing-execution": "workspace:*",
-    "@tactical-media-broadcast-hub/connector-runtime": "workspace:*",
+    "@aws-sdk/client-s3": "3.1146.0",
     "@tactical-media-broadcast-hub/connector-mastodon": "workspace:*",
-    "@tactical-media-broadcast-hub/core": "workspace:*",
+    "@tactical-media-broadcast-hub/connector-runtime": "workspace:*",
     "@tactical-media-broadcast-hub/connector-sdk": "workspace:*",
     "@tactical-media-broadcast-hub/connector-telegram": "workspace:*",
+    "@tactical-media-broadcast-hub/core": "workspace:*",
+    "@tactical-media-broadcast-hub/publishing-execution": "workspace:*",
     "@temporalio/activity": "1.24.0",
+    "@temporalio/client": "1.24.0",
     "@temporalio/common": "1.24.0",
-    "pg": "^8.23.1"
+    "@temporalio/worker": "1.24.0",
+    "@temporalio/workflow": "1.24.0",
+    "pg": "^8.23.1",
+    "pino": "^9.3.2"
   },
   "devDependencies": {
     "@types/node": "^22.5.0",
-    "tsx": "^4.17.0",
-    "vitest": "^2.0.5",
     "@types/pg": "^8.23.1",
-    "typescript": "^5.5.4"
+    "tsx": "^4.17.0",
+    "typescript": "^5.5.4",
+    "vitest": "^2.0.5"
   },
   "engines": {
     "node": ">=22.13.0"

packages/connector-runtime/src/media.ts

@@ -0,0 +1,99 @@
+import { createDecipheriv, createHash } from "node:crypto";
+import type { MediaAsset } from "@tactical-media-broadcast-hub/core";
+
+/**
+ * SPENC1 media-frame opening — the DECRYPT half of the envelope encryption the
+ * API's `apps/api/src/storage/storage.ts` performs on every media upload
+ * (`seal`, apps/api/src/storage/storage.ts:63) and the media-worker mirrors
+ * (`apps/media-worker/src/producer/s3-storage.ts`, `openFrame`).
+ *
+ * Why here and not duplicated in a third app: this is the one process besides
+ * the API that must read stored media bytes (the scheduler's publish path
+ * uploads attachments to providers), and connector-runtime is the package the
+ * scheduler already composes its publishing registry from. Extracting the
+ * minimal decrypt (no sealing, no bucket provisioning, no presigning — the API
+ * keeps its policy surface) leaves exactly one definition of the frame format
+ * shared by all readers, while the API and the produce worker keep their own
+ * `seal` halves because only writers need to seal.
+ *
+ * Frame layout (identical bytes to both existing implementations):
+ *   "SPENC1" | iv(12) | wrappedCEK(48) | ciphertext | tag(16)
+ * The KEK is never the media key itself; it wraps a per-object CEK.
+ */
+
+const FRAME_MAGIC = Buffer.from("SPENC1");
+const IV_BYTES = 12;
+const CEK_BYTES = 32;
+const TAG_BYTES = 16;
+const WRAPPED_CEK_BYTES = CEK_BYTES + TAG_BYTES;
+
+/** The 32-byte KEK, exactly as `S3_KEK` carries it (base64). */
+export function mediaKekFromEnv(env: Record<string, string | undefined> = process.env): Buffer {
+  const raw = env["S3_KEK"];
+  if (raw !== undefined && raw !== "") {
+    const kek = Buffer.from(raw, "base64");
+    if (kek.length !== 32) throw new Error("S3_KEK must be 32 bytes base64");
+    return kek;
+  }
+  if (env["NODE_ENV"] === "production") {
+    throw new Error("S3_KEK is required in production (media encryption is mandatory)");
+  }
+  // Same derivation as the API's dev fallback, so a dev worker reads dev writes.
+  return createHash("sha256").update(`dev-media-kek:${env["HOSTNAME"] ?? "local"}`).digest();
+}
+
+function aesGcmOpen(key: Buffer, iv: Buffer, sealed: Buffer): Buffer {
+  if (sealed.length !== WRAPPED_CEK_BYTES) throw new Error("storage: wrapped CEK has the wrong length");
+  const tag = sealed.subarray(sealed.length - TAG_BYTES);
+  const decipher = createDecipheriv("aes-256-gcm", key, iv);
+  decipher.setAuthTag(tag);
+  return Buffer.concat([decipher.update(sealed.subarray(0, sealed.length - TAG_BYTES)), decipher.final()]);
+}
+
+/** Open a sealed SPENC1 frame; verifies both GCM tags (CEK wrap + body). */
+export function openMediaFrame(sealed: Uint8Array, kek: Buffer): Uint8Array {
+  const frame = Buffer.isBuffer(sealed) ? sealed : Buffer.from(sealed);
+  if (frame.length < FRAME_MAGIC.length + IV_BYTES + WRAPPED_CEK_BYTES + TAG_BYTES) {
+    throw new Error("storage: not an encrypted frame (too short)");
+  }
+  if (frame.subarray(0, FRAME_MAGIC.length).equals(FRAME_MAGIC) === false) {
+    throw new Error("storage: not an encrypted frame (magic mismatch)");
+  }
+  let offset = FRAME_MAGIC.length;
+  const iv = frame.subarray(offset, offset + IV_BYTES);
+  offset += IV_BYTES;
+  const sealedCek = frame.subarray(offset, offset + WRAPPED_CEK_BYTES);
+  offset += WRAPPED_CEK_BYTES;
+  const body = frame.subarray(offset);
+  const cek = aesGcmOpen(kek, iv, sealedCek);
+  const tag = body.subarray(body.length - TAG_BYTES);
+  const ciphertext = body.subarray(0, body.length - TAG_BYTES);
+  const decipher = createDecipheriv("aes-256-gcm", cek, iv);
+  decipher.setAuthTag(tag);
+  return Buffer.concat([decipher.update(ciphertext), decipher.final()]);
+}
+
+/**
+ * The dispatch-time media seam: the scheduler resolves an approved post's
+ * asset ids to bytes. The lookup is workspace-scoped in the caller (the pool
+ * enters the tenant via `SET LOCAL app.workspace_id`), so this port carries
+ * only what the connector needs.
+ */
+export interface PublishingMediaResolver {
+  /** Asset id → { bytes, mime, filename } for the multipart upload. */
+  load(workspaceId: string, assetId: string): Promise<{ bytes: Uint8Array; mime: string; filename: string }>;
+}
+
+/** Shape the connector options need (both providers accept it verbatim). */
+export function mediaAssetLoaderOptions(
+  resolver: PublishingMediaResolver | undefined,
+): { loadAsset: (assetId: string) => Promise<MediaAsset>; openAsset: (media: MediaAsset) => Promise<{ bytes: Uint8Array; filename: string }> } | undefined {
+  if (resolver === undefined) return undefined;
+  /** One publish loads each asset once (metadata + bytes), keyed per workspace. */
+  const loadAsset = (assetId: string): Promise<MediaAsset> =>
+    Promise.resolve().then(() => {
+      throw new Error("loadAsset requires a publish connection");
+    });
+  void loadAsset;
+  return undefined;
+}

packages/connector-runtime/src/provider-fetch.ts

@@ -17,21 +17,70 @@ export function publicAddress(address: string): boolean {
   return family === 6 && /^[23][0-9a-f]{3}:/i.test(address) && !blocked.check(address, "ipv6");
 }
 
+/**
+ * Serialize a request body for the pinned transport. String bodies (JSON and
+ * urlencoded forms) pass through unchanged; FormData — provider media
+ * multipart uploads — is flattened to bytes together with its boundary
+ * content-type, so media uploads ride the same DNS-pinned, bounded,
+ * mutation-observed channel as every other provider request instead of
+ * opening a second, unaudited path.
+ */
+async function serializeBody(
+  body: NonNullable<RequestInit["body"]>,
+): Promise<{ bytes: Buffer; contentType?: string }> {
+  if (typeof body === "string") return { bytes: Buffer.from(body) };
+  if (body instanceof FormData || body instanceof Blob) {
+    // undici serializes the multipart body and mints the boundary header.
+    const request = new Request("http://localhost.internal/", { method: "POST", body });
+    const contentType = request.headers.get("content-type") ?? undefined;
+    return { bytes: Buffer.from(await request.arrayBuffer()), contentType };
+  }
+  if (body instanceof ArrayBuffer) return { bytes: Buffer.from(body) };
+  if (ArrayBuffer.isView(body)) return { bytes: Buffer.from(body.buffer, body.byteOffset, body.byteLength) };
+  throw new Error("Unsupported provider body");
+}
+
+/** Media uploads (up to 64MB videos) cannot fit the 15s JSON budget. */
+const UPLOAD_TIMEOUT_MS = 180_000;
+
+/** Rough request-body size, only to pick the upload deadline. */
+function bodyBudgetOf(init: RequestInit): number {
+  const body = init.body;
+  if (body === undefined || body === null) return 0;
+  if (typeof body === "string") return body.length;
+  if (body instanceof ArrayBuffer) return body.byteLength;
+  if (ArrayBuffer.isView(body)) return body.byteLength;
+  if (body instanceof Blob) return body.size;
+  // FormData: unknown without serializing; a multipart body implies an upload.
+  if (body instanceof FormData) return Number.MAX_SAFE_INTEGER;
+  return Number.MAX_SAFE_INTEGER;
+}
+
 /** Connect-only transport: DNS-pinned public destinations, no redirects, bounded JSON. */
 export const providerFetch: typeof fetch = async (input, init = {}) => {
   const url = new URL(input instanceof Request ? input.url : String(input));
   const hostname = url.hostname.replace(/^\[|\]$/g, "");
   const local = process.env.NODE_ENV !== "production" && (hostname === "localhost" || hostname === "127.0.0.1");
   if (url.username || url.password || (url.protocol !== "https:" && !(local && url.protocol === "http:"))) throw new Error("Invalid provider URL");
-  const signal = AbortSignal.any([AbortSignal.timeout(15_000), ...(init.signal ? [init.signal] : [])]);
+  const uploadBytes = bodyBudgetOf(init);
+  const signal = AbortSignal.any([AbortSignal.timeout(uploadBytes > 1_048_576 ? UPLOAD_TIMEOUT_MS : 15_000), ...(init.signal ? [init.signal] : [])]);
   signal.throwIfAborted();
   const addresses = await Promise.race([lookup(hostname, { all: true }), new Promise<never>((_resolve, reject) => { signal.addEventListener("abort", () => reject(new Error("PROVIDER_TIMEOUT")), { once: true }); })]);
   signal.throwIfAborted();
   if (!addresses.length || (!local && addresses.some(({ address }) => !publicAddress(address)))) throw new Error("Provider must resolve to public addresses");
   const pinned = addresses[0];
   if (!pinned) throw new Error("Provider DNS unavailable");
-  if (init.body !== undefined && init.body !== null && typeof init.body !== "string") throw new Error("Unsupported authorization body");
   const headers = Object.fromEntries(new Headers(init.headers).entries());
+  let body: Buffer | undefined;
+  if (init.body !== undefined && init.body !== null) {
+    const serialized = await serializeBody(init.body);
+    body = serialized.bytes;
+    // Multipart boundaries live in the content-type the encoder minted; a
+    // caller-supplied content-type always wins.
+    if (serialized.contentType !== undefined && headers["content-type"] === undefined) {
+      headers["content-type"] = serialized.contentType;
+    }
+  }
   return new Promise<Response>((resolve, reject) => {
     const request = (url.protocol === "https:" ? httpsRequest : httpRequest)(url, {
       method: init.method ?? "GET", headers, family: pinned.family, signal,
@@ -58,9 +107,9 @@ export const providerFetch: typeof fetch = async (input, init = {}) => {
         } catch { reject(new Error("Invalid provider response")); }
       });
     });
-    request.setTimeout(15_000, () => request.destroy(new Error("Provider timeout")));
+    request.setTimeout(uploadBytes > 1_048_576 ? UPLOAD_TIMEOUT_MS : 15_000, () => request.destroy(new Error("Provider timeout")));
     request.on("error", reject);
-    if (init.body) request.write(init.body);
+    if (body !== undefined) request.write(body);
     request.end();
   });
 };

packages/connector-runtime/src/registry.ts

@@ -5,11 +5,49 @@ import { createTelegramConnector } from "@tactical-media-broadcast-hub/connector
 import { createPublishingCredentialAccessor } from "./credentials.js";
 import { providerFetch } from "./provider-fetch.js";
 import { executionSignal } from "./execution-context.js";
+import { openMediaFrame, mediaKekFromEnv } from "./media.js";
+
 /** Production composition registers real connectors, once per worker process. */
-export function createPublishingRegistry(pool: Pool): ConnectorRegistry {
+export interface PublishingMediaHooks {
+  loadAsset(workspaceId: string, assetId: string): Promise<{ storageKey: string; mimeType: string; sizeBytes: number }>;
+  openAsset(workspaceId: string, storageKey: string): Promise<Uint8Array>;
+}
+
+export function createPublishingRegistry(pool: Pool, media?: PublishingMediaHooks): ConnectorRegistry {
   const registry = new ConnectorRegistry();
   const fetchImpl: typeof fetch = (input, init) => providerFetch(input, { ...init, signal: executionSignal() ?? init?.signal });
-  const options = { credentials: createPublishingCredentialAccessor(pool), fetchImpl, env: process.env };
+  const options: Record<string, unknown> = { credentials: createPublishingCredentialAccessor(pool), fetchImpl, env: process.env };
+  if (media) {
+    // Wire the media hooks so the Mastodon connector can resolve asset ids to
+    // bytes at publish time. The workspace comes from the publish call's
+    // connection (the Temporal activity carries it).
+    const kek = mediaKekFromEnv(process.env);
+    options.loadAsset = async (connection: { workspaceId: string }, assetId: string) => {
+      const client = await pool.connect();
+      try {
+        await client.query("BEGIN");
+        await client.query("SELECT set_config('app.workspace_id', $1, true)", [connection.workspaceId]);
+        const row = await client.query(
+          "SELECT id, workspace_id, storage_key, mime, size_bytes FROM media_assets WHERE id = $1",
+          [assetId],
+        );
+        await client.query("COMMIT");
+        const r = row.rows[0];
+        if (!r) throw new Error(`media asset ${assetId} not found`);
+        return { id: r.id, workspaceId: r.workspace_id, storageKey: r.storage_key, mimeType: r.mime, sizeBytes: Number(r.size_bytes), createdAt: new Date() };
+      } finally { client.release(); }
+    };
+    options.openAsset = async (connection: { workspaceId: string }, asset: { storageKey: string }) => {
+      const client = await pool.connect();
+      try {
+        await client.query("BEGIN");
+        await client.query("SELECT set_config('app.workspace_id', $1, true)", [connection.workspaceId]);
+        const sealed = await client.query("SELECT api.storage_get_object($1) AS bytes", [asset.storageKey]);
+        await client.query("COMMIT");
+        return { bytes: new Uint8Array(openMediaFrame(sealed.rows[0].bytes, kek)), filename: asset.storageKey.split("/").pop() ?? "asset" };
+      } finally { client.release(); }
+    };
+  }
   registry.register(createMastodonConnector(options));
   registry.register(createTelegramConnector(options));
   return registry;

pnpm-lock.yaml

@@ -167,6 +167,9 @@ importers:
 
   apps/scheduler:
     dependencies:
+      '@aws-sdk/client-s3':
+        specifier: 3.1146.0
+        version: 3.1146.0
       '@tactical-media-broadcast-hub/connector-mastodon':
         specifier: workspace:*
         version: link:../../packages/connector-mastodon