openwork/apps/app/tests/session-group-event-poller.test.ts

import { describe, expect, test } from "bun:test";

import type { OpenworkSessionGroupEvent } from "../src/app/lib/openwork-server";
import { SessionGroupEventPoller } from "../src/react-app/shell/session-group-event-poller";

function event(seq: number, workspaceId = "workspace-a"): OpenworkSessionGroupEvent {
  return {
    id: `event-${seq}`,
    seq,
    workspaceId,
    type: "session_groups.updated",
    action: "updated",
    timestamp: seq,
  };
}

describe("session group event poller", () => {
  test("sends the newest returned sequence on the next poll", async () => {
    const poller = new SessionGroupEventPoller();
    const requests: number[] = [];
    const events = [event(1), event(2)];
    const request = async ({ since }: { since: number }) => {
      requests.push(since);
      return { items: events.filter((item) => item.seq > since), cursor: 0 };
    };

    await poller.poll("workspace-a", request, async () => {});
    await poller.poll("workspace-a", request, async () => {});

    expect(requests).toEqual([0, 2]);
  });

  test("keeps its cursor when a poll returns no events", async () => {
    const poller = new SessionGroupEventPoller();
    const requests: number[] = [];
    const request = async ({ since }: { since: number }) => {
      requests.push(since);
      return { items: since === 0 ? [event(3)] : [] };
    };

    await poller.poll("workspace-a", request, async () => {});
    await poller.poll("workspace-a", request, async () => {});
    await poller.poll("workspace-a", request, async () => {});

    expect(requests).toEqual([0, 3, 3]);
  });

  test("resets removed workspaces without leaking another workspace cursor", async () => {
    const poller = new SessionGroupEventPoller();
    const requests: Array<{ key: string; since: number }> = [];
    const poll = (key: string, seq: number) => poller.poll(
      key,
      async ({ since }) => {
        requests.push({ key, since });
        return { items: since === 0 ? [event(seq, key)] : [] };
      },
      async () => {},
    );

    poller.setWorkspaces(["workspace-a"]);
    await poll("workspace-a", 4);
    poller.setWorkspaces(["workspace-b"]);
    await poll("workspace-b", 7);
    poller.setWorkspaces(["workspace-a"]);
    await poll("workspace-a", 8);

    expect(requests).toEqual([
      { key: "workspace-a", since: 0 },
      { key: "workspace-b", since: 0 },
      { key: "workspace-a", since: 0 },
    ]);
  });

  test("applies each event once across the exclusive cursor boundary", async () => {
    const poller = new SessionGroupEventPoller();
    const events = [event(1), event(2)];
    const applied: string[] = [];
    const request = async ({ since }: { since: number }) => ({
      items: events.filter((item) => item.seq > since),
    });
    const apply = async (items: OpenworkSessionGroupEvent[]) => {
      applied.push(...items.map((item) => item.id));
    };

    await poller.poll("workspace-a", request, apply);
    events.push(event(3));
    await poller.poll("workspace-a", request, apply);

    expect(applied).toEqual(["event-1", "event-2", "event-3"]);
  });

  test("falls back to zero after an error or sequence gap", async () => {
    const poller = new SessionGroupEventPoller();
    const requests: number[] = [];
    let call = 0;
    const request = async ({ since }: { since: number }) => {
      requests.push(since);
      call += 1;
      if (call === 2) throw new Error("offline");
      if (call === 4) return { items: [event(3)] };
      if (call === 5) return { items: [event(1), event(3)] };
      return { items: since === 0 ? [event(1)] : [] };
    };

    await poller.poll("workspace-a", request, async () => {});
    await expect(poller.poll("workspace-a", request, async () => {})).resolves.toBeUndefined();
    await poller.poll("workspace-a", request, async () => {});
    await poller.poll("workspace-a", request, async () => {});
    await poller.poll("workspace-a", request, async () => {});

    expect(requests).toEqual([0, 1, 0, 1, 0]);
  });
});