crank/apps/ui/scripts/stream-fixture-server.js

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('<email>user@example.com</email>')
        || buffer.includes('<m:email>user@example.com</m:email>');

      if (!hasEmail) {
        response.writeHead(400, {
          'Content-Type': 'text/xml; charset=utf-8',
          'Cache-Control': 'no-store',
        });
        response.end([
          '<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">',
          '<soap:Body>',
          '<soap:Fault>',
          '<faultcode>Client</faultcode>',
          '<faultstring>missing email</faultstring>',
          '</soap:Fault>',
          '</soap:Body>',
          '</soap:Envelope>',
        ].join(''));
        return;
      }

      response.writeHead(200, {
        'Content-Type': 'text/xml; charset=utf-8',
        'Cache-Control': 'no-store',
      });
      response.end([
        '<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">',
        '<soap:Body>',
        '<CreateLeadResponse>',
        '<id>lead_soap_123</id>',
        '<status>created</status>',
        '</CreateLeadResponse>',
        '</soap:Body>',
        '</soap:Envelope>',
      ].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}`);
});