sync-server-v2/test/routes/resources.test.ts

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

import { DataRoute } from "../../source/routes/resources.js";
import { UnauthorizedError } from "../../source/errors/index.js";
import { type BaseBroadcaster } from "../../source/services/broadcaster.js";
import { ApplicationRouteStream } from "../../source/services/route-stream.js";
import { Database } from "../../source/services/storage/database.js";
import { TestConnection } from "../helpers/test-connection.js";
import { HTTP_STATUS_CODE_NOT_ACCEPTED } from "../../source/constants.js";

function createBroadcasterStub() {
  return {
    subscribe: vi.fn(),
    unsubscribe: vi.fn().mockResolvedValue(undefined),
    publish: vi.fn(),
    sendEvent: vi.fn(),
  } as unknown as BaseBroadcaster;
}

describe("DataRoute subscriptions", () => {
  it("subscribes to future resource changes until removal", async () => {
    let resolveRemoved: () => void = () => undefined;
    const removed = new Promise<void>((resolve) => {
      resolveRemoved = resolve;
    });
    const storage = {
      db: {
        transaction: vi.fn(),
      },
    } as unknown as Database;
    const broadcaster = createBroadcasterStub();
    vi.mocked(broadcaster.subscribe).mockReturnValue(removed);
    const connection = new TestConnection(true, false);
    const stream = new ApplicationRouteStream(connection, {
      resourceId: ["a", "b"],
    });
    const route = new DataRoute(storage, broadcaster, 0);

    const execution = route.subscribeData(stream);

    expect(broadcaster.subscribe).toHaveBeenCalledWith(
      stream,
      ["resource:a", "resource:b"],
    );
    expect(storage.db.transaction).not.toHaveBeenCalled();
    expect(connection.messages).toEqual([]);
    await expect(
      Promise.race([
        execution.then(() => "settled"),
        Promise.resolve("pending"),
      ]),
    ).resolves.toBe("pending");

    resolveRemoved();
    await execution;
  });

  it("unsubscribes a bidirectional connection and acknowledges the request", async () => {
    const storage = {
      db: {
        transaction: vi.fn(),
      },
    } as unknown as Database;
    const broadcaster = createBroadcasterStub();
    const connection = new TestConnection(true, true);
    const stream = new ApplicationRouteStream(
      connection,
      { resourceId: ["a"] },
      "unsubscribe-1",
    );
    const route = new DataRoute(storage, broadcaster, 0);

    await route.unsubscribeData(stream);

    expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [
      "resource:a",
    ]);
    expect(connection.messages).toEqual([
      {
        id: "unsubscribe-1",
        type: "response",
        statusCode: 200,
        body: {},
      },
    ]);
  });

  it("rejects selective unsubscribe on a one-way connection", async () => {
    const storage = {
      db: {
        transaction: vi.fn(),
      },
    } as unknown as Database;
    const broadcaster = createBroadcasterStub();
    const stream = new ApplicationRouteStream(new TestConnection(true, false), {
      resourceId: ["a"],
    });
    const route = new DataRoute(storage, broadcaster, 0);

    await expect(route.unsubscribeData(stream)).rejects.toMatchObject({ statusCode: HTTP_STATUS_CODE_NOT_ACCEPTED });
    expect(broadcaster.unsubscribe).not.toHaveBeenCalled();
  });
});

describe("DataRoute resource write auth", () => {
  it("rejects an invalid resource signature before writing the batch", async () => {
    const storage = {
      db: {
        transaction: vi.fn(),
      },
    } as unknown as Database;

    const broadcaster = createBroadcasterStub()
    const route = new DataRoute(storage, broadcaster, 0);

    await expect(
      route.writeData({
        connection: new TestConnection(true, true),
        streaming: true,
        bidirectional: true,
        send: vi.fn(),
        body: {
          resources: [
            {
              id: "resource-a",
              publicKey: "not-a-public-key",
              timestamp: Date.now(),
              signature: "not-a-signature",
              value: new Uint8Array([1, 2, 3]),
            },
          ],
        },
      } as unknown as ApplicationRouteStream),
    ).rejects.toBeInstanceOf(UnauthorizedError);

    expect(storage.db.transaction).not.toHaveBeenCalled()
    expect(broadcaster.publish).not.toHaveBeenCalled();
  });
});