ProdDash/modules/vista-cues/rtpmidi.js

'use strict';

/**
 * A listen-only RTP-MIDI (AppleMIDI) session participant, in plain Node.
 *
 * This is the protocol behind a macOS "Network" MIDI session (Audio MIDI
 * Setup → MIDI Network Setup) and rtpMIDI on Windows: two UDP ports, a
 * control port P and a data port P+1. A session sends every MIDI message to
 * every participant it is connected to, so joining the same session Vista is
 * in hears exactly the notes Vista hears.
 *
 * Both ways of joining work:
 *  - someone invites us (we show up in Audio MIDI Setup's Directory, or they
 *    add our address by hand) and we accept anyone;
 *  - we invite them (`invite(host, port)`), and keep re-inviting until they
 *    accept, then keep the connection alive with clock syncs.
 *
 * We never send MIDI. Packet layouts follow Apple's AppleMIDI session
 * protocol and RFC 6295 (RTP payload for MIDI).
 */

const dgram = require('dgram');
const dns = require('dns');
const crypto = require('crypto');
const { EventEmitter } = require('events');

const SIG = 0xffff;
const CMD = { IN: 0x494e, OK: 0x4f4b, NO: 0x4e4f, BY: 0x4259, CK: 0x434b, RS: 0x5253 };
const PROTOCOL_VERSION = 2;

const INVITE_RETRY_MS = 1500;     // between invitation attempts while setting up
const INVITE_ATTEMPTS = 12;       // then give up for a while…
const REINVITE_MS = 5000;         // …and start over after this long
const SYNC_FAST_MS = 1500;        // first few clock syncs, close together
const SYNC_MS = 10000;            // then this often, as macOS does
const SYNC_TIMEOUT_MS = 25000;    // no sync answer for this long → connection is gone
const IDLE_MS = 90000;            // an inbound participant we have not heard from at all
const FEEDBACK_MS = 1000;         // receiver feedback (lets the sender trim its journal)

/** The session clock: 100 µs ticks since this process started the session. */
function makeClock() {
  const t0 = process.hrtime.bigint();
  return () => (process.hrtime.bigint() - t0) / 100000n;
}

function randomU32() {
  return crypto.randomBytes(4).readUInt32BE(0);
}

/* ── packets ─────────────────────────────────────────────────────────── */

function sessionPacket(cmd, token, ssrc, name) {
  const nameBuf = name == null ? Buffer.alloc(0) : Buffer.from(`${name}\0`, 'utf8');
  const b = Buffer.alloc(16 + nameBuf.length);
  b.writeUInt16BE(SIG, 0);
  b.writeUInt16BE(cmd, 2);
  b.writeUInt32BE(PROTOCOL_VERSION, 4);
  b.writeUInt32BE(token >>> 0, 8);
  b.writeUInt32BE(ssrc >>> 0, 12);
  nameBuf.copy(b, 16);
  return b;
}

function syncPacket(ssrc, count, ts) {
  const b = Buffer.alloc(36);
  b.writeUInt16BE(SIG, 0);
  b.writeUInt16BE(CMD.CK, 2);
  b.writeUInt32BE(ssrc >>> 0, 4);
  b.writeUInt8(count, 8);
  for (let i = 0; i < 3; i++) b.writeBigUInt64BE(BigInt.asUintN(64, ts[i] || 0n), 12 + i * 8);
  return b;
}

function feedbackPacket(ssrc, seq) {
  const b = Buffer.alloc(12);
  b.writeUInt16BE(SIG, 0);
  b.writeUInt16BE(CMD.RS, 2);
  b.writeUInt32BE(ssrc >>> 0, 4);
  b.writeUInt16BE(seq & 0xffff, 8);
  return b;
}

/** A session command (IN/OK/NO/BY/CK/RS), or null when it isn't one. */
function parseSession(buf) {
  if (buf.length < 4 || buf.readUInt16BE(0) !== SIG) return null;
  const cmd = buf.readUInt16BE(2);
  if (cmd === CMD.CK) {
    if (buf.length < 36) return null;
    return {
      cmd,
      ssrc: buf.readUInt32BE(4),
      count: buf.readUInt8(8),
      ts: [buf.readBigUInt64BE(12), buf.readBigUInt64BE(20), buf.readBigUInt64BE(28)],
    };
  }
  if (cmd === CMD.RS) {
    if (buf.length < 12) return null;
    return { cmd, ssrc: buf.readUInt32BE(4), seq: buf.readUInt16BE(8) };
  }
  if (cmd === CMD.IN || cmd === CMD.OK || cmd === CMD.NO || cmd === CMD.BY) {
    if (buf.length < 16) return null;
    let name = '';
    if (buf.length > 16) {
      const end = buf.indexOf(0, 16);
      name = buf.toString('utf8', 16, end < 0 ? buf.length : end);
    }
    return { cmd, version: buf.readUInt32BE(4), token: buf.readUInt32BE(8), ssrc: buf.readUInt32BE(12), name };
  }
  return null;
}

/** Data bytes that follow a status byte (sysex is handled separately). */
function dataLength(status) {
  const hi = status & 0xf0;
  if (hi === 0xc0 || hi === 0xd0) return 1;
  if (hi < 0xf0) return 2;
  if (status === 0xf1 || status === 0xf3) return 1;
  if (status === 0xf2) return 2;
  return 0;
}

/**
 * The MIDI commands in one RTP-MIDI packet (RFC 6295 §3), as
 * `{ status, data: [..] }` in order. Journal data is ignored: we only show
 * what is happening, and a note lost on the wire is one Vista also missed.
 */
function parseRtpMidi(buf) {
  if (buf.length < 13 || (buf[0] >> 6) !== 2) return null;
  const cc = buf[0] & 0x0f;
  const ext = Boolean(buf[0] & 0x10);
  const seq = buf.readUInt16BE(2);
  const ssrc = buf.readUInt32BE(8);
  let off = 12 + cc * 4;
  if (ext) {
    if (buf.length < off + 4) return null;
    off += 4 + buf.readUInt16BE(off + 2) * 4;
  }
  if (off >= buf.length) return null;
  const header = buf[off];
  const big = Boolean(header & 0x80);
  const zero = Boolean(header & 0x20);
  let len = header & 0x0f;
  if (big) {
    if (off + 1 >= buf.length) return null;
    len = (len << 8) | buf[off + 1];
    off += 2;
  } else {
    off += 1;
  }
  const end = Math.min(buf.length, off + len);
  const messages = [];
  let running = 0;
  let first = true;
  while (off < end) {
    if (!first || zero) {
      // Delta time: 1–4 octets, high bit means "more follow".
      let n = 0;
      while (off < end && (buf[off] & 0x80) && n < 3) { off++; n++; }
      off++;
      if (off >= end) break;
    }
    first = false;
    let status = buf[off];
    if (status & 0x80) {
      off++;
    } else if (running) {
      status = running;
    } else {
      break; // data byte with nothing to run on: give up on this packet
    }
    if (status === 0xf0 || status === 0xf7) {
      // A sysex, or a segment of one: runs to F7 (end), F0 (more follows) or F4 (cancelled).
      const start = off;
      while (off < end && buf[off] !== 0xf7 && buf[off] !== 0xf0 && buf[off] !== 0xf4) off++;
      const data = Array.from(buf.subarray(start, off));
      off++;
      messages.push({ status, data });
      running = 0;
      continue;
    }
    if (status < 0xf0) running = status;
    else if (status < 0xf8) running = 0; // system common cancels running status; realtime doesn't
    const n = dataLength(status);
    if (off + n > end) break;
    messages.push({ status, data: Array.from(buf.subarray(off, off + n)) });
    off += n;
  }
  return { seq, ssrc, messages };
}

/** `{ status, data }` → a plain event, or null for what nobody here cares about. */
function describe(msg) {
  const hi = msg.status & 0xf0;
  const channel = (msg.status & 0x0f) + 1;
  if (hi === 0x90 && msg.data[1] > 0) return { type: 'noteon', channel, note: msg.data[0], velocity: msg.data[1] };
  if (hi === 0x80 || (hi === 0x90 && msg.data[1] === 0)) return { type: 'noteoff', channel, note: msg.data[0] };
  if (hi === 0xc0) return { type: 'program', channel, program: msg.data[0] };
  if (hi === 0xb0) return { type: 'cc', channel, controller: msg.data[0], value: msg.data[1] };
  return null;
}

/* ── the session ─────────────────────────────────────────────────────── */

class Session extends EventEmitter {
  /**
   * @param {{ name: string, port: number, log?: Function }} opts
   * Events: 'midi' ({ participant, event }), 'change' (participants or
   * listening changed), 'listening' (both ports bound).
   */
  constructor({ name, port, log }) {
    super();
    this.name = name || 'ProdDash';
    this.port = port;
    this.log = log || (() => {});
    this.ssrc = randomU32();
    this.now = makeClock();
    this.control = null;
    this.data = null;
    this.participants = new Map(); // key → participant
    this.invites = new Map();      // key "host:port" → outbound invitation
    this.strangers = new Map();    // address → when we last told it we're not in its session
    this.timers = new Set();
    this.stopped = false;
    this.error = '';
  }

  /**
   * Open both ports. Resolves once both are bound — or once the first try
   * failed, with `error` set. A port that is busy (our own previous session
   * still closing after a config save, or another program) is retried every
   * few seconds for as long as the session runs; 'listening' fires when it
   * finally binds.
   */
  start() {
    this.every(FEEDBACK_MS, () => this.sendFeedback());
    this.every(5000, () => this.reap());
    return new Promise((resolve) => {
      let first = true;
      const attempt = (delay) => this.later(delay, async () => {
        const ok = await this.bind();
        if (ok) {
          this.error = '';
          this.emit('listening');
          this.emit('change');
        } else {
          this.emit('change');
          attempt(first ? 250 : 3000);
        }
        if (first) { first = false; resolve(ok); }
      });
      attempt(0);
    });
  }

  /** Bind control and data sockets together; on failure close both. */
  bind() {
    const open = (port, which) => new Promise((resolve) => {
      const sock = dgram.createSocket('udp4');
      const fail = (e) => {
        try { sock.close(); } catch { /* not open */ }
        resolve({ sock: null, port, e });
      };
      sock.once('error', fail);
      sock.bind(port, () => {
        sock.removeListener('error', fail);
        sock.on('error', (e) => this.log(`socket error on ${port}: ${e.message}`));
        sock.on('message', (buf, rinfo) => {
          try { this.onPacket(buf, rinfo, which); } catch (e) { this.log(`bad packet from ${rinfo.address}: ${e.message}`); }
        });
        resolve({ sock, port });
      });
    });
    return Promise.all([open(this.port, 'control'), open(this.port + 1, 'data')]).then(([c, d]) => {
      if (this.stopped || !c.sock || !d.sock) {
        for (const x of [c, d]) if (x.sock) { try { x.sock.close(); } catch { /* closed */ } }
        const bad = !c.sock ? c : d;
        if (bad.e) {
          const msg = bad.e.code === 'EADDRINUSE'
            ? `UDP port ${bad.port} is in use on this machine — pick another session port in /admin`
            : `UDP port ${bad.port}: ${bad.e.message}`;
          if (msg !== this.error) this.log(msg);
          this.error = msg;
        }
        return false;
      }
      this.control = c.sock;
      this.data = d.sock;
      return true;
    });
  }

  stop() {
    this.stopped = true;
    for (const t of this.timers) { clearTimeout(t); clearInterval(t); }
    this.timers.clear();
    const control = this.control;
    const data = this.data;
    const close = () => {
      for (const s of [control, data]) { try { s && s.close(); } catch { /* closed */ } }
    };
    // Say goodbye, then close once the goodbyes have left (a config save
    // re-inits straight away and wants these ports back).
    const byes = [...this.participants.values()].filter((p) => p.connected);
    let left = byes.length;
    if (!control || !left) close();
    else {
      for (const p of byes) {
        try {
          control.send(sessionPacket(CMD.BY, p.token, this.ssrc), p.controlPort, p.address, () => { if (--left === 0) close(); });
        } catch { if (--left === 0) close(); }
      }
    }
    this.control = null;
    this.data = null;
    this.participants.clear();
    this.invites.clear();
  }

  every(ms, fn) {
    const t = setInterval(fn, ms);
    t.unref?.();
    this.timers.add(t);
    return t;
  }

  later(ms, fn) {
    const t = setTimeout(() => { this.timers.delete(t); if (!this.stopped) fn(); }, ms);
    t.unref?.();
    this.timers.add(t);
    return t;
  }

  send(sock, buf, port, address) {
    if (!sock || this.stopped) return;
    try { sock.send(buf, port, address); } catch { /* socket closed */ }
  }

  /** Every participant, for display. */
  list() {
    return [...this.participants.values()].map((p) => ({
      name: p.name,
      address: p.address,
      port: p.controlPort,
      connected: p.connected,
      outbound: p.outbound,
      lastHeard: p.lastHeard || null,
      since: p.since || null,
    }));
  }

  /* — outbound invitations — */

  /**
   * Keep a connection to the session at host:port. Re-invites on its own
   * whenever the connection drops. Returns the invitation's state object.
   */
  invite(host, port, label) {
    const key = `${host}:${port}`;
    if (this.invites.has(key)) return this.invites.get(key);
    const inv = { key, host, port, label: label || host, address: '', token: 0, attempts: 0, stage: 'idle', error: '', timer: null };
    this.invites.set(key, inv);
    this.beginInvite(inv);
    return inv;
  }

  beginInvite(inv) {
    if (this.stopped || !this.invites.has(inv.key)) return;
    dns.lookup(inv.host, { family: 4 }, (err, address) => {
      if (this.stopped) return;
      if (err) {
        inv.error = `can't find ${inv.host}`;
        inv.stage = 'idle';
        this.emit('change');
        this.later(REINVITE_MS, () => this.beginInvite(inv));
        return;
      }
      inv.address = address;
      // Already in a session with that machine (it invited us)? Nothing to do.
      const existing = this.findByAddress(address, inv.port);
      if (existing && existing.connected) {
        existing.label = inv.label;
        inv.stage = 'connected';
        inv.error = '';
        this.emit('change');
        return;
      }
      inv.token = randomU32();
      inv.attempts = 0;
      inv.stage = 'control';
      this.sendInvite(inv);
    });
  }

  sendInvite(inv) {
    if (this.stopped || !this.invites.has(inv.key)) return;
    if (inv.stage !== 'control' && inv.stage !== 'data') return;
    if (inv.attempts >= INVITE_ATTEMPTS) {
      inv.error = inv.error || 'no answer';
      inv.stage = 'idle';
      this.emit('change');
      this.later(REINVITE_MS, () => this.beginInvite(inv));
      return;
    }
    inv.attempts++;
    const sock = inv.stage === 'control' ? this.control : this.data;
    const port = inv.stage === 'control' ? inv.port : inv.port + 1;
    this.send(sock, sessionPacket(CMD.IN, inv.token, this.ssrc, this.name), port, inv.address);
    clearTimeout(inv.timer);
    inv.timer = this.later(INVITE_RETRY_MS, () => this.sendInvite(inv));
  }

  inviteFor(rinfo, token) {
    for (const inv of this.invites.values()) {
      if (inv.token === token && inv.address === rinfo.address) return inv;
    }
    return null;
  }

  findByAddress(address, controlPort) {
    for (const p of this.participants.values()) {
      if (p.address === address && (controlPort == null || p.controlPort === controlPort)) return p;
    }
    return null;
  }

  /* — packets in — */

  onPacket(buf, rinfo, which) {
    const s = parseSession(buf);
    if (!s) {
      if (which === 'data') this.onRtp(buf, rinfo);
      return;
    }
    const sock = which === 'control' ? this.control : this.data;
    switch (s.cmd) {
      case CMD.IN: return this.onInvitation(s, rinfo, which, sock);
      case CMD.OK: return this.onAccepted(s, rinfo, which);
      case CMD.NO: return this.onRejected(s, rinfo);
      case CMD.BY: return this.onBye(s, rinfo);
      case CMD.CK: return this.onSync(s, rinfo);
      default: return undefined; // RS from a peer: nothing for a listener to do
    }
  }

  onInvitation(s, rinfo, which, sock) {
    // We accept anyone: being in more sessions only means hearing more cues.
    this.send(sock, sessionPacket(CMD.OK, s.token, this.ssrc, this.name), rinfo.port, rinfo.address);
    const key = `in:${s.ssrc}`;
    let p = this.participants.get(key);
    if (which === 'control') {
      if (!p) {
        p = {
          key, ssrc: s.ssrc, token: s.token, name: s.name || rinfo.address, label: '',
          address: rinfo.address, controlPort: rinfo.port, dataPort: rinfo.port + 1,
          outbound: false, connected: false, lastSeq: -1, fedBack: -1, lastHeard: Date.now(), since: 0,
        };
        this.participants.set(key, p);
        this.log(`invited by "${p.name}" at ${rinfo.address}:${rinfo.port}`);
      } else {
        p.token = s.token;
        p.controlPort = rinfo.port;
        p.name = s.name || p.name;
      }
    } else if (p) {
      p.dataPort = rinfo.port;
      p.connected = true;
      p.since = Date.now();
      p.lastHeard = Date.now();
      this.log(`in session with "${p.name}" (${p.address})`);
      for (const inv of this.invites.values()) {
        if (inv.address === p.address && inv.port === p.controlPort) {
          p.label = inv.label;
          inv.stage = 'connected';
          inv.error = '';
          clearTimeout(inv.timer);
        }
      }
      this.emit('change');
    }
  }

  onAccepted(s, rinfo, which) {
    const inv = this.inviteFor(rinfo, s.token);
    if (!inv) return;
    clearTimeout(inv.timer);
    if (which === 'control' && inv.stage === 'control') {
      inv.stage = 'data';
      inv.attempts = 0;
      inv.remoteName = s.name;
      this.sendInvite(inv);
      return;
    }
    if (which === 'data' && inv.stage === 'data') {
      const key = `out:${inv.key}`;
      // Drop any stale record of the same remote before taking the new one.
      for (const [k, q] of this.participants) if (q.ssrc === s.ssrc && k !== key) this.participants.delete(k);
      const p = {
        key, ssrc: s.ssrc, token: inv.token, name: inv.remoteName || s.name || inv.host, label: inv.label,
        address: rinfo.address, controlPort: inv.port, dataPort: rinfo.port,
        outbound: true, connected: true, lastSeq: -1, fedBack: -1,
        lastHeard: Date.now(), lastSyncAnswer: Date.now(), since: Date.now(), syncs: 0,
      };
      this.participants.set(key, p);
      inv.stage = 'connected';
      inv.error = '';
      this.log(`joined "${p.name}" at ${inv.host} (${rinfo.address}:${inv.port})`);
      this.emit('change');
      this.startSync(p, inv);
    }
  }

  onRejected(s, rinfo) {
    const inv = this.inviteFor(rinfo, s.token);
    if (!inv) return;
    clearTimeout(inv.timer);
    inv.stage = 'idle';
    inv.error = 'the session refused us — set "Who may connect" to Anyone in Audio MIDI Setup';
    this.log(`${inv.host}: invitation refused`);
    this.emit('change');
    this.later(REINVITE_MS * 2, () => this.beginInvite(inv));
  }

  onBye(s, rinfo) {
    for (const [k, p] of this.participants) {
      if (p.ssrc === s.ssrc && p.address === rinfo.address) this.drop(k, 'left the session');
    }
  }

  drop(key, why) {
    const p = this.participants.get(key);
    if (!p) return;
    this.participants.delete(key);
    clearInterval(p.syncTimer);
    this.timers.delete(p.syncTimer);
    this.log(`"${p.name}" ${why}`);
    // If we were asked to keep this one, go and get it back.
    for (const inv of this.invites.values()) {
      if (inv.address === p.address && inv.port === p.controlPort && inv.stage === 'connected') {
        inv.stage = 'idle';
        inv.error = why;
        this.later(p.outbound ? 1000 : REINVITE_MS, () => this.beginInvite(inv));
      }
    }
    this.emit('change');
  }

  /* — clock sync — */

  startSync(p) {
    const tick = () => {
      if (!this.participants.has(p.key)) return;
      if (Date.now() - p.lastSyncAnswer > SYNC_TIMEOUT_MS) {
        this.send(this.control, sessionPacket(CMD.BY, p.token, this.ssrc), p.controlPort, p.address);
        this.drop(p.key, 'stopped answering');
        return;
      }
      this.send(this.data, syncPacket(this.ssrc, 0, [this.now(), 0n, 0n]), p.dataPort, p.address);
      p.syncs++;
      if (p.syncs === 6) {
        clearInterval(p.syncTimer);
        this.timers.delete(p.syncTimer);
        p.syncTimer = this.every(SYNC_MS, tick);
      }
    };
    p.syncTimer = this.every(SYNC_FAST_MS, tick);
    tick();
  }

  onSync(s, rinfo) {
    const p = [...this.participants.values()].find((q) => q.ssrc === s.ssrc);
    if (p) p.lastHeard = Date.now();
    if (s.count === 0) {
      this.send(this.data, syncPacket(this.ssrc, 1, [s.ts[0], this.now(), 0n]), rinfo.port, rinfo.address);
    } else if (s.count === 1) {
      if (p) p.lastSyncAnswer = Date.now();
      this.send(this.data, syncPacket(this.ssrc, 2, [s.ts[0], s.ts[1], this.now()]), rinfo.port, rinfo.address);
    }
    // count 2 completes a sync someone else started; nothing to answer.
  }

  /** Drop inbound participants that have gone quiet (their machine slept or left without a BY). */
  reap() {
    const now = Date.now();
    for (const [k, p] of this.participants) {
      if (!p.outbound && now - (p.lastHeard || 0) > IDLE_MS) this.drop(k, 'went quiet');
    }
  }

  sendFeedback() {
    for (const p of this.participants.values()) {
      if (!p.connected || p.lastSeq < 0 || p.lastSeq === p.fedBack) continue;
      p.fedBack = p.lastSeq;
      this.send(this.control, feedbackPacket(this.ssrc, p.lastSeq), p.controlPort, p.address);
    }
  }

  /* — MIDI — */

  onRtp(buf, rinfo) {
    const pkt = parseRtpMidi(buf);
    if (!pkt) return;
    let p = [...this.participants.values()].find((q) => q.ssrc === pkt.ssrc && q.connected);
    if (!p) {
      // MIDI from someone not in a session with us (a stale session after our restart).
      // Ask them to start over so the session comes back properly — once in a while, not per note.
      const last = this.strangers.get(rinfo.address) || 0;
      if (Date.now() - last > 5000) {
        this.strangers.set(rinfo.address, Date.now());
        this.send(this.control, sessionPacket(CMD.BY, 0, this.ssrc), rinfo.port - 1, rinfo.address);
      }
      return;
    }
    p.lastHeard = Date.now();
    p.lastSeq = pkt.seq;
    for (const m of pkt.messages) {
      const event = describe(m);
      if (event) this.emit('midi', { participant: p, event });
    }
  }
}

module.exports = {
  Session,
  parseRtpMidi,
  parseSession,
  describe,
  sessionPacket,
  syncPacket,
  CMD,
};