const crypto = require('crypto'); const http = require('http'); const PORT = Number(process.env.CRANK_E2E_STREAM_FIXTURE_PORT || 3310); function writeJson(response, status, payload) { response.writeHead(status, { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-store', }); response.end(JSON.stringify(payload)); } function readJsonBody(request) { return new Promise((resolve, reject) => { let buffer = ''; request.setEncoding('utf8'); request.on('data', (chunk) => { buffer += chunk; }); request.on('end', () => { if (!buffer) { resolve({}); return; } try { resolve(JSON.parse(buffer)); } catch (error) { reject(error); } }); request.on('error', reject); }); } const server = http.createServer((request, response) => { const url = new URL(request.url, `http://127.0.0.1:${PORT}`); if (url.pathname === '/health') { writeJson(response, 200, { ok: true }); return; } if (url.pathname === '/sse/logs') { response.writeHead(200, { 'Content-Type': 'text/event-stream; charset=utf-8', 'Cache-Control': 'no-cache', Connection: 'keep-alive', }); const events = [ { level: 'info', message: 'billing started', cursor: 'c1' }, { level: 'warn', message: 'cache warmup slow', cursor: 'c2' }, { level: 'error', message: 'invoice timeout', cursor: 'c3' }, ]; let index = 0; const timer = setInterval(() => { if (index >= events.length) { clearInterval(timer); response.end(); return; } response.write('event: message\n'); response.write(`data: ${JSON.stringify(events[index])}\n\n`); index += 1; }, 150); request.on('close', () => { clearInterval(timer); }); return; } if (url.pathname === '/snapshot/metrics') { writeJson(response, 200, { summary: { service: 'billing', error_rate: 0.12, }, items: [ { timestamp: '2026-04-06T10:00:00Z', cpu: 0.41, memory: 0.67 }, { timestamp: '2026-04-06T10:00:05Z', cpu: 0.39, memory: 0.65 }, ], done: true, }); return; } if (url.pathname === '/crm/leads' && request.method === 'POST') { readJsonBody(request) .then((payload) => { writeJson(response, 200, { id: 'lead_123', email: payload.email || null, }); }) .catch(() => { writeJson(response, 400, { error: 'invalid_json' }); }); return; } if (url.pathname === '/soap/leads' && request.method === 'POST') { let buffer = ''; request.setEncoding('utf8'); request.on('data', (chunk) => { buffer += chunk; }); request.on('end', () => { const hasEmail = buffer.includes('user@example.com') || buffer.includes('user@example.com'); if (!hasEmail) { response.writeHead(400, { 'Content-Type': 'text/xml; charset=utf-8', 'Cache-Control': 'no-store', }); response.end([ '', '', '', 'Client', 'missing email', '', '', '', ].join('')); return; } response.writeHead(200, { 'Content-Type': 'text/xml; charset=utf-8', 'Cache-Control': 'no-store', }); response.end([ '', '', '', 'lead_soap_123', 'created', '', '', '', ].join('')); }); request.on('error', () => { response.writeHead(500, { 'Content-Type': 'text/plain; charset=utf-8', 'Cache-Control': 'no-store', }); response.end('fixture_error'); }); return; } writeJson(response, 404, { error: 'not_found' }); }); function writeWebSocketFrame(socket, payload) { const body = Buffer.from(payload, 'utf8'); const header = []; header.push(0x81); if (body.length < 126) { header.push(body.length); } else if (body.length < 65536) { header.push(126, (body.length >> 8) & 0xff, body.length & 0xff); } else { throw new Error('fixture payload is unexpectedly large'); } socket.write(Buffer.concat([Buffer.from(header), body])); } function acceptWebSocket(request, socket) { const key = request.headers['sec-websocket-key']; if (!key) { socket.destroy(); return; } const accept = crypto .createHash('sha1') .update(`${key}258EAFA5-E914-47DA-95CA-C5AB0DC85B11`) .digest('base64'); socket.write( [ 'HTTP/1.1 101 Switching Protocols', 'Upgrade: websocket', 'Connection: Upgrade', `Sec-WebSocket-Accept: ${accept}`, '\r\n', ].join('\r\n'), ); const messages = [ { type: 'tick', seq: 1, value: 101 }, { type: 'tick', seq: 2, value: 102 }, { type: 'tick', seq: 3, value: 103 }, ]; let index = 0; const timer = setInterval(() => { if (index >= messages.length) { clearInterval(timer); socket.end(); return; } writeWebSocketFrame(socket, JSON.stringify(messages[index])); index += 1; }, 150); socket.on('close', () => clearInterval(timer)); socket.on('end', () => clearInterval(timer)); socket.on('error', () => clearInterval(timer)); } server.on('upgrade', (request, socket, head) => { if (request.url !== '/events') { socket.destroy(); return; } if (head && head.length) { socket.unshift(head); } acceptWebSocket(request, socket); }); server.listen(PORT, '127.0.0.1', () => { console.log(`Streaming fixtures listening on http://127.0.0.1:${PORT}`); });