import WebSocket from 'ws';
// Shared by both server-side examples. No reconnect or automatic retry.
export async function openWebSocket(endpoint, query = {}) {
const key = process.env.API_KEY;
const model = process.env.API_MODEL;
const timeout = Number(process.env.API_WS_TIMEOUT_MS ?? 60000);
if (!key?.trim() || !model?.trim() || !Number.isInteger(timeout) || timeout < 100 || timeout > 120000) {
throw new Error('Set API_KEY, API_MODEL, and a 100..120000 ms timeout.');
}
let url;
try { url = new URL(process.env.API_BASE_URL); }
catch { throw new Error('Set a valid HTTP(S) API_BASE_URL including /v1.'); }
if (!['http:', 'https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash) {
throw new Error('API_BASE_URL must use HTTP(S) without credentials, query, or fragment.');
}
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
url.pathname = `${url.pathname.replace(/\/+$/, '')}/${endpoint}`;
for (const [name, value] of Object.entries(query)) url.searchParams.set(name, value);
const ws = new WebSocket(url, {
headers: { Authorization: `Bearer ${key}` },
followRedirects: false, perMessageDeflate: false,
handshakeTimeout: timeout, maxPayload: 1024 * 1024,
});
const queue = [];
let queuedBytes = 0;
let waiting;
let fault;
let closed = false;
let rejectOpen;
const opened = new Promise((resolve, reject) => {
rejectOpen = reject;
ws.once('open', resolve);
});
function fail(message) {
if (fault || closed) return;
fault = new Error(message);
rejectOpen(fault);
waiting?.reject(fault);
waiting = undefined;
ws.terminate();
}
const deadline = setTimeout(() => fail('WebSocket workflow deadline exceeded.'), timeout);
ws.on('unexpected-response', (_request, response) => {
response.resume();
fail(`WebSocket handshake failed (HTTP ${response.statusCode}).`);
});
ws.on('error', () => fail('WebSocket transport failed.'));
ws.on('close', () => fail('WebSocket closed before workflow completion.'));
ws.on('message', (data, isBinary) => {
if (closed || fault) return;
if (isBinary) { fail('Expected a JSON text message, not a binary frame.'); return; }
let event;
try { event = JSON.parse(data.toString()); }
catch { fail('Invalid JSON event.'); return; }
if (!event || typeof event !== 'object' || Array.isArray(event) || typeof event.type !== 'string') {
fail('Invalid event envelope.'); return;
}
if (event.type === 'error') { fail('Server returned a WebSocket error event.'); return; }
if (waiting) {
const pending = waiting;
waiting = undefined;
pending.resolve(event);
return;
}
queuedBytes += data.length;
if (queue.length >= 256 || queuedBytes > 4 * 1024 * 1024) {
fail('WebSocket receive queue limit exceeded.'); return;
}
queue.push({ event, bytes: data.length });
});
try { await opened; }
catch (error) { clearTimeout(deadline); throw error; }
return {
model,
async send(event) {
if (fault) throw fault;
if (closed || ws.readyState !== WebSocket.OPEN) throw new Error('WebSocket is not open.');
await new Promise((resolve, reject) => ws.send(JSON.stringify(event), (error) => {
if (error) reject(new Error('WebSocket send failed.'));
else resolve();
}));
},
async receive() {
if (fault) throw fault;
if (closed) throw new Error('WebSocket is closed.');
if (queue.length) {
const next = queue.shift();
queuedBytes -= next.bytes;
return next.event;
}
if (waiting) throw new Error('Only one event reader is allowed.');
return await new Promise((resolve, reject) => { waiting = { resolve, reject }; });
},
close() {
closed = true;
clearTimeout(deadline);
waiting?.reject(new Error('WebSocket reader closed.'));
waiting = undefined;
ws.close(1000);
// Bound shutdown even if the peer never acknowledges the close frame.
const cleanup = setTimeout(() => ws.terminate(), 1000);
cleanup.unref();
ws.once('close', () => clearTimeout(cleanup));
},
};
}