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> => { 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 { return resolvedRoutes; }, }, ]); const transport = new HttpTransportRouter(router, debug); const app = new Hono(); 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 => { const app = await createApp([ { url: '/echowtf', handler: async (stream): Promise => 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 => { 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 => { 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 => { const app = await createApp((broadcaster) => [ { url: '/items/subscribe', handler: async (stream): Promise => { 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 => { 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 => { const app = await createApp([ { url: '/echo', handler: (stream): Promise => 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 => { let removeSubscription: () => Promise = async () => undefined; let markSubscribed: () => void = () => undefined; const subscribed = new Promise((resolve) => { markSubscribed = resolve; }); const app = await createApp((broadcaster) => [ { url: '/items/subscribe', handler: async (stream): Promise => { const topics = [ 'items' ]; removeSubscription = (): Promise => 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 => { const app = await createApp((broadcaster) => [ { url: '/items/unsubscribe', handler: async (stream): Promise => { 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 => { const app = await createApp( [ { url: '/echo', handler: async (stream): Promise => 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', }); }); });