[&since_id=N] * events: connected {last_id, feed_last_id, server_time} · message {chat row} · feed {wt_signal row, feed-api shape} * · roster {reason, principal, present[], stale_after} · heartbeat {server_time, present_count} * · timeout (client reconnects; EventSource does so on its own) * * Presence: the credential resolves through wt-auth, which touches presence on connect; the 20s * heartbeat re-touches, so an open stream IS the reader's presence (no poll needed to stay in the * tower). Server-side it checks the table every second; that is a DB read, not a browser timer. * Auth: token in the query string because EventSource cannot set headers; wt_presented_token already * accepts ?token=. Watch-status stays 401 without one, as before. */ declare(strict_types=1); require_once '/var/www/data.finestoflines.net/citadel_2026/vendor/autoload.php'; require_once '/var/www/data.finestoflines.net/config/db-config.php'; require_once __DIR__ . '/wt-auth.php'; require_once __DIR__ . '/wt-roster.php'; // presentList(): the roster event is the roster poll's answer const WT_SSE_MAX_SECONDS = 300; // reconnect cadence; keeps PHP workers from living forever const WT_SSE_HEARTBEAT = 20; // seconds; also the presence re-touch interval function respond(int $status, array $body): never { // wt_require_reader's responder (pre-stream only) http_response_code($status); header('Content-Type: application/json'); echo json_encode($body); exit; } global $DB; $pdo = new PDO("mysql:host={$DB['host']};dbname={$DB['pathfinder']};charset=utf8mb4", $DB['user'], $DB['pass']); $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION); $reader = wt_require_reader($pdo, 'respond'); // 401 as JSON before any stream headers set_time_limit(WT_SSE_MAX_SECONDS + 10); @ini_set('zlib.output_compression', '0'); @ini_set('implicit_flush', '1'); header('Content-Type: text/event-stream'); header('Cache-Control: no-cache'); header('Connection: keep-alive'); header('X-Accel-Buffering: no'); while (ob_get_level()) ob_end_flush(); $emit = function (string $event, array $data, ?int $id = null): void { if ($id !== null) echo "id: {$id}\n"; echo "event: {$event}\ndata: " . json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES) . "\n\n"; if (ob_get_level()) ob_flush(); flush(); }; $lastId = isset($_GET['since_id']) ? (int)$_GET['since_id'] : (isset($_SERVER['HTTP_LAST_EVENT_ID']) ? (int)$_SERVER['HTTP_LAST_EVENT_ID'] : (int)$pdo->query("SELECT COALESCE(MAX(id),0) FROM wt_chat")->fetchColumn()); /** * FOUR CLOCKS, ONE STREAM (Solace, 08:3x 09-20): the client had a chat poll, a feed poll, a roster poll and a * presence beat. The broker line already carries `kind` (chat | signal | roster); this endpoint now READS it and * emits `message` / `feed` / `roster` on one stream, so the shell-level EventSource retires all four timers. * feed: wt_signal rows in feed-api's row shape incl. dedup_key: the client FOLDS on that key (the same GROUP BY * feed-api collapse=1 uses), so push and poll agree by construction (Solace 08:4x: 2.2x fan-out otherwise). Cursor = ?feed_since_id, else * MAX(id) at connect. Feed rows carry no SSE id (that field is the chat cursor); a reconnect gap is * the client's reconcile to cover, never a replay. * roster: presentList() — byte-for-byte the watch-status answer — on every presence TRANSITION (join, left, * stood_down, on_watch, assign/unassign, route, capped, delivery_failed). Never on a heartbeat. */ $feedId = isset($_GET['feed_since_id']) ? (int)$_GET['feed_since_id'] : (int)$pdo->query("SELECT COALESCE(MAX(id),0) FROM wt_signal")->fetchColumn(); $emit('connected', ['last_id' => $lastId, 'feed_last_id' => $feedId, 'server_time' => gmdate('c'), 'reader' => $reader['principal']]); echo "retry: 2000\n\n"; flush(); $sel = $pdo->prepare("SELECT id, principal, display_name, principal_kind, principal_kind AS kind, body, msg_kind, dm_to, via, dm_status, final, answered_by, answered_at, dismissed_by, dismiss_reason, posted_at FROM wt_chat WHERE id > ? ORDER BY id ASC LIMIT 100"); $fsel = $pdo->prepare("SELECT id, topic, msg_id, sender, recipient, body, sent_at, signal_class, signal_source, channel, is_failure, captured_at, dedup_key, 1 AS copies FROM wt_signal WHERE id > ? AND channel <> 'rd' ORDER BY id ASC LIMIT 200"); // channel <> 'rd': the SAME rule feed-api.php applies by default (Ray 09-12: lab talk is stored, never watched). Solace 09:0x measured 96.5% rd on the stream without it. $start = time(); $lastBeat = time(); /** * IMMEDIACY (Ray 22:3x): no table poll. Block on the broker's watchtower/events topic; a post wakes this * loop the instant chat-api publishes. The 20s heartbeat is LIVENESS ONLY (Solace's reading of the * doctrine: a clock may prove the path is alive; it may not carry data) and it re-touches presence. */ $sub = proc_open(['mosquitto_sub', '-h', '127.0.0.1', '-p', '1883', '-q', '1', '-t', WT_EVENT_TOPIC], [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes); if (!is_resource($sub)) { $emit('timeout', ['message' => 'broker unavailable']); exit; } stream_set_blocking($pipes[1], false); $drain = function () use ($sel, &$lastId, $emit) { $sel->execute([$lastId]); foreach ($sel->fetchAll(PDO::FETCH_ASSOC) as $row) { $lastId = (int)$row['id']; $emit('message', $row, $lastId); } }; $drainFeed = function () use ($fsel, &$feedId, $emit) { $fsel->execute([$feedId]); foreach ($fsel->fetchAll(PDO::FETCH_ASSOC) as $row) { $feedId = (int)$row['id']; $emit('feed', $row); } }; $emitRoster = function (string $reason, string $who) use ($pdo, $emit) { $emit('roster', ['reason' => $reason, 'principal' => $who, 'stale_after' => WT_PRESENCE_STALE_SECONDS, 'present' => presentList($pdo), 'server_time' => gmdate('c')]); }; $drain(); $drainFeed(); // anything that landed between the connected event and the subscription while (!connection_aborted() && (time() - $start) < WT_SSE_MAX_SECONDS) { $r = [$pipes[1]]; $w = null; $x = null; $wait = max(1, WT_SSE_HEARTBEAT - (time() - $lastBeat)); $n = @stream_select($r, $w, $x, $wait); if ($n > 0) { // Read the burst and SWITCH ON KIND. An unparseable line is "*": everything may have changed, drain it all, // so a malformed publish costs a redundant query, never a missed event. $kinds = []; while (($l = fgets($pipes[1])) !== false) { $ev = json_decode(trim($l), true); $k = is_array($ev) ? (string)($ev['kind'] ?? '') : ''; $kinds[$k === '' ? '*' : $k][] = is_array($ev) ? $ev : []; } $all = isset($kinds['*']); if ($all || isset($kinds['chat'])) $drain(); if ($all || isset($kinds['signal'])) $drainFeed(); if ($all || isset($kinds['roster'])) { $rl = $kinds['roster'] ?? [[]]; $last = end($rl) ?: []; $emitRoster((string)($last['reason'] ?? 'changed'), (string)($last['principal'] ?? '')); } } if (time() - $lastBeat >= WT_SSE_HEARTBEAT) { $lastBeat = time(); $drain(); $drainFeed(); // RECONCILE (Solace): a missed publish costs latency, never a message. Ray may strip this. wt_presence_touch($pdo, $reader); // an open stream is presence $st = $pdo->prepare("SELECT COUNT(*) FROM wt_presence WHERE left_at IS NULL AND (mode='push' OR last_seen > (NOW() - INTERVAL ? SECOND))"); $st->execute([WT_PRESENCE_STALE_SECONDS]); $emit('heartbeat', ['server_time' => gmdate('c'), 'present_count' => (int)$st->fetchColumn()]); } } proc_terminate($sub); $emit('timeout', ['message' => 'reconnect']);