dragonflight/services/mam-api/src/routes/cluster.js

import express from 'express';
import pool from '../db/pool.js';
import { requireAuth } from '../middleware/auth.js';

const router = express.Router();
router.use(requireAuth);

// If the agent reported Docker's default bridge IP (172.17.x) but the request
// itself came from a real LAN address, prefer the request source IP instead.
// We only check 172.17.x — the default docker0 bridge — not the full RFC1918
// 172.16/12 block, since real LANs (e.g. 172.18.91.x) fall in that range.
function pickIp(reportedIp, reqIp) {
  const clean = (s) => (s || '').replace(/^::ffff:/, '');
  const isDockerBridge = (ip) => /^172\.17\./.test(ip || '');
  const r = clean(reqIp);
  if (!reportedIp) return r || null;
  if (isDockerBridge(reportedIp) && r && !isDockerBridge(r)) return r;
  return reportedIp;
}

// GET / – list all registered cluster nodes with online status
router.get('/', async (req, res, next) => {
  try {
    const r = await pool.query(
      `SELECT *,
              EXTRACT(EPOCH FROM (NOW() - last_seen)) AS stale_seconds
       FROM cluster_nodes
       ORDER BY registered_at ASC`
    );
    res.json(r.rows.map(row => ({
      ...row,
      online: Number(row.stale_seconds) < 120,
    })));
  } catch (err) { next(err); }
});

// POST /heartbeat – upsert this node's registration (includes hardware capabilities)
router.post('/heartbeat', async (req, res, next) => {
  try {
    const {
      hostname, ip_address,
      role = 'worker', version, api_url,
      cpu_usage, mem_used_mb, mem_total_mb,
      capabilities, metadata,
    } = req.body;

    if (!hostname) return res.status(400).json({ error: 'hostname is required' });

    const effectiveIp = pickIp(ip_address, req.ip || req.socket?.remoteAddress);

    const r = await pool.query(
      `INSERT INTO cluster_nodes
         (hostname, ip_address, role, version, api_url,
          cpu_usage, mem_used_mb, mem_total_mb, last_seen, capabilities, metadata)
       VALUES ($1,$2,$3,$4,$5,$6,$7,$8,NOW(),$9,$10)
       ON CONFLICT (hostname) DO UPDATE SET
         ip_address   = EXCLUDED.ip_address,
         role         = EXCLUDED.role,
         version      = EXCLUDED.version,
         api_url      = EXCLUDED.api_url,
         cpu_usage    = EXCLUDED.cpu_usage,
         mem_used_mb  = EXCLUDED.mem_used_mb,
         mem_total_mb = EXCLUDED.mem_total_mb,
         last_seen    = NOW(),
         capabilities = EXCLUDED.capabilities,
         metadata     = EXCLUDED.metadata
       RETURNING *`,
      [
        hostname,
        effectiveIp,
        role,
        version      || null,
        api_url      || null,
        cpu_usage    != null ? cpu_usage    : null,
        mem_used_mb  != null ? mem_used_mb  : null,
        mem_total_mb != null ? mem_total_mb : null,
        capabilities != null ? JSON.stringify(capabilities) : '{}',
        metadata     != null ? JSON.stringify(metadata) : null,
      ]
    );
    res.json(r.rows[0]);
  } catch (err) { next(err); }
});

// GET /devices/blackmagic – flatten every node's DeckLink cards for the
// recorder picker. Returns one entry per device with the host node info.
router.get('/devices/blackmagic', async (req, res, next) => {
  try {
    const r = await pool.query(
      `SELECT id, hostname, ip_address, role, capabilities,
              EXTRACT(EPOCH FROM (NOW() - last_seen)) AS stale_seconds
       FROM cluster_nodes
       WHERE capabilities IS NOT NULL`
    );
    const out = [];
    for (const row of r.rows) {
      const online = Number(row.stale_seconds) < 120;
      const bm = (row.capabilities && row.capabilities.blackmagic) || [];
      const model = (row.capabilities && row.capabilities.blackmagic_model) || null;
      bm.forEach((d, idx) => {
        out.push({
          node_id:    row.id,
          hostname:   row.hostname,
          ip_address: row.ip_address,
          role:       row.role,
          online,
          model,
          index:      d.index !== undefined ? d.index : idx,
          device:     d.device,
        });
      });
    }
    res.json(out);
  } catch (err) { next(err); }
});

// GET /:id/ping – probe the node's api_url/health endpoint directly
router.get('/:id/ping', async (req, res, next) => {
  try {
    const r = await pool.query(
      'SELECT id, hostname, api_url FROM cluster_nodes WHERE id = $1',
      [req.params.id]
    );
    if (r.rowCount === 0) return res.status(404).json({ error: 'Node not found' });

    const node = r.rows[0];
    if (!node.api_url) return res.json({ reachable: false, reason: 'no api_url registered' });

    const start = Date.now();
    try {
      const upstream = await fetch(`${node.api_url}/health`, {
        signal: AbortSignal.timeout(4000),
      });
      const latency_ms = Date.now() - start;
      const body = await upstream.json().catch(() => ({}));
      res.json({ reachable: upstream.ok, latency_ms, status: upstream.status, agent: body });
    } catch (err) {
      res.json({ reachable: false, latency_ms: Date.now() - start, reason: err.message });
    }
  } catch (err) { next(err); }
});

// DELETE /:id – deregister a node
router.delete('/:id', async (req, res, next) => {
  try {
    const r = await pool.query(
      'DELETE FROM cluster_nodes WHERE id = $1 RETURNING id',
      [req.params.id]
    );
    if (r.rowCount === 0) return res.status(404).json({ error: 'Node not found' });
    res.json({ ok: true });
  } catch (err) { next(err); }
});

export default router;