dragonflight/services/mam-api/src/index.js

import 'dotenv/config';
import express from 'express';
import cors from 'cors';
import session from 'express-session';
import connectPgSimple from 'connect-pg-simple';
const PgStore = connectPgSimple(session);
import os from 'node:os';
import { exec } from 'node:child_process';
import pool from './db/pool.js';
import { errorHandler } from './middleware/errors.js';
import { requireAuth, requireUiHeader, requireAdmin } from './middleware/auth.js';
import { loadS3ConfigFromDb } from './s3/client.js';

import authRouter from './routes/auth.js';
import tokensRouter from './routes/tokens.js';
import usersRouter from './routes/users.js';
// Routes
import assetsRouter from './routes/assets.js';
import projectsRouter from './routes/projects.js';
import binsRouter from './routes/bins.js';
import jobsRouter from './routes/jobs.js';
import captureRouter from './routes/capture.js';
import uploadRouter from './routes/upload.js';
import recordersRouter from './routes/recorders.js';
import playoutRouter from './routes/playout.js';
import settingsRouter from './routes/settings.js';
import amppRouter from './routes/ampp.js';
import groupsRouter   from './routes/groups.js';
import sequencesRouter from './routes/sequences.js';
import systemRouter  from './routes/system.js';
import clusterRouter from './routes/cluster.js';
import sdkRouter     from './routes/sdk.js';
import schedulesRouter from './routes/schedules.js';
import metricsRouter  from './routes/metrics.js';
import commentsRouter from './routes/comments.js';
import importsRouter  from './routes/imports.js';
import storageRouter  from './routes/storage.js';
import { startSchedulerLoop, stopSchedulerLoop } from './scheduler.js';
import { startCleanupLoop } from './tasks/cleanupTempSegments.js';

const app  = express();
const PORT = process.env.PORT || 3000;

const allowedOrigins = (process.env.ALLOWED_ORIGINS || '')
  .split(',').map(s => s.trim()).filter(Boolean);
app.use(cors({
  origin: (origin, cb) => {
    if (!origin) return cb(null, true);
    if (allowedOrigins.length === 0 || allowedOrigins.includes(origin)) return cb(null, true);
    console.warn('[cors] rejected origin:', origin);
    return cb(null, false);
  },
  credentials: true,
}));
app.use(express.json({ limit: '50mb' }));

if (process.env.TRUST_PROXY === 'true') app.set('trust proxy', 1);

if (process.env.AUTH_ENABLED === 'true') {
  app.use((req, res, next) => {
    if (req.secure) res.setHeader('Strict-Transport-Security', 'max-age=31536000; includeSubDomains');
    next();
  });
}

if (process.env.AUTH_ENABLED === 'true' && !process.env.SESSION_SECRET) {
  console.error('[fatal] SESSION_SECRET is required when AUTH_ENABLED=true');
  process.exit(1);
}

app.use(session({
  store: new PgStore({ pool, tableName: 'sessions', pruneSessionInterval: 60 * 15 }),
  secret: process.env.SESSION_SECRET,
  name:   'dragonflight.sid',
  cookie: {
    httpOnly: true,
    sameSite: 'lax',
    secure:   process.env.TRUST_PROXY === 'true',
    path:     '/',
    maxAge:   8 * 3600 * 1000,
  },
  rolling: false,
  resave:  false,
  saveUninitialized: false,
}));

app.get('/health', (_req, res) => res.json({ status: 'ok' }));

const UNAUTH_PATHS = new Set([
  '/auth/login', '/auth/login/totp', '/auth/setup', '/auth/setup-required',
  '/auth/google', '/auth/google/callback', '/auth/google/enabled',
]);
app.use('/api/v1', requireUiHeader);
app.use('/api/v1', (req, res, next) => {
  if (UNAUTH_PATHS.has(req.path)) return next();
  return requireAuth(req, res, next);
});

app.use('/api/v1/auth',    authRouter);
app.use('/api/v1/auth/users', requireAdmin, usersRouter);
app.use('/api/v1/users', requireAdmin, usersRouter);
app.use('/api/v1/auth/tokens', requireAuth, tokensRouter);
app.use('/api/v1/assets',    assetsRouter);
app.use('/api/v1/projects',  projectsRouter);
app.use('/api/v1/bins',      binsRouter);
app.use('/api/v1/jobs',      jobsRouter);
app.use('/api/v1/capture',   captureRouter);
app.use('/api/v1/upload',    uploadRouter);
app.use('/api/v1/recorders', recordersRouter);
app.use('/api/v1/playout',   playoutRouter);
app.use('/api/v1/settings',  settingsRouter);
app.use('/api/v1/ampp',      amppRouter);
app.use('/api/v1/groups',    requireAdmin, groupsRouter);
app.use('/api/v1/sequences', sequencesRouter);
app.use('/api/v1/system',    systemRouter);
app.use('/api/v1/cluster',   clusterRouter);
app.use('/api/v1/sdk',       sdkRouter);
app.use('/api/v1/schedules', schedulesRouter);
app.use('/api/v1/metrics',   metricsRouter);
app.use('/api/v1/assets/:assetId/comments', commentsRouter);
app.use('/api/v1/imports',   importsRouter);
app.use('/api/v1/storage',   storageRouter);

app.use(errorHandler);

import { readdirSync, readFileSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { dirname, join } from 'node:path';

const __dirnameMig = dirname(fileURLToPath(import.meta.url));
async function runMigrations() {
  const dir = join(__dirnameMig, 'db', 'migrations');
  let files = [];
  try { files = readdirSync(dir).filter(f => f.endsWith('.sql')).sort(); } catch { return; }

  await pool.query(`
    CREATE TABLE IF NOT EXISTS schema_migrations (
      filename     TEXT PRIMARY KEY,
      applied_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
      checksum_sha TEXT
    )
  `);

  const force = process.env.MIGRATIONS_FORCE === '1';
  const allowFailures = process.env.MIGRATIONS_ALLOW_FAILURES === '1';

  const appliedRes = await pool.query('SELECT filename FROM schema_migrations');
  const applied = new Set(appliedRes.rows.map(r => r.filename));

  for (const f of files) {
    if (!force && applied.has(f)) continue;
    const sql = readFileSync(join(dir, f), 'utf8');
    const client = await pool.connect();
    try {
      await client.query('BEGIN');
      await client.query(sql);
      await client.query(
        `INSERT INTO schema_migrations (filename) VALUES ($1)
         ON CONFLICT (filename) DO UPDATE SET applied_at = NOW()`,
        [f]
      );
      await client.query('COMMIT');
      console.log('[migration] applied ' + f);
    } catch (err) {
      await client.query('ROLLBACK').catch(() => {});
      console.error('[migration] FAILED ' + f + ': ' + err.message);
      client.release();
      if (allowFailures) continue;
      console.error('[migration] aborting startup. Set MIGRATIONS_ALLOW_FAILURES=1 to override.');
      process.exit(1);
    }
    client.release();
  }
}
await runMigrations();

await loadS3ConfigFromDb();

function getLocalIp() {
  if (process.env.NODE_IP) return process.env.NODE_IP;

  const ifaces = os.networkInterfaces();
  for (const name of Object.keys(ifaces)) {
    for (const iface of (ifaces[name] || [])) {
      if (iface.family === 'IPv4' && !iface.internal) return iface.address;
    }
  }
  return '127.0.0.1';
}

function detectGpus() {
  return new Promise(resolve => {
    exec(
      'nvidia-smi --query-gpu=index,name,memory.total --format=csv,noheader,nounits',
      { timeout: 5000 },
      (err, stdout) => {
        if (err || !stdout.trim()) return resolve([]);
        const gpus = stdout.trim().split('\n').map(line => {
          const parts = line.split(',').map(s => s.trim());
          return {
            index:     parseInt(parts[0], 10),
            name:      parts[1] || 'Unknown GPU',
            memory_mb: parseInt(parts[2], 10) || 0,
          };
        }).filter(g => !isNaN(g.index));
        resolve(gpus);
      }
    );
  });
}

// Primary mam-api node self-registers in cluster_nodes every 30s. Must write
// BOTH last_seen (legacy column) and last_seen_at (added by mig 031, used by
// playout failover) — otherwise the primary appears stale to the failover
// query and channels get re-placed off it incorrectly.
async function selfHeartbeat() {
  const load  = os.loadavg()[0];
  const total = os.totalmem();
  const used  = total - os.freemem();
  const gpus  = await detectGpus();

  const capabilities = { gpus, blackmagic: [] };

  pool.query(
    `INSERT INTO cluster_nodes
       (hostname, ip_address, role, version, api_url,
        cpu_usage, mem_used_mb, mem_total_mb, capabilities, last_seen, last_seen_at)
     VALUES ($1,$2,'primary',$3,$4,$5,$6,$7,$8,NOW(),NOW())
     ON CONFLICT (hostname) DO UPDATE SET
       ip_address   = EXCLUDED.ip_address,
       cpu_usage    = EXCLUDED.cpu_usage,
       mem_used_mb  = EXCLUDED.mem_used_mb,
       mem_total_mb = EXCLUDED.mem_total_mb,
       capabilities = EXCLUDED.capabilities,
       last_seen_at = NOW(),
       last_seen    = NOW()`,
    [
      process.env.NODE_HOSTNAME || os.hostname(),
      getLocalIp(),
      process.env.npm_package_version || null,
      `http://${getLocalIp()}:${PORT}`,
      parseFloat(load.toFixed(2)),
      Math.round(used  / 1024 / 1024),
      Math.round(total / 1024 / 1024),
      JSON.stringify(capabilities),
    ]
  ).catch(err => console.error('[cluster] heartbeat failed:', err.message));
}

setInterval(selfHeartbeat, 30_000);
selfHeartbeat();

const server = app.listen(PORT, () => {
  const authMode = process.env.AUTH_ENABLED === 'true' ? 'ENABLED' : 'DISABLED — dev mode (dev user attached to every request)';
  console.log(`MAM API listening on port ${PORT}`);
  console.log(`Authentication: ${authMode}`);
  if (process.env.AUTH_ENABLED === 'true' && process.env.TRUST_PROXY !== 'true') {
    console.warn('[auth] WARNING: AUTH_ENABLED=true but TRUST_PROXY=false — req.ip will be the proxy IP, login rate-limit will throttle all clients together. Set TRUST_PROXY=true when behind nginx/HTTPS.');
  }
  startSchedulerLoop();
  startCleanupLoop();
});

let _shuttingDown = false;
async function gracefulShutdown(signal) {
  if (_shuttingDown) return;
  _shuttingDown = true;
  console.log(`[shutdown] received ${signal} — closing gracefully…`);

  try { stopSchedulerLoop(); } catch (_) {}

  const killSwitch = setTimeout(() => {
    console.error('[shutdown] forced exit after 25s timeout');
    process.exit(1);
  }, 25_000);
  killSwitch.unref();

  await new Promise(resolve => server.close(resolve));

  try { await pool.end(); } catch (e) { console.warn('[shutdown] pool.end:', e.message); }

  console.log('[shutdown] clean exit');
  process.exit(0);
}

process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
process.on('SIGINT',  () => gracefulShutdown('SIGINT'));
process.on('uncaughtException', (err) => {
  console.error('[fatal] uncaughtException:', err);
  gracefulShutdown('uncaughtException');
});
process.on('unhandledRejection', (reason) => {
  console.error('[fatal] unhandledRejection:', reason);
});