T-HUB/packages/connector-runtime/src/publish.ts

/**
 * Safe-replay wrapper around connector.publish (plan §4). It never retries by
 * itself: it injects the attempt's idempotency key, observes whether any
 * mutating request left the process, and classifies the result into a
 * PublishOutcome the caller can act on without resending ambiguous work.
 */
import type { ConnectorConnection, NormalizedPost, Publication } from "@tactical-media-broadcast-hub/core";
import {
  AuthError,
  CapabilityNotSupportedError,
  ConnectorError,
  deliveryCertaintyOf,
  ProviderRejectedError,
  ProviderServerError,
  RateLimitError,
  type HttpRequestEvent,
  type PlatformConnector,
} from "@tactical-media-broadcast-hub/connector-sdk";
import { runWithHttpHook } from "./execution-context.js";
import { ExecutionMetadataError, preparePostForDispatch } from "./idempotency.js";
import type { PublishOutcome, PublishReceipt } from "./outcome.js";
import { publishingPolicyFor, type PublishingPolicy } from "./policy.js";

export interface PublishAttempt {
  idempotencyKey: string;
  /** Defaults to the provider's registered policy. */
  policy?: PublishingPolicy;
  /** First dispatch time of this attempt, from the execution journal. */
  firstDispatchAt?: Date;
  /**
   * An earlier dispatch of this attempt has neither a receipt nor a definite
   * rejection. Only a verified token-dedupe provider inside its window may be
   * called again; otherwise the outcome is unknown with zero sends.
   */
  unsettledPriorDispatch?: boolean;
  /**
   * The deployment verified this origin's token dedupe (e.g. listed in
   * MASTODON_PUBLISH_DEDUPE_ORIGINS). Never derived from post metadata.
   */
  dedupeVerified?: boolean;
  /** Keep post.scheduledAt for native provider scheduling (default false). */
  allowProviderScheduling?: boolean;
  /**
   * Fence check run immediately before each mutating HTTP send, after
   * rate-limiter waits. Throwing blocks that send; the wrapper rethrows the
   * failure as DispatchFenceError.
   */
  beforeMutation?: (event: HttpRequestEvent) => void | Promise<void>;
  aroundMutation?: (event: HttpRequestEvent, send: () => Promise<Response>) => Promise<Response>;
  now?: () => Date;
}

/** The caller's fence refused a send; carries whether earlier sends happened. */
export class DispatchFenceError extends Error {
  constructor(
    readonly dispatched: boolean,
    override readonly cause: unknown,
  ) {
    super("publish dispatch blocked by execution fence");
    this.name = "DispatchFenceError";
  }
}

function canReplayAmbiguous(policy: PublishingPolicy, attempt: PublishAttempt, firstDispatchAt: Date | undefined, now: Date): boolean {
  if (policy.idempotency !== "token" || attempt.dedupeVerified !== true) return false;
  if (firstDispatchAt === undefined) return false;
  return now.getTime() - firstDispatchAt.getTime() < (policy.dedupeWindowMs ?? 0);
}

function receiptOf(connection: ConnectorConnection, idempotencyKey: string, publication: Publication): PublishReceipt {
  const receipt: PublishReceipt = {
    provider: connection.provider,
    connectionId: connection.id,
    idempotencyKey,
    remoteId: publication.remoteId,
    publishedAt: publication.publishedAt,
    providerMetadata: { ...publication.providerMetadata },
  };
  if (publication.remoteUrl !== undefined) receipt.remoteUrl = publication.remoteUrl;
  return receipt;
}

function successOutcome(connection: ConnectorConnection, idempotencyKey: string, publication: Publication): PublishOutcome {
  const receipt = receiptOf(connection, idempotencyKey, publication);
  const meta = publication.providerMetadata;
  const scheduled = publication.remoteId.startsWith("scheduled:") || meta["outcome"] === "provider_scheduled";
  if (!scheduled) return { kind: "published", receipt };
  const scheduledAt = typeof meta["scheduledAt"] === "string" ? new Date(meta["scheduledAt"]) : publication.publishedAt;
  return { kind: "provider_scheduled", receipt, scheduledRemoteId: publication.remoteId, scheduledAt };
}

function statusOf(error: unknown): number | undefined {
  if (error instanceof ProviderRejectedError || error instanceof ProviderServerError || error instanceof RateLimitError) {
    return error.status;
  }
  return undefined;
}

function retryAfterOf(error: unknown): number | undefined {
  if (error instanceof RateLimitError || error instanceof ProviderRejectedError || error instanceof ProviderServerError) {
    return error.retryAfterMs;
  }
  return undefined;
}

/**
 * Classify a publish failure. `dispatched`: a mutating request was sent in
 * this call; `replayAmbiguous`: a same-key resend is provider-deduped now;
 * `priorUnsettled`: an earlier send of this attempt is still ambiguous, so
 * even a definite refusal of this call cannot prove "not delivered".
 */
export function classifyPublishError(
  error: unknown,
  dispatched: boolean,
  replayAmbiguous: boolean,
  priorUnsettled = false,
): PublishOutcome {
  const outcome = classifyCurrentCall(error, dispatched, replayAmbiguous);
  if (priorUnsettled && outcome.kind === "retryable" && outcome.resend === "not_delivered") {
    return { ...outcome, resend: "deduplicated" };
  }
  return outcome;
}

function classifyCurrentCall(error: unknown, dispatched: boolean, replayAmbiguous: boolean): PublishOutcome {
  const base = {
    dispatched,
    errorName: error instanceof Error ? error.name : "UnknownError",
    ...(statusOf(error) === undefined ? {} : { status: statusOf(error) }),
  };
  const retryAfterMs = retryAfterOf(error);
  const retryAfter = retryAfterMs === undefined ? {} : { retryAfterMs };
  const definite = deliveryCertaintyOf(error) === "rejected" || !dispatched;

  if (!definite) {
    if (replayAmbiguous) {
      return { kind: "retryable", code: "PUBLISH_AMBIGUOUS_DEDUPED", resend: "deduplicated", ...base, ...retryAfter };
    }
    return { kind: "unknown", code: "PUBLISH_OUTCOME_UNKNOWN", ...base, dispatched: true };
  }

  if (error instanceof RateLimitError) {
    return { kind: "retryable", code: "RATE_LIMITED", resend: "not_delivered", ...base, ...retryAfter };
  }
  if (error instanceof AuthError) {
    return { kind: "rejected_permanent", code: "CREDENTIAL_REJECTED", needsReauthorization: true, ...base };
  }
  if (error instanceof ProviderRejectedError) {
    // 408/425: the provider refused for timing reasons, not content.
    if (error.status === 408 || error.status === 425) {
      return { kind: "retryable", code: "PROVIDER_UNAVAILABLE", resend: "not_delivered", ...base, ...retryAfter };
    }
    return { kind: "rejected_permanent", code: "PROVIDER_REJECTED", needsReauthorization: false, ...base };
  }
  if (error instanceof ProviderServerError) {
    return { kind: "retryable", code: "PROVIDER_UNAVAILABLE", resend: "not_delivered", ...base, ...retryAfter };
  }
  if (error instanceof CapabilityNotSupportedError) {
    return { kind: "rejected_permanent", code: "UNSUPPORTED", needsReauthorization: false, ...base };
  }
  if (error instanceof ExecutionMetadataError) {
    return { kind: "rejected_permanent", code: "INVALID_POST", needsReauthorization: false, ...base };
  }
  if (error instanceof ConnectorError) {
    // Connector-side validation (limits, missing loaders) before any send.
    return { kind: "rejected_permanent", code: "INVALID_POST", needsReauthorization: false, ...base };
  }
  // Raw transport/runtime failure before any mutating send: nothing delivered.
  return { kind: "retryable", code: "TRANSPORT_FAILURE", resend: "not_delivered", ...base };
}

/**
 * Publish one attempt through `connector.publish` (signature unchanged).
 * Rethrows only DispatchFenceError; every provider result is an outcome.
 */
export async function publishWithPolicy(
  connector: PlatformConnector,
  connection: ConnectorConnection,
  post: NormalizedPost,
  attempt: PublishAttempt,
): Promise<PublishOutcome> {
  const now = attempt.now ?? (() => new Date());
  const policy = attempt.policy ?? publishingPolicyFor(connection.provider);

  if (connector.publish === undefined) {
    return { kind: "rejected_permanent", code: "UNSUPPORTED", needsReauthorization: false, dispatched: false };
  }
  if (attempt.unsettledPriorDispatch === true && !canReplayAmbiguous(policy, attempt, attempt.firstDispatchAt, now())) {
    return { kind: "unknown", code: "PUBLISH_OUTCOME_UNKNOWN", dispatched: true };
  }

  let prepared: NormalizedPost;
  try {
    prepared = preparePostForDispatch(post, {
      idempotencyKey: attempt.idempotencyKey,
      allowProviderScheduling: attempt.allowProviderScheduling,
    });
  } catch (error) {
    return classifyPublishError(error, false, false);
  }

  let mutations = 0;
  let firstMutationAt: Date | undefined;
  let fenceFailure: DispatchFenceError | undefined;
  const publish = connector.publish.bind(connector);
  try {
    const publication = await runWithHttpHook(
      {
        aroundMutation: async (event, send) => {
          const observedSend = async (): Promise<Response> => {
            mutations += 1;
            firstMutationAt ??= now();
            return send();
          };
          try {
            await attempt.beforeMutation?.(event);
          } catch (cause) {
            fenceFailure = new DispatchFenceError(mutations > 0 || attempt.unsettledPriorDispatch === true, cause);
            throw fenceFailure;
          }
          return attempt.aroundMutation ? attempt.aroundMutation(event, observedSend) : observedSend();
        },
      },
      () => publish(connection, prepared),
    );
    return successOutcome(connection, attempt.idempotencyKey, publication);
  } catch (error) {
    if (fenceFailure !== undefined) throw fenceFailure;
    if (error instanceof DispatchFenceError) throw error;
    const dispatched = mutations > 0;
    const firstDispatchAt = attempt.firstDispatchAt ?? firstMutationAt;
    return classifyPublishError(
      error,
      dispatched,
      canReplayAmbiguous(policy, attempt, firstDispatchAt, now()),
      attempt.unsettledPriorDispatch === true,
    );
  }
}