apress/deploy/lan/reporter/index.mjs

// APESS reporter sidecar — mirrors one local (edge) stack's state UP to the
// central control plane. Outbound-only (NAT-friendly): it never accepts inbound
// traffic, it just consumes the LOCAL api's WS feed and replays the public write
// contract (PUT /teams/:id, POST /submissions) to CENTRAL, tagged with `site`.
//
// Failed central writes queue in an in-memory outbox and retry, so a flaky or
// offline uplink never blocks — the workshop runs fully on the local stack and
// central backfills when the link returns.
//
// Env:
//   SITE_ID        stable id for this instance (namespaces every id at central)
//   SITE_NAME      human label for the fleet dashboard (defaults to SITE_ID)
//   LOCAL_WS       ws url of the local api feed   (default ws://apess-api-lan:3000/ws)
//   LOCAL_API      http base of the local api     (default http://apess-api-lan:3000)
//   LOCAL_CODE     admin code — auths the WS + the full-submission read
//   CENTRAL_API    http base of the central api   (e.g. https://apess.redclaw.dev/api)
//   FLEET_SECRET   shared secret central gates federation on
//   HEARTBEAT_MS   instance heartbeat cadence     (default 30000)
import WebSocket from 'ws'

const SITE_ID = must('SITE_ID')
const SITE_NAME = process.env.SITE_NAME || SITE_ID
const LOCAL_WS = process.env.LOCAL_WS || 'ws://apess-api-lan:3000/ws'
const LOCAL_API = (process.env.LOCAL_API || 'http://apess-api-lan:3000').replace(/\/$/, '')
const LOCAL_CODE = must('LOCAL_CODE')
const CENTRAL_API = must('CENTRAL_API').replace(/\/$/, '')
const FLEET_SECRET = must('FLEET_SECRET')
const HEARTBEAT_MS = Number(process.env.HEARTBEAT_MS || 30000)

function must(name) {
  const v = process.env[name]
  if (!v) {
    console.error(`[reporter] missing required env ${name}`)
    process.exit(1)
  }
  return v
}
const log = (...a) => console.log('[reporter]', ...a)
const nsId = (id) => `${SITE_ID}:${id}` // namespace a local id so it never collides at central

// --- offline outbox: {key, run: () => fetch-promise} ------------------------
// keyed so a newer team snapshot collapses the older one still queued.
const outbox = new Map()
function enqueue(key, run) {
  outbox.set(key, run)
}
async function flush() {
  for (const [key, run] of [...outbox]) {
    try {
      await run()
      outbox.delete(key)
    } catch (e) {
      // leave it queued; try again next tick
      log(`outbox retry pending (${outbox.size}) — ${key}: ${e.message}`)
      break // preserve order; stop on first failure
    }
  }
}

async function central(path, body, method = 'POST') {
  const res = await fetch(`${CENTRAL_API}${path}`, {
    method,
    headers: { 'content-type': 'application/json', 'x-fleet-secret': FLEET_SECRET },
    body: JSON.stringify(body),
  })
  if (!res.ok) throw new Error(`${method} ${path} → ${res.status}`)
  return res
}

// --- replay the public write contract UP, namespaced + site-tagged ----------
function pushTeam(team) {
  const id = nsId(team.id)
  enqueue(`team:${id}`, () =>
    central(`/teams/${encodeURIComponent(id)}`, { ...team, id, site: SITE_ID }, 'PUT'),
  )
}
function pushSubmission(sub) {
  const teamId = nsId(sub.teamId)
  enqueue(`sub:${teamId}`, () =>
    central('/submissions', { ...sub, teamId, site: SITE_ID }),
  )
}

// The WS submission:new event only carries a summary — fetch the full record
// (add layers + code) from the LOCAL api before replaying it up.
async function fetchFullSubmission(localTeamId) {
  const res = await fetch(`${LOCAL_API}/submissions/${encodeURIComponent(localTeamId)}`, {
    headers: { 'x-access-code': LOCAL_CODE },
  })
  if (!res.ok) throw new Error(`local GET /submissions/${localTeamId} → ${res.status}`)
  return res.json()
}

// --- instance registration + heartbeat --------------------------------------
async function register() {
  enqueue('instance', () => central('/instances/register', { id: SITE_ID, name: SITE_NAME }))
}
function startHeartbeat() {
  setInterval(() => {
    enqueue('instance', () =>
      central(`/instances/${encodeURIComponent(SITE_ID)}/heartbeat`, { name: SITE_NAME }),
    )
    void flush()
  }, HEARTBEAT_MS)
}

// --- local WS subscription (auto-reconnecting) ------------------------------
function connect() {
  const ws = new WebSocket(`${LOCAL_WS}?code=${encodeURIComponent(LOCAL_CODE)}`)
  ws.on('open', () => log(`connected to local feed ${LOCAL_WS}`))
  ws.on('message', async (raw) => {
    let ev
    try {
      ev = JSON.parse(raw.toString())
    } catch {
      return
    }
    if (ev.type === 'snapshot') {
      // backfill: replay every known team + submission on (re)connect
      for (const t of ev.teams ?? []) pushTeam(t)
      for (const s of ev.submissions ?? []) {
        try {
          pushSubmission(await fetchFullSubmission(s.teamId))
        } catch (e) {
          log(`snapshot submission skip ${s.teamId}: ${e.message}`)
        }
      }
    } else if (ev.type === 'team:update') {
      pushTeam(ev.team)
    } else if (ev.type === 'submission:new') {
      try {
        pushSubmission(await fetchFullSubmission(ev.submission.teamId))
      } catch (e) {
        log(`submission fetch failed ${ev.submission.teamId}: ${e.message}`)
      }
    }
    void flush()
  })
  ws.on('close', () => {
    log('local feed closed — reconnecting in 3s')
    setTimeout(connect, 3000)
  })
  ws.on('error', (e) => log(`local feed error: ${e.message}`))
}

log(`starting — site=${SITE_ID} → central ${CENTRAL_API}`)
await register()
void flush()
startHeartbeat()
connect()
setInterval(() => void flush(), 15000) // periodic drain even when idle