import { BadRequestException, ForbiddenException, Inject, Injectable, NotFoundException, Optional, ConflictException } from "@nestjs/common";
// eslint-disable-next-line no-restricted-syntax -- Connector composition boundary.
import { createInstagramConnector } from "@tactical-media-broadcast-hub/connector-instagram";
// eslint-disable-next-line no-restricted-syntax -- Connector composition boundary.
import { createMastodonConnector } from "@tactical-media-broadcast-hub/connector-mastodon";
import { isPublishingProvider, providerFetch } from "@tactical-media-broadcast-hub/connector-runtime";
import { ConnectorRegistry, type CredentialAccessor, type PlatformConnector } from "@tactical-media-broadcast-hub/connector-sdk";
// eslint-disable-next-line no-restricted-syntax -- Connector composition boundary.
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";
import { markDraftPollsStale, autoOpenDraftPoll } from "../chatgov/poll.service.js";
import { PgConnectionStore } from "../connections/connection-store.js";
import { scrub } from "../audit/audit.service.js";
import { ApprovalGateService, ApprovalGateBlockedError } from "./approval-gate.service.js";
import { checkPublishGate, isPublishGateMode, onDraftApproved, readPublishGovernance, type ApprovalFollowUp, type PublishHold } from "./publish-governance.js";
import { pool, withTenant } from "../db/client.js";
import { userCanUseConnection } from "../groups/groups.service.js";
import { vault } from "../vault/vault.js";

export type DraftState = "editing" | "in_review" | "approved" | "rejected" | "scheduled" | "published";

export interface DraftPost {
  text: string;
  /** media_assets ids (0..4); existence is checked against this workspace on write. */
  mediaAssetIds: string[];
  /** Optional alt texts, positionally aligned with mediaAssetIds. */
  altTexts?: (string | null)[];
  visibility: "public" | "unlisted" | "private" | "followers" | "provider_default";
  scheduledAt?: string;
  providerMetadata: Record<string, unknown>;
}

export interface DraftRecord {
  id: string;
  workspaceId: string;
  authorUserId: string;
  post: DraftPost;
  state: DraftState;
  reviewerUserId?: string;
  createdAt: Date;
  updatedAt: Date;
}

export interface ScheduledPublicationRecord {
  id: string;
  workspaceId: string;
  draftId: string;
  connectionId: string;
  publishAt: Date;
  idempotencyKey: string;
  generation: string;
  state: "pending" | "in_flight" | "done" | "failed" | "needs_reauthorization" | "cancelled";
}

/** The publish/schedule routes either schedule, or hold for the publish gate. */
export type PublishOutcome = ScheduledPublicationRecord | PublishHold;

export interface Actor {
  userId: string;
  role: string;
}

export interface WorkflowStarter {
  startScheduledPublish(ref: AttemptRef, publishAt: Date): Promise<void>;
}

type TenantTransaction = <T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>) => Promise<T>;

export interface PublishingDependencies {
  tenantTransaction: TenantTransaction;
  workflowStarter: WorkflowStarter;
  repository: Pick<PublishingExecutionRepository, "scheduleAttempt" | "cancelAttempt">;
  assertReady: () => Promise<void>;
  /**
   * E1 chatgov approval gate (chatgov-design.md §4). Optional: a plain
   * `new PublishingService()` keeps today's ungated behavior; AppModule
   * injects the real one. `null` disables it explicitly.
   */
  approvalGate?: ApprovalGateService | null;
}

export const PUBLISHING_DEPENDENCIES = Symbol("PUBLISHING_DEPENDENCIES");

let publishingConnectors: ConnectorRegistry | undefined;

/**
 * Vault-backed credential accessor for the amplify path. Unlike the fenced
 * execution accessor (connector-runtime), this runs in the API process where
 * the trigger is the user request itself — same trust level as the archive
 * amplify route's bot-token path.
 */
function publishingCredentialAccessor(): CredentialAccessor {
  const store = new PgConnectionStore();
  return {
    async getAccessToken(connection) {
      return publishingCredentialAccessor().getSecret(connection, "access_token");
    },
    async getSecret(connection, key) {
      const stored = await store.get(connection.workspaceId, connection.id);
      if (!stored.credential) throw new Error("Credential missing");
      const bytes = await vault.open(connection.workspaceId, connection.id, stored.credential);
      const secrets = JSON.parse(bytes.toString()) as Record<string, unknown>;
      const value = secrets[key];
      if (typeof value !== "string" || value === "") throw new Error("Credential missing");
      return value;
    },
    async put(connection, key, value) {
      const stored = await store.get(connection.workspaceId, connection.id);
      if (!stored.credential) throw new Error("Credential missing");
      const bytes = await vault.open(connection.workspaceId, connection.id, stored.credential);
      const secrets = JSON.parse(bytes.toString()) as Record<string, unknown>;
      secrets[key] = value;
      const sealed = await vault.seal(connection.workspaceId, connection.id, Buffer.from(JSON.stringify(secrets)));
      await store.rotateCredential(connection.workspaceId, connection.id, stored.credential, sealed);
    },
  };
}

/** Publishing-side connectors with vault-backed credentials, once per process. */
function connectorFor(provider: string): PlatformConnector {
  publishingConnectors ??= (() => {
    const registry = new ConnectorRegistry();
    const options = {
      credentials: publishingCredentialAccessor(),
      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));
    registry.register(createInstagramConnector(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;
  private connecting?: Promise<Client>;
  private async client(): Promise<Client> {
    this.connecting ??= Connection.connect({ address: process.env.TEMPORAL_ADDRESS ?? "localhost:7233" })
      .then((connection) => { this.connection = connection; return new Client({ connection, namespace: process.env.TEMPORAL_NAMESPACE ?? "default" }); })
      .catch((error: unknown) => { this.connecting = undefined; throw error; });
    return this.connecting;
  }
  async startScheduledPublish(ref: AttemptRef, publishAt: Date): Promise<void> {
    // Future work belongs to the due sweep; no long-lived API timer is needed.
    if (publishAt.getTime() > Date.now()) return;
    try {
      const client = await this.client();
      await client.workflow.start("scheduledPublish", {
        args: [ref], workflowId: `scheduled-publication-${ref.idempotencyKey}`,
        taskQueue: process.env.TEMPORAL_TASK_QUEUE ?? "tmbh-publishing",
      });
    } catch (error) {
      if (error instanceof WorkflowExecutionAlreadyStartedError) return;
      // Committed pending work will be retried by the worker sweep.
    }
  }
  async onModuleDestroy(): Promise<void> { await this.connecting?.catch(() => undefined); await this.connection?.close(); }
}

const DEFAULT_DEPENDENCIES: PublishingDependencies = {
  tenantTransaction: withTenant,
  workflowStarter: new TemporalWorkflowStarter(),
  repository: new PublishingExecutionRepository(pool),
  assertReady: () => assertPublishingSchema(pool),
  approvalGate: new ApprovalGateService(),
};

@Injectable()
export class PublishingService {
  private readonly dependencies: PublishingDependencies;

  constructor(
    @Optional() @Inject(PUBLISHING_DEPENDENCIES) dependencies?: Partial<PublishingDependencies>,
  ) {
    this.dependencies = { ...DEFAULT_DEPENDENCIES, ...dependencies };
  }

  async createDraft(workspaceId: string, actor: Actor, post: DraftPost): Promise<DraftRecord> {
    await this.dependencies.assertReady();
    const draft = await this.dependencies.tenantTransaction(workspaceId, async (client) => {
      await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]);
      // FK checks bypass RLS, so a foreign tenant's asset id would satisfy the
      // constraint; this RLS-scoped count is the actual tenancy check.
      await assertMediaAssetsExist(client, post.mediaAssetIds);
      const result = await client.query(
        `INSERT INTO drafts (workspace_id, author_user_id, post)
         VALUES ($1, $2, $3) RETURNING *`,
        [workspaceId, actor.userId, post],
      );
      const draft = draftFromRow(requiredRow(result.rows[0], "created draft"));
      await this.audit(client, workspaceId, actor, "draft.created", draft.id, { state: draft.state });
      return draft;
    });
    return draft;
  }

  /**
   * Execute an approved draft's amplification (repost/quote of the source post
   * 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,
    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");
      // 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 {
        provider: String(connection.provider),
        fields: (connection.fields ?? {}) as Record<string, string>,
        source,
      };
    });
    if ("held" in gate) return gate.held;

    // 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 },
  ): Promise<DraftRecord[]> {
    return this.dependencies.tenantTransaction(workspaceId, async (client) => {
      const result = await client.query(
        `SELECT * FROM drafts
         WHERE ($1::text IS NULL OR state = $1)
         ORDER BY updated_at DESC
         LIMIT $2`,
        [options.state ?? null, options.limit],
      );
      return result.rows.map((row: unknown) => draftFromRow(requiredRow(row, "draft")));
    });
  }

  /** One entry per business attempt, including provider evidence before finalization. */
  async listPublications(workspaceId: string, draftId: string): Promise<{
    publications: Array<{
      id: string;
      connectionId: string;
      attemptNo: string;
      dispatchCount: number;
      state: "ready" | "in_progress" | "completed" | "failed" | "unknown" | "abandoned";
      createdAt: Date;
      updatedAt: Date;
      firstDispatchStartedAt?: Date;
      dispatchStartedAt?: Date;
      providerRetryDeadline?: Date;
      receivedAt?: Date;
      publishedAt?: Date;
      remoteId?: string;
      remoteUrl?: string;
      error?: string;
    }>;
  }> {
    return this.dependencies.tenantTransaction(workspaceId, async (client) => {
      const draft = await client.query("SELECT id FROM drafts WHERE id = $1", [draftId]);
      if (draft.rowCount !== 1) throw new NotFoundException(`draft ${draftId} not found`);
      const result = await client.query(
        `SELECT a.id, a.connection_id, a.attempt_no, a.dispatch_count, a.status,
                a.created_at, a.updated_at, a.first_dispatch_started_at,
                a.dispatch_started_at, a.provider_retry_deadline, a.error_code,
                r.received_at, COALESCE(p.published_at, r.published_at) AS published_at,
                COALESCE(p.remote_id, r.remote_id) AS remote_id,
                COALESCE(p.remote_url, r.remote_url) AS remote_url
         FROM publishing_attempts a
         LEFT JOIN publishing_receipts r ON r.attempt_id = a.id AND r.workspace_id = a.workspace_id
         LEFT JOIN publications p ON p.idempotency_key = a.idempotency_key AND p.workspace_id = a.workspace_id
         WHERE a.draft_id = $1
         ORDER BY a.attempt_no DESC`,
        [draftId],
      );
      return {
        publications: result.rows.map((value: unknown) => {
          const row = requiredRow(value, "publishing attempt");
          const timestamp = (value: unknown): Date => value instanceof Date ? value : new Date(String(value));
          return {
            id: String(row.id),
            connectionId: String(row.connection_id),
            attemptNo: String(row.attempt_no),
            dispatchCount: Number(row.dispatch_count),
            state: row.status as "ready" | "in_progress" | "completed" | "failed" | "unknown" | "abandoned",
            createdAt: timestamp(row.created_at),
            updatedAt: timestamp(row.updated_at),
            ...(row.first_dispatch_started_at === null || row.first_dispatch_started_at === undefined ? {} : { firstDispatchStartedAt: timestamp(row.first_dispatch_started_at) }),
            ...(row.dispatch_started_at === null || row.dispatch_started_at === undefined ? {} : { dispatchStartedAt: timestamp(row.dispatch_started_at) }),
            ...(row.provider_retry_deadline === null || row.provider_retry_deadline === undefined ? {} : { providerRetryDeadline: timestamp(row.provider_retry_deadline) }),
            ...(row.received_at === null || row.received_at === undefined ? {} : { receivedAt: timestamp(row.received_at) }),
            ...(row.published_at === null || row.published_at === undefined ? {} : { publishedAt: timestamp(row.published_at) }),
            ...(row.remote_id === null || row.remote_id === undefined ? {} : { remoteId: String(row.remote_id) }),
            ...(row.remote_url === null || row.remote_url === undefined ? {} : { remoteUrl: String(row.remote_url) }),
            ...(row.error_code === null || row.error_code === undefined ? {} : { error: String(row.error_code) }),
          };
        }),
      };
    });
  }

  async updateDraft(
    workspaceId: string,
    actor: Actor,
    draftId: string,
    post: DraftPost,
  ): Promise<DraftRecord> {
    await this.dependencies.assertReady();
    const draft = await this.dependencies.tenantTransaction(workspaceId, async (client) => {
      await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]);
      await assertMediaAssetsExist(client, post.mediaAssetIds);
      const result = await client.query(
        `UPDATE drafts SET post = $2, updated_at = now()
         WHERE id = $1 AND (state = 'editing' OR (state = 'in_review' AND EXISTS (SELECT 1 FROM polls WHERE draft_id = $1 AND state = 'open'))) RETURNING *`,
        [draftId, post],
      );
      if (result.rowCount !== 1) {
        const current = await client.query("SELECT state FROM drafts WHERE id = $1", [draftId]);
        if (current.rowCount === 0) throw new NotFoundException(`draft ${draftId} not found`);
        throw new BadRequestException(`cannot edit draft in state ${String(current.rows[0]?.state)}`);
      }
      const draft = draftFromRow(requiredRow(result.rows[0], "updated draft"));
      await markDraftPollsStale(client, workspaceId, draftId);
      await this.audit(client, workspaceId, actor, "draft.updated", draftId, { state: draft.state });
      return draft;
    });
    return draft;
  }

  async submitForReview(workspaceId: string, actor: Actor, draftId: string): Promise<DraftRecord> {
    return this.transition(workspaceId, actor, draftId, "editing", "in_review", "draft.submitted", {
      from: "editing",
      to: "in_review",
    });
  }

  /**
   * `review → approved` with the E1 chatgov gate (chatgov-design.md §4).
   *
   * E1 evaluates the destination's compiled posting_policy_snapshots PER
   * DESTINATION (fan-out law D8): immediate fires (no gate), admin_only fires
   * only for an Owner/Admin approver, quorum blocks with INSUFFICIENT_QUORUM
   * + PENDING_POLL until the linked chat poll passes (P2 owns the poll state
   * machine). Workspaces with no live posting_rules keep today's behavior
   * exactly — the gate is a no-op for them; once rules exist, a destination
   * with no compiled snapshot fails closed with OUTSIDE_POLICY (P0 hand-off).
   *
   * Fired decisions commit in the SAME transaction as the approval (the hook
   * runs under the draft row lock, before the transition UPDATE). A blocked
   * destination aborts that transaction (Conflict) and its verdicts are then
   * recorded after the rollback — the blocked queue (gate_decisions) is the
   * observable surface for them.
   */
  async approve(workspaceId: string, actor: Actor, draftId: string): Promise<DraftRecord> {
    const gate = this.dependencies.approvalGate ?? null;
    let followUp: ApprovalFollowUp | null = null;
    try {
      const approved = await this.transition(workspaceId, actor, draftId, "in_review", "approved", "draft.approved", {
        from: "in_review",
        to: "approved",
      }, actor.userId, gate === null ? undefined : async (client: PoolClient, draft: DraftRecord) => {
        const evaluation = await gate.evaluate(client, workspaceId, draftId, draft.post, actor.userId);
        if (evaluation.blocked.length > 0) throw new ApprovalGateBlockedError(evaluation.blocked);
        await gate.recordFiredInTxn(client, workspaceId, draftId, actor.userId, evaluation.verdicts);
      }, async (client: PoolClient, draft: DraftRecord) => {
        // Publish governance of the target connection: auto schedules now,
        // gate modes open the publish poll; manual connections are untouched.
        followUp = await onDraftApproved(client, {
          workspaceId, draftId, providerMetadata: draft.post.providerMetadata ?? {}, preferredUserId: actor.userId, actor,
        }, this.dependencies.repository);
      });
      // Assigned inside the transaction callback (TS cannot see that write).
      const dispatched = followUp as ApprovalFollowUp | null;
      if (dispatched !== null && dispatched.kind === "dispatched") {
        await this.dependencies.workflowStarter
          .startScheduledPublish({ workspaceId, scheduleId: dispatched.attempt.scheduleId, generation: dispatched.attempt.generation, idempotencyKey: dispatched.attempt.idempotencyKey }, dispatched.publishAt)
          .catch(() => undefined);
      }
      return approved;
    } catch (error) {
      if (error instanceof ApprovalGateBlockedError) {
        // The approval rolled back; record the blocked verdicts with no draft
        // lock held (gate_decisions' FK takes a KEY SHARE on the draft row,
        // which would deadlock against the still-held FOR UPDATE).
        await gate?.recordBlocked(workspaceId, draftId, actor.userId, error.verdicts).catch(() => undefined);
      }
      throw error;
    }
  }

  async reject(workspaceId: string, actor: Actor, draftId: string, reason?: string): Promise<DraftRecord> {
    return this.transition(workspaceId, actor, draftId, "in_review", "rejected", "draft.rejected", {
      from: "in_review",
      to: "rejected",
      ...(reason === undefined ? {} : { reason }),
    }, actor.userId);
  }

  /**
   * Programmatic scheduling. A connection under an admin publish gate cannot
   * be scheduled around it: while its publish poll is pending this refuses
   * with 409 PUBLISH_AWAITING_APPROVAL (the poll is opened/kept). The HTTP
   * routes use requestPublish, which answers 202 with the hold instead.
   */
  async schedule(
    workspaceId: string,
    actor: Actor,
    draftId: string,
    connectionId: string,
    publishAt: Date,
    requiredPermission: "publication.schedule" | "publication.publish" = "publication.schedule",
    expectedPrevious?: { state: "failed" | "needs_reauthorization"; generation: string },
  ): Promise<ScheduledPublicationRecord> {
    const outcome = await this.requestPublish(workspaceId, actor, draftId, connectionId, publishAt, requiredPermission, expectedPrevious);
    if (outcome.state === "awaiting_publish_approval") {
      throw new ConflictException({ code: "PUBLISH_AWAITING_APPROVAL", message: "publish is held for the connection's admin publish gate", hold: outcome });
    }
    return outcome;
  }

  /** Schedule, or hold behind the target connection's publish gate. */
  async requestPublish(
    workspaceId: string,
    actor: Actor,
    draftId: string,
    connectionId: string,
    publishAt: Date,
    requiredPermission: "publication.schedule" | "publication.publish" = "publication.schedule",
    expectedPrevious?: { state: "failed" | "needs_reauthorization"; generation: string },
  ): Promise<PublishOutcome> {
    if (!Number.isFinite(publishAt.getTime())) throw new BadRequestException("publishAt must be a valid timestamp");

    await this.dependencies.assertReady();
    const scheduled = await this.dependencies.tenantTransaction(workspaceId, async (client) => {
      await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]);
      // FK checks bypass RLS, so a foreign tenant's connection id would satisfy
      // the constraint; this RLS-scoped read is the actual tenancy check.
      const connection = await client.query("SELECT provider, fields FROM connections WHERE id = $1 AND disconnected_at IS NULL", [connectionId]);
      if (connection.rowCount !== 1) throw new NotFoundException(`connection ${connectionId} not found`);
      if (!isPublishingProvider(String(connection.rows[0]?.provider))) throw new BadRequestException("provider is not enabled for publishing");
      if (!(await userCanUseConnection(client, workspaceId, actor.userId, connectionId))) {
        throw new ForbiddenException("connection is not shared with any of your groups");
      }

      // Publish-phase gate: a gated connection schedules only once the admins'
      // publish poll for the current approved revision has passed.
      const governance = readPublishGovernance(connection.rows[0]?.fields);
      if (isPublishGateMode(governance)) {
        const gate = await checkPublishGate(client, {
          workspaceId, draftId, connectionId, mode: governance, requestedBy: actor.userId, publishAt, requiredPermission,
        });
        if (gate.kind === "held") {
          await this.audit(client, workspaceId, actor, "publication.publish_held", draftId, {
            connectionId, publishGovernance: governance, pollId: gate.hold.pollId, opened: gate.opened,
            publishAt: publishAt.toISOString(),
          });
          return gate.hold;
        }
      }

      let attempt;
      try {
        attempt = await this.dependencies.repository.scheduleAttempt(client, {
          workspaceId, draftId, connectionId, scheduledByUserId: actor.userId, requiredPermission, publishAt, expectedPrevious,
        });
      } catch (error) {
        if (error instanceof PublishingStateError) throw new ConflictException({ code: error.code, message: error.message });
        throw error;
      }
      const result = await client.query("SELECT * FROM scheduled_publications WHERE id = $1", [attempt.scheduleId]);
      const scheduled = scheduledFromRow(requiredRow(result.rows[0], "scheduled publication"));
      await this.audit(client, workspaceId, actor, "publication.scheduled", draftId, {
        connectionId,
        publishAt: publishAt.toISOString(),
      });
      return scheduled;
    });
    if (scheduled.state === "awaiting_publish_approval") return scheduled;
    await this.dependencies.workflowStarter
      .startScheduledPublish({ workspaceId, scheduleId: scheduled.id, generation: scheduled.generation, idempotencyKey: scheduled.idempotencyKey }, publishAt)
      .catch(() => undefined);
    return scheduled;
  }

  async cancelSchedule(workspaceId: string, actor: Actor, draftId: string): Promise<DraftRecord> {
    await this.dependencies.assertReady();
    const draft = await this.dependencies.tenantTransaction(workspaceId, async (client) => {
      await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]);
      try { await this.dependencies.repository.cancelAttempt(client, workspaceId, draftId); }
      catch (error) {
        if (error instanceof PublishingStateError) throw new ConflictException({ code: error.code, message: error.message });
        throw error;
      }
      const changed = await client.query(
        "SELECT * FROM drafts WHERE id = $1 AND state IN ('approved', 'scheduled') FOR UPDATE",
        [draftId],
      );
      if (changed.rowCount !== 1) await this.throwTransitionError(client, draftId, "approved", "editing");
      const oldState = String(changed.rows[0]?.state);
      // Legacy pending schedules were marked scheduled before this claim gate
      // existed. Reopen them for review instead of treating them as approved.
      const reopened = oldState === "scheduled"
        ? await client.query(
          "UPDATE drafts SET state = 'editing', reviewer_user_id = NULL, updated_at = now() WHERE id = $1 AND state = 'scheduled' RETURNING *",
          [draftId],
        )
        : changed;
      if (reopened.rowCount !== 1) throw new BadRequestException(`cannot reopen draft ${draftId}`);
      const draft = draftFromRow(requiredRow(reopened.rows[0], "cancelled draft"));
      await this.audit(client, workspaceId, actor, "publication.schedule_cancelled", draftId, {
        from: oldState,
        to: draft.state,
      });
      return draft;
    });
    return draft;
  }

  private async transition(
    workspaceId: string,
    actor: Actor,
    draftId: string,
    from: DraftState,
    to: DraftState,
    action: string,
    diff: Record<string, unknown>,
    reviewerUserId?: string,
    /** E1 seam: runs inside the transition's transaction, under the draft row
     * lock, before the UPDATE — may abort the whole transition by throwing. */
    afterLock?: (client: PoolClient, draft: DraftRecord) => Promise<void>,
    /** Runs after the UPDATE + audit, same transaction (publish governance). */
    afterTransition?: (client: PoolClient, draft: DraftRecord) => Promise<void>,
  ): Promise<DraftRecord> {
    await this.dependencies.assertReady();
    return this.dependencies.tenantTransaction(workspaceId, async (client) => {
      await client.query("SELECT id FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]);
      const current = await client.query("SELECT * FROM drafts WHERE id = $1 FOR UPDATE", [draftId]);
      if (current.rowCount !== 1) throw new NotFoundException("draft not found");
      if (afterLock !== undefined) await afterLock(client, draftFromRow(current.rows[0]));
      const hash = canonicalPostHash(current.rows[0].post as DraftPost);
      const result = await client.query(
        `UPDATE drafts
         SET state = $2, reviewer_user_id = COALESCE($3, reviewer_user_id), updated_at = now(),
             approved_revision = CASE WHEN $2 = 'approved' THEN revision ELSE NULL END,
             approved_hash = CASE WHEN $2 = 'approved' THEN $5 ELSE NULL END,
             approved_at = CASE WHEN $2 = 'approved' THEN now() ELSE NULL END
         WHERE id = $1 AND state = $4 RETURNING *`,
        [draftId, to, reviewerUserId ?? null, from, hash],
      );
      if (result.rowCount !== 1) await this.throwTransitionError(client, draftId, from, to);
      const draft = draftFromRow(requiredRow(result.rows[0], "transitioned draft"));
      if (to === "in_review") await autoOpenDraftPoll(client, workspaceId, draftId, actor, draft.post);
      await this.audit(client, workspaceId, actor, action, draftId, diff);
      if (afterTransition !== undefined) await afterTransition(client, draft);
      return draft;
    });
  }

  private async throwTransitionError(
    client: PoolClient,
    draftId: string,
    expected: DraftState,
    target: DraftState,
  ): Promise<never> {
    const current = await client.query("SELECT state FROM drafts WHERE id = $1", [draftId]);
    if (current.rowCount === 0) throw new NotFoundException(`draft ${draftId} not found`);
    throw new BadRequestException(
      `illegal draft transition ${String(current.rows[0]?.state)} -> ${target}; expected ${expected}`,
    );
  }

  private async audit(
    client: PoolClient,
    workspaceId: string,
    actor: Actor,
    action: string,
    targetId: string,
    diff: Record<string, unknown>,
  ): Promise<void> {
    await client.query("SELECT audit_append($1, $2, $3, $4, 'draft', $5, $6)", [
      workspaceId, actor.userId, actor.role, action, targetId, scrub(diff),
    ]);
  }
}

/**
 * Every mediaAssetId must be a media_assets row in THIS workspace. Runs inside
 * the caller's tenant transaction so the RLS tenant_isolation policy is the
 * fence -- a foreign tenant's id simply counts as absent.
 */
async function assertMediaAssetsExist(client: PoolClient, mediaAssetIds: string[]): Promise<void> {
  if (mediaAssetIds.length === 0) return;
  const result = await client.query<{ count: string | number }>(
    "SELECT count(*)::integer AS count FROM media_assets WHERE id = ANY($1::uuid[]) AND workspace_id = NULLIF(current_setting('app.workspace_id', true), '')::uuid",
    [mediaAssetIds],
  );
  if (Number(result.rows[0]?.count ?? 0) !== mediaAssetIds.length) {
    throw new BadRequestException("one or more post.mediaAssetIds do not exist in this workspace");
  }
}

function requiredRow(row: unknown, description: string): Record<string, unknown> {
  if (typeof row !== "object" || row === null) throw new Error(`database did not return ${description}`);
  return row as Record<string, unknown>;
}

function draftFromRow(row: Record<string, unknown>): DraftRecord {
  return {
    id: String(row.id),
    workspaceId: String(row.workspace_id),
    authorUserId: String(row.author_user_id),
    post: row.post as DraftPost,
    state: row.state as DraftState,
    ...(row.reviewer_user_id === null || row.reviewer_user_id === undefined
      ? {}
      : { reviewerUserId: String(row.reviewer_user_id) }),
    createdAt: new Date(String(row.created_at)),
    updatedAt: new Date(String(row.updated_at)),
  };
}

function scheduledFromRow(row: Record<string, unknown>): ScheduledPublicationRecord {
  return {
    id: String(row.id),
    workspaceId: String(row.workspace_id),
    draftId: String(row.draft_id),
    connectionId: String(row.connection_id),
    publishAt: new Date(String(row.publish_at)),
    idempotencyKey: String(row.idempotency_key),
    generation: String(row.generation),
    state: row.state as ScheduledPublicationRecord["state"],
  };
}