feat: gate amplify boosts behind per-connection publish governance

A boost to a gated connection (admin1/admin_majority/admin_consensus)
now holds with awaiting_publish_approval on the same publish-phase poll
the publish route uses, instead of executing the reblog. The holder
re-invokes amplify after the pass (hold-and-retry: no durable amplify
queue exists and the poll-close transaction must not do network I/O).
Unset/auto connections behave exactly as before; the route answers 202
on hold, 200 on execute. Adds amplify-governance IT suite (3 live-DB
tests).

apps/api/src/publishing/amplify-governance.integration.test.ts

@@ -0,0 +1,188 @@
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { Pool, type PoolClient } from "pg";
+import { PublishingExecutionRepository, assertPublishingSchema, type AttemptRef } from "@tactical-media-broadcast-hub/publishing-execution";
+
+/**
+ * Amplify behind per-connection publish governance (live PostgreSQL):
+ *  - gated (admin1): amplify holds with the 202 body, reuses the
+ *    approval-time publish poll, and executes nothing (fetch is stubbed to
+ *    throw, so any execution attempt explodes the test)
+ *  - release: after the poll passes, amplify proceeds past the gate (it then
+ *    stops at the vault credential read — there is no real credential here —
+ *    which proves the gate released instead of holding)
+ *  - manual (unset): amplify proceeds with no poll and no hold, as before
+ *
+ * Owns its workspace in the chatgov IT database (CHATGOV_IT_DATABASE_URL +
+ * CHATGOV_IT_SUPER_DATABASE_URL, migrated through 033). Skips clean.
+ */
+const appUrl = process.env.CHATGOV_IT_DATABASE_URL;
+const ownerUrl = process.env.CHATGOV_IT_SUPER_DATABASE_URL;
+
+type Modules = {
+  publishing: typeof import("./publishing.service.js");
+  polls: typeof import("../chatgov/poll.service.js");
+  db: typeof import("../db/client.js");
+};
+
+describe.skipIf(appUrl === undefined || ownerUrl === undefined)("amplify publish governance (live DB)", () => {
+  let m: Modules;
+  let ownerPool: Pool;
+  let appPool: Pool;
+  const suffix = `${Date.now()}${Math.floor(Math.random() * 1e6)}`;
+  let ws: string;
+  let ownerId: string;
+  let admin2: string;
+  let admin3: string;
+  let editorId: string;
+  let gatedConn: string;
+  let manualConn: string;
+
+  async function inTenant<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
+    const client = await ownerPool.connect();
+    try {
+      await client.query("BEGIN");
+      await client.query("SELECT set_config('app.workspace_id', $1, true)", [ws]);
+      const result = await fn(client);
+      await client.query("COMMIT");
+      return result;
+    } catch (error) {
+      await client.query("ROLLBACK");
+      throw error;
+    } finally {
+      client.release();
+    }
+  }
+
+  beforeAll(async () => {
+    process.env.DATABASE_URL = appUrl;
+    process.env.SUPER_DATABASE_URL = ownerUrl;
+    m = {
+      publishing: await import("./publishing.service.js"),
+      polls: await import("../chatgov/poll.service.js"),
+      db: await import("../db/client.js"),
+    };
+    ownerPool = new Pool({ connectionString: ownerUrl });
+    appPool = new Pool({ connectionString: appUrl });
+    // Any network attempt means the gate failed to hold: explode loudly.
+    vi.stubGlobal("fetch", () => { throw new Error("amplify must not touch the network while held"); });
+    const c = await ownerPool.connect();
+    try {
+      ws = (await c.query("INSERT INTO workspaces (name, slug) VALUES ($1, $2) RETURNING id", ["ampgate", `ampgate-${suffix}`])).rows[0].id as string;
+      const user = async (name: string): Promise<string> => (await c.query(
+        "INSERT INTO users (external_id, username, email, display_name) VALUES ($1, $2, $3, $4) RETURNING id",
+        [`ampgate:${name}:${suffix}`, `ampgate-${name}-${suffix}`, `ampgate-${name}-${suffix}@example.com`, name],
+      )).rows[0].id as string;
+      ownerId = await user("owner");
+      admin2 = await user("admin2");
+      admin3 = await user("admin3");
+      editorId = await user("editor");
+      for (const [id, role] of [[ownerId, "Owner"], [admin2, "Admin"], [admin3, "Admin"], [editorId, "Editor"]] as const) {
+        await c.query("INSERT INTO memberships (workspace_id, user_id, role, status) VALUES ($1, $2, $3, 'active')", [ws, id, role]);
+      }
+    } finally {
+      c.release();
+    }
+    await inTenant(async (client) => {
+      for (const [key, fields] of [["gated", { handle: "gated", publishGovernance: "admin1" }], ["manual", { handle: "manual" }]] as const) {
+        const id = (await client.query("INSERT INTO connections (workspace_id, provider, channel_id, fields) VALUES ($1, 'mastodon', $2, $3) RETURNING id",
+          [ws, `ampgate-${key}`, fields])).rows[0].id as string;
+        if (key === "gated") gatedConn = id; else manualConn = id;
+        await client.query("INSERT INTO credentials (workspace_id, connection_id, kind, ciphertext) VALUES ($1, $2, 'connector-oauth', '\\x00')", [ws, id]);
+        await client.query(`INSERT INTO posting_policy_snapshots (workspace_id, destination_kind, account_id, effective, policy_version, resolved_from)
+          VALUES ($1, 'account', $2, '{"version":1,"mode":"immediate"}', 'pv-immediate', '{}')`, [ws, id]);
+      }
+    });
+  }, 60_000);
+
+  afterAll(async () => {
+    vi.unstubAllGlobals();
+    if (ownerPool !== undefined) {
+      await ownerPool.query("DELETE FROM workspaces WHERE id = $1", [ws]);
+      await ownerPool.query("DELETE FROM users WHERE external_id LIKE $1", [`ampgate:%:${suffix}`]);
+      await ownerPool.end();
+    }
+    await appPool?.end();
+    await m?.db.pool.end();
+  });
+
+  const owner = () => ({ userId: ownerId, role: "Owner" });
+
+  function service() {
+    const starts: AttemptRef[] = [];
+    return new m.publishing.PublishingService({
+      tenantTransaction: m.db.withTenant,
+      workflowStarter: { startScheduledPublish: vi.fn(async (ref: AttemptRef) => { starts.push(ref); }) },
+      repository: new PublishingExecutionRepository(m.db.pool),
+      assertReady: () => assertPublishingSchema(m.db.pool),
+    });
+  }
+
+  async function boostDraft(connectionId: string): Promise<string> {
+    const publishing = service();
+    const draft = await publishing.createDraft(ws, { userId: editorId, role: "Editor" }, {
+      text: `boost ${connectionId} ${Math.random()}`, mediaAssetIds: [], visibility: "public",
+      providerMetadata: { connectionId, sourceRemoteRef: "123456789" },
+    });
+    await publishing.submitForReview(ws, { userId: editorId, role: "Editor" }, draft.id);
+    await publishing.approve(ws, owner(), draft.id);
+    return draft.id;
+  }
+
+  const vote = (pollId: string, userId: string, ballot: "yes" | "no" | "veto") =>
+    new m.polls.PollService().vote(ws, pollId, { ballot }, { kind: "user", userId });
+
+  async function publishPolls(draftId: string) {
+    return inTenant(async (c) => (await c.query(
+      "SELECT id, state FROM polls WHERE draft_id = $1 AND subject_kind = 'draft_publish' ORDER BY created_at",
+      [draftId])).rows as Array<{ id: string; state: string }>);
+  }
+
+  async function heldAudits(draftId: string) {
+    return inTenant(async (c) => (await c.query(
+      "SELECT diff FROM audit_log WHERE target_id = $1 AND action = 'draft.amplify_held'", [draftId])).rows);
+  }
+
+  it("gated: amplify holds with the 202 body on the approval-time poll and executes nothing", async () => {
+    const draftId = await boostDraft(gatedConn);
+    const [poll] = await publishPolls(draftId);
+    expect(poll?.state).toBe("open");
+
+    const held = await service().amplifyDraft(ws, owner(), draftId, gatedConn, "repost");
+    expect(held).toMatchObject({
+      state: "awaiting_publish_approval", draftId, connectionId: gatedConn,
+      publishGovernance: "admin1", pollId: poll?.id,
+    });
+    // The approval-time poll is reused, not duplicated.
+    expect(await publishPolls(draftId)).toHaveLength(1);
+    const audits = await heldAudits(draftId);
+    expect(audits).toHaveLength(1);
+    expect(audits[0]?.diff).toMatchObject({ connectionId: gatedConn, pollId: poll?.id, opened: false });
+  });
+
+  it("release: after the pass, amplify proceeds past the gate (stops at credentials, not the hold)", async () => {
+    const draftId = await boostDraft(gatedConn);
+    const [poll] = await publishPolls(draftId);
+    await service().amplifyDraft(ws, owner(), draftId, gatedConn, "repost");
+    await vote(poll!.id, admin2, "yes");
+    expect((await publishPolls(draftId))[0]?.state).toBe("passed");
+
+    // Past the gate the call reaches the connector, which fails here on the
+    // dummy vault credential — the assertion is that it no longer holds.
+    const outcome = await service().amplifyDraft(ws, owner(), draftId, gatedConn, "repost").then(
+      (result) => ({ held: typeof result === "object" && result !== null && (result as { state?: unknown }).state === "awaiting_publish_approval" }),
+      () => ({ held: false }),
+    );
+    expect(outcome.held).toBe(false);
+    expect(await heldAudits(draftId)).toHaveLength(1);
+  });
+
+  it("manual: amplify proceeds with no poll and no hold, as before", async () => {
+    const draftId = await boostDraft(manualConn);
+    expect(await publishPolls(draftId)).toEqual([]);
+    // Unset governance behaves exactly as before the gate existed: straight
+    // to the connector (dummy credential here, so it throws past the gate).
+    await expect(service().amplifyDraft(ws, owner(), draftId, manualConn, "repost")).rejects.toThrow();
+    expect(await publishPolls(draftId)).toEqual([]);
+    expect(await heldAudits(draftId)).toEqual([]);
+  });
+});

apps/api/src/publishing/publishing.controller.ts

@@ -11,9 +11,10 @@ import {
   Post,
   Query,
   Req,
+  Res,
   UseGuards,
 } from "@nestjs/common";
-import type { Request } from "express";
+import type { Request, Response } from "express";
 import { RbacGuard, RequireGroupAccess, RequirePermission, type AuthenticatedRequest } from "../rbac/rbac.guard.js";
 import { ApprovalGateService } from "./approval-gate.service.js";
 import type { GateDecisionRecord } from "../chatgov/gate.service.js";
@@ -149,10 +150,13 @@ export class PublishingController {
     @Req() request: Request & AuthenticatedRequest,
     @Param("draftId") draftId: string,
     @Body() body: unknown,
+    @Res({ passthrough: true }) res: Response,
   ): Promise<unknown> {
     const session = requireSession(request);
     const dto = parseAmplifyDraft(body);
-    return this.publishing.amplifyDraft(session.workspaceId, actorFrom(session), parseId(draftId), dto.connectionId, dto.kind);
+    const result = await this.publishing.amplifyDraft(session.workspaceId, actorFrom(session), parseId(draftId), dto.connectionId, dto.kind);
+    if (typeof result === "object" && result !== null && (result as { state?: unknown }).state === "awaiting_publish_approval") res.status(202);
+    return result;
   }
 
   @Post(":draftId/schedule")

apps/api/src/publishing/publishing.service.ts

@@ -194,9 +194,13 @@ export class PublishingService {
 
   /**
    * Execute an approved draft's amplification (repost/quote of the source post
-   * recorded at creation) through the given connection. The governance gate is
-   * the draft state: only `approved` drafts amplify, and the target is the
-   * source reference the draft carries — never a caller-supplied id.
+   * recorded at creation) through the given connection. The governance gates
+   * are the draft state (only `approved` drafts amplify; the target is the
+   * source reference the draft carries — never a caller-supplied id) and the
+   * connection's publish governance: a gated connection holds with
+   * `awaiting_publish_approval` until its publish poll for the current
+   * approved revision passes, and the holder re-invokes amplify after the
+   * pass (hold-and-retry).
    */
   async amplifyDraft(
     workspaceId: string,
@@ -223,6 +227,24 @@ export class PublishingService {
       const connection = (await client.query(
         "SELECT provider, fields FROM connections WHERE id = $1", [connectionId])).rows[0];
       if (connection === undefined) throw new NotFoundException("connection not found");
+      // Publish-phase gate: a gated connection amplifies only once the admins'
+      // publish poll for the current approved revision has passed. A held
+      // amplify opens (or reuses) the same poll the publish route uses; the
+      // holder re-invokes amplify after the pass.
+      const governance = readPublishGovernance(connection.fields);
+      if (isPublishGateMode(governance)) {
+        const held = await checkPublishGate(client, {
+          workspaceId, draftId, connectionId, mode: governance,
+          requestedBy: actor.userId, publishAt: new Date(), requiredPermission: "publication.schedule",
+        });
+        if (held.kind === "held") {
+          await this.audit(client, workspaceId, actor, "draft.amplify_held", draftId, {
+            connectionId, kind, target: source, publishGovernance: governance,
+            pollId: held.hold.pollId, opened: held.opened,
+          });
+          return { held: held.hold };
+        }
+      }
       await this.audit(client, workspaceId, actor, "draft.amplify_requested", draftId,
         { connectionId, kind, target: source });
       return {
@@ -231,6 +253,7 @@ export class PublishingService {
         source,
       };
     });
+    if ("held" in gate) return gate.held;
 
     // The network call stays outside the tenant transaction.
     const connection: ConnectorConnection = {