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

import { describe, expect, it } from "vitest";
import type { ConnectorConnection, NormalizedPost, Publication } from "@tactical-media-broadcast-hub/core";
import {
  AuthError,
  ConnectorError,
  ProviderRejectedError,
  RateLimitedHttpClient,
  type PlatformConnector,
} from "@tactical-media-broadcast-hub/connector-sdk";
import {
  DispatchFenceError,
  ExecutionMetadataError,
  MASTODON_DEDUPE_WINDOW_MS,
  preparePostForDispatch,
  publishingIdempotencyKey,
  publishingPolicyFor,
  publishWithPolicy,
  stripExecutionMetadata,
} from "./index.js";

const KEY = "wp3:ws-1:draft-1:a1:publish";
const ORIGIN = "https://provider.example";

/** Records sends independently of the responses the connector sees. */
interface Wire {
  posts: { key: string | null; body: Record<string, unknown> }[];
  gets: number;
}

type Reply = "ok" | "scheduled" | "500" | "429" | "401" | "422" | "drop";

/**
 * Minimal connector over the real SDK HTTP client: an identity GET, then the
 * publish POST. The scripted reply for each POST is consumed in order.
 */
function fakeConnector(replies: Reply[], wire: Wire): PlatformConnector {
  const client = new RateLimitedHttpClient({
    requestsPerSecond: 1000,
    sleep: async () => Promise.resolve(),
    fetchImpl: (async (input: Parameters<typeof fetch>[0], init?: RequestInit) => {
      const method = init?.method ?? "GET";
      if (method === "GET") {
        wire.gets += 1;
        return new Response("{}");
      }
      const body = JSON.parse(String(init?.body)) as Record<string, unknown>;
      wire.posts.push({ key: new Headers(init?.headers).get("idempotency-key"), body });
      const reply = replies.shift() ?? "ok";
      switch (reply) {
        case "ok":
          return Response.json({ id: `r${wire.posts.length}`, created_at: "2026-10-09T12:00:00Z" });
        case "scheduled":
          return Response.json({ id: "s9", scheduled_at: "2026-10-09T15:00:00.000Z" });
        case "500":
          return new Response("", { status: 500 });
        case "429":
          return new Response("", { status: 429, headers: { "retry-after": "12" } });
        case "401":
          return new Response("", { status: 401 });
        case "422":
          return new Response("", { status: 422 });
        case "drop":
          throw new TypeError("socket hang up");
      }
      void input;
      throw new Error("unreachable");
    }) as typeof fetch,
  });

  return {
    manifest: () => {
      throw new Error("not needed");
    },
    authorize: async () => {
      throw new Error("not needed");
    },
    getIdentity: async () => {
      throw new Error("not needed");
    },
    async publish(connection: ConnectorConnection, post: NormalizedPost): Promise<Publication> {
      if (post.text === "too-long") throw new ConnectorError("text over limit", connection.provider);
      await client.request(connection.provider, `${ORIGIN}/me`);
      const key = post.providerMetadata["idempotencyKey"];
      const response = await client.request(connection.provider, `${ORIGIN}/statuses`, {
        method: "POST",
        headers: typeof key === "string" ? { "idempotency-key": key } : {},
        body: JSON.stringify({ status: post.text, scheduled_at: post.scheduledAt?.toISOString() }),
      });
      if (response.status === 401) throw new AuthError("revoked", connection.provider, false);
      if (!response.ok) throw new ProviderRejectedError("refused", connection.provider, response.status);
      const data = (await response.json()) as { id: string; created_at?: string; scheduled_at?: string };
      const scheduled = data.scheduled_at !== undefined;
      return {
        id: "local-placeholder",
        workspaceId: connection.workspaceId,
        draftId: "direct",
        connectionId: connection.id,
        remoteId: scheduled ? `scheduled:${data.id}` : data.id,
        publishedAt: new Date(data.scheduled_at ?? data.created_at ?? 0),
        providerMetadata: scheduled
          ? { outcome: "provider_scheduled", scheduledAt: data.scheduled_at }
          : { outcome: "published" },
      };
    },
  };
}

function connectionFor(provider: string): ConnectorConnection {
  return {
    id: `conn-${provider}`,
    workspaceId: "ws-1",
    provider,
    channelId: "c",
    credentialId: { kind: "credential-ref", id: "cred" },
    fields: {},
    createdAt: new Date(),
  };
}

const post = (text = "hello", extra: Partial<NormalizedPost> = {}): NormalizedPost => ({
  text,
  mediaAssetIds: [],
  visibility: "public",
  providerMetadata: { connectionId: "conn-x", draftId: "user-draft", idempotencyKey: "user-key" },
  ...extra,
});

function setup(provider: string, replies: Reply[]): { wire: Wire; connector: PlatformConnector; connection: ConnectorConnection } {
  const wire: Wire = { posts: [], gets: 0 };
  return { wire, connector: fakeConnector(replies, wire), connection: connectionFor(provider) };
}

describe("idempotency key + dispatch preparation", () => {
  it("formats the per-attempt key and validates its parts", () => {
    expect(publishingIdempotencyKey({ workspaceId: "ws-1", draftId: "draft-1", attemptNo: 1 })).toBe(KEY);
    expect(() => publishingIdempotencyKey({ workspaceId: "ws:1", draftId: "d", attemptNo: 1 })).toThrow(RangeError);
    expect(() => publishingIdempotencyKey({ workspaceId: "ws", draftId: "d", attemptNo: 0 })).toThrow(RangeError);
  });

  it("injects the key, drops author execution keys and strips scheduledAt", () => {
    const original = post("x", { scheduledAt: new Date("2026-12-01T00:00:00Z") });
    const prepared = preparePostForDispatch(original, { idempotencyKey: KEY });
    expect(prepared.providerMetadata).toEqual({ connectionId: "conn-x", idempotencyKey: KEY });
    expect(prepared.scheduledAt).toBeUndefined();
    // Input untouched.
    expect(original.providerMetadata["draftId"]).toBe("user-draft");
    expect(original.scheduledAt).toBeDefined();
    expect(preparePostForDispatch(original, { idempotencyKey: KEY, allowProviderScheduling: true }).scheduledAt).toEqual(
      original.scheduledAt,
    );
  });

  it("refuses author-supplied remote media ids", () => {
    expect(() => preparePostForDispatch(post("x", { providerMetadata: { mediaIds: ["m1"] } }), { idempotencyKey: KEY })).toThrow(
      ExecutionMetadataError,
    );
  });

  it("stripExecutionMetadata removes only runtime-owned keys (approval hash input)", () => {
    expect(stripExecutionMetadata(post()).providerMetadata).toEqual({ connectionId: "conn-x" });
  });

  it("declares token dedupe for Mastodon and none for Telegram/unknown providers", () => {
    expect(publishingPolicyFor("mastodon")).toEqual({ idempotency: "token", dedupeWindowMs: MASTODON_DEDUPE_WINDOW_MS });
    expect(publishingPolicyFor("telegram")).toEqual({ idempotency: "none" });
    expect(publishingPolicyFor("bluesky")).toEqual({ idempotency: "none" });
  });
});

describe("publishWithPolicy outcomes", () => {
  it("published: receipt carries the injected key, never the local id/draftId", async () => {
    const { wire, connector, connection } = setup("telegram", ["ok"]);
    const outcome = await publishWithPolicy(connector, connection, post(), { idempotencyKey: KEY });
    expect(wire.posts).toHaveLength(1);
    expect(wire.posts[0]?.key).toBe(KEY);
    expect(wire.posts[0]?.body["scheduled_at"]).toBeUndefined();
    expect(outcome).toMatchObject({ kind: "published", receipt: { remoteId: "r1", idempotencyKey: KEY, connectionId: "conn-telegram" } });
    expect(outcome.kind === "published" && "id" in outcome.receipt).toBe(false);
    expect(outcome.kind === "published" && "draftId" in outcome.receipt).toBe(false);
  });

  it("provider_scheduled when native scheduling is explicitly allowed", async () => {
    const { connector, connection } = setup("mastodon", ["scheduled"]);
    const outcome = await publishWithPolicy(connector, connection, post("x", { scheduledAt: new Date("2026-10-09T15:00:00Z") }), {
      idempotencyKey: KEY,
      allowProviderScheduling: true,
    });
    expect(outcome).toMatchObject({
      kind: "provider_scheduled",
      scheduledRemoteId: "scheduled:s9",
      scheduledAt: new Date("2026-10-09T15:00:00.000Z"),
    });
  });

  it("429 → retryable RATE_LIMITED with Retry-After, one send", async () => {
    const { wire, connector, connection } = setup("telegram", ["429"]);
    const outcome = await publishWithPolicy(connector, connection, post(), { idempotencyKey: KEY });
    expect(wire.posts).toHaveLength(1);
    expect(outcome).toMatchObject({ kind: "retryable", code: "RATE_LIMITED", retryAfterMs: 12_000, resend: "not_delivered", dispatched: true });
  });

  it("401 → rejected_permanent needing reauthorization; 422 → PROVIDER_REJECTED", async () => {
    const auth = setup("telegram", ["401"]);
    expect(await publishWithPolicy(auth.connector, auth.connection, post(), { idempotencyKey: KEY })).toMatchObject({
      kind: "rejected_permanent",
      code: "CREDENTIAL_REJECTED",
      needsReauthorization: true,
    });
    const invalid = setup("telegram", ["422"]);
    expect(await publishWithPolicy(invalid.connector, invalid.connection, post(), { idempotencyKey: KEY })).toMatchObject({
      kind: "rejected_permanent",
      code: "PROVIDER_REJECTED",
      status: 422,
      needsReauthorization: false,
    });
  });

  it("connector validation before any send → INVALID_POST, not dispatched", async () => {
    const { wire, connector, connection } = setup("telegram", []);
    const outcome = await publishWithPolicy(connector, connection, post("too-long"), { idempotencyKey: KEY });
    expect(outcome).toMatchObject({ kind: "rejected_permanent", code: "INVALID_POST", dispatched: false });
    expect(wire.posts).toHaveLength(0);
  });

  it("author-supplied media ids → INVALID_POST with zero HTTP", async () => {
    const { wire, connector, connection } = setup("telegram", []);
    const outcome = await publishWithPolicy(connector, connection, post("x", { providerMetadata: { mediaIds: ["1"] } }), {
      idempotencyKey: KEY,
    });
    expect(outcome).toMatchObject({ kind: "rejected_permanent", code: "INVALID_POST" });
    expect(wire.gets + wire.posts.length).toBe(0);
  });
});

describe("ambiguous sends", () => {
  it("no-token provider: 5xx and dropped responses are unknown, never retryable", async () => {
    for (const reply of ["500", "drop"] as const) {
      const { wire, connector, connection } = setup("telegram", [reply]);
      const outcome = await publishWithPolicy(connector, connection, post(), { idempotencyKey: KEY });
      expect(outcome).toMatchObject({ kind: "unknown", code: "PUBLISH_OUTCOME_UNKNOWN", dispatched: true });
      expect(wire.posts).toHaveLength(1);
    }
  });

  it("token provider on a verified origin inside the window: retryable deduplicated", async () => {
    const { connector, connection } = setup("mastodon", ["500"]);
    const outcome = await publishWithPolicy(connector, connection, post(), { idempotencyKey: KEY, dedupeVerified: true });
    expect(outcome).toMatchObject({ kind: "retryable", code: "PUBLISH_AMBIGUOUS_DEDUPED", resend: "deduplicated", status: 500 });
  });

  it("token provider on an unverified origin is unknown", async () => {
    const { connector, connection } = setup("mastodon", ["500"]);
    const outcome = await publishWithPolicy(connector, connection, post(), { idempotencyKey: KEY });
    expect(outcome.kind).toBe("unknown");
  });

  it("token provider past the window (measured from first dispatch) is unknown", async () => {
    const { connector, connection } = setup("mastodon", ["drop"]);
    const outcome = await publishWithPolicy(connector, connection, post(), {
      idempotencyKey: KEY,
      dedupeVerified: true,
      firstDispatchAt: new Date(Date.now() - MASTODON_DEDUPE_WINDOW_MS - 1000),
    });
    expect(outcome.kind).toBe("unknown");
  });

  it("unsettled prior dispatch without dedupe: unknown with zero HTTP", async () => {
    const { wire, connector, connection } = setup("telegram", ["ok"]);
    const outcome = await publishWithPolicy(connector, connection, post(), {
      idempotencyKey: KEY,
      unsettledPriorDispatch: true,
      firstDispatchAt: new Date(),
      dedupeVerified: true,
    });
    expect(outcome).toMatchObject({ kind: "unknown", dispatched: true });
    expect(wire.gets + wire.posts.length).toBe(0);
  });

  it("unsettled prior dispatch with verified dedupe: replays the same key", async () => {
    const { wire, connector, connection } = setup("mastodon", ["ok"]);
    const outcome = await publishWithPolicy(connector, connection, post(), {
      idempotencyKey: KEY,
      unsettledPriorDispatch: true,
      firstDispatchAt: new Date(Date.now() - 60_000),
      dedupeVerified: true,
    });
    expect(outcome.kind).toBe("published");
    expect(wire.posts.map((p) => p.key)).toEqual([KEY]);
  });

  it("a definite refusal of a replay does not claim the earlier send was not delivered", async () => {
    const { connector, connection } = setup("mastodon", ["429"]);
    const outcome = await publishWithPolicy(connector, connection, post(), {
      idempotencyKey: KEY,
      unsettledPriorDispatch: true,
      firstDispatchAt: new Date(),
      dedupeVerified: true,
    });
    expect(outcome).toMatchObject({ kind: "retryable", code: "RATE_LIMITED", resend: "deduplicated" });
  });
});

describe("dispatch fence", () => {
  it("a failing fence blocks the mutating send; reads are not fenced", async () => {
    const { wire, connector, connection } = setup("telegram", ["ok"]);
    const error = await publishWithPolicy(connector, connection, post(), {
      idempotencyKey: KEY,
      beforeMutation: () => {
        throw new Error("lease lost");
      },
    }).catch((e: unknown) => e);
    expect(error).toBeInstanceOf(DispatchFenceError);
    expect(error).toMatchObject({ dispatched: false });
    expect(wire.gets).toBe(1);
    expect(wire.posts).toHaveLength(0);
  });

  it("concurrent executions keep their own fences (AsyncLocalStorage isolation)", async () => {
    const blocked = setup("telegram", ["ok"]);
    const allowed = setup("telegram", ["ok"]);
    const seen: string[] = [];
    const [a, b] = await Promise.allSettled([
      publishWithPolicy(blocked.connector, blocked.connection, post(), {
        idempotencyKey: "wp3:ws-1:a:a1:publish",
        beforeMutation: async () => {
          await new Promise((r) => setTimeout(r, 5));
          seen.push("blocked");
          throw new Error("revoked");
        },
      }),
      publishWithPolicy(allowed.connector, allowed.connection, post(), {
        idempotencyKey: "wp3:ws-1:b:a1:publish",
        beforeMutation: () => void seen.push("allowed"),
      }),
    ]);
    expect(a.status).toBe("rejected");
    expect(b).toMatchObject({ status: "fulfilled", value: { kind: "published" } });
    expect(blocked.wire.posts).toHaveLength(0);
    expect(allowed.wire.posts.map((p) => p.key)).toEqual(["wp3:ws-1:b:a1:publish"]);
    expect(seen.sort()).toEqual(["allowed", "blocked"]);
  });

  it("calls outside a runtime execution are unaffected by the hook", async () => {
    const { wire, connector, connection } = setup("telegram", ["ok"]);
    await publishWithPolicy(connector, connection, post(), {
      idempotencyKey: KEY,
      beforeMutation: () => {
        throw new Error("only inside");
      },
    }).catch(() => undefined);
    const direct = await connector.publish?.(connection, post());
    expect(direct?.remoteId).toBe("r1");
    expect(wire.posts).toHaveLength(1);
  });
});