sync-server-v2/source/services/transport/http-transport.ts

import type { Context as HonoContext, Hono, MiddlewareHandler, ErrorHandler } from 'hono';
import { streamSSE } from 'hono/streaming';
import { toExtendedJson, fromExtendedJson } from '@xo-cash/utils';

import type { Logger } from '../../utils/logger.ts';
import type { ApplicationRequest, ApplicationRouter } from '../router.ts';
import type { StreamResponse } from '../stream/base-stream.ts';

import { ApplicationError, normalizePublicError } from '../../errors/index.ts';
import { HTTP_STATUS_CODE_BAD_REQUEST, HTTP_STATUS_CODE_NO_CONTENT } from '../../constants.ts';
import { HonoSSEStream } from '../stream/hono-sse-stream.ts';
import { HttpRequestStream } from '../stream/http-request-stream.ts';
import { type AppEnv, BaseTransport } from './base-transport.ts';

/** Hono context key where decoded Extended JSON bodies are stored. */
const PARSED_BODY_KEY = 'parsedBody';

/**
 * HTTP protocol adapter.
 *
 * Normal HTTP and SSE both enter the same application router with different
 * connection-stream capabilities.
 */
export class HttpTransportRouter extends BaseTransport {
    private readonly debug: Logger;

    /** SSE connections retained until their final subscription or peer closes. */
    private readonly activeSseStreams = new Set<HonoSSEStream>();

    /**
     * @param router - Shared application router for route dispatch.
     * @param debug - Root logger extended with an http-transport namespace.
     */
    constructor(
        private readonly router: ApplicationRouter,
        debug: Logger,
    ) {
        super();

        this.debug = debug.extend('http-transport');
    }

    /**
     * Register the single HTTP application entry point.
     *
     * @param app - Hono instance to attach the POST catch-all handler to.
     */
    register(app: Hono<AppEnv>): void {
        // Create HTTP error handler middleware
        app.onError(HttpTransportRouter.createErrorHandler(this.debug));

        // Create a middleware to decode Extended JSON once at the HTTP boundary before route dispatch.
        app.use('*', HttpTransportRouter.createExtJsonMiddleware(this.debug));

        // Handle HTTP POST requests
        app.post('*', async (context) => {
            const request = this.createRequest(context);

            // Branch to SSE when the client negotiates an event stream.
            if (HttpTransportRouter.acceptsSse(context)) {
                return this.openSse(context, request);
            }

            return this.handleRequest(request);
        });
    }

    /**
     * Decode Extended JSON once at the HTTP boundary before route dispatch.
     *
     * @param debug - Logger used to record request metadata and parse failures.
     * @returns Hono middleware that populates parsedBody on the context.
     */
    static createExtJsonMiddleware(debug: Logger): MiddlewareHandler<AppEnv> {
        return async (c: HonoContext<AppEnv>, next: () => Promise<void>) => {
            debug('request: %s %s', c.req.method, c.req.url);

            // ServerHost's Hono bodyLimit middleware has already accepted this body.
            // Hono reconstructs streamed bodies after counting them, so this remains
            // the only body read and requires no custom stream-management code.
            const rawJsonBody = await c.req.text();

            // Preserve exact JSON text for any future request-signature middleware.
            c.set('rawJsonBody', rawJsonBody);

            // Application routes decode bodies only when the client declares JSON.
            const contentType = c.req.header('content-type');
            if (contentType?.includes('application/json') && rawJsonBody.trim().length > 0) {
                try {
                    // Extended JSON revives typed values (Uint8Array blobs, etc.) that
                    // plain JSON cannot represent. This is the single HTTP decode point.
                    const parsed = fromExtendedJson(rawJsonBody);
                    debug('request body: %O', parsed);
                    c.set(PARSED_BODY_KEY, parsed);
                } catch (error) {
                    debug('invalid Extended JSON request: %O', error);
                    throw new ApplicationError(HTTP_STATUS_CODE_BAD_REQUEST, 'Invalid JSON in request body');
                }
            }

            await next();
        };
    }

    /**
     * Build a transport-neutral application request from the Hono context.
     *
     * @param context - Active request context with optional parsed body.
     */
    private createRequest(context: HonoContext<AppEnv>): ApplicationRequest {
        const body = context.get(PARSED_BODY_KEY);

        return {
            path: context.req.path,
            ...(body === undefined ? {} : { body }),
            headers: context.req.header(),
        };
    }

    /**
     * Run normal HTTP through a non-streaming connection and close it when the
     * handler returns. The connection buffers at most one route response.
     *
     * @param request - Application request derived from the HTTP context.
     */
    private async handleRequest(request: ApplicationRequest): Promise<Response> {
        const stream = new HttpRequestStream();

        try {
            await this.router.dispatch(request, stream);

            return HttpTransportRouter.toResponse(stream.getResponse());
        } finally {
            stream.close();
        }
    }

    /**
     * Keep SSE open until the dispatched route finishes.
     *
     * Subscription routes remain active for their subscription lifetime. Once
     * streaming starts, failures are events because HTTP status and headers have
     * already been committed.
     *
     * @param context - Active Hono context for the SSE response.
     * @param request - Application request derived from the HTTP context.
     */
    private openSse(context: HonoContext<AppEnv>, request: ApplicationRequest): Response {
        return streamSSE(context, async (streamApi) => {
            const stream = new HonoSSEStream(streamApi);
            this.activeSseStreams.add(stream);
            stream.onClose(() => this.activeSseStreams.delete(stream));

            // Close promptly when the client aborts the underlying HTTP request.
            context.req.raw.signal.addEventListener('abort', () => stream.close(), {
                once: true,
            });

            try {
                // Subscription routes remain pending until their broadcaster
                // registration is removed. Ordinary routes return immediately.
                await this.router.dispatch(request, stream);
            } catch (error) {
                this.debug('SSE dispatch failed for %s: %O', request.path, error);

                // SSE is a one-way stream: once streamSSE opens the response, the HTTP
                // status (200) and Content-Type (text/event-stream) are already sent.
                // Unlike normal HTTP, we cannot replace them with a 4xx/5xx Response.
                // The client must learn about failures from an SSE event instead.
                try {
                    await stream.send({
                        type: 'error',
                        // Same PublicError shape as HTTP and WebSocket so clients handle
                        // validation, auth, and application failures uniformly.
                        data: normalizePublicError(error),
                    });
                } catch (sendError) {
                    // The client may have disconnected before we could deliver the error
                    // event — there is no further recovery path for one-way SSE.
                    this.debug('failed to send SSE error event: %O', sendError);
                }
            } finally {
                // Hono closes SSE when this callback returns. Closing here is
                // idempotent if the client already ended a subscription.
                await stream.close();
            }
        });
    }

    /** Close every retained SSE response during application shutdown. */
    async stop(): Promise<void> {
        this.debug('closing %d active SSE stream(s)', this.activeSseStreams.size);

        await Promise.all(Array.from(this.activeSseStreams).map((stream) => stream.close()));
    }

    /**
     * Detect whether the client requested Server-Sent Events.
     *
     * @param context - Active Hono request context.
     */
    private static acceptsSse(context: HonoContext<AppEnv>): boolean {
        return (
            context.req
                .header('accept')
                ?.split(',')
                .some((value) => value.trim().startsWith('text/event-stream')) ?? false
        );
    }

    /**
     * Convert a buffered route response into an HTTP Response.
     *
     * @param response - Buffered response from HttpRequestStream, if any.
     */
    private static toResponse(response: StreamResponse | undefined): Response {
        if (!response || response.statusCode === HTTP_STATUS_CODE_NO_CONTENT) {
            return new Response(null, { status: HTTP_STATUS_CODE_NO_CONTENT });
        }

        return new Response(toExtendedJson(response.body), {
            status: response.statusCode,
            headers: { 'content-type': 'application/json' },
        });
    }

    /**
     * Handle failures which occur before an SSE response has been opened.
     *
     * @param debug - Logger used to record the underlying failure.
     * @returns Hono onError handler returning the public error contract.
     */
    static createErrorHandler(debug: Logger): ErrorHandler<AppEnv> {
        return (error: Error) => {
            debug('HTTP dispatch failed: %O', error);
            const normalized = normalizePublicError(error);

            return new Response(toExtendedJson(normalized), {
                status: normalized.statusCode,
                headers: { 'content-type': 'application/json' },
            });
        };
    }
}