feat(federation): reporter sidecar + central ingest (edge → control plane)

Phase 2 of the edge + control-plane architecture. A per-team local stack
can now mirror its state UP to a central instance so judges get a fleet
view — outbound-only, so it works from behind the room NAT.

Central ingest (same api/, central-mode):
- POST /instances/register + /instances/:id/heartbeat (fleet-secret
  gated) + an `instances` table and GET /instances (admin) fleet read.
- PUT /teams/:id and POST /submissions accept an optional `site` tag;
  `site` column added to teams + submissions (grouping/filtering).
- WS snapshot now carries instances; instance:update broadcast added.

Reporter sidecar (deploy/lan/reporter/, opt-in `federated` compose
profile): subscribes to the local WS feed and replays team/submission
writes up to central, namespaced by SITE_ID (ids never collide) and
site-tagged. Registers + heartbeats; failed writes queue in an in-memory
outbox and backfill on reconnect. Proven end-to-end (local→reporter→
central) before commit.

Also fixes a latent bug: listSubmissions selected snake_case columns but
mapped camelCase, so GET /submissions summaries were missing teamId/
teamName/submittedAt. Aliased the columns; added a regression assertion.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Omar Sobh
2026-07-21 23:18:03 -07:00
co-authored by Claude Opus 4.8
parent 60416cb8c7
commit 5ef14655b9
12 changed files with 429 additions and 17 deletions
+51 -1
View File
@@ -99,7 +99,9 @@ describe('collective API', () => {
it('lists submission summaries with a scored flag', async () => { it('lists submission summaries with a scored flag', async () => {
await request(app).post('/submissions').send({ teamId: 't1', code: 'c', add: fullAdd }) await request(app).post('/submissions').send({ teamId: 't1', code: 'c', add: fullAdd })
const before = await request(app).get('/submissions').set('X-Access-Code', JUDGE) const before = await request(app).get('/submissions').set('X-Access-Code', JUDGE)
expect(before.body[0].scored).toBe(false) // the summary must carry the identifying fields, not just `scored`
expect(before.body[0]).toMatchObject({ teamId: 't1', scored: false })
expect(typeof before.body[0].submittedAt).toBe('string')
await request(app).post('/scores').set('X-Access-Code', JUDGE).send({ teamId: 't1', total: 8 }) await request(app).post('/scores').set('X-Access-Code', JUDGE).send({ teamId: 't1', total: 8 })
const after = await request(app).get('/submissions').set('X-Access-Code', JUDGE) const after = await request(app).get('/submissions').set('X-Access-Code', JUDGE)
expect(after.body[0].scored).toBe(true) expect(after.body[0].scored).toBe(true)
@@ -121,4 +123,52 @@ describe('collective API', () => {
expect(events.some((e) => e.type === 'score:new')).toBe(true) expect(events.some((e) => e.type === 'score:new')).toBe(true)
}) })
}) })
describe('federation (central-mode)', () => {
const FLEET = 'fleet-secret'
let central: ReturnType<typeof createApp>
let cEvents: WsEvent[]
beforeEach(() => {
cEvents = []
central = createApp({
store: openStore(':memory:'),
broadcast: (e) => cEvents.push(e),
adminCode: ADMIN,
judgeCode: JUDGE,
fleetSecret: FLEET,
now: () => '2026-07-27T14:00:00.000Z',
})
})
it('registers an instance (fleet-secret gated) and broadcasts instance:update', async () => {
await request(central).post('/instances/register').send({ id: 'site-a', name: 'Team A laptop' }).expect(401)
const res = await request(central)
.post('/instances/register')
.set('x-fleet-secret', FLEET)
.send({ id: 'site-a', name: 'Team A laptop' })
.expect(201)
expect(res.body).toMatchObject({ id: 'site-a', name: 'Team A laptop', lastSeen: '2026-07-27T14:00:00.000Z' })
expect(cEvents.some((e) => e.type === 'instance:update')).toBe(true)
const list = await request(central).get('/instances').set('X-Access-Code', ADMIN).expect(200)
expect(list.body).toEqual([{ id: 'site-a', name: 'Team A laptop', lastSeen: '2026-07-27T14:00:00.000Z' }])
})
it('heartbeat upserts the same instance by id', async () => {
await request(central).post('/instances/register').set('x-fleet-secret', FLEET).send({ id: 'site-a' })
await request(central).post('/instances/site-a/heartbeat').set('x-fleet-secret', FLEET).send({}).expect(201)
const list = await request(central).get('/instances').set('X-Access-Code', ADMIN)
expect(list.body).toHaveLength(1) // still one instance, not duplicated
expect(list.body[0]).toMatchObject({ id: 'site-a', name: 'site-a' }) // name falls back to id
})
it('ingests site-tagged team + submission writes so central can group by site', async () => {
await request(central).put('/teams/site-a:team-1').send(team('team-1', { site: 'site-a' })).expect(204)
await request(central)
.post('/submissions')
.send({ teamId: 'site-a:team-1', code: 'c', add: fullAdd, site: 'site-a' })
.expect(201)
const teams = await request(central).get('/teams').set('X-Access-Code', ADMIN)
expect(teams.body[0]).toMatchObject({ id: 'site-a:team-1', site: 'site-a' })
})
})
}) })
+31
View File
@@ -49,6 +49,7 @@ export function createApp(opts: AppOptions): Express {
stats: { ...emptyStats, ...(b.stats ?? {}) }, stats: { ...emptyStats, ...(b.stats ?? {}) },
deviceConnected: !!b.deviceConnected, deviceConnected: !!b.deviceConnected,
updatedAt: typeof b.updatedAt === 'string' ? b.updatedAt : now(), updatedAt: typeof b.updatedAt === 'string' ? b.updatedAt : now(),
site: typeof b.site === 'string' ? b.site : '',
} }
store.upsertTeam(team) store.upsertTeam(team)
broadcast({ type: 'team:update', team }) broadcast({ type: 'team:update', team })
@@ -66,6 +67,7 @@ export function createApp(opts: AppOptions): Express {
code: b.code, code: b.code,
add: b.add, add: b.add,
submittedAt: typeof b.submittedAt === 'string' ? b.submittedAt : now(), submittedAt: typeof b.submittedAt === 'string' ? b.submittedAt : now(),
site: typeof b.site === 'string' ? b.site : '',
} }
const summary = store.upsertSubmission(dto) const summary = store.upsertSubmission(dto)
broadcast({ type: 'submission:new', submission: summary }) broadcast({ type: 'submission:new', submission: summary })
@@ -110,6 +112,11 @@ export function createApp(opts: AppOptions): Express {
res.json(store.leaderboard()) res.json(store.leaderboard())
}) })
// Central-mode fleet view: the local instances that have phoned home.
app.get('/instances', requireAnyCode(adminCode, judgeCode), (_req, res) => {
res.json(store.listInstances())
})
// --- ZeroClaw nodes ----------------------------------------------------- // --- ZeroClaw nodes -----------------------------------------------------
// Registration holds bearer tokens → admin only. Prompting is a public // Registration holds bearer tokens → admin only. Prompting is a public
// participant action (like PUT /teams/:id). List/broadcasts never leak tokens. // participant action (like PUT /teams/:id). List/broadcasts never leak tokens.
@@ -175,6 +182,30 @@ export function createApp(opts: AppOptions): Express {
res.status(201).json({ kitId: b.kitId, url, claimed: r.claimed }) res.status(201).json({ kitId: b.kitId, url, claimed: r.claimed })
}) })
// --- federation: local instances phone home (central-mode) --------------
// A per-team local (edge) stack's reporter sidecar registers itself and then
// heartbeats. Fleet-secret gated (same shared secret as self-register). The
// sidecar replays team/submission writes UP via the public PUT/POST routes,
// tagged with `site` (= the instance id), so central becomes the fleet view.
const registerInstance = (req: express.Request, res: express.Response) => {
if (!fleetSecret || !matches(req.header('x-fleet-secret') ?? '', fleetSecret)) {
return res.status(401).json({ error: 'unauthorized' })
}
const b = req.body ?? {}
const id = typeof b.id === 'string' ? b.id : req.params.id
if (typeof id !== 'string' || !id) return res.status(400).json({ error: 'id is required' })
const instance = {
id,
name: typeof b.name === 'string' && b.name ? b.name : id,
lastSeen: now(),
}
store.upsertInstance(instance)
broadcast({ type: 'instance:update', instance })
return res.status(201).json(instance)
}
app.post('/instances/register', registerInstance)
app.post('/instances/:id/heartbeat', registerInstance)
// A participant claims their powered-on board to their team by proving // A participant claims their powered-on board to their team by proving
// possession of the kit's claim code. Public + rate-limited (no operator in // possession of the kit's claim code. Public + rate-limited (no operator in
// the loop); the bearer token moves straight from the pool into the bridge // the loop); the bearer token moves straight from the pool into the bridge
+54 -15
View File
@@ -6,6 +6,7 @@ import type {
ScoreInput, ScoreInput,
ScoreDTO, ScoreDTO,
LeaderboardRow, LeaderboardRow,
InstanceDTO,
} from './types' } from './types'
import type { Board } from './claim' import type { Board } from './claim'
@@ -22,6 +23,10 @@ export interface Store {
saveBoard(b: Board): void saveBoard(b: Board): void
/** All persisted boards, to seed the in-memory registry on boot. */ /** All persisted boards, to seed the in-memory registry on boot. */
listBoards(): Board[] listBoards(): Board[]
/** Register/refresh a federated local instance (central-mode). */
upsertInstance(i: InstanceDTO): void
/** All known instances (central-mode fleet view). */
listInstances(): InstanceDTO[]
close(): void close(): void
} }
@@ -44,6 +49,7 @@ interface TeamRow {
stats: string stats: string
device_connected: number device_connected: number
updated_at: string updated_at: string
site: string
} }
function rowToTeam(r: TeamRow): TeamSnapshot { function rowToTeam(r: TeamRow): TeamSnapshot {
@@ -57,6 +63,7 @@ function rowToTeam(r: TeamRow): TeamSnapshot {
stats: JSON.parse(r.stats), stats: JSON.parse(r.stats),
deviceConnected: !!r.device_connected, deviceConnected: !!r.device_connected,
updatedAt: r.updated_at, updatedAt: r.updated_at,
site: r.site ?? '',
} }
} }
@@ -74,14 +81,16 @@ export function openStore(path = ':memory:'): Store {
phases TEXT NOT NULL DEFAULT '{}', phases TEXT NOT NULL DEFAULT '{}',
stats TEXT NOT NULL DEFAULT '{}', stats TEXT NOT NULL DEFAULT '{}',
device_connected INTEGER NOT NULL DEFAULT 0, device_connected INTEGER NOT NULL DEFAULT 0,
updated_at TEXT NOT NULL updated_at TEXT NOT NULL,
site TEXT NOT NULL DEFAULT ''
); );
CREATE TABLE IF NOT EXISTS submissions ( CREATE TABLE IF NOT EXISTS submissions (
team_id TEXT PRIMARY KEY, team_id TEXT PRIMARY KEY,
team_name TEXT NOT NULL DEFAULT '', team_name TEXT NOT NULL DEFAULT '',
code TEXT NOT NULL, code TEXT NOT NULL,
add_json TEXT NOT NULL, add_json TEXT NOT NULL,
submitted_at TEXT NOT NULL submitted_at TEXT NOT NULL,
site TEXT NOT NULL DEFAULT ''
); );
CREATE TABLE IF NOT EXISTS scores ( CREATE TABLE IF NOT EXISTS scores (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -100,35 +109,46 @@ export function openStore(path = ':memory:'): Store {
claim_code TEXT NOT NULL DEFAULT '', claim_code TEXT NOT NULL DEFAULT '',
claimed_by TEXT claimed_by TEXT
); );
CREATE TABLE IF NOT EXISTS instances (
id TEXT PRIMARY KEY,
name TEXT NOT NULL DEFAULT '',
last_seen TEXT NOT NULL
);
`) `)
// Migration for DBs created before the `domain` column existed. CREATE TABLE // Migrations for DBs created before a column existed. CREATE TABLE IF NOT
// IF NOT EXISTS won't add it to an existing table, so add it defensively. // EXISTS won't add columns to an existing table, so add them defensively.
try { for (const stmt of [
db.exec(`ALTER TABLE teams ADD COLUMN domain TEXT NOT NULL DEFAULT ''`) `ALTER TABLE teams ADD COLUMN domain TEXT NOT NULL DEFAULT ''`,
} catch { `ALTER TABLE teams ADD COLUMN site TEXT NOT NULL DEFAULT ''`,
/* column already exists — fine */ `ALTER TABLE submissions ADD COLUMN site TEXT NOT NULL DEFAULT ''`,
]) {
try {
db.exec(stmt)
} catch {
/* column already exists — fine */
}
} }
const upsertTeamStmt = db.prepare(` const upsertTeamStmt = db.prepare(`
INSERT INTO teams (id, name, kit, domain, members, phases, stats, device_connected, updated_at) INSERT INTO teams (id, name, kit, domain, members, phases, stats, device_connected, updated_at, site)
VALUES (@id, @name, @kit, @domain, @members, @phases, @stats, @device_connected, @updated_at) VALUES (@id, @name, @kit, @domain, @members, @phases, @stats, @device_connected, @updated_at, @site)
ON CONFLICT(id) DO UPDATE SET ON CONFLICT(id) DO UPDATE SET
name=excluded.name, kit=excluded.kit, domain=excluded.domain, members=excluded.members, name=excluded.name, kit=excluded.kit, domain=excluded.domain, members=excluded.members,
phases=excluded.phases, stats=excluded.stats, phases=excluded.phases, stats=excluded.stats,
device_connected=excluded.device_connected, updated_at=excluded.updated_at device_connected=excluded.device_connected, updated_at=excluded.updated_at, site=excluded.site
`) `)
const listTeamsStmt = db.prepare('SELECT * FROM teams ORDER BY name') const listTeamsStmt = db.prepare('SELECT * FROM teams ORDER BY name')
const getTeamStmt = db.prepare('SELECT * FROM teams WHERE id = ?') const getTeamStmt = db.prepare('SELECT * FROM teams WHERE id = ?')
const upsertSubStmt = db.prepare(` const upsertSubStmt = db.prepare(`
INSERT INTO submissions (team_id, team_name, code, add_json, submitted_at) INSERT INTO submissions (team_id, team_name, code, add_json, submitted_at, site)
VALUES (@team_id, @team_name, @code, @add_json, @submitted_at) VALUES (@team_id, @team_name, @code, @add_json, @submitted_at, @site)
ON CONFLICT(team_id) DO UPDATE SET ON CONFLICT(team_id) DO UPDATE SET
team_name=excluded.team_name, code=excluded.code, team_name=excluded.team_name, code=excluded.code,
add_json=excluded.add_json, submitted_at=excluded.submitted_at add_json=excluded.add_json, submitted_at=excluded.submitted_at, site=excluded.site
`) `)
const listSubsStmt = db.prepare(` const listSubsStmt = db.prepare(`
SELECT s.team_id, s.team_name, s.submitted_at, SELECT s.team_id AS teamId, s.team_name AS teamName, s.submitted_at AS submittedAt,
EXISTS(SELECT 1 FROM scores sc WHERE sc.team_id = s.team_id) AS scored EXISTS(SELECT 1 FROM scores sc WHERE sc.team_id = s.team_id) AS scored
FROM submissions s ORDER BY s.submitted_at FROM submissions s ORDER BY s.submitted_at
`) `)
@@ -158,6 +178,13 @@ export function openStore(path = ':memory:'): Store {
claim_code=excluded.claim_code, claimed_by=excluded.claimed_by claim_code=excluded.claim_code, claimed_by=excluded.claimed_by
`) `)
const listBoardsStmt = db.prepare('SELECT * FROM boards') const listBoardsStmt = db.prepare('SELECT * FROM boards')
const upsertInstanceStmt = db.prepare(`
INSERT INTO instances (id, name, last_seen)
VALUES (@id, @name, @last_seen)
ON CONFLICT(id) DO UPDATE SET
name=excluded.name, last_seen=excluded.last_seen
`)
const listInstancesStmt = db.prepare('SELECT * FROM instances ORDER BY name')
return { return {
upsertTeam(t) { upsertTeam(t) {
@@ -171,6 +198,7 @@ export function openStore(path = ':memory:'): Store {
stats: JSON.stringify(t.stats), stats: JSON.stringify(t.stats),
device_connected: t.deviceConnected ? 1 : 0, device_connected: t.deviceConnected ? 1 : 0,
updated_at: t.updatedAt, updated_at: t.updatedAt,
site: t.site ?? '',
}) })
}, },
listTeams() { listTeams() {
@@ -187,6 +215,7 @@ export function openStore(path = ':memory:'): Store {
code: s.code, code: s.code,
add_json: JSON.stringify(s.add), add_json: JSON.stringify(s.add),
submitted_at: s.submittedAt, submitted_at: s.submittedAt,
site: s.site ?? '',
}) })
// mark the team's final phase complete if we know the team // mark the team's final phase complete if we know the team
if (getTeamStmt.get(s.teamId)) setPhaseAddStmt.run(s.teamId) if (getTeamStmt.get(s.teamId)) setPhaseAddStmt.run(s.teamId)
@@ -252,6 +281,16 @@ export function openStore(path = ':memory:'): Store {
claimedBy: r.claimed_by, claimedBy: r.claimed_by,
})) }))
}, },
upsertInstance(i) {
upsertInstanceStmt.run({ id: i.id, name: i.name, last_seen: i.lastSeen })
},
listInstances() {
return (listInstancesStmt.all() as Array<{ id: string; name: string; last_seen: string }>).map((r) => ({
id: r.id,
name: r.name,
lastSeen: r.last_seen,
}))
},
close() { close() {
db.close() db.close()
}, },
+24 -1
View File
@@ -27,6 +27,9 @@ export interface TeamSnapshot {
stats: SessionStats stats: SessionStats
deviceConnected: boolean deviceConnected: boolean
updatedAt: string updatedAt: string
/** Federation tag: the local instance (site) this team belongs to. '' on a
* single-fleet deploy; set by the reporter sidecar on a central deploy. */
site?: string
} }
export interface SubmissionDTO { export interface SubmissionDTO {
@@ -35,6 +38,18 @@ export interface SubmissionDTO {
code: string code: string
add: AddLayers add: AddLayers
submittedAt: string submittedAt: string
/** Federation tag — see {@link TeamSnapshot.site}. */
site?: string
}
/**
* A running local (edge) stack that has registered with the central control
* plane. Identified by a stable `site` id; `lastSeen` drives online/offline.
*/
export interface InstanceDTO {
id: string
name: string
lastSeen: string
} }
export interface SubmissionSummary { export interface SubmissionSummary {
@@ -72,7 +87,13 @@ export interface LeaderboardRow {
export type NodeActivityKind = 'thinking' | 'tool' | 'flash' | 'error' | 'response' | 'fallback' export type NodeActivityKind = 'thinking' | 'tool' | 'flash' | 'error' | 'response' | 'fallback'
export type WsEvent = export type WsEvent =
| { type: 'snapshot'; teams: TeamSnapshot[]; submissions: SubmissionSummary[]; unclaimed?: string[] } | {
type: 'snapshot'
teams: TeamSnapshot[]
submissions: SubmissionSummary[]
unclaimed?: string[]
instances?: InstanceDTO[]
}
| { type: 'team:update'; team: TeamSnapshot } | { type: 'team:update'; team: TeamSnapshot }
| { type: 'submission:new'; submission: SubmissionSummary } | { type: 'submission:new'; submission: SubmissionSummary }
| { type: 'score:new'; teamId: string; total: number } | { type: 'score:new'; teamId: string; total: number }
@@ -80,3 +101,5 @@ export type WsEvent =
| { type: 'node:activity'; teamId: string; kind: NodeActivityKind; label: string; ts: string } | { type: 'node:activity'; teamId: string; kind: NodeActivityKind; label: string; ts: string }
// Kit ids of boards that have self-registered but aren't claimed yet. // Kit ids of boards that have self-registered but aren't claimed yet.
| { type: 'unclaimed:update'; kits: string[] } | { type: 'unclaimed:update'; kits: string[] }
// A local instance (site) registered or heartbeated on the central plane.
| { type: 'instance:update'; instance: InstanceDTO }
+1
View File
@@ -36,6 +36,7 @@ export function attachWs(
teams: store.listTeams(), teams: store.listTeams(),
submissions: store.listSubmissions(), submissions: store.listSubmissions(),
unclaimed: boards?.unclaimedKits().map((u) => u.kitId) ?? [], unclaimed: boards?.unclaimedKits().map((u) => u.kitId) ?? [],
instances: store.listInstances(),
}), }),
) )
hub.add(ws) hub.add(ws)
+9
View File
@@ -6,3 +6,12 @@ JUDGE_CODE=jdg-xxxxxxxx
FLEET_SECRET=change-me FLEET_SECRET=change-me
# Host port for the web UI. Use 8080 if the box can't bind privileged :80. # Host port for the web UI. Use 8080 if the box can't bind privileged :80.
WEB_PORT=80 WEB_PORT=80
# --- Federated mode (optional) ----------------------------------------------
# Only needed if you run the reporter sidecar (`--profile federated`) to mirror
# this instance UP to the central control plane. Omit for a purely local room.
# SITE_ID must be unique per team/instance — it namespaces every id at central.
SITE_ID=team-01
SITE_NAME=Team 01
# Central control-plane API base (its /api origin).
CENTRAL_API=https://apess.redclaw.dev/api
+25
View File
@@ -78,6 +78,31 @@ work): a rebooted board re-announces and the API refreshes its binding; a team
that lost its browser just re-scans the QR + re-enters the code to resume. For that lost its browser just re-scans the QR + re-enters the code to resume. For
anything stuck, release the kit from `/admin` and let the team re-claim. anything stuck, release the kit from `/admin` and let the team re-claim.
## 4. Federated mode (optional) — phone home to a central dashboard
For a multi-team event you can run **one local stack per team** (each with its
board) and have every instance mirror its state UP to a **central control
plane** — a fleet dashboard where judges see the whole cohort at once. This is
outbound-only, so it works from behind the room's NAT.
```sh
# in .env, set a UNIQUE SITE_ID per team + the central api base:
# SITE_ID=team-07 SITE_NAME="Team 07" CENTRAL_API=https://apess.redclaw.dev/api
docker compose --env-file .env -f docker-compose.yml --profile federated up -d --build
```
The **reporter** sidecar joins the local network, subscribes to the local API's
WS feed, and replays every `team:update` / submission UP to `CENTRAL_API`,
**namespaced by `SITE_ID`** (so ids never collide across instances) and
`site`-tagged (so central groups by team/site). It registers the instance and
heartbeats on a timer. If the uplink is down, writes queue in an in-memory
outbox and backfill on reconnect — the local workshop never blocks on it.
Central is **observe-only**: it never touches a board (it can't reach the NAT'd
boards — only the local stack drives them). Run the central instance from
`deploy/docker-compose.yml` with the same `FLEET_SECRET`; the reporter presents
it on every federation call.
## Notes ## Notes
- **Data** persists in the `apess-lan-data` volume (`docker compose down` keeps - **Data** persists in the `apess-lan-data` volume (`docker compose down` keeps
+25
View File
@@ -50,6 +50,31 @@ services:
# Not published: boards + browsers reach the API through the web's /api proxy. # Not published: boards + browsers reach the API through the web's /api proxy.
networks: [apess-lan] networks: [apess-lan]
# Optional edge→central reporter. Enable with `--profile federated`; mirrors
# this instance's state UP to the central control plane (outbound-only). Needs
# SITE_ID + CENTRAL_API set (see .env.example). Omit the profile for a purely
# local, offline single-room workshop.
reporter:
build:
context: ../..
dockerfile: deploy/lan/reporter/Dockerfile
image: apess-reporter:lan
container_name: apess-reporter-lan
restart: unless-stopped
profiles: [federated]
environment:
SITE_ID: ${SITE_ID:?set SITE_ID for federated mode}
SITE_NAME: ${SITE_NAME:-}
CENTRAL_API: ${CENTRAL_API:?set CENTRAL_API for federated mode}
FLEET_SECRET: ${FLEET_SECRET:?set FLEET_SECRET}
LOCAL_WS: ws://apess-api-lan:3000/ws
LOCAL_API: http://apess-api-lan:3000
# LOCAL_CODE auths the reporter's read of the local feed — reuse ADMIN_CODE.
LOCAL_CODE: ${ADMIN_CODE:?set ADMIN_CODE}
depends_on:
- apess-api
networks: [apess-lan]
networks: networks:
apess-lan: apess-lan:
driver: bridge driver: bridge
+8
View File
@@ -0,0 +1,8 @@
# APESS reporter sidecar — tiny outbound-only state mirror (edge → central).
FROM node:22-alpine
WORKDIR /app
ENV NODE_ENV=production
COPY deploy/lan/reporter/package.json ./
RUN npm install --omit=dev
COPY deploy/lan/reporter/index.mjs ./
CMD ["node", "index.mjs"]
+151
View File
@@ -0,0 +1,151 @@
// 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
+36
View File
@@ -0,0 +1,36 @@
{
"name": "apess-reporter",
"version": "1.0.0",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "apess-reporter",
"version": "1.0.0",
"dependencies": {
"ws": "^8.18.0"
}
},
"node_modules/ws": {
"version": "8.21.1",
"resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz",
"integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==",
"license": "MIT",
"engines": {
"node": ">=10.0.0"
},
"peerDependencies": {
"bufferutil": "^4.0.1",
"utf-8-validate": ">=5.0.2"
},
"peerDependenciesMeta": {
"bufferutil": {
"optional": true
},
"utf-8-validate": {
"optional": true
}
}
}
}
}
+14
View File
@@ -0,0 +1,14 @@
{
"name": "apess-reporter",
"version": "1.0.0",
"private": true,
"type": "module",
"description": "APESS edge→central reporter sidecar (outbound-only state mirror)",
"main": "index.mjs",
"scripts": {
"start": "node index.mjs"
},
"dependencies": {
"ws": "^8.18.0"
}
}