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.ts';
import { ApplicationError } from '../../../source/errors/index.ts';
import { ApplicationRouter } from '../../../source/services/router.ts';
import { Broadcaster } from '../../../source/services/broadcaster.ts';
import { HttpTransportRouter } from '../../../source/services/transport/http-transport.ts';
import type { AppEnv } from '../../../source/services/transport/transport-router.ts';
import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
import { Logger } from '../../../source/utils/logger.ts';
import { ServerHost } from '../../../source/services/server-host.ts';
import { AuthSecp256k1 } from '../../../source/auth/auth.ts';

/**
 * A mock of the AuthSecp256k1 service
 */
const auth = {
    verifySignature: vi.fn().mockResolvedValue(true),
    verifyUniqueRequest: vi.fn().mockResolvedValue(true),
    assertTimestampFreshness: vi.fn().mockResolvedValue(true),
} as unknown as AuthSecp256k1;

const mockAuthHeaders = {
    'x-public-key': 'public-key',
    'x-signature': 'signature',
    'x-timestamp': '1000',
};

const createApp = async (
    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({
        auth,
    }, [
        {
            async getRoutes(): Promise<RouteDefinition[]> {
                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', (): void => {
    it('runs normal HTTP through a non-streaming route stream', async (): Promise<void> => {
        const app = await createApp([
            {
                url: '/echowtf',
                handler: async (stream): Promise<void> => stream.send(stream.body),
            },
        ]);
        const value = new Uint8Array([ 1, 2, 3 ]);

        const request = new Request('http://localhost/echowtf', {
            method: 'POST',
            headers: { 'content-type': 'application/json', ...mockAuthHeaders },
            body: toExtendedJson({ value }),
        });

        const response = await app.request(request);

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

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

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

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

    it('returns normalized errors for non-streaming requests', async (): Promise<void> => {
        const app = await createApp([]);

        const missing = await app.request('/missing', { method: 'POST', headers: { ...mockAuthHeaders } });
        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', ...mockAuthHeaders },
            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 (): Promise<void> => {
        const app = await createApp((broadcaster) => [
            {
                url: '/items/subscribe',
                handler: async (stream): Promise<void> => {
                    await broadcaster.subscribe(stream, [ 'items' ]);
                },
            },
        ]);

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

        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 (): Promise<void> => {
        const app = await createApp([
            {
                url: '/items/subscribe',
                handler: (): void => {
                    throw new Error('private storage failure');
                },
            },
        ]);

        const response = await app.request('/items/subscribe', {
            method: 'POST',
            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
        });
        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 (): Promise<void> => {
        const app = await createApp([
            {
                url: '/echo',
                handler: (stream): Promise<void> => stream.send({ ok: true }),
            },
        ]);

        const response = await app.request('/echo', {
            method: 'POST',
            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
        });
        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 (): Promise<void> => {
        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): Promise<void> => {
                    const topics = [ 'items' ];
                    removeSubscription = (): Promise<void> => 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', ...mockAuthHeaders },
        });
        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 (): Promise<void> => {
        const app = await createApp((broadcaster) => [
            {
                url: '/items/unsubscribe',
                handler: async (stream): Promise<void> => {
                    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', ...mockAuthHeaders },
        });
        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 (): Promise<void> => {
        const app = await createApp(
            [
                {
                    url: '/echo',
                    handler: async (stream): Promise<void> => stream.send(stream.body),
                },
            ],
            32,
        );

        const response = await app.request('/echo', {
            method: 'POST',
            headers: { 'content-type': 'application/json', ...mockAuthHeaders },
            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',
        });
    });
});