feat: 기능 추가 과정중

tests/unit/realtime/event-consumer.test.ts

@@ -0,0 +1,211 @@
+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);
+  });
+});