publishing: governed amplify route — approved drafts can boost their recorded source

POST /publishing/drafts/:draftId/amplify { connectionId, kind?: repost|quote }:
the governance gate is the draft state (only approved drafts amplify), the
target is providerMetadata.sourceRemoteRef recorded at creation — never a
caller-supplied id. Connectors resolve from a process-level publishing
registry (vault-backed credentials); capability-checked per kind. Closes
the gap the governance sim documented (boosts fell back to replies).

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

@@ -131,6 +131,20 @@ export class PublishingController {
     return this.publishing.reject(session.workspaceId, actorFrom(session), parseId(draftId), dto.reason);
   }
 
+  @Post(":draftId/amplify")
+  @RequirePermission("publication.schedule")
+  @RequireGroupAccess("body")
+  @HttpCode(200)
+  async amplify(
+    @Req() request: Request & AuthenticatedRequest,
+    @Param("draftId") draftId: string,
+    @Body() body: unknown,
+  ): Promise<unknown> {
+    const session = requireSession(request);
+    const dto = parseAmplifyDraft(body);
+    return this.publishing.amplifyDraft(session.workspaceId, actorFrom(session), parseId(draftId), dto.connectionId, dto.kind);
+  }
+
   @Post(":draftId/schedule")
   @HttpCode(202)
   @RequirePermission("publication.schedule")
@@ -236,6 +250,14 @@ function parseRejectDraft(value: unknown): RejectDraftDto {
   return { reason: requireString(body.reason, "reason") };
 }
 
+function parseAmplifyDraft(value: unknown): { connectionId: string; kind: "repost" | "quote" } {
+  const body = requireObject(value, "body");
+  const connectionId = parseId(requireString(body.connectionId, "connectionId"));
+  const kind = body.kind === undefined ? "repost" : requireString(body.kind, "kind");
+  if (kind !== "repost" && kind !== "quote") throw new BadRequestException("kind must be repost or quote");
+  return { connectionId, kind };
+}
+
 function parseScheduleDraft(value: unknown): ScheduleDraftDto {
   const body = requireObject(value, "body");
   return {

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

@@ -1,5 +1,13 @@
 import { BadRequestException, ForbiddenException, Inject, Injectable, NotFoundException, Optional, ConflictException } from "@nestjs/common";
-import { isPublishingProvider } from "@tactical-media-broadcast-hub/connector-runtime";
+import { createMastodonConnector } from "@tactical-media-broadcast-hub/connector-mastodon";
+import {
+  createPublishingCredentialAccessor,
+  isPublishingProvider,
+  providerFetch,
+} from "@tactical-media-broadcast-hub/connector-runtime";
+import { ConnectorRegistry, type PlatformConnector } from "@tactical-media-broadcast-hub/connector-sdk";
+import { createTelegramConnector } from "@tactical-media-broadcast-hub/connector-telegram";
+import type { ConnectorConnection } from "@tactical-media-broadcast-hub/core";
 import { Client, Connection, WorkflowExecutionAlreadyStartedError } from "@temporalio/client";
 import { PublishingExecutionRepository, PublishingStateError, canonicalPostHash, assertPublishingSchema, type AttemptRef } from "@tactical-media-broadcast-hub/publishing-execution";
 import type { PoolClient } from "pg";
@@ -67,6 +75,25 @@ export interface PublishingDependencies {
 
 export const PUBLISHING_DEPENDENCIES = Symbol("PUBLISHING_DEPENDENCIES");
 
+let publishingConnectors: ConnectorRegistry | undefined;
+
+/** Publishing-side connectors with vault-backed credentials, once per process. */
+function connectorFor(provider: string): PlatformConnector {
+  publishingConnectors ??= (() => {
+    const registry = new ConnectorRegistry();
+    const options = {
+      credentials: createPublishingCredentialAccessor(pool),
+      fetchImpl: (input: Parameters<typeof fetch>[0], init?: Parameters<typeof fetch>[1]) =>
+        providerFetch(input, init),
+      env: process.env,
+    };
+    registry.register(createMastodonConnector(options));
+    registry.register(createTelegramConnector(options));
+    return registry;
+  })();
+  return publishingConnectors.get(provider);
+}
+
 /** The pending row is durable; a failed hint is repaired by discovery. */
 export class TemporalWorkflowStarter implements WorkflowStarter {
   private connection?: Connection;
@@ -128,6 +155,68 @@ export class PublishingService {
     return draft;
   }
 
+  /**
+   * 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.
+   */
+  async amplifyDraft(
+    workspaceId: string,
+    actor: Actor,
+    draftId: string,
+    connectionId: string,
+    kind: "repost" | "quote",
+  ): Promise<unknown> {
+    await this.dependencies.assertReady();
+    const gate = await this.dependencies.tenantTransaction(workspaceId, async (client) => {
+      const result = await client.query(
+        "SELECT * FROM drafts WHERE id = $1 AND state = 'approved'", [draftId]);
+      if (result.rowCount !== 1) {
+        const current = await client.query("SELECT state FROM drafts WHERE id = $1", [draftId]);
+        const state = current.rows[0]?.state;
+        if (state === undefined) throw new NotFoundException(`draft ${draftId} not found`);
+        throw new BadRequestException(`cannot amplify a draft in state ${String(state)}`);
+      }
+      const draft = draftFromRow(requiredRow(result.rows[0], "draft"));
+      const source = draft.post.providerMetadata.sourceRemoteRef;
+      if (typeof source !== "string" || source === "") {
+        throw new BadRequestException("draft has no amplifiable source reference (providerMetadata.sourceRemoteRef)");
+      }
+      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");
+      await this.audit(client, workspaceId, actor, "draft.amplify_requested", draftId,
+        { connectionId, kind, target: source });
+      return {
+        provider: String(connection.provider),
+        fields: (connection.fields ?? {}) as Record<string, string>,
+        source,
+      };
+    });
+
+    // The network call stays outside the tenant transaction.
+    const connection: ConnectorConnection = {
+      id: connectionId,
+      workspaceId,
+      provider: gate.provider,
+      channelId: connectionId,
+      credentialId: { kind: "credential-ref", id: connectionId },
+      fields: gate.fields,
+      createdAt: new Date(),
+    };
+    const connector = connectorFor(gate.provider);
+    const capabilities = connector.manifest().capabilities;
+    if (kind === "repost" && capabilities.amplifyRepost !== true) {
+      throw new BadRequestException(`${gate.provider} cannot repost`);
+    }
+    if (kind === "quote" && capabilities.amplifyQuote !== true) {
+      throw new BadRequestException(`${gate.provider} cannot quote`);
+    }
+    if (connector.amplify === undefined) throw new BadRequestException(`${gate.provider} cannot amplify`);
+    return connector.amplify(connection, { kind, targetRemoteId: gate.source });
+  }
+
   async listDrafts(
     workspaceId: string,
     options: { state?: DraftState; limit: number },