tech-log-frontend/tests/unit/realtime/event-consumer.test.ts

import { describe, expect, it, vi } from "vitest";

import type {
  RealtimeAcceptDisposition,
  RealtimeRecoveryCheckpoint,
} from "../../../src/application/ports/realtime/event-authority.ts";
import type {
  RealtimeResult,
} from "../../../src/application/ports/realtime/shared.ts";
import {
  createRealtimeEventConsumer,
} from "../../../src/adapters/realtime/event-consumer.ts";
import type {
  RealtimeEventCodec,
  ValidatedRealtimeEventDto,
} from "../../../src/adapters/realtime/event-codec.ts";
import {
  realtimeFailure,
  realtimeSuccess,
} from "../../../src/adapters/realtime/result.ts";
import type {
  RealtimeStreamCoordinator,
} from "../../../src/adapters/realtime/stream-coordinator.ts";

function setup(
  recoveryMode:
    | "CURSOR"
    | "SNAPSHOT_ONLY"
    | "SESSION_REBUILD",
  resumeCursor: string | null,
) {
  const dto = {
    envelope: {
      recoveryMode,
      resumeCursor,
      streamId: "REFERENCE_STREAM",
    },
  } as ValidatedRealtimeEventDto;
  const codec: RealtimeEventCodec = {
    decode: vi.fn(() => realtimeSuccess(dto)),
  };
  const accept = vi.fn<
    (
      event: ValidatedRealtimeEventDto,
      signal?: AbortSignal,
    ) => Promise<RealtimeResult<RealtimeAcceptDisposition>>
  >(() =>
    Promise.resolve(
      realtimeSuccess({
        outcome: "DROPPED" as const,
        reason: "DUPLICATE_EVENT" as const,
      }),
    ),
  );
  const consumer = createRealtimeEventConsumer({
    codec,
    coordinator: {
      accept,
    } as Pick<RealtimeStreamCoordinator, "accept">,
  });
  return { consumer, accept, codec };
}

describe("realtime transport event consumer", () => {
  it("requires exact equality between SSE id and CURSOR envelope", async () => {
    const matching = setup("CURSOR", "cursor.0001");
    await expect(
      matching.consumer.consume("{}", {
        kind: "SSE_DIRECT_CURSOR",
        eventId: "cursor.0001",
      }),
    ).resolves.toMatchObject({ ok: true });
    expect(matching.accept).toHaveBeenCalledTimes(1);

    const advanced = setup("CURSOR", "cursor.0002");
    await expect(
      advanced.consumer.consume("{}", {
        kind: "SSE_DIRECT_CURSOR",
        eventId: "cursor.0001",
      }),
    ).resolves.toEqual(
      realtimeFailure("PROTOCOL_MISMATCH", "RECEIVE"),
    );
    expect(advanced.accept).not.toHaveBeenCalled();
  });

  it("forbids SSE id semantics for non-CURSOR recovery", async () => {
    const runtime = setup("SNAPSHOT_ONLY", null);
    await expect(
      runtime.consumer.consume("{}", {
        kind: "SSE_NO_CURSOR",
      }),
    ).resolves.toMatchObject({ ok: true });

    await expect(
      runtime.consumer.consume("{}", {
        kind: "SSE_DIRECT_CURSOR",
        eventId: "cursor.0001",
      }),
    ).resolves.toEqual(
      realtimeFailure("PROTOCOL_MISMATCH", "RECEIVE"),
    );
  });

  it("serializes an encapsulated WebSocket envelope through the codec", async () => {
    const runtime = setup("SESSION_REBUILD", null);
    await expect(
      runtime.consumer.consumeEncapsulated(
        Object.freeze({ protocol: "REALTIME_EVENT_V1" }),
      ),
    ).resolves.toMatchObject({ ok: true });
    expect(runtime.codec.decode).toHaveBeenCalledWith(
      '{"protocol":"REALTIME_EVENT_V1"}',
    );
  });

  it("projects common dispositions into the canonical transport outcome", async () => {
    const runtime = setup("SNAPSHOT_ONLY", null);

    await expect(
      runtime.consumer.consumeForTransport("{}", {
        kind: "SSE_NO_CURSOR",
      }),
    ).resolves.toEqual({
      ok: true,
      value: { kind: "CONTINUE" },
    });

    const checkpoint = Object.freeze({
      recoveryMode: "SNAPSHOT_ONLY" as const,
      streamEpoch: "stream-epoch.0001",
      lastAppliedSequence: "0",
      resumeCursor: null,
    }) as RealtimeRecoveryCheckpoint;
    runtime.accept.mockResolvedValueOnce(
      realtimeSuccess({
        outcome: "RECOVERED",
        reason: "INITIALIZE",
        resumeState: checkpoint,
      }),
    );
    await expect(
      runtime.consumer.consumeForTransport("{}", {
        kind: "SSE_NO_CURSOR",
      }),
    ).resolves.toEqual({
      ok: true,
      value: {
        kind: "RECOVERY_COMMITTED",
        streamId: "REFERENCE_STREAM",
        checkpoint,
      },
    });

    runtime.accept.mockResolvedValueOnce(
      realtimeSuccess({
        outcome: "DROPPED",
        reason: "RECOVERY_IN_PROGRESS",
      }),
    );
    await expect(
      runtime.consumer.consumeForTransport("{}", {
        kind: "SSE_NO_CURSOR",
      }),
    ).resolves.toEqual(
      realtimeFailure("PROTOCOL_MISMATCH", "RECEIVE"),
    );

    runtime.accept.mockResolvedValueOnce(
      realtimeSuccess({
        outcome: "DROPPED",
        reason: "SCOPE_FENCED",
      }),
    );
    await expect(
      runtime.consumer.consumeForTransport("{}", {
        kind: "SSE_NO_CURSOR",
      }),
    ).resolves.toEqual(
      realtimeFailure("SCOPE_FENCED", "RECEIVE"),
    );
  });

  it("threads transport cancellation into the common authority", async () => {
    const runtime = setup("SNAPSHOT_ONLY", null);
    const active = new AbortController();

    await runtime.consumer.consume(
      "{}",
      { kind: "SSE_NO_CURSOR" },
      active.signal,
    );
    expect(runtime.accept).toHaveBeenLastCalledWith(
      expect.anything(),
      active.signal,
    );

    const aborted = new AbortController();
    aborted.abort();
    await expect(
      runtime.consumer.consume(
        "{}",
        { kind: "SSE_NO_CURSOR" },
        aborted.signal,
      ),
    ).resolves.toEqual(
      realtimeFailure("ABORTED", "RECEIVE"),
    );
    expect(runtime.accept).toHaveBeenCalledTimes(1);
  });
});