Station 이벤트 지켜보기

#!/usr/bin/env node
// Station 이벤트 지켜보기 — endue OS 샘플.
//   입장권을 받아 SSE 로 붙고, 이벤트를 한 줄씩 찍는다. 끊기면 마지막 seq 부터 이어 받는다.
//
//   node station-events.mjs stn_…                 (토큰은 ENDUE_OS_TOKEN, 없으면 로컬 가짜 IdP 의 sub=ENDUE_OS_SUB)
//   node station-events.mjs stn_… device.status_changed,command.completed
const [stationId, types] = process.argv.slice(2);
if (!stationId) throw new Error('사용법: node station-events.mjs <station_id> [types]');
const BASE = (process.env.ENDUE_OS_URL ?? 'http://127.0.0.1:5173').replace(/\/+$/, '');
const API = `${BASE}/api/v1`;

// 로컬에서는 Station 을 만든 사용자와 같은 sub 로 토큰을 받아야 그 Station 이 보인다 (남의 것은 404).
const token =
  process.env.ENDUE_OS_TOKEN ??
  (await (await fetch(`${BASE}/idp/token?sub=${process.env.ENDUE_OS_SUB ?? 'alice'}&ttl=3600`, { method: 'POST' })).json()).access_token;

let lastSeq = null;

async function connect() {
  // 1회용 입장권 — 60초 안에 한 번. 브라우저의 EventSource 도 같은 방법으로 붙는다.
  const t = await fetch(`${API}/stations/${stationId}/event-tickets`, { method: 'POST', headers: { Authorization: `Bearer ${token}` } });
  if (!t.ok) throw new Error(`event-tickets → ${t.status}`);
  const { ticket } = (await t.json()).data;
  const qs = new URLSearchParams({ ticket });
  if (types) qs.set('types', types);
  if (lastSeq != null) qs.set('last_event_id', String(lastSeq));

  const res = await fetch(`${API}/stations/${stationId}/events?${qs}`);
  if (res.status !== 200) throw new Error(`events → ${res.status}`);
  console.log(`붙었다${lastSeq != null ? ` — seq ${lastSeq} 다음부터` : ''}`);

  const decoder = new TextDecoder();
  let buf = '';
  for await (const chunk of res.body) {
    buf += decoder.decode(chunk, { stream: true });
    let i;
    while ((i = buf.indexOf('\n\n')) >= 0) {
      const block = buf.slice(0, i);
      buf = buf.slice(i + 2);
      const field = (name) => block.split('\n').find((l) => l.startsWith(`${name}:`))?.slice(name.length + 1).trim();
      if (field('event') === 'resync') {
        // 링버퍼(1000개·15분) 밖 — REST 로 상태를 다시 읽고 처음부터 붙는다.
        console.log('resync — 상태를 다시 읽어야 한다');
        lastSeq = null;
        continue;
      }
      const data = field('data');
      if (!data) continue;
      const ev = JSON.parse(data);
      lastSeq = ev.seq;
      // resource.* 의 resource_id 가 곧 device_id 다.
      const brief = ev.type.startsWith('command.')
        ? `${ev.data.command} ${ev.data.status}${ev.data.duration_ms != null ? ` ${ev.data.duration_ms}ms` : ''}`
        : ev.type === 'device.status_changed'
          ? `${ev.resource_id ?? ''} ${ev.data.compute_status}${ev.data.status_reason ? ` (${ev.data.status_reason})` : ''} online=${ev.data.online}`
          : JSON.stringify(ev.data);
      console.log(`#${ev.seq} ${ev.occurred_at.slice(11, 19)} ${ev.type.padEnd(22)} ${brief}`);
    }
  }
}

for (;;) {
  try {
    await connect();
  } catch (e) {
    console.warn(String(e.message ?? e));
  }
  await new Promise((r) => setTimeout(r, 2000)); // 끊기면 잠시 뒤 이어 받는다
}