fix: wire scheduler media resolver into publishing registry

The registry accepted media hooks but nothing supplied them, so Mastodon
media posts failed uploadAll (INVALID_POST attempts). PublishingMediaResolver
in the scheduler resolves workspace-scoped media_assets rows via RLS tenant
context and reads plaintext bytes from S3 with SPENC1 open (S3_KEK), matching
the api's S3Storage frame format.

apps/scheduler/src/media-resolver.ts

@@ -0,0 +1,80 @@
+/**
+ * Scheduler-side implementation of the publishing media seam: resolves a
+ * workspace-scoped media asset id to its row (via RLS tenant context) and its
+ * plaintext bytes (S3 download + SPENC1 open, identical to the api's
+ * S3Storage.getObject so either side reads what the other wrote).
+ */
+import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3";
+import type { Pool } from "pg";
+import type { MediaAsset } from "@tactical-media-broadcast-hub/core";
+import { openMediaFrame, mediaKekFromEnv, type PublishingMediaHooks } from "@tactical-media-broadcast-hub/connector-runtime";
+
+interface MediaAssetRow {
+  id: string;
+  workspace_id: string;
+  storage_key: string;
+  mime: string;
+  size_bytes: string | number;
+  created_at: Date;
+}
+
+export class PublishingMediaResolver implements PublishingMediaHooks {
+  private readonly pool: Pool;
+  private readonly client: S3Client;
+  private readonly bucket: string;
+  private readonly kek: Buffer;
+
+  constructor(pool: Pool) {
+    this.pool = pool;
+    this.bucket = process.env.S3_BUCKET ?? "tactical-media-broadcast-hub-media";
+    this.kek = mediaKekFromEnv(process.env);
+    this.client = new S3Client({
+      endpoint: process.env.S3_ENDPOINT,
+      region: process.env.S3_REGION ?? "us-east-1",
+      credentials:
+        process.env.S3_ACCESS_KEY !== undefined && process.env.S3_SECRET_KEY !== undefined
+          ? { accessKeyId: process.env.S3_ACCESS_KEY, secretAccessKey: process.env.S3_SECRET_KEY }
+          : undefined,
+      forcePathStyle: process.env.S3_FORCE_PATH_STYLE !== "false",
+    });
+  }
+
+  private async loadRow(workspaceId: string, assetId: string): Promise<MediaAssetRow> {
+    const client = await this.pool.connect();
+    try {
+      await client.query("BEGIN");
+      await client.query("SELECT set_config('app.workspace_id', $1, true)", [workspaceId]);
+      const result = await client.query<MediaAssetRow>(
+        "SELECT id, workspace_id, storage_key, mime, size_bytes, created_at FROM media_assets WHERE id = $1",
+        [assetId],
+      );
+      await client.query("COMMIT");
+      const row = result.rows[0];
+      if (row === undefined) throw new Error(`media asset ${assetId} not found in workspace ${workspaceId}`);
+      return row;
+    } finally {
+      client.release();
+    }
+  }
+
+  async loadAsset(workspaceId: string, assetId: string): Promise<MediaAsset> {
+    const row = await this.loadRow(workspaceId, assetId);
+    return {
+      id: row.id,
+      workspaceId: row.workspace_id,
+      storageKey: row.storage_key,
+      mimeType: row.mime,
+      sizeBytes: Number(row.size_bytes),
+      createdAt: new Date(row.created_at),
+    };
+  }
+
+  async openAsset(workspaceId: string, asset: MediaAsset): Promise<Uint8Array> {
+    await this.loadRow(workspaceId, asset.id);
+    const res = await this.client.send(new GetObjectCommand({ Bucket: this.bucket, Key: asset.storageKey }));
+    if (res.Body === undefined) throw new Error(`empty object body for ${asset.storageKey}`);
+    const stored = Buffer.from(await res.Body.transformToByteArray());
+    const plaintext = openMediaFrame(stored, this.kek);
+    return new Uint8Array(plaintext);
+  }
+}

apps/scheduler/src/worker.ts

@@ -9,6 +9,7 @@ import pino from "pino";
 import { createPublishingRegistry } from "@tactical-media-broadcast-hub/connector-runtime";
 import { PublishingExecutionRepository, assertPublishingSchema, type AttemptRef } from "@tactical-media-broadcast-hub/publishing-execution";
 import { PublishActivity } from "./activities/publish.activity.js";
+import { PublishingMediaResolver } from "./media-resolver.js";
 import { DirectorActivity } from "./director/director.activity.js";
 import { DirectorHttpClient } from "./director/http.client.js";
 import { HttpDirectorStatePort, HttpDesiredStatePort } from "./director/adapters.js";
@@ -61,7 +62,7 @@ export async function main(): Promise<void> {
   try {
     await assertPublishingSchema(pool);
     const repository = new PublishingExecutionRepository(pool);
-    const registry = createPublishingRegistry(pool);
+    const registry = createPublishingRegistry(pool, new PublishingMediaResolver(pool));
     native = await NativeConnection.connect({ address });
     connection = await Connection.connect({ address });
     const temporal = new Client({ connection, namespace });

packages/connector-runtime/src/index.ts

@@ -6,4 +6,5 @@ export * from "./publish.js";
 export * from "./vault.js";
 export * from "./provider-fetch.js";
 export * from "./credentials.js";
+export * from "./media.js";
 export * from "./registry.js";

packages/connector-runtime/src/registry.ts

@@ -1,52 +1,43 @@
 import type { Pool } from "pg";
 import { ConnectorRegistry } from "@tactical-media-broadcast-hub/connector-sdk";
+import type { MediaAsset } from "@tactical-media-broadcast-hub/core";
 import { createMastodonConnector } from "@tactical-media-broadcast-hub/connector-mastodon";
 import { createTelegramConnector } from "@tactical-media-broadcast-hub/connector-telegram";
 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. */
+/**
+ * Publish-time media seam. The scheduler provides an implementation that
+ * resolves a workspace-scoped media asset id to its metadata and bytes
+ * (S3 download + SPENC1 open); the registry adapts it to the connector
+ * options, threading the publish connection's workspace through.
+ */
 export interface PublishingMediaHooks {
-  loadAsset(workspaceId: string, assetId: string): Promise<{ storageKey: string; mimeType: string; sizeBytes: number }>;
-  openAsset(workspaceId: string, storageKey: string): Promise<Uint8Array>;
+  loadAsset(workspaceId: string, assetId: string): Promise<MediaAsset>;
+  openAsset(workspaceId: string, asset: MediaAsset): Promise<Uint8Array>;
 }
 
+const EXT_FOR_MIME: Record<string, string> = {
+  "image/png": "png",
+  "image/jpeg": "jpg",
+  "image/webp": "webp",
+  "video/mp4": "mp4",
+  "video/webm": "webm",
+};
+
+/** Production composition registers real connectors, once per worker process. */
 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: 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(); }
-    };
+    options.loadAsset = (connection: { workspaceId: string }, assetId: string): Promise<MediaAsset> =>
+      media.loadAsset(connection.workspaceId, assetId);
+    options.openAsset = async (connection: { workspaceId: string }, asset: MediaAsset) => ({
+      bytes: await media.openAsset(connection.workspaceId, asset),
+      filename: `${EXT_FOR_MIME[asset.mimeType] ?? "bin"}`,
+    });
   }
   registry.register(createMastodonConnector(options));
   registry.register(createTelegramConnector(options));