server.js

const http = require('http');
const fs = require('fs');
const path = require('path');
const zlib = require('zlib');
const { WebSocketServer } = require('ws');
const { Pool } = require('pg');
const { translateOtlpLogs, translateOtlpTraces } = require('./translate');


const dbPool = new Pool({
  connectionString: process.env.AGENT_GATEWAY_DATABASE_URL ||
    'postgres://agent_gateway_admin:agent_gateway_dev@localhost:5432/agent_gateway',
});

const PORT = 3000;
const MAX_EVENTS = 1000;

// In-memory event store (bounded)
let events = [];
const wsClients = new Set();

async function parseBody(req) {
  const buf = await parseBodyBuffer(req);
  return buf.length ? JSON.parse(buf.toString('utf8')) : {};
}

function parseBodyBuffer(req) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    req.on('data', chunk => { chunks.push(chunk); });
    req.on('end', () => {
      const buf = Buffer.concat(chunks);
      const enc = req.headers['content-encoding'] || '';
      if (enc.includes('gzip')) {
        zlib.gunzip(buf, (err, result) => err ? reject(err) : resolve(result));
      } else {
        resolve(buf);
      }
    });
    req.on('error', reject);
  });
}

function sendJson(res, statusCode, data) {
  const json = JSON.stringify(data);
  res.writeHead(statusCode, {
    'Content-Type': 'application/json',
    'Content-Length': Buffer.byteLength(json),
  });
  res.end(json);
}

function broadcastEvent(event) {
  const msg = JSON.stringify(event);
  for (const client of wsClients) {
    if (client.readyState === 1) { // OPEN
      client.send(msg);
    }
  }
}

// Serve static files from /app/public (+ shared theme.css)
function serveStatic(req, res) {
  const publicDir = path.join(__dirname, 'public');

  // Decode and strip the query/fragment before touching the filesystem so
  // encoded traversal (e.g. %2e%2e) can't slip past the boundary check.
  let pathname;
  try {
    pathname = decodeURIComponent(new URL(req.url, 'http://localhost').pathname);
  } catch (_err) {
    res.writeHead(400);
    return res.end('Bad request');
  }

  let filePath;
  if (pathname === '/theme.css') {
    filePath = path.join(__dirname, 'theme.css');
  } else if (pathname === '/registry-panel.js') {
    filePath = path.join(__dirname, 'registry-panel.js');
  } else {
    // Resolve the request against publicDir and confirm it stays inside it.
    const rel = pathname === '/' ? 'index.html' : '.' + path.posix.normalize(pathname);
    filePath = path.resolve(publicDir, rel);
    if (filePath !== publicDir && !filePath.startsWith(publicDir + path.sep)) {
      res.writeHead(403);
      return res.end('Forbidden');
    }
  }

  const ext = path.extname(filePath);
  const contentTypes = {
    '.html': 'text/html',
    '.js': 'text/javascript',
    '.css': 'text/css',
    '.json': 'application/json',
    '.png': 'image/png',
    '.svg': 'image/svg+xml',
  };

  fs.readFile(filePath, (err, data) => {
    if (err) {
      res.writeHead(404);
      return res.end('Not found');
    }
    res.writeHead(200, { 'Content-Type': contentTypes[ext] || 'application/octet-stream' });
    res.end(data);
  });
}

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url, `http://localhost:${PORT}`);
  const pathName = url.pathname;
  const method = req.method;

  try {
    // Health check
    if (pathName === '/health' && method === 'GET') {
      return sendJson(res, 200, { status: 'ok' });
    }

    // GET /events - retrieve event history
    if (pathName === '/events' && method === 'GET') {
      return sendJson(res, 200, events);
    }

    // POST /api/events - simple event ingestion (no OTLP wrapping required)
    // Body: { "event_type": "agent.prompted", "source": "agent-platform", "attributes": { ... } }
    if (pathName === '/api/events' && method === 'POST') {
      const body = await parseBody(req);
      const ev = {
        source: body.source || 'unknown',
        event_type: body.event_type || 'unknown',
        timestamp: body.timestamp || new Date().toISOString(),
        attributes: body.attributes || {},
        received_at: new Date().toISOString(),
      };
      events.push(ev);
      if (events.length > MAX_EVENTS) events = events.slice(-MAX_EVENTS);
      broadcastEvent(ev);
      return sendJson(res, 200, {});
    }

    // POST /v1/logs - OTLP HTTP/JSON logs ingestion
    if (pathName === '/v1/logs' && method === 'POST') {
      const body = await parseBody(req);
      for (const ev of translateOtlpLogs(body)) {
        ev.received_at = new Date().toISOString();
        events.push(ev);
        if (events.length > MAX_EVENTS) events = events.slice(-MAX_EVENTS);
        broadcastEvent(ev);
      }
      return sendJson(res, 200, { partialSuccess: {} });
    }

    // POST /v1/traces - OTLP HTTP/JSON traces ingestion (optionally gzip-encoded).
    if (pathName === '/v1/traces' && method === 'POST') {
      const body = await parseBody(req);
      for (const ev of translateOtlpTraces(body)) {
        ev.received_at = new Date().toISOString();
        events.push(ev);
        if (events.length > MAX_EVENTS) events = events.slice(-MAX_EVENTS);
        broadcastEvent(ev);
      }
      return sendJson(res, 200, { partialSuccess: {} });
    }

    // GET /api/users - list of active principal signing keys as user nodes
    if (pathName === '/api/users' && method === 'GET') {
      try {
        const result = await dbPool.query(
          `SELECT key_id AS id FROM principal_signing_keys
           WHERE revoked_at IS NULL AND not_before <= now() AND not_after > now()
           ORDER BY key_id`
        );
        return sendJson(res, 200, result.rows);
      } catch (_err) {
        return sendJson(res, 200, []);
      }
    }

    // GET /api/registry - snapshot of principals and agent permissions from Postgres
    if (pathName === '/api/registry' && method === 'GET') {
      try {
        const [keysResult, delegationsResult, permsResult] = await Promise.all([
          dbPool.query(
            `SELECT key_id FROM principal_signing_keys
             WHERE revoked_at IS NULL AND not_before <= now() AND not_after > now()
             ORDER BY key_id`
          ),
          dbPool.query(
            `SELECT signing_key_id, destination
             FROM principal_key_permissions
             WHERE revoked_at IS NULL AND not_before <= now() AND not_after > now()
             ORDER BY signing_key_id, destination`
          ),
          dbPool.query(
            `SELECT DISTINCT signing_key_id, subject_identity, destination
             FROM permission_registry
             WHERE revoked_at IS NULL AND not_before <= now() AND not_after > now()
             ORDER BY subject_identity, destination`
          ),
        ]);

        const humanMap = new Map();
        for (const row of keysResult.rows) {
          humanMap.set(row.key_id, { id: row.key_id, permissions: [] });
        }
        for (const row of delegationsResult.rows) {
          const h = humanMap.get(row.signing_key_id);
          if (h) h.permissions.push(row.destination);
        }

        const agentMap = new Map();
        for (const row of permsResult.rows) {
          if (!agentMap.has(row.subject_identity)) {
            agentMap.set(row.subject_identity, {
              id: row.subject_identity,
              created_by: row.signing_key_id,
              permissions: [],
            });
          }
          agentMap.get(row.subject_identity).permissions.push(row.destination);
        }

        return sendJson(res, 200, {
          humans: Array.from(humanMap.values()),
          agents: Array.from(agentMap.values()),
        });
      } catch (_err) {
        return sendJson(res, 502, { error: 'registry unavailable' });
      }
    }

    // Serve static files for everything else
    serveStatic(req, res);
  } catch (e) {
    sendJson(res, 400, { error: e.message });
  }
});

// WebSocket server on /ws path
const wss = new WebSocketServer({ server, path: '/ws' });

wss.on('connection', (ws) => {
  wsClients.add(ws);

  // Send event history on connect, sorted chronologically
  const sortedEvents = [...events].sort((a, b) => (a.timestamp < b.timestamp ? -1 : a.timestamp > b.timestamp ? 1 : 0));
  ws.send(JSON.stringify({ type: 'history', events: sortedEvents }));

  ws.on('close', () => {
    wsClients.delete(ws);
  });

  ws.on('error', () => {
    wsClients.delete(ws);
  });
});

server.listen(PORT, () => {
  console.log(`Dashboard listening on port ${PORT}`);
});