sync-server-v2/test/services/transports/http-transport.test.ts

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

import type { RouteDefinition } from "../../../source/routes/types.js";
import { ApplicationError } from "../../../source/errors/index.js";
import { ApplicationRouter } from "../../../source/services/router.js";
import { Broadcaster } from "../../../source/services/broadcaster.js";
import { HttpTransportRouter } from "../../../source/services/transport/http-transport.js";
import type { AppEnv } from "../../../source/services/transport/transport-router.js";
import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils";
import { Logger } from "../../../source/utils/logger.js";
import { ServerHost } from "../../../source/services/server-host.js";

async function createApp(
  routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
  maxRequestBodyBytes = 1024 * 1024,
): Promise<Hono<AppEnv>> {
  const debug = new Logger("http-transport-test");
  const broadcaster = new Broadcaster(debug);
  const resolvedRoutes =
    typeof routes === "function" ? routes(broadcaster) : routes;
  const router = await ApplicationRouter.create([
    {
      async getRoutes() {
        return resolvedRoutes;
      },
    },
  ]);
  const transport = new HttpTransportRouter(router, debug);
  const app = new Hono<AppEnv>();

  app.onError(HttpTransportRouter.createErrorHandler(debug));
  app.use("*", ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug));
  app.use("*", HttpTransportRouter.createExtJsonMiddleware(debug));
  transport.register(app);
  return app;
}

describe("HttpTransportRouter", () => {
  it("runs normal HTTP through a non-streaming route stream", async () => {
    const app = await createApp([
      {
        url: "/echo",
        handler: async (stream) => stream.send(stream.body),
      },
    ]);
    const value = new Uint8Array([1, 2, 3]);

    const response = await app.request("/echo", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: toExtendedJson({ value }),
    });

    expect(response.status).toBe(200);
    expect(fromExtendedJson(await response.text())).toEqual({ value });
  });

  it("returns 204 when a normal HTTP route sends nothing", async () => {
    const app = await createApp([
      {
        url: "/nothing",
        handler: () => undefined,
      },
    ]);

    const response = await app.request("/nothing", { method: "POST" });

    expect(response.status).toBe(204);
    expect(await response.text()).toBe("");
  });

  it("returns normalized errors for non-streaming requests", async () => {
    const app = await createApp([]);

    const missing = await app.request("/missing", { method: "POST" });
    expect(missing.status).toBe(404);
    expect(await missing.json()).toEqual({
      statusCode: 404,
      error: "No route found for /missing",
    });

    const invalid = await app.request("/missing", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: "{",
    });
    expect(invalid.status).toBe(400);
    expect(await invalid.json()).toEqual({
      statusCode: 400,
      error: "Invalid JSON in request body",
    });
  });

  it("rejects subscribe when normal HTTP has no streaming capability", async () => {
    const app = await createApp((broadcaster) => [
      {
        url: "/items/subscribe",
        handler: async (stream) => {
          await broadcaster.subscribe(stream, ["items"]);
        },
      },
    ]);

    const response = await app.request("/items/subscribe", { method: "POST" });

    expect(response.status).toBe(406);
    expect(await response.json()).toMatchObject({ statusCode: 406 });
  });

  it("sends SSE route errors as events and closes only that stream", async () => {
    const app = await createApp([
      {
        url: "/items/subscribe",
        handler: () => {
          throw new Error("private storage failure");
        },
      },
    ]);

    const response = await app.request("/items/subscribe", {
      method: "POST",
      headers: { accept: "text/event-stream" },
    });
    const events = await response.text();

    expect(response.status).toBe(200);
    expect(events).toContain("event: error");
    expect(events).toContain(
      'data: {"statusCode":500,"error":"Internal Server Error"}',
    );
    expect(events).not.toContain("private storage failure");
  });

  it("sends a normal route as one SSE response event and then closes", async () => {
    const app = await createApp([
      {
        url: "/echo",
        handler: (stream) => stream.send({ ok: true }),
      },
    ]);

    const response = await app.request("/echo", {
      method: "POST",
      headers: { accept: "text/event-stream" },
    });
    const events = await response.text();

    expect(response.status).toBe(200);
    expect(events).toContain("event: response");
    expect(events).toContain('data: {"ok":true}');
  });

  it("keeps SSE open until the route's subscription promise resolves", async () => {
    let removeSubscription: () => Promise<void> = async () => undefined;
    let markSubscribed: () => void = () => undefined;
    const subscribed = new Promise<void>((resolve) => {
      markSubscribed = resolve;
    });
    const app = await createApp((broadcaster) => [
      {
        url: "/items/subscribe",
        handler: async (stream) => {
          const topics = ["items"];
          removeSubscription = () => broadcaster.unsubscribe(stream, topics);

          const removed = broadcaster.subscribe(stream, topics);
          markSubscribed();
          await removed;
        },
      },
    ]);

    const response = await app.request("/items/subscribe", {
      method: "POST",
      headers: { accept: "text/event-stream" },
    });
    const body = response.text();
    const completed = vi.fn();
    void body.then(completed);

    await subscribed;
    await Promise.resolve();
    expect(completed).not.toHaveBeenCalled();

    await removeSubscription();

    expect(await body).toBe("");
    expect(completed).toHaveBeenCalledOnce();
  });

  it("rejects unsubscribe over non-bidirectional SSE", async () => {
    const app = await createApp((broadcaster) => [
      {
        url: "/items/unsubscribe",
        handler: async (stream) => {
          if (!stream.bidirectional) {
            throw new ApplicationError(
              400,
              "This route requires an existing bidirectional stream",
            );
          }

          await broadcaster.unsubscribe(stream, ["items"]);
        },
      },
    ]);

    const response = await app.request("/items/unsubscribe", {
      method: "POST",
      headers: { accept: "text/event-stream" },
    });
    const events = await response.text();

    expect(response.status).toBe(200);
    expect(events).toContain("event: error");
    expect(events).toContain('"statusCode":400');
  });
  it("rejects HTTP bodies larger than the configured byte limit", async () => {
    const app = await createApp(
      [
        {
          url: "/echo",
          handler: async (stream) => stream.send(stream.body),
        },
      ],
      32,
    );

    const response = await app.request("/echo", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ value: "x".repeat(64) }),
    });

    expect(response.status).toBe(413);
    expect(await response.json()).toEqual({
      statusCode: 413,
      error: "Request body exceeds the 32 byte limit",
    });
  });
});