PDQ/bin/events_poll.php

<?php
// Long-poll endpoint. Holds the connection open for up to ~28 seconds until
// either (a) the job_history or messages tables receive a row beyond the
// caller's since markers, or (b) it times out. Either way returns the current
// max IDs so the client can update its trackers.
//
// One PHP-FPM worker is parked per connected client while this runs. For 2-3
// users that's fine; if more users start using this app, raise pm.max_children.

require_once __DIR__ . '/../lib/identity.php';

[$actor, $vendor_id] = resolve_request_actor();

$since_h = (int) ($_GET['since_h'] ?? 0);

// Messages "since" markers.
//   Vendor caller: a single int (their own vendor's last seen message id).
//   ICG caller:    "vid:id,vid:id" pairs (one per vendor thread on PDQ.php).
$since_m = [];
if ($actor === 'ICG') {
    foreach (array_filter(explode(',', (string) ($_GET['since_m'] ?? ''))) as $pair) {
        $p = explode(':', $pair);
        if (count($p) === 2) $since_m[(int) $p[0]] = (int) $p[1];
    }
} else {
    $since_m[$vendor_id] = (int) ($_GET['since_m'] ?? 0);
}

// PHP's default max_execution_time is 30s; give ourselves slightly more so the
// loop can finish writing the response before the deadline.
@set_time_limit(45);

$pdo = db();
$vfilter = $actor === 'ICG' ? null : $vendor_id;
$deadline = microtime(true) + 28.0;

header('Content-Type: application/json');

while (true) {
    [$cur_h, $cur_m] = events_snapshot($pdo, $vfilter);

    $history_changed = $cur_h > $since_h;
    $changed_vendors = [];
    foreach ($cur_m as $vid => $maxid) {
        if ($maxid > ($since_m[$vid] ?? 0)) $changed_vendors[] = $vid;
    }

    if ($history_changed || $changed_vendors) {
        echo json_encode([
            'history'          => $cur_h,
            'messages'         => $cur_m,
            'history_changed'  => $history_changed,
            'changed_vendors'  => $changed_vendors,
        ]);
        return;
    }
    if (microtime(true) >= $deadline) {
        echo json_encode([
            'history'  => $cur_h,
            'messages' => $cur_m,
            'timeout'  => true,
        ]);
        return;
    }
    // Bail early if the client has gone away (cheap check — flush updates the
    // connection state). Otherwise sleep 1s and loop.
    if (connection_aborted()) return;
    usleep(1000000);
}

function events_snapshot(PDO $pdo, ?int $vendor_filter): array {
    if ($vendor_filter === null) {
        $h = (int) $pdo->query('SELECT COALESCE(MAX(id), 0) FROM job_history')->fetchColumn();
        $m = [];
        foreach ($pdo->query('SELECT vendor_id, COALESCE(MAX(id), 0) AS maxid FROM messages GROUP BY vendor_id')->fetchAll() as $r) {
            $m[(int) $r['vendor_id']] = (int) $r['maxid'];
        }
        // Vendors with zero messages get a 0 entry so the client tracks them.
        foreach ($pdo->query('SELECT id FROM vendors WHERE active = 1')->fetchAll() as $v) {
            if (!isset($m[(int) $v['id']])) $m[(int) $v['id']] = 0;
        }
    } else {
        $stmt = $pdo->prepare(
            'SELECT COALESCE(MAX(h.id), 0)
               FROM job_history h
               JOIN jobs j ON j.id = h.job_id
              WHERE j.vendor_id = ?'
        );
        $stmt->execute([$vendor_filter]);
        $h = (int) $stmt->fetchColumn();

        $stmt = $pdo->prepare('SELECT COALESCE(MAX(id), 0) FROM messages WHERE vendor_id = ?');
        $stmt->execute([$vendor_filter]);
        $m = [$vendor_filter => (int) $stmt->fetchColumn()];
    }
    return [$h, $m];
}