import type { IncomingMessage } from 'http';
import type { Duplex } from 'stream';
import { WebSocketServer, type WebSocket } from 'ws';
import { MAX_FRAME_SIZE_BYTES, decodeBinaryFrame, decodeJsonFrame, wsDataToBuffer, wsDataToString } from '../pilot/protocol';
import {
    TcpStreamSwitchboard,
    attachTcpStreamSwitchboard,
    resolveByComposeLabels,
    type ReverseTcpStreamHandle,
} from '../mesh/tcpStreamSwitchboard';
import { sanitizeForLog } from '../utils/safeLog';
import { isDebugEnabled } from '../utils/debug';
import { rejectUpgrade as reject } from './reject';

/**
 * Mesh proxy-tunnel ingress.
 *
 * The peer side of a proxy-mode mesh tunnel. Central dials
 * `WSS <api_url>/api/mesh/proxy-tunnel` using the fleet credential as a
 * Bearer header (either the `node_proxy` JWT generated by the remote's
 * Settings → Nodes → Generate Token flow, or a `full-admin` api_token
 * for non-fleet integrations); this handler upgrades the connection and
 * wires the shared `TcpStreamSwitchboard` to handle `tcp_open` /
 * `tcp_open_ack` / `tcp_open_reverse` / `tcp_close` and `TcpData` frames.
 *
 * Auth + scope gating happens in `upgradeHandler.ts` before this handler
 * runs: it accepts node_proxy and full-admin api_token, and rejects
 * session cookies plus restricted api_token scopes. The handler itself
 * trusts the upgrade; the WS credential is the only trust boundary.
 *
 * Bidirectional: when the tunnel opens, the handler registers itself as
 * the reverse-dialer on the local MeshService so meshed containers on
 * this Sencho can dial cross-node aliases via `tcp_open_reverse` over
 * the same WS. On disconnect the reverse-dialer registration is
 * cleared. Central maintains the bridge persistently (no idle close;
 * reconciled on a timer), so peer→central traffic always has a live
 * channel in steady state.
 */
const wss = new WebSocketServer({ noServer: true, maxPayload: MAX_FRAME_SIZE_BYTES });

interface SwitchboardReverseDialer {
    openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null;
}

export async function handleMeshProxyTunnel(req: IncomingMessage, socket: Duplex, head: Buffer): Promise<void> {
    // Mesh service is only available on the central deployment (SENCHO_MODE
    // unset or 'central'). Pilot-mode Sencho receives mesh traffic via the
    // pilot tunnel and has no use for the proxy-mode WS path.
    if (process.env.SENCHO_MODE === 'pilot') {
        return reject(socket, 404, 'Not Found');
    }

    // Central appends `?nodeId=<central-namespace-id>` so this peer can tag
    // its own MeshService with the right nodeId for `handleAccept` dispatch.
    // The value is unsigned, but the Bearer credential at the upgrade has
    // already proven the caller is the upstream central (auth + scope gate
    // in `upgradeHandler.ts`); the query param's trust ceiling is the
    // token's trust ceiling.
    // Strict regex (no parseInt) so `?nodeId=14.5`, `?nodeId=00014`,
    // `?nodeId=1e2`, leading whitespace, etc. are rejected rather than
    // silently truncated.
    let peerNodeId: number | null = null;
    try {
        const parsed = new URL(req.url ?? '/', `http://${req.headers.host || 'localhost'}`);
        const raw = parsed.searchParams.get('nodeId');
        if (raw != null && /^[1-9][0-9]*$/.test(raw)) {
            const n = Number(raw);
            if (Number.isSafeInteger(n)) peerNodeId = n;
        }
    } catch {
        // Malformed URL; treat as missing param.
    }
    if (peerNodeId === null) {
        console.warn('[MeshProxy] proxy-tunnel upgrade missing or malformed nodeId query param; reverse-direction mesh dispatch will be undefined until central upgrades');
    }

    await new Promise<void>((resolve) => {
        wss.handleUpgrade(req, socket as Parameters<typeof wss.handleUpgrade>[1], head, (ws) => {
            void attachSwitchboard(ws, peerNodeId).finally(resolve);
        });
    });
}

async function attachSwitchboard(ws: WebSocket, peerNodeId: number | null): Promise<void> {
    let switchboard: TcpStreamSwitchboard | null = null;
    let meshServiceCleanup: (() => void) | null = null;

    const onMessage = (data: unknown, isBinary: boolean): void => {
        if (!switchboard) return;
        try {
            if (isBinary) {
                const buf = wsDataToBuffer(data);
                if (!buf) return;
                switchboard.handleBinaryFrame(decodeBinaryFrame(buf));
                return;
            }
            const text = wsDataToString(data);
            if (text == null) return;
            switchboard.handleJsonFrame(decodeJsonFrame(text));
        } catch (err) {
            if (isDebugEnabled()) {
                console.warn('[MeshProxy:diag] malformed frame:', sanitizeForLog((err as Error).message));
            }
        }
    };

    const teardown = (): void => {
        ws.off('message', onMessage);
        if (switchboard) {
            switchboard.cleanup('mesh proxy-tunnel closed');
            switchboard = null;
        }
        if (meshServiceCleanup) {
            meshServiceCleanup();
            meshServiceCleanup = null;
        }
    };

    try {
        switchboard = attachTcpStreamSwitchboard({
            ws,
            resolveTarget: resolveByComposeLabels,
            logLabel: 'MeshProxy',
        });

        ws.on('message', onMessage);
        ws.once('close', teardown);
        ws.once('error', (err) => {
            if (isDebugEnabled()) {
                console.warn('[MeshProxy:diag] ws error:', sanitizeForLog(err.message));
            }
            teardown();
        });

        // Register a reverse dialer so this side's MeshForwarder can dial
        // cross-node aliases via `tcp_open_reverse` over the same WS. The
        // CAS swap refuses to overwrite a dialer that another caller
        // (a concurrent proxy-tunnel upgrade, or a pilot agent in a
        // misconfigured deployment) has already installed.
        const { MeshService } = await import('../services/MeshService');
        const meshService = MeshService.getInstance();
        const localSwitchboard = switchboard;
        const localDialer: SwitchboardReverseDialer = {
            openMeshTcpStream(target) {
                return localSwitchboard.openReverseStream(target);
            },
        };
        const installed = meshService.setReverseDialer(localDialer, null);
        if (!installed) {
            console.warn('[MeshProxy] reverse dialer already installed; rejecting concurrent tunnel');
            try { ws.close(1013, 'reverse dialer already installed'); } catch { /* ignore */ }
            return;
        }
        // Install central's view of this peer's nodeId so handleAccept
        // dispatches cross-node aliases correctly. Done after setReverseDialer
        // succeeds so a CAS-rejected tunnel does not leak nodeId state.
        //
        // The install persists across bridge lifecycles: the peer's identity
        // in central's namespace is stable for the enrollment, not per-WS.
        if (peerNodeId !== null) {
            meshService.setProxyTunnelSelfCentralNodeId(peerNodeId);
        }
        meshServiceCleanup = () => {
            meshService.setReverseDialer(null, localDialer);
        };
    } catch (err) {
        if (isDebugEnabled()) {
            console.warn('[MeshProxy:diag] failed to attach switchboard:', sanitizeForLog((err as Error).message));
        }
        try { ws.close(1011, 'switchboard attach failed'); } catch { /* ignore */ }
        return;
    }
}
