From d6eb39ff27d798ed07c22f4bd823181c5a01aa6d Mon Sep 17 00:00:00 2001 From: glass beadian Date: Wed, 14 Oct 2026 10:26:27 +0000 Subject: [PATCH] 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). --- .../amplify-governance.integration.test.ts | 188 ++++++++++++++++++ .../src/publishing/publishing.controller.ts | 8 +- apps/api/src/publishing/publishing.service.ts | 29 ++- 3 files changed, 220 insertions(+), 5 deletions(-) create mode 100644 apps/api/src/publishing/amplify-governance.integration.test.ts diff --git a/apps/api/src/publishing/amplify-governance.integration.test.ts b/apps/api/src/publishing/amplify-governance.integration.test.ts new file mode 100644 index 0000000..f7404c8 --- /dev/null +++ b/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(fn: (client: PoolClient) => Promise): Promise { + 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 => (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 { + 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([]); + }); +}); diff --git a/apps/api/src/publishing/publishing.controller.ts b/apps/api/src/publishing/publishing.controller.ts index e6aa5d9..ca66e14 100644 --- a/apps/api/src/publishing/publishing.controller.ts +++ b/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 { 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") diff --git a/apps/api/src/publishing/publishing.service.ts b/apps/api/src/publishing/publishing.service.ts index c0eae1e..1a2af6f 100644 --- a/apps/api/src/publishing/publishing.service.ts +++ b/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 = {