tech-log-frontend/tests/unit/realtime/realtime-stream-registry.test.ts

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

import {
  createRealtimePolicyRegistry,
  defineEventTypeId,
  defineStreamRegistrationId,
  REALTIME_HARD_LIMITS,
  type RealtimeLimits,
  type RealtimeStreamRegistration,
} from "../../../src/contracts/realtime-streams.ts";
import {
  EFFECT_PROFILE_ID,
  ENDPOINT_ID,
  EVENT_TYPE,
  KILL_SWITCH_ID,
  STREAM_ID,
  TEST_API_OPERATIONS,
  TEST_LIMITS,
  TEST_MAPPERS,
  TEST_SCHEMA_CODECS,
  createTestRealtimeRegistry,
} from "./fixture.ts";

describe("realtime policy registry", () => {
  it("deep-snapshots registrations and resolves only stream-owned event types", () => {
    const baseline = createTestRealtimeRegistry();
    const sourceEventIds = [EVENT_TYPE];
    const sourceLimits = { ...TEST_LIMITS };
    const sourceStream: RealtimeStreamRegistration = {
      ...baseline.listStreams()[0]!,
      eventTypeIds: sourceEventIds,
      limits: sourceLimits,
    };
    const sourceEventType = {
      ...baseline.listEventTypes()[0]!,
    };
    const registry = createRealtimePolicyRegistry({
      streams: [sourceStream],
      eventTypes: [sourceEventType],
      bindings: bindings(),
    });

    sourceEventIds[0] = defineEventTypeId("MUTATED_EVENT");
    sourceLimits.maxQueueEvents = 1;
    sourceEventType.owner = "mutated-owner";

    const installed = registry.findStream(STREAM_ID);
    expect(installed).toMatchObject({
      id: STREAM_ID,
      eventTypeIds: [EVENT_TYPE],
      limits: { maxQueueEvents: TEST_LIMITS.maxQueueEvents },
    });
    expect(registry.findEventType(EVENT_TYPE)?.owner).toBe(
      "sample-owner",
    );
    expect(
      registry.findStreamEventType(STREAM_ID, EVENT_TYPE)?.id,
    ).toBe(EVENT_TYPE);
    expect(
      registry.findStreamEventType(STREAM_ID, "MUTATED_EVENT"),
    ).toBeUndefined();
    expect(Object.isFrozen(installed)).toBe(true);
    expect(Object.isFrozen(installed?.eventTypeIds)).toBe(true);
    expect(Object.isFrozen(installed?.limits)).toBe(true);
  });

  it("rejects duplicates, unknown references and extra keys", () => {
    const baseline = createTestRealtimeRegistry();
    const stream = baseline.listStreams()[0]!;
    const eventType = baseline.listEventTypes()[0]!;

    expect(() =>
      createRealtimePolicyRegistry({
        streams: [stream, stream],
        eventTypes: [eventType],
        bindings: bindings(),
      }),
    ).toThrow("stream is duplicated");

    expect(() =>
      createRealtimePolicyRegistry({
        streams: [
          {
            ...stream,
            eventTypeIds: [defineEventTypeId("UNKNOWN_EVENT")],
          },
        ],
        eventTypes: [eventType],
        bindings: bindings(),
      }),
    ).toThrow("stream registration is invalid");

    expect(() =>
      createRealtimePolicyRegistry({
        streams: [
          {
            ...stream,
            unregisteredOverride: true,
          } as RealtimeStreamRegistration,
        ],
        eventTypes: [eventType],
        bindings: bindings(),
      }),
    ).toThrow("stream registration is invalid");

    expect(() =>
      createTestRealtimeRegistry({
        eventTypeMutator: (candidate) => ({
          ...candidate,
          mapperId: "MissingMapper",
        }),
      }),
    ).toThrow("event type registration is invalid");
  });

  it("closes state-bearing, ephemeral and recovery contradictions", () => {
    expect(() =>
      createTestRealtimeRegistry({
        recovery: {
          mode: "SNAPSHOT_ONLY",
          snapshotOperationId: "GET_REFERENCE_REALTIME_SNAPSHOT",
          checkpointCodecId: "ReferenceRealtimeCheckpoint",
          barrier: "NONE",
        },
        delivery: "INVALIDATION_HINT",
        stateBearing: true,
      }),
    ).toThrow("recovery contract is contradictory");

    expect(() =>
      createTestRealtimeRegistry({
        recovery: {
          mode: "SESSION_REBUILD",
          rebuildInputId: "referenceRealtimeRebuild",
        },
        delivery: "EPHEMERAL",
        stateBearing: true,
      }),
    ).toThrow("recovery contract is contradictory");

    expect(() =>
      createTestRealtimeRegistry({
        recovery: {
          mode: "SESSION_REBUILD",
          rebuildInputId: "referenceRealtimeRebuild",
        },
        delivery: "INVALIDATION_HINT",
        stateBearing: false,
      }),
    ).toThrow("recovery contract is contradictory");

    const registry = createTestRealtimeRegistry({
      recovery: {
        mode: "SESSION_REBUILD",
        rebuildInputId: "referenceRealtimeRebuild",
      },
      delivery: "EPHEMERAL",
      stateBearing: false,
    });
    expect(registry.findStream(STREAM_ID)?.recovery.mode).toBe(
      "SESSION_REBUILD",
    );
  });

  it("allows only reductions of implementation ceilings", () => {
    expect(() =>
      createTestRealtimeRegistry({
        limits: {
          ...TEST_LIMITS,
          maxQueueEvents: REALTIME_HARD_LIMITS.maxQueueEvents + 1,
        },
      }),
    ).toThrow("exceed implementation ceilings");

    expect(() =>
      createTestRealtimeRegistry({
        limits: {
          ...TEST_LIMITS,
          maxQueueBytes: TEST_LIMITS.maxEventBytes - 1,
        },
      }),
    ).toThrow("cannot hold one event");

    const strictLimits: RealtimeLimits = {
      ...TEST_LIMITS,
      maxQueueEvents: 1,
      maxDedupeEntries: 1,
    };
    expect(
      createTestRealtimeRegistry({ limits: strictLimits }).findStream(
        STREAM_ID,
      )?.limits,
    ).toMatchObject({
      maxQueueEvents: 1,
      maxDedupeEntries: 1,
    });
  });

  it("issues only bounded closed registry IDs", () => {
    expect(defineStreamRegistrationId("VALID_STREAM")).toBe(
      "VALID_STREAM",
    );
    expect(() => defineStreamRegistrationId("arbitrary-channel")).toThrow(
      "stream ID is invalid",
    );
    expect(() => defineEventTypeId("X")).toThrow(
      "event type ID is invalid",
    );
  });
});

function bindings() {
  return {
    schemaCodecs: TEST_SCHEMA_CODECS,
    mappers: TEST_MAPPERS,
    apiOperations: TEST_API_OPERATIONS,
    endpointIds: [ENDPOINT_ID],
    effectProfileIds: [EFFECT_PROFILE_ID],
    killSwitchIds: [KILL_SWITCH_ID],
    rebuildInputIds: ["referenceRealtimeRebuild"],
  } as const;
}