first commit

server.mjs

@@ -0,0 +1,273 @@
+import { createServer } from 'node:http'
+import { readFile } from 'node:fs/promises'
+import { extname, join, normalize } from 'node:path'
+import { fileURLToPath } from 'node:url'
+
+const port = Number(process.env.PORT ?? 8090)
+const host = process.env.HOST ?? '0.0.0.0'
+const forwardUrl = process.env.FORWARD_URL
+const maxBodyBytes = Number(process.env.MAX_BODY_BYTES ?? 5 * 1024 * 1024)
+const maxEvents = Number(process.env.MAX_EVENTS ?? 500)
+const publicDir = join(fileURLToPath(new URL('.', import.meta.url)), 'public')
+
+const events = []
+const clients = new Set()
+
+let receivedBatches = 0
+let receivedItems = 0
+let receivedBytes = 0
+let startedAt = Date.now()
+
+const contentTypes = {
+  '.css': 'text/css; charset=utf-8',
+  '.html': 'text/html; charset=utf-8',
+  '.js': 'text/javascript; charset=utf-8',
+  '.json': 'application/json; charset=utf-8',
+  '.svg': 'image/svg+xml',
+}
+
+function sendJson(res, status, body) {
+  res.writeHead(status, {
+    'content-type': 'application/json; charset=utf-8',
+    'access-control-allow-origin': '*',
+  })
+  res.end(JSON.stringify(body))
+}
+
+function readRequestBody(req) {
+  return new Promise((resolve, reject) => {
+    const chunks = []
+    let total = 0
+
+    req.on('data', (chunk) => {
+      total += chunk.length
+
+      if (total > maxBodyBytes) {
+        reject(new Error(`Request body is larger than ${maxBodyBytes} bytes`))
+        req.destroy()
+        return
+      }
+
+      chunks.push(chunk)
+    })
+
+    req.on('end', () => resolve(Buffer.concat(chunks)))
+    req.on('error', reject)
+  })
+}
+
+function normalizePayload(payload) {
+  const items = Array.isArray(payload) ? payload : [payload]
+  return items.map((item, index) => ({
+    edge: typeof item?.edge === 'string' ? item.edge : '',
+    timestamp:
+      typeof item?.timestamp === 'number' && Number.isFinite(item.timestamp)
+        ? item.timestamp
+        : Date.now(),
+    tag: typeof item?.tag === 'string' ? item.tag : `item.${index + 1}`,
+    value:
+      typeof item?.value === 'number' && Number.isFinite(item.value)
+        ? item.value
+        : Number(item?.value ?? 0),
+    raw: item,
+  }))
+}
+
+function buildSnapshot() {
+  return {
+    ...buildSummary(),
+    events,
+  }
+}
+
+function buildSummary() {
+  const latest = events[events.length - 1]
+  const now = Date.now()
+  const uptimeSeconds = Math.round((now - startedAt) / 1000)
+  const recent = events.filter((event) => now - event.receivedAt <= 60_000)
+
+  return {
+    startedAt,
+    uptimeSeconds,
+    receivedBatches,
+    receivedItems,
+    receivedBytes,
+    recentBatchesPerMinute: recent.length,
+    maxEvents,
+    latest,
+  }
+}
+
+function broadcast(message) {
+  const data = `data: ${JSON.stringify(message)}\n\n`
+
+  for (const client of clients) {
+    client.write(data)
+  }
+}
+
+async function handleIngest(req, res) {
+  let body
+
+  try {
+    body = await readRequestBody(req)
+  } catch (err) {
+    sendJson(res, 413, { ok: false, error: err.message })
+    return
+  }
+
+  let payload
+
+  try {
+    payload = JSON.parse(body.toString('utf8'))
+  } catch {
+    sendJson(res, 400, { ok: false, error: 'Expected JSON payload' })
+    return
+  }
+
+  const items = normalizePayload(payload)
+  const event = {
+    id: `${Date.now()}-${Math.random().toString(16).slice(2)}`,
+    receivedAt: Date.now(),
+    bytes: body.length,
+    count: items.length,
+    forward: forwardUrl ? { status: 'pending', url: forwardUrl } : undefined,
+    items,
+    payload,
+  }
+
+  events.push(event)
+  while (events.length > maxEvents) {
+    events.shift()
+  }
+
+  receivedBatches += 1
+  receivedItems += items.length
+  receivedBytes += body.length
+
+  console.log('ingest.received', {
+    id: event.id,
+    bytes: event.bytes,
+    count: event.count,
+  })
+
+  if (forwardUrl) {
+    try {
+      const forwardResponse = await fetch(forwardUrl, {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+        },
+        body,
+      })
+
+      event.forward = {
+        status: forwardResponse.ok ? 'ok' : 'error',
+        url: forwardUrl,
+        httpStatus: forwardResponse.status,
+        statusText: forwardResponse.statusText,
+      }
+    } catch (err) {
+      event.forward = {
+        status: 'error',
+        url: forwardUrl,
+        error: err.message,
+      }
+    }
+  }
+
+  broadcast({ type: 'ingest', event, summary: buildSummary() })
+  sendJson(res, 202, { ok: true, id: event.id, count: event.count })
+}
+
+function handleEvents(req, res) {
+  res.writeHead(200, {
+    'content-type': 'text/event-stream; charset=utf-8',
+    'cache-control': 'no-cache, no-transform',
+    connection: 'keep-alive',
+    'access-control-allow-origin': '*',
+  })
+
+  res.write(`data: ${JSON.stringify({ type: 'snapshot', snapshot: buildSnapshot() })}\n\n`)
+  clients.add(res)
+
+  req.on('close', () => {
+    clients.delete(res)
+  })
+}
+
+function resetMonitor(res) {
+  events.length = 0
+  receivedBatches = 0
+  receivedItems = 0
+  receivedBytes = 0
+  startedAt = Date.now()
+  broadcast({ type: 'snapshot', snapshot: buildSnapshot() })
+  sendJson(res, 200, { ok: true })
+}
+
+async function serveStatic(req, res) {
+  const url = new URL(req.url ?? '/', `http://${req.headers.host ?? 'localhost'}`)
+  const requestedPath = url.pathname === '/' ? '/index.html' : url.pathname
+  const normalized = normalize(requestedPath)
+    .replace(/^([/\\])+/, '')
+    .replace(/^(\.\.[/\\])+/, '')
+  const filePath = join(publicDir, normalized)
+
+  try {
+    const file = await readFile(filePath)
+    res.writeHead(200, {
+      'content-type': contentTypes[extname(filePath)] ?? 'application/octet-stream',
+    })
+    res.end(file)
+  } catch {
+    sendJson(res, 404, { ok: false, error: 'Not found' })
+  }
+}
+
+const server = createServer(async (req, res) => {
+  if (req.method === 'OPTIONS') {
+    res.writeHead(204, {
+      'access-control-allow-origin': '*',
+      'access-control-allow-methods': 'GET,POST,DELETE,OPTIONS',
+      'access-control-allow-headers': 'content-type',
+    })
+    res.end()
+    return
+  }
+
+  if (req.method === 'POST' && req.url?.startsWith('/api/ingest')) {
+    await handleIngest(req, res)
+    return
+  }
+
+  if (req.method === 'GET' && req.url?.startsWith('/api/events')) {
+    sendJson(res, 200, buildSnapshot())
+    return
+  }
+
+  if (req.method === 'DELETE' && req.url?.startsWith('/api/events')) {
+    resetMonitor(res)
+    return
+  }
+
+  if (req.method === 'GET' && req.url?.startsWith('/events')) {
+    handleEvents(req, res)
+    return
+  }
+
+  if (req.method === 'GET') {
+    await serveStatic(req, res)
+    return
+  }
+
+  sendJson(res, 405, { ok: false, error: 'Method not allowed' })
+})
+
+server.listen(port, host, () => {
+  console.log('ingest-monitor.started', {
+    url: `http://localhost:${port}`,
+    ingestUrl: `http://localhost:${port}/api/ingest`,
+    forwardUrl,
+  })
+})