diff --git a/apps/api/src/publishing/publishing.controller.ts b/apps/api/src/publishing/publishing.controller.ts index 127b0ee..4918221 100644 --- a/apps/api/src/publishing/publishing.controller.ts +++ b/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 { + 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 { diff --git a/apps/api/src/publishing/publishing.service.ts b/apps/api/src/publishing/publishing.service.ts index 26dc122..5d6bc5f 100644 --- a/apps/api/src/publishing/publishing.service.ts +++ b/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[0], init?: Parameters[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 { + 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, + 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 },