ingest-monitor/server.mjs

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,
  })
})