feat(runtime): publish approved media from the scheduler

Approved posts may carry mediaAssetIds and altTexts since the repository
gate was lifted, but the scheduler built connectors without an asset
loader, so Mastodon and Telegram rejected every post with media as
INVALID_POST and alt texts never reached them.

- media.ts: loadAsset reads media_assets inside the execution's tenant
  (app.workspace_id, RLS forced); openAsset fetches the sealed object,
  opens the SPENC1 frame with S3_KEK and checks the content hash.
  Read-only; bytes go worker to provider and never through the API.
- preparePostForDispatch maps positional altTexts onto
  providerMetadata.altText keyed by asset id.
- Instagram joins the scheduler registry so its posts fail with an
  explicit connector error instead of UnknownProviderError.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

packages/connector-runtime/package.json

@@ -19,7 +19,9 @@
     "@tactical-media-broadcast-hub/core": "workspace:*",
     "pg": "^8.23.1",
     "@tactical-media-broadcast-hub/connector-mastodon": "workspace:*",
-    "@tactical-media-broadcast-hub/connector-telegram": "workspace:*"
+    "@tactical-media-broadcast-hub/connector-telegram": "workspace:*",
+    "@tactical-media-broadcast-hub/connector-instagram": "workspace:*",
+    "@aws-sdk/client-s3": "^3.1146.0"
   },
   "devDependencies": {
     "@types/node": "^22.20.5",

packages/connector-runtime/src/execution-context.ts

@@ -39,3 +39,10 @@ export function publishingContext(connection: ConnectorConnection): PublishingEx
   return context;
 }
 export function executionSignal(): AbortSignal | undefined { return executionStorage.getStore()?.signal; }
+/** The running execution's context; media access has no connection to fence on, only the tenant. */
+export function currentPublishingContext(): PublishingExecutionContext {
+  const context = executionStorage.getStore();
+  if (!context) throw new Error("EXECUTION_CONTEXT_MISSING");
+  context.signal?.throwIfAborted();
+  return context;
+}

packages/connector-runtime/src/idempotency.ts

@@ -37,6 +37,22 @@ export interface DispatchPreparation {
  * drop author-supplied execution keys, inject the attempt's idempotency key
  * and strip scheduledAt unless provider scheduling is explicitly allowed.
  */
+/**
+ * The approved snapshot carries alt texts positionally (altTexts[i] describes
+ * mediaAssetIds[i]); connectors read them as providerMetadata.altText keyed by
+ * asset id. Pipeline-owned, so it replaces any author-set map.
+ */
+function altTextByAsset(post: NormalizedPost): { altText?: Record<string, string> } {
+  const altTexts = (post as NormalizedPost & { altTexts?: unknown }).altTexts;
+  if (!Array.isArray(altTexts) || altTexts.length !== post.mediaAssetIds.length) return {};
+  const map: Record<string, string> = {};
+  post.mediaAssetIds.forEach((id, index) => {
+    const text = altTexts[index];
+    if (typeof text === "string" && text.trim() !== "") map[id] = text;
+  });
+  return Object.keys(map).length === 0 ? {} : { altText: map };
+}
+
 export function preparePostForDispatch(post: NormalizedPost, prep: DispatchPreparation): NormalizedPost {
   for (const key of FORBIDDEN_POST_METADATA_KEYS) {
     if (Object.prototype.hasOwnProperty.call(post.providerMetadata, key)) throw new ExecutionMetadataError(key);
@@ -45,7 +61,7 @@ export function preparePostForDispatch(post: NormalizedPost, prep: DispatchPrepa
   const prepared: NormalizedPost = {
     ...stripped,
     mediaAssetIds: [...post.mediaAssetIds],
-    providerMetadata: { ...stripped.providerMetadata, idempotencyKey: prep.idempotencyKey },
+    providerMetadata: { ...stripped.providerMetadata, ...altTextByAsset(post), idempotencyKey: prep.idempotencyKey },
   };
   if (prep.allowProviderScheduling !== true) delete prepared.scheduledAt;
   return prepared;

packages/connector-runtime/src/index.ts

@@ -7,3 +7,4 @@ export * from "./vault.js";
 export * from "./provider-fetch.js";
 export * from "./credentials.js";
 export * from "./registry.js";
+export * from "./media.js";

packages/connector-runtime/src/media.test.ts

@@ -0,0 +1,86 @@
+import { createCipheriv, createHash, randomBytes } from "node:crypto";
+import type { Pool } from "pg";
+import { describe, expect, it } from "vitest";
+import type { MediaAsset, NormalizedPost } from "@tactical-media-broadcast-hub/core";
+import { runWithPublishingContext } from "./execution-context.js";
+import { preparePostForDispatch } from "./idempotency.js";
+import { createPublishingMediaAccess, openSealedFrame } from "./media.js";
+
+const KEK = randomBytes(32);
+const WS = "0d9a1b40-6f0f-4b1e-8f0e-3f3c9e6b4a10";
+const context = { workspaceId: WS, connectionId: "conn-1", credentialId: "cred-1", credentialGeneration: "1" };
+
+/** The API's frame: magic | iv | AES-GCM(kek, cek) | AES-GCM(cek, body). */
+function seal(plaintext: Buffer, kek = KEK): Buffer {
+  const cek = randomBytes(32);
+  const iv = randomBytes(12);
+  const wrap = createCipheriv("aes-256-gcm", kek, iv);
+  const sealedCek = Buffer.concat([wrap.update(cek), wrap.final(), wrap.getAuthTag()]);
+  const cipher = createCipheriv("aes-256-gcm", cek, iv);
+  const body = Buffer.concat([cipher.update(plaintext), cipher.final(), cipher.getAuthTag()]);
+  return Buffer.concat([Buffer.from("SPENC1"), iv, sealedCek, body]);
+}
+
+const JPEG = Buffer.from([0xff, 0xd8, 0xff, 0xe0, 1, 2, 3, 4]);
+const KEY = `sha256/${createHash("sha256").update(JPEG).digest("hex")}.jpg`;
+const asset: MediaAsset = { id: "11111111-2222-4333-8444-555555555555", workspaceId: WS, storageKey: KEY, mimeType: "image/jpeg", sizeBytes: JPEG.length, createdAt: new Date() };
+
+function fakePool(rows: unknown[]): { pool: Pool; queries: Array<{ sql: string; values?: unknown[] }> } {
+  const queries: Array<{ sql: string; values?: unknown[] }> = [];
+  const client = {
+    query: async (sql: string, values?: unknown[]) => { queries.push({ sql, values }); return { rows: sql.startsWith("SELECT id") ? rows : [] }; },
+    release: () => undefined,
+  };
+  return { pool: { connect: async () => client } as unknown as Pool, queries };
+}
+
+describe("openSealedFrame", () => {
+  it("opens the API frame and rejects tampering or a wrong key", () => {
+    const sealed = seal(JPEG);
+    expect(openSealedFrame(sealed, KEK).equals(JPEG)).toBe(true);
+    const flipped = Buffer.from(sealed); flipped[flipped.length - 20]! ^= 1;
+    expect(() => openSealedFrame(flipped, KEK)).toThrow();
+    expect(() => openSealedFrame(sealed, randomBytes(32))).toThrow();
+    expect(() => openSealedFrame(Buffer.from("plain"), KEK)).toThrow(/MEDIA_FRAME_INVALID/);
+  });
+});
+
+describe("createPublishingMediaAccess", () => {
+  it("loads the asset inside the execution's tenant", async () => {
+    const { pool, queries } = fakePool([{ id: asset.id, workspace_id: WS, storage_key: KEY, mime: "image/jpeg", size_bytes: "8", created_at: asset.createdAt }]);
+    const media = createPublishingMediaAccess(pool, { read: async () => seal(JPEG) }, KEK);
+    const loaded = await runWithPublishingContext(context, () => media.loadAsset(asset.id));
+    expect(loaded).toMatchObject({ id: asset.id, workspaceId: WS, storageKey: KEY, mimeType: "image/jpeg", sizeBytes: 8 });
+    expect(queries[1]).toEqual({ sql: "SELECT set_config('app.workspace_id', $1, true)", values: [WS] });
+    expect(queries[2]?.sql).toContain("workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid");
+  });
+
+  it("refuses outside a publishing execution and for unknown ids", async () => {
+    const { pool } = fakePool([]);
+    const media = createPublishingMediaAccess(pool, { read: async () => seal(JPEG) }, KEK);
+    await expect(media.loadAsset(asset.id)).rejects.toThrow(/EXECUTION_CONTEXT_MISSING/);
+    await expect(runWithPublishingContext(context, () => media.loadAsset(asset.id))).rejects.toThrow(/MEDIA_ASSET_NOT_FOUND/);
+  });
+
+  it("opens bytes, checks the content address and the tenant", async () => {
+    const media = createPublishingMediaAccess(fakePool([]).pool, { read: async () => seal(JPEG) }, KEK);
+    const opened = await runWithPublishingContext(context, () => media.openAsset(asset));
+    expect(Buffer.from(opened.bytes).equals(JPEG)).toBe(true);
+    expect(opened.filename).toBe(`${asset.id}.jpg`);
+    await expect(runWithPublishingContext(context, () => media.openAsset({ ...asset, workspaceId: "other" }))).rejects.toThrow(/MEDIA_TENANT_FENCE/);
+    const swapped = createPublishingMediaAccess(fakePool([]).pool, { read: async () => seal(Buffer.from("other bytes")) }, KEK);
+    await expect(runWithPublishingContext(context, () => swapped.openAsset(asset))).rejects.toThrow(/MEDIA_INTEGRITY_FAILURE/);
+  });
+});
+
+describe("alt text at dispatch", () => {
+  it("maps positional altTexts onto providerMetadata.altText by asset id", () => {
+    const post: NormalizedPost & { altTexts?: Array<string | null> } = { text: "x", mediaAssetIds: ["a", "b", "c"], altTexts: ["first", null, " "], visibility: "public", providerMetadata: { altText: { a: "author-set", b: "author-set" } } };
+    const prepared = preparePostForDispatch(post, { idempotencyKey: "k" });
+    expect(prepared.providerMetadata.altText).toEqual({ a: "first" });
+    expect(prepared.providerMetadata.idempotencyKey).toBe("k");
+    const withoutAltTexts: NormalizedPost = { text: "x", mediaAssetIds: ["a"], visibility: "public", providerMetadata: {} };
+    const none = preparePostForDispatch(withoutAltTexts, { idempotencyKey: "k" });
+    expect(none.providerMetadata.altText).toBeUndefined();
+  });
+});

packages/connector-runtime/src/media.ts

@@ -0,0 +1,137 @@
+/**
+ * Publishing-side media access: the `loadAsset` / `openAsset` pair the
+ * connectors need to attach an approved post's media.
+ *
+ * - loadAsset reads the `media_assets` row inside the current execution's
+ *   tenant (app.workspace_id set from the publishing context, RLS forced),
+ *   so a job can only ever resolve its own workspace's assets.
+ * - openAsset downloads the sealed object and opens the SPENC1 frame with the
+ *   media KEK, checking the content address. Read-only: publishing never
+ *   writes to storage.
+ *
+ * Bytes travel worker → provider; they never pass through the API process.
+ */
+import { createDecipheriv, createHash } from "node:crypto";
+import type { Pool } from "pg";
+import { GetObjectCommand, S3Client } from "@aws-sdk/client-s3";
+import type { MediaAsset } from "@tactical-media-broadcast-hub/core";
+import { currentPublishingContext } from "./execution-context.js";
+
+const FRAME_MAGIC = Buffer.from("SPENC1");
+/** Same per-asset caps the API enforces at upload (media.service.ts). */
+const MAX_ASSET_BYTES = 64 * 1024 * 1024;
+
+export interface SealedObjectReader {
+  /** Returns the stored (sealed) object body. */
+  read(key: string): Promise<Buffer>;
+}
+
+export interface PublishingMediaAccess {
+  loadAsset(assetId: string): Promise<MediaAsset>;
+  openAsset(media: MediaAsset): Promise<{ bytes: Uint8Array; filename: string }>;
+}
+
+function aesGcmOpen(key: Buffer, iv: Buffer, sealed: Buffer): Buffer {
+  const decipher = createDecipheriv("aes-256-gcm", key, iv);
+  decipher.setAuthTag(sealed.subarray(sealed.length - 16));
+  return Buffer.concat([decipher.update(sealed.subarray(0, sealed.length - 16)), decipher.final()]);
+}
+
+/** Open [magic | iv(12) | wrapped CEK(48) | body+tag], the API's storage frame. */
+export function openSealedFrame(sealed: Buffer, kek: Buffer): Buffer {
+  if (!sealed.subarray(0, FRAME_MAGIC.length).equals(FRAME_MAGIC)) throw new Error("MEDIA_FRAME_INVALID");
+  const iv = sealed.subarray(6, 18);
+  const cek = aesGcmOpen(kek, iv, sealed.subarray(18, 66));
+  return aesGcmOpen(cek, iv, sealed.subarray(66));
+}
+
+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)");
+  // Dev mirror of the API's fallback so a local scheduler reads local uploads.
+  return createHash("sha256").update(`dev-media-kek:${env.HOSTNAME ?? "local"}`).digest();
+}
+
+export function s3ObjectReaderFromEnv(env: Record<string, string | undefined> = process.env): SealedObjectReader {
+  const bucket = env.S3_BUCKET ?? "tactical-media-broadcast-hub-media";
+  const client = new S3Client({
+    endpoint: env.S3_ENDPOINT,
+    region: env.S3_REGION ?? "us-east-1",
+    credentials: env.S3_ACCESS_KEY !== undefined && env.S3_SECRET_KEY !== undefined
+      ? { accessKeyId: env.S3_ACCESS_KEY, secretAccessKey: env.S3_SECRET_KEY }
+      : undefined,
+    forcePathStyle: env.S3_FORCE_PATH_STYLE !== "false",
+  });
+  return {
+    async read(key) {
+      const response = await client.send(new GetObjectCommand({ Bucket: bucket, Key: key }));
+      if (response.Body === undefined) throw new Error("MEDIA_OBJECT_EMPTY");
+      return Buffer.from(await response.Body.transformToByteArray());
+    },
+  };
+}
+
+const EXTENSIONS: Record<string, string> = {
+  "image/jpeg": "jpg", "image/png": "png", "image/webp": "webp", "video/mp4": "mp4", "video/webm": "webm",
+};
+
+interface MediaRow {
+  id: string;
+  workspace_id: string;
+  storage_key: string;
+  mime: string;
+  size_bytes: string | number;
+  created_at: Date;
+}
+
+/** Caller owns pool lifecycle (same contract as the credential accessor). */
+export function createPublishingMediaAccess(pool: Pool, reader: SealedObjectReader, kek: Buffer): PublishingMediaAccess {
+  async function loadAsset(assetId: string): Promise<MediaAsset> {
+    const context = currentPublishingContext();
+    const client = await pool.connect();
+    try {
+      await client.query("BEGIN");
+      await client.query("SELECT set_config('app.workspace_id', $1, true)", [context.workspaceId]);
+      const result = await client.query<MediaRow>(
+        `SELECT id, workspace_id, storage_key, mime, size_bytes, created_at FROM media_assets
+         WHERE id = $1 AND workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid`,
+        [assetId],
+      );
+      await client.query("COMMIT");
+      const row = result.rows[0];
+      if (row === undefined) throw new Error("MEDIA_ASSET_NOT_FOUND");
+      return {
+        id: row.id,
+        workspaceId: row.workspace_id,
+        storageKey: row.storage_key,
+        mimeType: row.mime,
+        sizeBytes: Number(row.size_bytes),
+        createdAt: row.created_at,
+      };
+    } catch (error) {
+      await client.query("ROLLBACK").catch(() => undefined);
+      throw error;
+    } finally {
+      client.release();
+    }
+  }
+
+  async function openAsset(media: MediaAsset): Promise<{ bytes: Uint8Array; filename: string }> {
+    const context = currentPublishingContext();
+    if (media.workspaceId !== context.workspaceId) throw new Error("MEDIA_TENANT_FENCE");
+    if (media.sizeBytes > MAX_ASSET_BYTES) throw new Error("MEDIA_ASSET_TOO_LARGE");
+    const plaintext = openSealedFrame(await reader.read(media.storageKey), kek);
+    const expected = /^sha256\/([0-9a-f]{64})\./.exec(media.storageKey)?.[1];
+    if (expected === undefined || createHash("sha256").update(plaintext).digest("hex") !== expected) {
+      throw new Error("MEDIA_INTEGRITY_FAILURE");
+    }
+    return { bytes: plaintext, filename: `${media.id}.${EXTENSIONS[media.mimeType] ?? "bin"}` };
+  }
+
+  return { loadAsset, openAsset };
+}

packages/connector-runtime/src/registry.test.ts

@@ -6,10 +6,10 @@ import type { ConnectorConnection } from "@tactical-media-broadcast-hub/core";
 import { runWithPublishingContext } from "./execution-context.js";
 const connection: ConnectorConnection = { id: "connection", workspaceId: "workspace", provider: "mastodon", channelId: "channel", fields: {}, createdAt: new Date(), credentialId: { kind: "credential-ref", id: "connection" } };
 describe("production publishing registry", () => {
-  it("registers real Mastodon and Telegram connectors and rejects duplicates", () => {
+  it("registers real Mastodon, Telegram and Instagram connectors and rejects duplicates", () => {
     const pool = {} as Pool;
     const registry = createPublishingRegistry(pool);
-    expect(registry.list().map((c) => c.manifest().id).sort()).toEqual(["mastodon", "telegram"]);
+    expect(registry.list().map((c) => c.manifest().id).sort()).toEqual(["instagram", "mastodon", "telegram"]);
     expect(registry.get("mastodon").publish).toBeTypeOf("function");
     expect(() => registry.register(registry.get("mastodon"))).toThrow(/already registered/);
   });

packages/connector-runtime/src/registry.ts

@@ -1,17 +1,22 @@
 import type { Pool } from "pg";
 import { ConnectorRegistry } from "@tactical-media-broadcast-hub/connector-sdk";
+import { createInstagramConnector } from "@tactical-media-broadcast-hub/connector-instagram";
 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 { createPublishingMediaAccess, mediaKekFromEnv, s3ObjectReaderFromEnv, type PublishingMediaAccess } from "./media.js";
 /** Production composition registers real connectors, once per worker process. */
-export function createPublishingRegistry(pool: Pool): ConnectorRegistry {
+export function createPublishingRegistry(pool: Pool, media?: PublishingMediaAccess): 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 assets = media ?? createPublishingMediaAccess(pool, s3ObjectReaderFromEnv(), mediaKekFromEnv());
+  const options = { credentials: createPublishingCredentialAccessor(pool), fetchImpl, env: process.env, loadAsset: assets.loadAsset, openAsset: assets.openAsset };
   registry.register(createMastodonConnector(options));
   registry.register(createTelegramConnector(options));
+  // Connects and refuses text-only posts; image publishing needs a public media URL (not provided yet).
+  registry.register(createInstagramConnector({ credentials: options.credentials, fetchImpl, env: process.env, loadAsset: assets.loadAsset }));
   return registry;
 }
 export { verifiedDedupeOrigin } from "./policy.js";

pnpm-lock.yaml

@@ -502,6 +502,12 @@ importers:
 
   packages/connector-runtime:
     dependencies:
+      '@aws-sdk/client-s3':
+        specifier: ^3.1146.0
+        version: 3.1146.0
+      '@tactical-media-broadcast-hub/connector-instagram':
+        specifier: workspace:*
+        version: link:../connector-instagram
       '@tactical-media-broadcast-hub/connector-mastodon':
         specifier: workspace:*
         version: link:../connector-mastodon