WebSocket hub: device GPS/SOS, PTT floor control, PCM audio relay

server/src/realtime.js

@@ -0,0 +1,202 @@
+import jwt from 'jsonwebtoken';
+import { JWT_SECRET } from './auth.js';
+import db from './db.js';
+
+// Binary audio frame: [1B type=0x01][4B groupId BE][4B memberId BE][PCM Int16LE @16kHz]
+const AUDIO_TYPE = 0x01;
+const HDR = 9;
+
+const memberById = (id) => db.prepare('SELECT * FROM members WHERE id = ?').get(id);
+const allMembers = () =>
+  db.prepare(`SELECT m.*, g.name AS group_name FROM members m
+              LEFT JOIN talkgroups g ON g.id = m.group_id ORDER BY m.id`).all();
+
+export default class Hub {
+  constructor(wss) {
+    this.wss = wss;
+    this.clients = new Set();
+    this.floors = new Map(); // groupId -> { client, memberId, name }
+    setInterval(() => this.heartbeat(), 30000);
+    wss.on('connection', (ws) => this.onConnection(ws));
+  }
+
+  heartbeat() {
+    for (const ws of this.clients) {
+      if (ws.isAlive === false) { ws.terminate(); this.onClose(ws); continue; }
+      ws.isAlive = false;
+      ws.ping();
+    }
+  }
+
+  onConnection(ws) {
+    ws.isAlive = true;
+    ws.client = null;
+    ws.on('pong', () => { ws.isAlive = true; });
+    ws.on('message', (data, isBinary) => {
+      try {
+        if (isBinary) return this.onAudio(ws, Buffer.from(data));
+        const msg = JSON.parse(data.toString('utf8'));
+        if (!ws.client) return this.onHello(ws, msg);
+        this.onMessage(ws, msg);
+      } catch {
+        this.send(ws, { type: 'error', message: 'malformed message' });
+      }
+    });
+    ws.on('close', () => this.onClose(ws));
+    ws.on('error', () => {});
+  }
+
+  verify(token) {
+    try { return jwt.verify(token, JWT_SECRET); } catch { return null; }
+  }
+
+  onHello(ws, msg) {
+    if (msg.type !== 'auth' || !msg.token) { ws.close(); return; }
+    const claims = this.verify(msg.token);
+    if (!claims) { this.send(ws, { type: 'error', message: 'invalid token' }); ws.close(); return; }
+
+    if (msg.memberId != null) {
+      // device client — speaks as one member
+      const member = memberById(msg.memberId);
+      if (!member) { this.send(ws, { type: 'error', message: 'unknown member' }); ws.close(); return; }
+      ws.client = { kind: 'device', memberId: member.id, name: member.name, listens: new Set() };
+      if (member.group_id != null) ws.client.listens.add(member.group_id);
+      db.prepare("UPDATE members SET status = 'online', last_seen = ? WHERE id = ?").run(Date.now(), member.id);
+    } else {
+      // dispatcher client — hears every talkgroup
+      ws.client = { kind: 'dispatcher', memberId: 0, name: claims.username, listens: new Set() };
+      for (const g of db.prepare('SELECT id FROM talkgroups').all()) ws.client.listens.add(g.id);
+    }
+    this.clients.add(ws);
+    this.send(ws, this.snapshot());
+    this.broadcastMembers();
+  }
+
+  onMessage(ws, msg) {
+    const c = ws.client;
+    switch (msg.type) {
+      case 'sync':         return this.send(ws, this.snapshot());
+      case 'listen':
+        if (Array.isArray(msg.groupIds)) c.listens = new Set(msg.groupIds.filter(Number.isInteger));
+        return;
+      case 'gps':          return this.onGps(ws, msg);
+      case 'sos':          return this.onSos(ws, msg);
+      case 'sos_ack':      return this.onSosState(ws, msg.id, 'ack', c.name);
+      case 'sos_resolve':  return this.onSosState(ws, msg.id, 'resolved', c.name);
+      case 'ptt_start':    return this.onPttStart(ws, msg);
+      case 'ptt_stop':     return this.onPttStop(ws);
+      default:             return this.send(ws, { type: 'error', message: `unknown type: ${msg.type}` });
+    }
+  }
+
+  onGps(ws, msg) {
+    if (ws.client.kind !== 'device') return;
+    const { lat, lng } = msg;
+    if (!Number.isFinite(lat) || !Number.isFinite(lng)) {
+      return this.send(ws, { type: 'error', message: 'bad gps' });
+    }
+    const battery = Number.isFinite(msg.battery) ? Math.max(0, Math.min(100, Math.round(msg.battery))) : null;
+    const speed = Number.isFinite(msg.speed) ? msg.speed : null;
+    db.prepare(`UPDATE members SET lat = ?, lng = ?, battery = COALESCE(?, battery),
+                last_seen = ?, status = 'online' WHERE id = ?`)
+      .run(lat, lng, battery, Date.now(), ws.client.memberId);
+    db.prepare('INSERT INTO gps_pings (member_id, lat, lng, speed, ts) VALUES (?, ?, ?, ?, ?)')
+      .run(ws.client.memberId, lat, lng, speed, Date.now());
+    this.broadcastMembers();
+  }
+
+  onSos(ws) {
+    if (ws.client.kind !== 'device') return;
+    const m = memberById(ws.client.memberId);
+    const info = db.prepare(`INSERT INTO sos_alerts (member_id, lat, lng, state, created_at)
+                             VALUES (?, ?, ?, 'open', ?)`).run(m.id, m.lat, m.lng, Date.now());
+    const alert = db.prepare(`SELECT a.*, mm.name AS member_name FROM sos_alerts a
+                              JOIN members mm ON mm.id = a.member_id WHERE a.id = ?`)
+      .get(info.lastInsertRowid);
+    this.broadcast({ type: 'sos_new', alert });
+  }
+
+  onSosState(ws, id, state, name) {
+    if (ws.client.kind !== 'dispatcher') return;
+    db.prepare('UPDATE sos_alerts SET state = ?, ack_by = ?, updated_at = ? WHERE id = ?')
+      .run(state, name, Date.now(), id);
+    const alert = db.prepare(`SELECT a.*, mm.name AS member_name FROM sos_alerts a
+                              JOIN members mm ON mm.id = a.member_id WHERE a.id = ?`).get(id);
+    if (alert) this.broadcast({ type: 'sos_update', alert });
+  }
+
+  onPttStart(ws, msg) {
+    const gid = Number.isInteger(msg?.groupId)
+      ? msg.groupId
+      : memberById(ws.client.memberId)?.group_id ?? null;
+    if (gid == null) return this.send(ws, { type: 'error', message: 'no talkgroup' });
+    const current = this.floors.get(gid);
+    if (current && current.client !== ws) {
+      return this.send(ws, { type: 'ptt_denied', groupId: gid, holder: current.name });
+    }
+    this.floors.set(gid, { client: ws, memberId: ws.client.memberId, name: ws.client.name });
+    db.prepare('INSERT INTO call_log (group_id, member_id, kind, ts) VALUES (?, ?, ?, ?)')
+      .run(gid, ws.client.memberId, 'ptt_start', Date.now());
+    this.broadcast({ type: 'floor', groupId: gid, memberId: ws.client.memberId, name: ws.client.name });
+  }
+
+  onPttStop(ws) {
+    for (const [gid, f] of this.floors) {
+      if (f.client !== ws) continue;
+      this.floors.delete(gid);
+      db.prepare('INSERT INTO call_log (group_id, member_id, kind, ts) VALUES (?, ?, ?, ?)')
+        .run(gid, ws.client.memberId, 'ptt_stop', Date.now());
+      this.broadcast({ type: 'floor', groupId: gid, memberId: null });
+    }
+  }
+
+  onAudio(ws, buf) {
+    if (!ws.client || buf.length < HDR) return;
+    if (buf.readUInt8(0) !== AUDIO_TYPE) return;
+    const gid = buf.readUInt32BE(1);
+    const floor = this.floors.get(gid);
+    if (!floor || floor.client !== ws) return; // only the floor holder may transmit
+    for (const other of this.clients) {
+      if (other === ws || other.readyState !== 1) continue;
+      if (!other.client.listens.has(gid)) continue;
+      other.send(buf);
+    }
+  }
+
+  snapshot() {
+    return {
+      type: 'welcome',
+      groups: db.prepare('SELECT * FROM talkgroups ORDER BY id').all(),
+      members: allMembers(),
+      sos: db.prepare(`SELECT a.*, mm.name AS member_name FROM sos_alerts a
+                       JOIN members mm ON mm.id = a.member_id
+                       WHERE a.state != 'resolved' ORDER BY a.created_at DESC`).all(),
+      floors: [...this.floors.entries()].map(([groupId, f]) => ({ groupId, memberId: f.memberId, name: f.name })),
+    };
+  }
+
+  send(ws, obj) { if (ws.readyState === 1) ws.send(JSON.stringify(obj)); }
+  broadcast(obj) { for (const ws of this.clients) this.send(ws, obj); }
+
+  broadcastMembers() { this.broadcast({ type: 'members', members: allMembers() }); }
+  broadcastGroups() { this.broadcast({ type: 'groups', groups: db.prepare('SELECT * FROM talkgroups ORDER BY id').all() }); }
+
+  onClose(ws) {
+    this.clients.delete(ws);
+    for (const [gid, f] of this.floors) {
+      if (f.client !== ws) continue;
+      this.floors.delete(gid);
+      this.broadcast({ type: 'floor', groupId: gid, memberId: null });
+    }
+    if (ws.client?.kind === 'device') {
+      const stillOnline = [...this.clients].some(
+        (c) => c.client?.kind === 'device' && c.client.memberId === ws.client.memberId
+      );
+      if (!stillOnline) {
+        db.prepare("UPDATE members SET status = 'offline', last_seen = ? WHERE id = ?")
+          .run(Date.now(), ws.client.memberId);
+        this.broadcastMembers();
+      }
+    }
+  }
+}