import { publishingIdempotencyKey } from "@tactical-media-broadcast-hub/core";
import { canonicalPostHash } from "./approval.js";
import type { Pool, PoolClient } from "pg";

/**
 * WP3 execution spine: the tenant-scoped claim / dispatch-gate / receipt /
 * finalization repository behind the scheduler's publish activity (plan §4).
 *
 * PostgreSQL owns the fences. Every method enters the tenant through the same
 * transaction-local `set_config('app.workspace_id', …, true)` as withTenant,
 * so RLS binds as sp_app. Mutations lock in one order — workspace row, draft,
 * schedule, attempt, then the audit chain — matching team/group/credential
 * mutations, which take the workspace row FOR UPDATE before domain rows.
 *
 * Bigint columns cross this boundary as decimal strings (pg's int8 default).
 */

export const PUBLISH_ERROR_CODES = [
  "AUTHZ_REVOKED",
  "CREDENTIAL_CHANGED",
  "SUPERSEDED",
  "CANCELLED",
  "PUBLISH_OUTCOME_UNKNOWN",
  "RETRY_EXHAUSTED",
  "EXECUTION_FENCE_MISSING",
  "LEGACY_AUTHORIZATION_MISSING",
] as const;
export type PublishErrorCode = (typeof PUBLISH_ERROR_CODES)[number] | (string & {});

/**
 * Provider-compatible ceiling on attachments per post (the API's
 * media.service MAX_MEDIA_PER_POST; Mastodon's connector MAX_MEDIA). The API
 * validated it at intake; the repository re-checks the approved snapshot so a
 * forged snapshot can never reach a connector.
 */
export const MAX_MEDIA_ATTACHMENTS = 4;

/** Roles that may use every connection without group sharing (groups.service WORKSPACE_WIDE_ROLES). */
const WORKSPACE_WIDE_ROLES = ["Owner", "Admin"];
const DEFAULT_LEASE_SECONDS = 90;
export const MAX_DISPATCHES = 5;
const SCHEDULER_ROLE = "sp_scheduler";

/** Identity of one business attempt as carried by the workflow input. */
export interface AttemptRef {
  workspaceId: string;
  scheduleId: string;
  generation: string;
  idempotencyKey: string;
}

/** Proof of current ownership; every journal write CASes owner_token + claim_epoch. */
export interface Lease {
  workspaceId: string;
  attemptId: string;
  scheduleId: string;
  draftId: string;
  idempotencyKey: string;
  generation: string;
  ownerToken: string;
  claimEpoch: string;
  leaseUntil: Date;
}

export interface ExecutionPost {
  text: string;
  /** media_assets ids (0..4). Existence was proven at intake; dispatch resolves them. */
  mediaAssetIds: string[];
  /** Optional alt texts, positionally aligned with mediaAssetIds (null = none). */
  altTexts?: (string | null)[];
  visibility: string;
  providerMetadata: Record<string, unknown>;
}

export interface ClaimedJob {
  lease: Lease;
  attemptNo: string;
  retryCount: number;
  firstDispatchStartedAt?: Date;
  providerRetryDeadline?: Date;
  post: ExecutionPost;
  connection: { id: string; provider: string; channelId: string; fields: Record<string, unknown>; createdAt: Date };
  credential: { id: string; generation: string };
  /** A previous owner stamped a dispatch whose outcome is not settled. */
  priorDispatch: { startedAt: Date; count: number; providerRetryDeadline?: Date } | undefined;
}

export interface StoredPublication {
  id: string;
  workspaceId: string;
  draftId: string;
  connectionId: string;
  idempotencyKey: string;
  remoteId: string;
  remoteUrl?: string;
  publishedAt: Date;
  providerMetadata: Record<string, unknown>;
}

export interface StoredReceipt {
  id: string;
  workspaceId: string;
  attemptId: string;
  remoteId: string;
  remoteUrl?: string;
  publishedAt?: Date;
  providerMetadata: Record<string, unknown>;
  receivedAt: Date;
}

export interface RemoteResult {
  remoteId: string;
  remoteUrl?: string;
  publishedAt?: Date;
  /** Caller-sanitized, allowlisted provider metadata (never tokens or raw bodies). */
  providerMetadata?: Record<string, unknown>;
}

export type ClaimOutcome =
  | { kind: "claimed"; job: ClaimedJob }
  | { kind: "finalize"; job: ClaimedJob; receipt: StoredReceipt }
  | { kind: "done"; publication: StoredPublication }
  | { kind: "wait"; until: Date; reason: "not_due" | "busy" }
  | { kind: "paused"; code: PublishErrorCode }
  | { kind: "superseded"; code: PublishErrorCode }
  | { kind: "failed"; code: PublishErrorCode }
  | { kind: "unknown"; code: PublishErrorCode };

export type GateOutcome<T> =
  | { kind: "sent"; value: T }
  | { kind: "finalize" }
  | { kind: "lost" }
  | { kind: "paused"; code: PublishErrorCode }
  | { kind: "superseded"; code: PublishErrorCode }
  | { kind: "failed"; code: PublishErrorCode }
  | { kind: "unknown"; code: PublishErrorCode };

export type FinalizeOutcome = { kind: "done"; publication: StoredPublication } | { kind: "stale" };
export type SettleOutcome = { kind: "settled"; until?: Date } | { kind: "stale" };

export interface ScheduleAttemptInput {
  workspaceId: string;
  draftId: string;
  connectionId: string;
  scheduledByUserId: string;
  requiredPermission: "publication.schedule" | "publication.publish";
  publishAt: Date;
  expectedPrevious?: { state: "failed" | "needs_reauthorization"; generation: string };
  timezone?: string;
  localPublishAt?: string;
  tzdataVersion?: string;
}

export interface ScheduledAttempt {
  scheduleId: string;
  attemptId: string;
  attemptNo: string;
  generation: string;
  idempotencyKey: string;
}

export interface DueWork {
  evidenceOnly: boolean;
  workspaceId: string;
  scheduleId: string;
  idempotencyKey: string;
  generation: string;
  publishAt: Date;
}

/** Conflicting or missing evidence: operator-visible, never auto-repaired. */
export class PublishingIntegrityError extends Error {
  constructor(readonly code: string, message: string) {
    super(message);
    this.name = "PublishingIntegrityError";
  }
}

/** A scheduling command that the current draft/schedule/credential state refuses. */
export class PublishingStateError extends Error {
  constructor(readonly code: string, message: string) {
    super(message);
    this.name = "PublishingStateError";
  }
}

/** `wp3:<workspaceUUID>:<draftUUID>:a<attemptNo>:publish` — reused verbatim by every replay. */
export function publishIdempotencyKey(workspaceId: string, draftId: string, attemptNo: string): string {
  return publishingIdempotencyKey({ workspaceId, draftId, attemptNo });
}

type Row = Record<string, unknown>;

interface Locked {
  epoch: string;
  connection: Row;
  draft: Row;
  schedule: Row;
  attempt: Row;
}

export interface PublishingRepositoryOptions {
  leaseSeconds?: number;
}

export class PublishingExecutionRepository {
  private readonly leaseSeconds: number;

  constructor(
    private readonly pool: Pool,
    options: PublishingRepositoryOptions = {},
  ) {
    this.leaseSeconds = options.leaseSeconds ?? DEFAULT_LEASE_SECONDS;
  }

  // -------------------------------------------------------------------------
  // Discovery (SECURITY DEFINER, identifiers only)
  // -------------------------------------------------------------------------

  async listDue(limit = 100): Promise<DueWork[]> {
    const res = await this.pool.query("SELECT * FROM publishing_due_schedules($1)", [limit]);
    return (res.rows as Row[]).map((r) => ({
      evidenceOnly: r.evidence_only === true,
      workspaceId: String(r.workspace_id),
      scheduleId: String(r.schedule_id),
      idempotencyKey: String(r.idempotency_key),
      generation: String(r.generation),
      publishAt: new Date(String(r.publish_at)),
    }));
  }

  async resolveWorkspace(idempotencyKey: string): Promise<string | undefined> {
    const res = await this.pool.query("SELECT publishing_schedule_workspace($1) AS workspace_id", [idempotencyKey]);
    const id = (res.rows[0] as Row | undefined)?.workspace_id;
    return id === null || id === undefined ? undefined : String(id);
  }

  async findPublication(workspaceId: string, idempotencyKey: string): Promise<StoredPublication | undefined> {
    return this.inTenant(workspaceId, async (client) => {
      const res = await client.query("SELECT * FROM publications WHERE idempotency_key = $1", [idempotencyKey]);
      const row = res.rows[0] as Row | undefined;
      return row === undefined ? undefined : publicationFromRow(row);
    });
  }

  // -------------------------------------------------------------------------
  // Scheduling: a new business attempt (called inside the service's tenant tx)
  // -------------------------------------------------------------------------

  /**
   * Creates or re-arms the draft's single schedule row for a NEW business
   * attempt: allocates attempt_no from the locked draft counter, snapshots
   * principal / approval proof / exact credential generation, bumps the
   * schedule generation (voiding every older workflow) and journals a ready
   * attempt. Must run in a tenant transaction for `input.workspaceId`.
   */
  async scheduleAttempt(client: PoolClient, input: ScheduleAttemptInput): Promise<ScheduledAttempt> {
    const epoch = await lockWorkspace(client, input.workspaceId);
    const draftRes = await client.query("SELECT * FROM drafts WHERE id = $1 FOR UPDATE", [input.draftId]);
    const draft = draftRes.rows[0] as Row | undefined;
    if (draft === undefined) throw new PublishingStateError("DRAFT_NOT_FOUND", "draft not found");
    if (
      draft.state !== "approved" ||
      draft.approved_revision === null ||
      String(draft.approved_revision) !== String(draft.revision) ||
      draft.approved_hash === null ||
      draft.reviewer_user_id === null
    ) {
      throw new PublishingStateError("DRAFT_NOT_APPROVED", "draft has no current approval");
    }

    const approvedPost = draft.post as ExecutionPost;
    assertApprovedMediaShape(approvedPost);
    if (approvedPost.providerMetadata.connectionId !== input.connectionId || canonicalPostHash(approvedPost) !== draft.approved_hash) {
      throw new PublishingStateError("APPROVAL_MISMATCH", "approved content or target does not match");
    }

    const existingRes = await client.query("SELECT * FROM scheduled_publications WHERE draft_id = $1 FOR UPDATE", [
      input.draftId,
    ]);
    const existing = existingRes.rows[0] as Row | undefined;
    if (input.expectedPrevious && (!existing || existing.state !== input.expectedPrevious.state || String(existing.generation) !== input.expectedPrevious.generation)) throw new PublishingStateError("SCHEDULE_CHANGED", "schedule state or generation changed");
    if (existing !== undefined && ["pending", "in_flight", "done"].includes(String(existing.state))) {
      if (existing.state === "pending" && String(existing.connection_id) === input.connectionId && toDate(existing.publish_at).getTime() === input.publishAt.getTime() && String(existing.scheduled_by_user_id) === input.scheduledByUserId && existing.required_permission === input.requiredPermission && String(existing.post_hash) === String(draft.approved_hash)) {
        const prior = await client.query("SELECT id FROM publishing_attempts WHERE idempotency_key = $1", [existing.idempotency_key]);
        if (prior.rows[0]) return { scheduleId: String(existing.id), attemptId: String(prior.rows[0].id), attemptNo: String(existing.attempt_no), generation: String(existing.generation), idempotencyKey: String(existing.idempotency_key) };
      }
      throw new PublishingStateError("SCHEDULE_ACTIVE", `schedule is ${String(existing.state)}`);
    }
    if (existing !== undefined) {
      const unsettled = await client.query(
        `SELECT 1 FROM publishing_attempts
          WHERE schedule_id = $1
            AND (status IN ('in_progress', 'unknown') OR (status = 'failed' AND error_code = 'PUBLISH_OUTCOME_UNKNOWN') OR EXISTS (SELECT 1 FROM publishing_receipts r WHERE r.attempt_id = publishing_attempts.id AND r.workspace_id = publishing_attempts.workspace_id))`,
        [existing.id],
      );
      if ((unsettled.rowCount ?? 0) > 0) {
        throw new PublishingStateError("PUBLISH_OUTCOME_UNKNOWN", "a previous attempt has an unsettled outcome");
      }
    }

    const credRes = await client.query(
      `SELECT cr.id, cr.generation FROM connections c
        JOIN credentials cr ON cr.connection_id = c.id AND cr.kind = 'connector-oauth' AND cr.revoked_at IS NULL
       WHERE c.id = $1 AND c.disconnected_at IS NULL`,
      [input.connectionId],
    );
    const credential = credRes.rows[0] as Row | undefined;
    if (credential === undefined) throw new PublishingStateError("CREDENTIAL_CHANGED", "connection has no active credential");

    const authz = await authorize(client, {
      workspaceId: input.workspaceId,
      principal: input.scheduledByUserId,
      permission: input.requiredPermission,
      reviewer: String(draft.reviewer_user_id),
      connectionId: input.connectionId,
      credentialId: String(credential.id),
      credentialGeneration: String(credential.generation),
    });
    if (authz !== undefined) throw new PublishingStateError(authz, "scheduling principal is not authorized");

    const counter = await client.query(
      "UPDATE drafts SET publication_attempt = publication_attempt + 1 WHERE id = $1 RETURNING publication_attempt",
      [input.draftId],
    );
    const attemptNo = String((counter.rows[0] as Row).publication_attempt);
    const key = publishIdempotencyKey(input.workspaceId, input.draftId, attemptNo);
    const fence = [
      input.connectionId,
      input.publishAt,
      key,
      input.scheduledByUserId,
      input.requiredPermission,
      epoch,
      String(draft.approved_revision),
      String(draft.approved_hash),
      String(credential.id),
      String(credential.generation),
      attemptNo,
      input.timezone ?? "UTC",
      input.localPublishAt ?? null,
      input.tzdataVersion ?? null,
    ];

    let schedule: Row;
    if (existing === undefined) {
      const ins = await client.query(
        `INSERT INTO scheduled_publications
           (workspace_id, draft_id, connection_id, publish_at, idempotency_key, scheduled_by_user_id,
            required_permission, authorized_epoch, approved_revision, post_hash, credential_id,
            credential_generation, attempt_no, timezone, local_publish_at, tzdata_version,
            generation, execution_version, state, next_attempt_at)
         VALUES ($15, $16, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, 1, 1, 'pending', $2)
         RETURNING id, generation`,
        [...fence, input.workspaceId, input.draftId],
      );
      schedule = ins.rows[0] as Row;
    } else {
      const upd = await client.query(
        `UPDATE scheduled_publications SET
           connection_id = $1, publish_at = $2, idempotency_key = $3, scheduled_by_user_id = $4,
           required_permission = $5, authorized_epoch = $6, approved_revision = $7, post_hash = $8,
           credential_id = $9, credential_generation = $10, attempt_no = $11, timezone = $12,
           local_publish_at = $13, tzdata_version = $14, generation = generation + 1,
           execution_version = 1, state = 'pending', next_attempt_at = $2, retry_count = 0,
           error = NULL, error_code = NULL
         WHERE id = $15
         RETURNING id, generation`,
        [...fence, existing.id],
      );
      schedule = upd.rows[0] as Row;
    }

    const attempt = await client.query(
      `INSERT INTO publishing_attempts
         (workspace_id, schedule_id, draft_id, connection_id, attempt_no, idempotency_key,
          schedule_generation, post_snapshot, post_hash, scheduled_by_user_id, required_permission,
          approved_revision, reviewer_user_id, credential_id, credential_generation, authorized_epoch)
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
       RETURNING id`,
      [
        input.workspaceId,
        schedule.id,
        input.draftId,
        input.connectionId,
        attemptNo,
        key,
        schedule.generation,
        draft.post,
        draft.approved_hash,
        input.scheduledByUserId,
        input.requiredPermission,
        draft.approved_revision,
        draft.reviewer_user_id,
        credential.id,
        credential.generation,
        epoch,
      ],
    );
    return {
      scheduleId: String(schedule.id),
      attemptId: String((attempt.rows[0] as Row).id),
      attemptNo,
      generation: String(schedule.generation),
      idempotencyKey: key,
    };
  }

  /** Cancellation retains journal and evidence; only undispatched intent can be cancelled. */
  async cancelAttempt(client: PoolClient, workspaceId: string, draftId: string): Promise<void> {
    await lockWorkspace(client, workspaceId);
    await client.query("SELECT id FROM drafts WHERE id = $1 FOR UPDATE", [draftId]);
    const schedules = await client.query("SELECT * FROM scheduled_publications WHERE draft_id = $1 FOR UPDATE", [draftId]);
    const schedule = schedules.rows[0] as Row | undefined;
    if (!schedule) throw new PublishingStateError("SCHEDULE_NOT_FOUND", "schedule not found");
    const attempts = await client.query("SELECT * FROM publishing_attempts WHERE schedule_id = $1 ORDER BY attempt_no FOR UPDATE", [schedule.id]);
    if ((attempts.rows as Row[]).some((a) => a.status === "unknown" || (a.dispatch_started_at !== null && a.status !== "completed" && a.status !== "failed"))) throw new PublishingStateError("PUBLISH_OUTCOME_UNKNOWN", "dispatch is unsettled");
    if (!["pending", "in_flight", "needs_reauthorization", "cancelled"].includes(String(schedule.state))) throw new PublishingStateError("SCHEDULE_ACTIVE", "schedule cannot be cancelled");
    await client.query("UPDATE publishing_attempts SET status = 'abandoned', owner_token = NULL, lease_until = NULL, error_code = 'CANCELLED' WHERE schedule_id = $1 AND status IN ('ready', 'in_progress')", [schedule.id]);
    await client.query("UPDATE scheduled_publications SET state = 'cancelled', generation = generation + 1, error_code = 'CANCELLED' WHERE id = $1 AND state <> 'cancelled'", [schedule.id]);
  }

  // -------------------------------------------------------------------------
  // Claim
  // -------------------------------------------------------------------------

  /**
   * Atomic claim under workspace FOR UPDATE + draft/schedule/attempt row locks.
   * A live owner yields `wait`; an expired lease is recoverable by exactly one
   * claimer (owner_token change advances claim_epoch in the DB). Existing
   * evidence short-circuits to `done` / `finalize` — never a second send.
   */
  async claim(ref: AttemptRef): Promise<ClaimOutcome> {
    return this.inTenant(ref.workspaceId, async (client) => {
      const epoch = await lockWorkspace(client, ref.workspaceId);
      const pub = await client.query("SELECT * FROM publications WHERE idempotency_key = $1", [ref.idempotencyKey]);
      if (pub.rows[0] !== undefined) return { kind: "done", publication: publicationFromRow(pub.rows[0] as Row) };

      const locked = await lockAttemptByKey(client, ref.idempotencyKey, epoch);
      if (locked === undefined) return { kind: "failed", code: "EXECUTION_FENCE_MISSING" };
      const { draft, schedule, attempt } = locked;
      if (String(schedule.id) !== ref.scheduleId) return { kind: "failed", code: "EXECUTION_FENCE_MISSING" };

      const now = await dbNow(client);
      const status = String(attempt.status);
      if (status === "completed") {
        throw new PublishingIntegrityError("PUBLICATION_MISSING", `attempt ${String(attempt.id)} completed without a publication`);
      }

      // Durable evidence wins over every other fence: finalize, never resend.
      const receipt = await readReceipt(client, String(attempt.id));
      if (receipt !== undefined) {
        if (status === "abandoned") {
          throw new PublishingIntegrityError("RECEIPT_ON_ABANDONED_ATTEMPT", `attempt ${String(attempt.id)} is abandoned but has a receipt`);
        }
        if (status === "in_progress" && toDate(attempt.lease_until) > now) {
          return { kind: "wait", until: toDate(attempt.lease_until), reason: "busy" };
        }
        const lease = await this.takeOwnership(client, locked, epoch);
        return { kind: "finalize", job: jobFromLocked(locked, lease), receipt };
      }

      // A ref whose generation disagrees with its own key's attempt is not
      // this attempt's workflow: refuse it without touching the real attempt.
      if (String(attempt.schedule_generation) !== ref.generation) return { kind: "superseded", code: "SUPERSEDED" };
      if (
        String(schedule.generation) !== String(attempt.schedule_generation) ||
        String(schedule.idempotency_key) !== String(attempt.idempotency_key)
      ) {
        return supersede(client, locked, "SUPERSEDED");
      }
      if (Number(schedule.execution_version) !== 1) return { kind: "failed", code: "EXECUTION_FENCE_MISSING" };

      switch (status) {
        case "abandoned":
          return { kind: "superseded", code: String(attempt.error_code ?? "SUPERSEDED") };
        case "failed":
          return { kind: "failed", code: String(attempt.error_code ?? schedule.error_code ?? "FAILED") };
        case "unknown":
          return { kind: "unknown", code: "PUBLISH_OUTCOME_UNKNOWN" };
        case "ready": {
          const scheduleState = String(schedule.state);
          if (scheduleState === "cancelled") return supersede(client, locked, "CANCELLED");
          if (scheduleState === "needs_reauthorization") return { kind: "paused", code: String(schedule.error_code) };
          if (scheduleState !== "pending") return { kind: "failed", code: String(schedule.error_code ?? "FAILED") };
          const dueAt = maxDate(toDate(schedule.next_attempt_at), toDate(schedule.publish_at));
          if (dueAt > now) return { kind: "wait", until: dueAt, reason: "not_due" };
          if (draft.state !== "approved") return supersede(client, locked, "SUPERSEDED");
          break;
        }
        case "in_progress": {
          if (toDate(attempt.lease_until) > now) {
            return { kind: "wait", until: toDate(attempt.lease_until), reason: "busy" };
          }
          // Recovery: only this exact in-flight schedule's scheduled draft.
          if (schedule.state !== "in_flight" || draft.state !== "scheduled") {
            throw new PublishingIntegrityError("RECOVERY_STATE", `attempt ${String(attempt.id)} in progress outside an in-flight schedule`);
          }
          break;
        }
        default:
          throw new PublishingIntegrityError("ATTEMPT_STATUS", `unexpected attempt status ${status}`);
      }

      if (!contentFenceHolds(draft, attempt)) return supersede(client, locked, "SUPERSEDED");

      const denied = await authorize(client, authzInput(attempt));
      if (denied !== undefined) {
        if (attempt.dispatch_started_at !== null) {
          await settleUnknown(client, locked, "PUBLISH_OUTCOME_UNKNOWN");
          return { kind: "unknown", code: "PUBLISH_OUTCOME_UNKNOWN" };
        }
        await pause(client, locked, denied);
        return { kind: "paused", code: denied };
      }

      const lease = await this.takeOwnership(client, locked, epoch);
      return { kind: "claimed", job: jobFromLocked(locked, lease) };
    });
  }

  /** Renews a live lease; false once the lease expired or ownership moved on. */
  async heartbeat(lease: Lease): Promise<boolean> {
    return this.inTenant(lease.workspaceId, async (client) => {
      const res = await client.query(
        `UPDATE publishing_attempts SET lease_until = now() + make_interval(secs => $4::double precision)
          WHERE id = $1 AND owner_token = $2 AND claim_epoch = $3
            AND status = 'in_progress' AND lease_until > now()
          RETURNING lease_until`,
        [lease.attemptId, lease.ownerToken, lease.claimEpoch, this.leaseSeconds],
      );
      const row = res.rows[0] as Row | undefined;
      if (row === undefined) return false;
      lease.leaseUntil = toDate(row.lease_until);
      return true;
    });
  }

  // -------------------------------------------------------------------------
  // Dispatch gate
  // -------------------------------------------------------------------------

  /**
   * Holds the workspace row FOR SHARE in a dedicated READ COMMITTED tenant
   * transaction across `send`, so any approval/authz/credential mutation
   * (which takes workspace FOR UPDATE first) either committed before the
   * fence re-read — and blocks the send — or serializes after it. The dispatch
   * stamp commits in a separate short transaction (owner/claim_epoch CAS,
   * dispatch_count < MAX) so receipts and heartbeats never wait on the gate.
   *
   * `allowRedispatch` with `dedupeWindowMs` is the connector's declared
   * token-dedupe policy (e.g. verified Mastodon origins). Without it an
   * unsettled earlier dispatch becomes PUBLISH_OUTCOME_UNKNOWN, never a resend.
   */
  async withDispatchGate<T>(
    lease: Lease,
    policy: { allowRedispatch?: boolean; dedupeWindowMs?: number },
    send: () => Promise<T>,
  ): Promise<GateOutcome<T>> {
    const client = await this.pool.connect();
    let open = false;
    const end = async (sql: "COMMIT" | "ROLLBACK"): Promise<void> => {
      if (!open) return;
      open = false;
      await client.query(sql).catch(() => undefined);
    };
    try {
      await client.query("BEGIN ISOLATION LEVEL READ COMMITTED");
      open = true;
      await client.query("SELECT set_config('app.workspace_id', $1, true)", [lease.workspaceId]);
      await client.query("SELECT 1 FROM workspaces WHERE id = $1 FOR SHARE", [lease.workspaceId]);

      const attemptRes = await client.query(
        `SELECT a.*, clock_timestamp() AS db_now FROM publishing_attempts a WHERE a.id = $1`,
        [lease.attemptId],
      );
      const attempt = attemptRes.rows[0] as Row | undefined;
      if (
        attempt === undefined ||
        attempt.status !== "in_progress" ||
        attempt.owner_token !== lease.ownerToken ||
        String(attempt.claim_epoch) !== lease.claimEpoch ||
        toDate(attempt.lease_until) <= toDate(attempt.db_now)
      ) {
        await end("ROLLBACK");
        return { kind: "lost" };
      }
      if ((await readReceipt(client, lease.attemptId)) !== undefined) {
        await end("ROLLBACK");
        return { kind: "finalize" };
      }

      const scheduleRes = await client.query("SELECT * FROM scheduled_publications WHERE id = $1", [attempt.schedule_id]);
      const draftRes = await client.query("SELECT * FROM drafts WHERE id = $1", [attempt.draft_id]);
      const schedule = scheduleRes.rows[0] as Row;
      const draft = draftRes.rows[0] as Row;
      const fenced =
        schedule.state === "in_flight" &&
        String(schedule.generation) === String(attempt.schedule_generation) &&
        String(schedule.idempotency_key) === String(attempt.idempotency_key) &&
        contentFenceHolds(draft, attempt);
      const denied = fenced ? await authorize(client, authzInput(attempt)) : undefined;
      const priorDispatch = attempt.dispatch_started_at !== null;
      const redispatchAllowed =
        policy.allowRedispatch === true &&
        attempt.provider_retry_deadline !== null &&
        toDate(attempt.provider_retry_deadline) > toDate(attempt.db_now);
      const exhausted = Number(attempt.dispatch_count) >= MAX_DISPATCHES;

      if (!fenced || denied !== undefined || (priorDispatch && !redispatchAllowed) || exhausted) {
        await end("ROLLBACK");
        if (priorDispatch) {
          const settled = await this.settle(lease, (c, l) => settleUnknown(c, l, "PUBLISH_OUTCOME_UNKNOWN"));
          return settled.kind === "stale" ? { kind: "lost" } : { kind: "unknown", code: "PUBLISH_OUTCOME_UNKNOWN" };
        }
        if (!fenced) {
          const settled = await this.settle(lease, (c, l) => supersede(c, l, "SUPERSEDED").then(() => undefined));
          return settled.kind === "stale" ? { kind: "lost" } : { kind: "superseded", code: "SUPERSEDED" };
        }
        if (denied !== undefined) {
          const settled = await this.settle(lease, (c, l) => pause(c, l, denied));
          return settled.kind === "stale" ? { kind: "lost" } : { kind: "paused", code: denied };
        }
        const settled = await this.settle(lease, (c, l) => settleFailed(c, l, "RETRY_EXHAUSTED"));
        return settled.kind === "stale" ? { kind: "lost" } : { kind: "failed", code: "RETRY_EXHAUSTED" };
      }

      const windowSeconds = Math.max(0, (policy.dedupeWindowMs ?? 0) / 1000);
      const stamped = await this.pool.connect().then(async (stampClient) => {
        try {
          await stampClient.query("BEGIN");
          await stampClient.query("SELECT set_config('app.workspace_id', $1, true)", [lease.workspaceId]);
          const res = await stampClient.query(
            `UPDATE publishing_attempts SET
               dispatch_started_at = clock_timestamp(),
               first_dispatch_started_at = COALESCE(first_dispatch_started_at, clock_timestamp()),
               dispatch_count = dispatch_count + 1,
               provider_retry_deadline = COALESCE(provider_retry_deadline,
                 CASE WHEN $4::double precision > 0 THEN clock_timestamp() + make_interval(secs => $4::double precision) END)
             WHERE id = $1 AND owner_token = $2 AND claim_epoch = $3 AND status = 'in_progress'
               AND dispatch_count < ${MAX_DISPATCHES} AND lease_until > clock_timestamp()`,
            [lease.attemptId, lease.ownerToken, lease.claimEpoch, windowSeconds],
          );
          await stampClient.query("COMMIT");
          return (res.rowCount ?? 0) === 1;
        } catch (err) {
          await stampClient.query("ROLLBACK").catch(() => undefined);
          throw err;
        } finally {
          stampClient.release();
        }
      });
      if (!stamped) {
        await end("ROLLBACK");
        return { kind: "lost" };
      }
      try {
        return { kind: "sent", value: await send() };
      } finally {
        await end("COMMIT");
      }
    } finally {
      await end("ROLLBACK");
      client.release();
    }
  }

  // -------------------------------------------------------------------------
  // Receipt + finalization
  // -------------------------------------------------------------------------

  /**
   * Append-only provider evidence in its own transaction. Any (even deposed)
   * owner may record matching evidence; a different remote id for the same
   * attempt is an integrity failure, never an overwrite.
   */
  async recordReceipt(
    ref: { workspaceId: string; attemptId: string },
    remote: RemoteResult,
  ): Promise<{ receipt: StoredReceipt; inserted: boolean }> {
    const remoteId = remote.remoteId.trim();
    if (remoteId === "" || remoteId.length > 512) {
      throw new PublishingIntegrityError("RECEIPT_SHAPE", "provider returned an invalid remote id");
    }
    const metadata = remote.providerMetadata ?? {};
    if (typeof metadata !== "object" || metadata === null || Array.isArray(metadata)) {
      throw new PublishingIntegrityError("RECEIPT_SHAPE", "provider metadata must be an object");
    }
    return this.inTenant(ref.workspaceId, async (client) => {
      const attempt = await client.query("SELECT dispatch_count FROM publishing_attempts WHERE id = $1", [ref.attemptId]);
      const row = attempt.rows[0] as Row | undefined;
      if (row === undefined) throw new PublishingIntegrityError("ATTEMPT_MISSING", "receipt for an unknown attempt");
      if (Number(row.dispatch_count) === 0) {
        throw new PublishingIntegrityError("RECEIPT_WITHOUT_DISPATCH", "receipt for an attempt that was never dispatched");
      }
      const ins = await client.query(
        `INSERT INTO publishing_receipts (workspace_id, attempt_id, remote_id, remote_url, published_at, provider_metadata)
         VALUES ($1, $2, $3, $4, $5, $6)
         ON CONFLICT (workspace_id, attempt_id) DO NOTHING
         RETURNING *`,
        [ref.workspaceId, ref.attemptId, remoteId, remote.remoteUrl ?? null, remote.publishedAt ?? null, metadata],
      );
      if (ins.rows[0] !== undefined) return { receipt: receiptFromRow(ins.rows[0] as Row), inserted: true };
      const existing = await readReceipt(client, ref.attemptId);
      if (existing === undefined) throw new PublishingIntegrityError("RECEIPT_MISSING", "receipt conflict without a row");
      if (existing.remoteId !== remoteId) {
        throw new PublishingIntegrityError("RECEIPT_CONFLICT", `attempt ${ref.attemptId} already has different remote evidence`);
      }
      return { receipt: existing, inserted: false };
    });
  }

  /**
   * Current owner turns the receipt into exactly one canonical Publication:
   * CAS owner/epoch, same generation/key, insert ON CONFLICT(key) then verify
   * identity, mark done/completed/published, append audit last. A deposed
   * owner gets `stale` and must not report success; its receipt stays for
   * the new owner.
   */
  async finalize(lease: Lease): Promise<FinalizeOutcome> {
    return this.inTenant(lease.workspaceId, async (client) => {
      const epoch = await lockWorkspace(client, lease.workspaceId);
      const locked = await lockAttemptByKey(client, lease.idempotencyKey, epoch);
      if (locked === undefined || !ownsLease(locked.attempt, lease)) return { kind: "stale" };
      const { draft, schedule, attempt } = locked;
      if (
        String(schedule.generation) !== String(attempt.schedule_generation) ||
        String(schedule.idempotency_key) !== String(attempt.idempotency_key) ||
        String(draft.revision) !== String(attempt.approved_revision) ||
        canonicalPostHash(draft.post as ExecutionPost) !== attempt.post_hash ||
        canonicalPostHash(attempt.post_snapshot as ExecutionPost) !== attempt.post_hash
      ) {
        throw new PublishingIntegrityError("FINALIZE_FENCE", `attempt ${lease.attemptId} fences changed while dispatched`);
      }
      const receipt = await readReceipt(client, lease.attemptId);
      if (receipt === undefined) throw new PublishingIntegrityError("RECEIPT_MISSING", "finalize without a durable receipt");

      await client.query(
        `INSERT INTO publications
           (workspace_id, draft_id, connection_id, idempotency_key, remote_id, remote_url, published_at, provider_metadata)
         VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7::timestamptz, $8::timestamptz), $9)
         ON CONFLICT (idempotency_key) DO NOTHING`,
        [
          lease.workspaceId,
          attempt.draft_id,
          attempt.connection_id,
          attempt.idempotency_key,
          receipt.remoteId,
          receipt.remoteUrl ?? null,
          receipt.publishedAt ?? null,
          receipt.receivedAt,
          receipt.providerMetadata,
        ],
      );
      const pubRes = await client.query("SELECT * FROM publications WHERE idempotency_key = $1", [attempt.idempotency_key]);
      const publication = publicationFromRow(pubRes.rows[0] as Row);
      if (
        publication.remoteId !== receipt.remoteId ||
        publication.draftId !== String(attempt.draft_id) ||
        publication.connectionId !== String(attempt.connection_id)
      ) {
        throw new PublishingIntegrityError("PUBLICATION_CONFLICT", `publication for ${String(attempt.idempotency_key)} disagrees with its receipt`);
      }

      await client.query(
        `UPDATE publishing_attempts SET status = 'completed', owner_token = NULL, lease_until = NULL, error_code = NULL
          WHERE id = $1`,
        [attempt.id],
      );
      await client.query(
        "UPDATE scheduled_publications SET state = 'done', error = NULL, error_code = NULL WHERE id = $1",
        [schedule.id],
      );
      await client.query(
        "UPDATE drafts SET state = 'published', updated_at = now() WHERE id = $1 AND state IN ('scheduled', 'approved')",
        [draft.id],
      );
      const authorizationChanged = await authorize(client, authzInput(attempt));
      await audit(client, lease.workspaceId, "publication.published", "publication", publication.id, {
        ...(authorizationChanged ? { authorizationChanged } : {}),
        draftId: String(draft.id),
        attemptNo: String(attempt.attempt_no),
        remoteId: publication.remoteId,
      });
      return { kind: "done", publication };
    });
  }

  // -------------------------------------------------------------------------
  // Settlements (owner-CAS'd)
  // -------------------------------------------------------------------------

  /**
   * Definite provider non-delivery (e.g. 429 with retry_after): same attempt
   * and key, dispatch marker cleared, schedule pending at `notBefore`.
   */
  async retryLater(lease: Lease, code: string, notBefore: Date): Promise<SettleOutcome> {
    return this.retry(lease, code, notBefore, false);
  }

  /** Ambiguous deduplicated retry retains the original evidence and deadline. */
  async retryDeduplicated(lease: Lease, code: string, notBefore: Date): Promise<SettleOutcome> {
    return this.retry(lease, code, notBefore, true);
  }

  async pauseForReauthorization(lease: Lease, code = "CREDENTIAL_REJECTED"): Promise<SettleOutcome> {
    return this.settle(lease, (client, locked) => pause(client, locked, code));
  }

  private async retry(lease: Lease, code: string, notBefore: Date, ambiguous: boolean): Promise<SettleOutcome> {
    return this.settle(lease, async (client, locked) => {
      if (Number(locked.attempt.dispatch_count) >= MAX_DISPATCHES || Number(locked.schedule.retry_count) >= MAX_DISPATCHES - 1) {
        if (ambiguous) await settleUnknown(client, locked, "PUBLISH_OUTCOME_UNKNOWN");
        else await settleFailed(client, locked, "RETRY_EXHAUSTED");
        return undefined;
      }
      await client.query(
        `UPDATE publishing_attempts SET status = 'ready', owner_token = NULL, lease_until = NULL,
                dispatch_started_at = CASE WHEN $3::boolean THEN dispatch_started_at ELSE NULL END, error_code = $2
          WHERE id = $1`,
        [locked.attempt.id, code, ambiguous],
      );
      const res = await client.query(
        `UPDATE scheduled_publications SET state = 'pending', next_attempt_at = GREATEST($2::timestamptz, now()),
                retry_count = retry_count + 1, error_code = $3
          WHERE id = $1 RETURNING next_attempt_at`,
        [locked.schedule.id, notBefore, code],
      );
      await client.query("UPDATE drafts SET state = 'approved', updated_at = now() WHERE id = $1 AND state = 'scheduled'", [
        locked.draft.id,
      ]);
      return toDate((res.rows[0] as Row).next_attempt_at);
    });
  }

  /** Permanent provider rejection or exhausted safe retries (known no-send). */
  async fail(lease: Lease, code: string): Promise<SettleOutcome> {
    return this.settle(lease, (client, locked) => settleFailed(client, locked, code));
  }

  /** Send may have happened and cannot be reconciled: lock, never resend. */
  async markUnknown(lease: Lease, evidence?: { remoteId: string; scheduledAt: string }): Promise<SettleOutcome> {
    return this.settle(lease, async (client, locked) => {
      if (evidence) await audit(client, lease.workspaceId, "publication.unexpected_remote_schedule", "draft", lease.draftId, { remoteId: evidence.remoteId.slice(0, 512), scheduledAt: evidence.scheduledAt });
      await settleUnknown(client, locked, "PUBLISH_OUTCOME_UNKNOWN");
    });
  }

  private async settle(
    lease: Lease,
    fn: (client: PoolClient, locked: Locked) => Promise<Date | undefined | void>,
  ): Promise<SettleOutcome> {
    return this.inTenant(lease.workspaceId, async (client) => {
      const epoch = await lockWorkspace(client, lease.workspaceId);
      const locked = await lockAttemptByKey(client, lease.idempotencyKey, epoch);
      if (locked === undefined || !ownsLease(locked.attempt, lease)) return { kind: "stale" } as const;
      const until = await fn(client, locked);
      return until instanceof Date ? { kind: "settled", until } : { kind: "settled" };
    });
  }

  private async takeOwnership(client: PoolClient, locked: Locked, epoch: string): Promise<Lease> {
    const { attempt, schedule, draft } = locked;
    const res = await client.query(
      `UPDATE publishing_attempts SET status = 'in_progress', owner_token = gen_random_uuid(),
              lease_until = now() + make_interval(secs => $3::double precision), authorized_epoch = $4, error_code = NULL
        WHERE id = $1 AND claim_epoch = $2
        RETURNING owner_token, claim_epoch, lease_until`,
      [attempt.id, attempt.claim_epoch, this.leaseSeconds, epoch],
    );
    const row = res.rows[0] as Row | undefined;
    if (row === undefined) throw new PublishingIntegrityError("CLAIM_CAS", "claim lost its compare-and-set under row lock");
    await client.query(
      "UPDATE scheduled_publications SET state = 'in_flight', authorized_epoch = $2, error = NULL, error_code = NULL WHERE id = $1",
      [schedule.id, epoch],
    );
    await client.query("UPDATE drafts SET state = 'scheduled', updated_at = now() WHERE id = $1 AND state = 'approved'", [
      draft.id,
    ]);
    return {
      workspaceId: String(attempt.workspace_id),
      attemptId: String(attempt.id),
      scheduleId: String(schedule.id),
      draftId: String(draft.id),
      idempotencyKey: String(attempt.idempotency_key),
      generation: String(attempt.schedule_generation),
      ownerToken: String(row.owner_token),
      claimEpoch: String(row.claim_epoch),
      leaseUntil: toDate(row.lease_until),
    };
  }

  private async inTenant<T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>): Promise<T> {
    const client = await this.pool.connect();
    try {
      await client.query("BEGIN");
      await client.query("SELECT set_config('app.workspace_id', $1, true)", [workspaceId]);
      const result = await fn(client);
      await client.query("COMMIT");
      return result;
    } catch (err) {
      await client.query("ROLLBACK").catch(() => undefined);
      throw err;
    } finally {
      client.release();
    }
  }
}

// ---------------------------------------------------------------------------
// Tenant-transaction helpers (caller holds the tenant context)
// ---------------------------------------------------------------------------

async function lockWorkspace(client: PoolClient, workspaceId: string): Promise<string> {
  const res = await client.query("SELECT authz_epoch FROM workspaces WHERE id = $1 FOR UPDATE", [workspaceId]);
  const row = res.rows[0] as Row | undefined;
  if (row === undefined) throw new PublishingIntegrityError("WORKSPACE_MISSING", "workspace not found");
  return String(row.authz_epoch);
}

/** Locks draft -> schedule -> attempt for one key (workspace already locked). */
async function lockAttemptByKey(client: PoolClient, idempotencyKey: string, epoch: string): Promise<Locked | undefined> {
  const ids = await client.query("SELECT id, schedule_id, draft_id FROM publishing_attempts WHERE idempotency_key = $1", [
    idempotencyKey,
  ]);
  const ref = ids.rows[0] as Row | undefined;
  if (ref === undefined) return undefined;
  const draft = (await client.query("SELECT * FROM drafts WHERE id = $1 FOR UPDATE", [ref.draft_id])).rows[0] as Row;
  const schedule = (await client.query("SELECT * FROM scheduled_publications WHERE id = $1 FOR UPDATE", [ref.schedule_id]))
    .rows[0] as Row;
  const attempt = (await client.query("SELECT *, clock_timestamp() AS db_now FROM publishing_attempts WHERE id = $1 FOR UPDATE", [ref.id])).rows[0] as Row;
  const connection = (
    await client.query("SELECT id, provider, channel_id, fields, created_at FROM connections WHERE id = $1", [
      attempt.connection_id,
    ])
  ).rows[0] as Row;
  return { epoch, connection, draft, schedule, attempt };
}

function ownsLease(attempt: Row, lease: Lease): boolean {
  return (
    attempt.status === "in_progress" &&
    attempt.owner_token === lease.ownerToken &&
    String(attempt.claim_epoch) === lease.claimEpoch &&
    toDate(attempt.lease_until) > toDate(attempt.db_now)
  );
}

/**
 * The approved snapshot may carry media, but its shape must still be exactly
 * what the API admits at intake: 0..4 asset ids plus (optionally) an
 * index-aligned list of alt texts. Asset EXISTENCE is not re-proven here — the
 * api validated it against media_assets inside the workspace at intake, and
 * the approved snapshot is immutable afterwards; the dispatch-time resolver
 * re-derives the row (workspace-scoped) when it loads bytes. Anything else
 * (wrong types, misaligned alt texts, remote media ids smuggled into
 * providerMetadata) is a forged snapshot and refuses to schedule.
 */
function assertApprovedMediaShape(post: ExecutionPost): void {
  const ids = post.mediaAssetIds;
  if (!Array.isArray(ids) || !ids.every((id) => typeof id === "string" && id.length > 0)) {
    throw new PublishingStateError("UNSUPPORTED_MEDIA", "post.mediaAssetIds must be an array of non-empty strings");
  }
  if (ids.length > MAX_MEDIA_ATTACHMENTS) {
    throw new PublishingStateError(
      "UNSUPPORTED_MEDIA",
      `post.mediaAssetIds supports at most ${MAX_MEDIA_ATTACHMENTS} attachments`,
    );
  }
  const altTexts = post.altTexts;
  if (altTexts !== undefined) {
    if (!Array.isArray(altTexts) || !altTexts.every((entry) => entry === null || typeof entry === "string")) {
      throw new PublishingStateError("UNSUPPORTED_MEDIA", "post.altTexts must be an array of strings or nulls");
    }
    if (altTexts.length !== ids.length) {
      throw new PublishingStateError("UNSUPPORTED_MEDIA", "post.altTexts must align with post.mediaAssetIds (same length)");
    }
  }
  if (Object.prototype.hasOwnProperty.call(post.providerMetadata, "mediaIds")) {
    // Remote media ids bypass asset ownership; the runtime refuses them at
    // dispatch too (idempotency.FORBIDDEN_POST_METADATA_KEYS).
    throw new PublishingStateError("UNSUPPORTED_MEDIA", "providerMetadata.mediaIds is reserved and may not be supplied by the author");
  }
}

function contentFenceHolds(draft: Row, attempt: Row): boolean {
  return (
    String(draft.revision) === String(attempt.approved_revision) &&
    String(draft.approved_revision) === String(attempt.approved_revision) &&
    draft.approved_hash === attempt.post_hash &&
    canonicalPostHash(draft.post as ExecutionPost) === attempt.post_hash &&
    canonicalPostHash(attempt.post_snapshot as ExecutionPost) === attempt.post_hash &&
    draft.reviewer_user_id === attempt.reviewer_user_id &&
    String(draft.id) === String(attempt.draft_id)
  );
}

interface AuthzInput {
  workspaceId: string;
  principal: string;
  permission: string;
  reviewer: string;
  connectionId: string;
  credentialId: string;
  credentialGeneration: string;
}

function authzInput(attempt: Row): AuthzInput {
  return {
    workspaceId: String(attempt.workspace_id),
    principal: String(attempt.scheduled_by_user_id),
    permission: String(attempt.required_permission),
    reviewer: String(attempt.reviewer_user_id),
    connectionId: String(attempt.connection_id),
    credentialId: String(attempt.credential_id),
    credentialGeneration: String(attempt.credential_generation),
  };
}

/**
 * Re-reads every live authorization input: the recorded principal's active
 * membership + role permission + connection access (userCanUseConnection),
 * the approver's live approval.grant, the connection (not a tombstone) and
 * the exact credential row/generation, unrevoked. Never a stored role.
 */
async function authorize(client: PoolClient, input: AuthzInput): Promise<PublishErrorCode | undefined> {
  const res = await client.query(
    `SELECT
       EXISTS (
         SELECT 1 FROM memberships m
           JOIN role_permissions rp ON rp.role = m.role AND rp.permission = $3
          WHERE m.workspace_id = $1 AND m.user_id = $2 AND m.status = 'active'
            AND (m.role = ANY($8::text[])
                 OR EXISTS (SELECT 1 FROM group_connections gc
                              JOIN group_members gm ON gm.group_id = gc.group_id
                             WHERE gc.connection_id = $5 AND gm.user_id = m.user_id))
       ) AS principal_ok,
       EXISTS (
         SELECT 1 FROM memberships m
           JOIN role_permissions rp ON rp.role = m.role AND rp.permission = 'approval.grant'
          WHERE m.workspace_id = $1 AND m.user_id = $4 AND m.status = 'active'
       ) AS approver_ok,
       EXISTS (
         SELECT 1 FROM connections c
           JOIN credentials cr ON cr.connection_id = c.id
          WHERE c.id = $5 AND c.disconnected_at IS NULL
            AND cr.id = $6 AND cr.kind = 'connector-oauth' AND cr.generation = $7 AND cr.revoked_at IS NULL
       ) AS credential_ok`,
    [
      input.workspaceId,
      input.principal,
      input.permission,
      input.reviewer,
      input.connectionId,
      input.credentialId,
      input.credentialGeneration,
      WORKSPACE_WIDE_ROLES,
    ],
  );
  const row = res.rows[0] as { principal_ok: boolean; approver_ok: boolean; credential_ok: boolean };
  if (!row.credential_ok) return "CREDENTIAL_CHANGED";
  if (!row.principal_ok || !row.approver_ok) return "AUTHZ_REVOKED";
  return undefined;
}

/** Durable pause: the attempt is closed; resume is a new explicit attempt. */
async function pause(client: PoolClient, locked: Locked, code: PublishErrorCode): Promise<void> {
  await client.query(
    "UPDATE publishing_attempts SET status = 'abandoned', owner_token = NULL, lease_until = NULL, error_code = $2 WHERE id = $1",
    [locked.attempt.id, code],
  );
  await client.query(
    `UPDATE scheduled_publications SET state = 'needs_reauthorization', error_code = $3
      WHERE id = $1 AND generation = $2`,
    [locked.schedule.id, locked.attempt.schedule_generation, code],
  );
  await client.query("UPDATE drafts SET state = 'approved', updated_at = now() WHERE id = $1 AND state = 'scheduled'", [
    locked.draft.id,
  ]);
  await audit(client, String(locked.attempt.workspace_id), "publication.paused", "draft", String(locked.draft.id), {
    code,
    attemptNo: String(locked.attempt.attempt_no),
  });
}

/**
 * Old generation / changed content / cancelled: abandon the undispatched
 * attempt; invalidate the schedule only if its generation still matches. A
 * newer draft or schedule is never overwritten.
 */
async function supersede(client: PoolClient, locked: Locked, code: PublishErrorCode): Promise<ClaimOutcome> {
  const { attempt, schedule } = locked;
  if (attempt.dispatch_started_at !== null) {
    await settleUnknown(client, locked, "PUBLISH_OUTCOME_UNKNOWN");
    return { kind: "unknown", code: "PUBLISH_OUTCOME_UNKNOWN" };
  }
  if (attempt.status === "ready" || attempt.status === "in_progress") {
    await client.query(
      "UPDATE publishing_attempts SET status = 'abandoned', owner_token = NULL, lease_until = NULL, error_code = $2 WHERE id = $1",
      [attempt.id, code],
    );
  }
  if (String(schedule.generation) === String(attempt.schedule_generation) && ["pending", "in_flight"].includes(String(schedule.state))) {
    await client.query(
      "UPDATE scheduled_publications SET state = 'cancelled', error_code = $2 WHERE id = $1",
      [schedule.id, code],
    );
    await client.query("UPDATE drafts SET state = 'approved', updated_at = now() WHERE id = $1 AND state = 'scheduled'", [
      locked.draft.id,
    ]);
  }
  await audit(client, String(attempt.workspace_id), "publication.superseded", "draft", String(attempt.draft_id), {
    code,
    attemptNo: String(attempt.attempt_no),
  });
  return { kind: "superseded", code };
}

async function settleFailed(client: PoolClient, locked: Locked, code: string): Promise<void> {
  await client.query(
    "UPDATE publishing_attempts SET status = 'failed', owner_token = NULL, lease_until = NULL, error_code = $2 WHERE id = $1",
    [locked.attempt.id, code],
  );
  await client.query("UPDATE scheduled_publications SET state = 'failed', error_code = $2 WHERE id = $1", [
    locked.schedule.id,
    code,
  ]);
  await client.query("UPDATE drafts SET state = 'approved', updated_at = now() WHERE id = $1 AND state = 'scheduled'", [
    locked.draft.id,
  ]);
  await audit(client, String(locked.attempt.workspace_id), "publication.failed", "draft", String(locked.draft.id), {
    code,
    attemptNo: String(locked.attempt.attempt_no),
  });
}

/** Unknown outcome keeps the draft scheduled (locked) and never resends automatically. */
async function settleUnknown(client: PoolClient, locked: Locked, code: string): Promise<void> {
  await client.query(
    "UPDATE publishing_attempts SET status = 'unknown', owner_token = NULL, lease_until = NULL, error_code = $2 WHERE id = $1",
    [locked.attempt.id, code],
  );
  await client.query("UPDATE scheduled_publications SET state = 'failed', error_code = $2 WHERE id = $1", [
    locked.schedule.id,
    code,
  ]);
  await audit(client, String(locked.attempt.workspace_id), "publication.outcome_unknown", "draft", String(locked.draft.id), {
    code,
    attemptNo: String(locked.attempt.attempt_no),
  });
}

async function audit(
  client: PoolClient,
  workspaceId: string,
  action: string,
  targetType: string,
  targetId: string,
  diff: Record<string, unknown>,
): Promise<void> {
  await client.query("SELECT audit_append($1, NULL, $2, $3, $4, $5, $6)", [
    workspaceId,
    SCHEDULER_ROLE,
    action,
    targetType,
    targetId,
    diff,
  ]);
}

async function readReceipt(client: PoolClient, attemptId: string): Promise<StoredReceipt | undefined> {
  const res = await client.query("SELECT * FROM publishing_receipts WHERE attempt_id = $1", [attemptId]);
  const row = res.rows[0] as Row | undefined;
  return row === undefined ? undefined : receiptFromRow(row);
}

async function dbNow(client: PoolClient): Promise<Date> {
  const res = await client.query("SELECT clock_timestamp() AS now");
  return toDate((res.rows[0] as Row).now);
}

// ---------------------------------------------------------------------------
// Row mapping
// ---------------------------------------------------------------------------

function jobFromLocked(locked: Locked, lease: Lease): ClaimedJob {
  return {
    lease,
    retryCount: Number(locked.schedule.retry_count),
    ...(locked.attempt.first_dispatch_started_at == null ? {} : { firstDispatchStartedAt: toDate(locked.attempt.first_dispatch_started_at) }),
    ...(locked.attempt.provider_retry_deadline == null ? {} : { providerRetryDeadline: toDate(locked.attempt.provider_retry_deadline) }),
    attemptNo: String(locked.attempt.attempt_no),
    post: executionPost(locked.attempt.post_snapshot as Row, String(locked.attempt.draft_id), lease.idempotencyKey),
    connection: {
      id: String(locked.connection.id),
      provider: String(locked.connection.provider),
      channelId: String(locked.connection.channel_id),
      fields: (locked.connection.fields ?? {}) as Record<string, unknown>,
      createdAt: toDate(locked.connection.created_at),
    },
    credential: { id: String(locked.attempt.credential_id), generation: String(locked.attempt.credential_generation) },
    priorDispatch:
      locked.attempt.dispatch_started_at === null
        ? undefined
        : {
            startedAt: toDate(locked.attempt.dispatch_started_at),
            count: Number(locked.attempt.dispatch_count),
            ...(locked.attempt.provider_retry_deadline === null
              ? {}
              : { providerRetryDeadline: toDate(locked.attempt.provider_retry_deadline) }),
          },
  };
}

/**
 * The immutable approved snapshot as the connector sees it: pipeline-owned
 * timing (no scheduledAt) and server-injected reserved metadata that a user
 * cannot pre-set.
 */
function executionPost(snapshot: Row, draftId: string, idempotencyKey: string): ExecutionPost {
  const metadata = snapshot.providerMetadata;
  const mediaAssetIds = Array.isArray(snapshot.mediaAssetIds) ? snapshot.mediaAssetIds.map((v) => String(v)) : [];
  const rawAltTexts = snapshot.altTexts;
  const altTexts =
    Array.isArray(rawAltTexts) && rawAltTexts.length === mediaAssetIds.length
      ? rawAltTexts.map((v) => (v === null || v === undefined ? null : typeof v === "string" ? v : null))
      : undefined;
  return {
    text: String(snapshot.text ?? ""),
    mediaAssetIds,
    ...(altTexts === undefined ? {} : { altTexts }),
    visibility: String(snapshot.visibility ?? "provider_default"),
    providerMetadata: {
      ...(typeof metadata === "object" && metadata !== null && !Array.isArray(metadata) ? (metadata as Row) : {}),
      draftId,
      idempotencyKey,
    },
  };
}

function publicationFromRow(row: Row): StoredPublication {
  return {
    id: String(row.id),
    workspaceId: String(row.workspace_id),
    draftId: String(row.draft_id),
    connectionId: String(row.connection_id),
    idempotencyKey: String(row.idempotency_key),
    remoteId: String(row.remote_id),
    ...(row.remote_url === null || row.remote_url === undefined ? {} : { remoteUrl: String(row.remote_url) }),
    publishedAt: toDate(row.published_at),
    providerMetadata: (row.provider_metadata ?? {}) as Record<string, unknown>,
  };
}

function receiptFromRow(row: Row): StoredReceipt {
  return {
    id: String(row.id),
    workspaceId: String(row.workspace_id),
    attemptId: String(row.attempt_id),
    remoteId: String(row.remote_id),
    ...(row.remote_url === null || row.remote_url === undefined ? {} : { remoteUrl: String(row.remote_url) }),
    ...(row.published_at === null || row.published_at === undefined ? {} : { publishedAt: toDate(row.published_at) }),
    providerMetadata: (row.provider_metadata ?? {}) as Record<string, unknown>,
    receivedAt: toDate(row.received_at),
  };
}

function toDate(value: unknown): Date {
  return value instanceof Date ? value : new Date(String(value));
}

function maxDate(a: Date, b: Date): Date {
  return a.getTime() >= b.getTime() ? a : b;
}