88 lines
3.0 KiB
TypeScript
88 lines
3.0 KiB
TypeScript
import type { FastifyPluginAsync } from "fastify";
|
|
import type { WebSocket } from "@fastify/websocket";
|
|
import { createConnection } from "./connection.js";
|
|
import type { GatewayRegistry, GatewaySocket } from "./registry.js";
|
|
import type { OpencodeProbe } from "../../opencode.js";
|
|
import type { NetworkId } from "../../auth.js";
|
|
|
|
let nextConnectionId = 1;
|
|
|
|
// WS keepalive. The gateway is idle between requests, so without protocol-level
|
|
// traffic the CC:T client's socket hits its idle timeout and drops (code 1006
|
|
// "Timed out"), churning reconnects. We ping every interval; the client auto-pongs
|
|
// at the protocol level. A socket that misses a full interval (no pong) is presumed
|
|
// dead and terminated, which also reaps a rebooted machine's stale half-open entry
|
|
// before its new daemon connects (so newest-wins doesn't fire on a phantom).
|
|
const HEARTBEAT_INTERVAL_MS = 20_000;
|
|
|
|
export type GatewayModuleOptions = {
|
|
registry: GatewayRegistry;
|
|
probe: OpencodeProbe;
|
|
resolveNetwork: (secret: string | undefined) => NetworkId | null;
|
|
};
|
|
|
|
// Adapter: the only place that touches the real `ws` socket. Bridges socket events
|
|
// into the (pure) connection state machine. @fastify/websocket must be registered
|
|
// on a parent scope before this plugin.
|
|
const gatewayModule: FastifyPluginAsync<GatewayModuleOptions> = (fastify, opts) => {
|
|
// Open liveness check for the HTTP service itself — no auth, no opencode probe.
|
|
fastify.get("/health", () => ({ ok: true }));
|
|
|
|
fastify.get("/gateway", { websocket: true }, (socket: WebSocket, req) => {
|
|
const connId = `gw-${nextConnectionId++}`;
|
|
const log = fastify.log.child({ connId });
|
|
const gatewaySocket: GatewaySocket = {
|
|
send: (data) => socket.send(data),
|
|
close: (code, reason) => socket.close(code, reason),
|
|
};
|
|
|
|
log.info({ remoteAddress: req.ip }, "gateway connection opened");
|
|
|
|
const connection = createConnection({
|
|
socket: gatewaySocket,
|
|
registry: opts.registry,
|
|
probe: opts.probe,
|
|
resolveNetwork: opts.resolveNetwork,
|
|
log,
|
|
});
|
|
|
|
// Heartbeat: a pong (or any inbound frame) marks the socket live for the next
|
|
// interval; a silent interval terminates it.
|
|
let isAlive = true;
|
|
socket.on("pong", () => {
|
|
isAlive = true;
|
|
});
|
|
const heartbeat = setInterval(() => {
|
|
if (!isAlive) {
|
|
log.info("gateway connection heartbeat timeout; terminating");
|
|
socket.terminate();
|
|
return;
|
|
}
|
|
isAlive = false;
|
|
try {
|
|
socket.ping();
|
|
} catch {
|
|
// socket already closing; the close handler clears the interval
|
|
}
|
|
}, HEARTBEAT_INTERVAL_MS);
|
|
|
|
socket.on("message", (raw) => {
|
|
isAlive = true;
|
|
void connection.onMessage(raw);
|
|
});
|
|
socket.on("close", (code, reason) => {
|
|
clearInterval(heartbeat);
|
|
connection.onClose(code, reason.toString());
|
|
});
|
|
socket.on("error", (err) => {
|
|
clearInterval(heartbeat);
|
|
log.warn({ err }, "gateway socket error");
|
|
connection.onClose();
|
|
});
|
|
});
|
|
|
|
return Promise.resolve();
|
|
};
|
|
|
|
export default gatewayModule;
|