T-HUB/packages/connector-runtime/src/media.ts

/**
 * 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 };
}