Statum/frontend-demo/server.php

<?php
/**
 * Statum frontend demo backend.
 *
 * Run from this directory with PHP's built-in server. Because the realtime
 * channel is a long-lived Server-Sent Events connection, you MUST start the
 * server with several workers so an open SSE stream does not block other
 * requests:
 *
 *     PHP_CLI_SERVER_WORKERS=5 php -S localhost:8000 server.php
 *
 * Then open http://localhost:8000/ in your browser (and a second tab to see
 * realtime pushes).
 *
 * This is a DEMO only: there is no real signing or encryption. Snapshots are
 * wrapped in trivial frames, and per-item action context is carried in the
 * action's `private` blob as plain base64. State lives in the PHP session so
 * that actions are stateful across requests. Realtime subscriptions are shared
 * between workers through small files in the system temp dir.
 */

session_start();

// Static product catalog used by the demo.
$CATALOG = [
    ['id' => 1, 'name' => 'Widget',  'price' => 9.99,  'featured' => true,  'inStock' => true],
    ['id' => 2, 'name' => 'Gadget',  'price' => 19.99, 'featured' => false, 'inStock' => true],
    ['id' => 3, 'name' => 'Sprocket', 'price' => 4.50, 'featured' => false, 'inStock' => false],
    ['id' => 4, 'name' => 'Gizmo',   'price' => 29.99, 'featured' => true,  'inStock' => true],
];

$TYPES = ['counter', 'products', 'cart', 'user', 'notice'];

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/** Send a JSON array of Statum directives and stop. */
function statum_respond(array $directives): void
{
    header('Content-Type: application/vnd.statum+json');
    header('Cache-Control: no-store');
    echo json_encode($directives, JSON_PRETTY_PRINT);
    exit;
}

/** Wrap a snapshot in a trivial (unsigned) frame. The signature is a SHA256
 *  hash of the content so that identical snapshots share a signature (used by
 *  the client as an idempotency key) while distinct ones do not. */
function frame(array $snapshot): array
{
    $content = json_encode($snapshot);
    return [
        'content' => $content,
        'signer' => 'demo',
        'signature' => base64_encode(hash('sha256', $content, true)),
    ];
}

/** Stable per-type slot key, kept in the session. */
function key_for(string $type): string
{
    if (!isset($_SESSION['keys'][$type])) {
        $_SESSION['keys'][$type] = 'k_' . $type;
    }
    return $_SESSION['keys'][$type];
}

/**
 * Build a snapshot for a type. Options:
 *  - `always`: transmit_after == as_at, so the client sends the base64 frame on
 *    every request (the always-send path).
 *  - `stale`: transmit_after is set 10s in the future (with a far-future
 *    invalid_after). The slot is fresh for 10s, after which the client must
 *    pre-flight via /_statum/slots before each request. Load the page, wait ~10s,
 *    then trigger an action to see the pre-flight request.
 */
function make_snapshot(string $type, string $scope, array $public, array $opts = [], ?string $key = null): array
{
    $now = time();
    $asAt = gmdate('Y-m-d\TH:i:s\Z', $now);
    if (!empty($opts['always'])) {
        $transmit = $asAt;
    } elseif (!empty($opts['stale'])) {
        $transmit = gmdate('Y-m-d\TH:i:s\Z', $now + 10);
    } else {
        $transmit = gmdate('Y-m-d\TH:i:s\Z', $now + 60);
    }
    return [
        'as_at' => $asAt,
        'slot' => ['key' => $key ?? key_for($type), 'scope' => $scope, 'type' => $type],
        'transmit_after' => $transmit,
        'invalid_after' => gmdate('Y-m-d\TH:i:s\Z', $now + 600),
        'private' => $opts['private'] ?? '',
        'public' => $public,
    ];
}

function set_dir(array $snapshot): array
{
    return ['type' => 'set', 'snapshot_frame' => frame($snapshot)];
}

function clear_dir(string $key): array
{
    return ['type' => 'clear', 'key' => $key];
}

function product_by_id($id): ?array
{
    global $CATALOG;
    foreach ($CATALOG as $p) {
        if ((string)$p['id'] === (string)$id) return $p;
    }
    return null;
}

// ---------------------------------------------------------------------------
// Public-data builders
// ---------------------------------------------------------------------------

function build_counter(): array
{
    $v = $_SESSION['counter'] ?? 0;
    return [
        'value' => $v,
        'high' => $v > 5,
        'increment' => ['uri' => '/api/counter/inc', 'method' => 'POST'],
        'decrement' => ['uri' => '/api/counter/dec', 'method' => 'POST'],
        'reset' => ['uri' => '/api/counter/reset', 'method' => 'POST'],
        'add' => ['uri' => '/api/counter/add', 'method' => 'POST'],
    ];
}

function build_products(): array
{
    global $CATALOG;
    $items = [];
    foreach ($CATALOG as $p) {
        $pid = (string)$p['id'];
        $items[] = [
            'id' => $pid,
            'name' => $p['name'],
            'price' => $p['price'],
            'featured' => $p['featured'],
            'inStock' => $p['inStock'],
            // Per-item context rides in `private`; the client echoes it back
            // via the X-Statum-Private header (no stm-data expression needed).
            'addToCartAction' => [
                'uri' => '/api/cart/add',
                'method' => 'POST',
                'private' => base64_encode('id=' . $pid),
            ],
        ];
    }
    return ['items' => $items];
}

function build_cart(): array
{
    $cart = $_SESSION['cart'] ?? [];
    $count = 0;
    $total = 0.0;
    $lines = [];
    foreach ($cart as $id => $qty) {
        $p = product_by_id($id);
        if (!$p) continue;
        $count += $qty;
        $total += $p['price'] * $qty;
        $lines[] = ['id' => (string)$id, 'name' => $p['name'], 'qty' => $qty, 'price' => $p['price']];
    }
    return [
        'count' => $count,
        'total' => round($total, 2),
        'empty' => $count === 0,
        'items' => $lines,
        'checkoutAction' => ['uri' => '/api/cart/checkout', 'method' => 'POST'],
        'clearAction' => ['uri' => '/api/cart/clear', 'method' => 'POST'],
    ];
}

function build_user(): array
{
    $role = $_SESSION['role'] ?? 'guest';
    $names = ['guest' => 'Guest', 'member' => 'Member Mike', 'admin' => 'Admin Alice'];
    return [
        'role' => $role,
        'name' => $names[$role] ?? 'Guest',
        'toggleAction' => ['uri' => '/api/user/toggle', 'method' => 'POST'],
        'resetAllAction' => ['uri' => '/api/reset-all', 'method' => 'POST'],
    ];
}

function build_notice(): array
{
    return [
        'html' => $_SESSION['notice'] ?? '<em>No notice set.</em>',
        'setAction' => ['uri' => '/api/notice/set', 'method' => 'POST'],
        'failAction' => ['uri' => '/api/notice/fail', 'method' => 'POST'],
    ];
}

/** Build the snapshot for a type with its demo scope/options. */
function snap(string $type): ?array
{
    switch ($type) {
        case 'counter':  return make_snapshot('counter', 'session', build_counter());
        case 'products': return make_snapshot('products', 'session', build_products());
        case 'cart':     return make_snapshot('cart', 'session', build_cart(), ['always' => true]);
        case 'user':     return make_snapshot('user', 'session', build_user(), ['stale' => true]);
        case 'notice':   return make_snapshot('notice', 'flow', build_notice());
    }
    return null;
}

function all_directives(): array
{
    global $TYPES;
    $dirs = [];
    foreach ($TYPES as $t) {
        $s = snap($t);
        if ($s) $dirs[] = set_dir($s);
    }
    return $dirs;
}

// ---------------------------------------------------------------------------
// Realtime channel helpers (demo pub/sub via the filesystem)
// ---------------------------------------------------------------------------

/** A `set` directive wrapping an already-framed snapshot. */
function set_dir_frame(array $frame): array
{
    return ['type' => 'set', 'snapshot_frame' => $frame];
}

/** Directory holding per-channel subscription and event-log files. */
function channel_dir(): string
{
    $dir = rtrim(sys_get_temp_dir(), '/') . '/statum_demo';
    if (!is_dir($dir)) @mkdir($dir, 0777, true);
    return $dir;
}

function channel_subs_path(string $id): string   { return channel_dir() . '/ch_' . $id . '.subs'; }
function channel_events_path(string $id): string { return channel_dir() . '/ch_' . $id . '.events'; }
function gen_channel_id(): string                { return bin2hex(random_bytes(8)); }

/** Read a channel's subscribed slot keys (empty array if unknown). */
function read_channel_subs(string $id): array
{
    $f = channel_subs_path($id);
    if (!is_file($f)) return [];
    $arr = json_decode((string)@file_get_contents($f), true);
    return is_array($arr) ? array_values($arr) : [];
}

/** Overwrite a channel's subscribed slot keys. */
function set_channel_subs(string $id, array $subs): void
{
    @file_put_contents(channel_subs_path($id), json_encode(array_values(array_unique($subs))), LOCK_EX);
}

/** Append one event to a channel's event log (a JSON line). */
function append_channel_event(string $id, string $type, $data): void
{
    $line = json_encode(['t' => $type, 'd' => $data]) . "\n";
    @file_put_contents(channel_events_path($id), $line, FILE_APPEND | LOCK_EX);
}

/** Channel ids currently subscribed to a given slot key. */
function channels_subscribed_to(string $key): array
{
    $ids = [];
    foreach ((array)glob(channel_dir() . '/ch_*.subs') as $f) {
        if (!is_file($f)) continue;
        $id = substr(basename($f), 3, -5); // ch_{ID}.subs
        if (in_array($key, read_channel_subs($id), true)) $ids[] = $id;
    }
    return $ids;
}

/** Push a `set` frame to every channel subscribed to a slot key. */
function fanout_set(string $key, array $frame): void
{
    foreach (channels_subscribed_to($key) as $id) append_channel_event($id, 'set', $frame);
}

/** Shared "live" broadcast value, stored on disk so all workers agree. */
function live_path(): string { return channel_dir() . '/live.json'; }

function live_read(): array
{
    if (!is_file(live_path())) return ['text' => 'No live messages yet.', 'at' => ''];
    $d = json_decode((string)@file_get_contents(live_path()), true);
    return is_array($d) ? $d : ['text' => 'No live messages yet.', 'at' => ''];
}

/** Snapshot for the shared `live` slot (transient scope, fixed key `k_live`). */
function snap_live(): array
{
    $d = live_read();
    return make_snapshot('live', 'transient', [
        'text' => $d['text'],
        'at' => $d['at'],
        'sendAction' => ['uri' => '/api/live/send', 'method' => 'POST'],
        'muteAction' => ['uri' => '/api/live/mute', 'method' => 'POST'],
        'resumeAction' => ['uri' => '/api/live/resume', 'method' => 'POST'],
    ], [], 'k_live');
}

/** The initial frame to push to a channel when it subscribes to a key. */
function initial_frame_for_key(string $key): ?array
{
    if ($key === 'k_live') return frame(snap_live());
    return null;
}

// ---------------------------------------------------------------------------
// Router
// ---------------------------------------------------------------------------

$method = $_SERVER['REQUEST_METHOD'];
$path = parse_url($_SERVER['REQUEST_URI'], PHP_URL_PATH);

// Static assets.
if ($path === '/statum.js') {
    header('Content-Type: application/javascript');
    readfile(__DIR__ . '/../js/statum.js');
    exit;
}

if ($path === '/_statum/worker.js') {
    header('Content-Type: application/javascript');
    readfile(__DIR__ . '/../js/statum-worker.js');
    exit;
}

// The page itself.
if ($path === '/' || $path === '/index.html') {
    header('Content-Type: text/html; charset=utf-8');
    readfile(__DIR__ . '/index.html');
    exit;
}

// Landing page after a (post) checkout navigation.
if ($path === '/api/checkout-done') {
    header('Content-Type: text/html; charset=utf-8');
    $order = htmlspecialchars($_GET['order'] ?? 'UNKNOWN');
    echo "<!doctype html><meta charset=utf-8><title>Checkout</title>" .
         "<body style='font-family:sans-serif;padding:2rem'>" .
         "<h1>Order $order placed</h1><p>This page was reached via a statum <code>post</code> directive.</p>" .
         "<p><a href='/'>Back to the demo</a></p>";
    exit;
}

// --- Statum endpoints -------------------------------------------------------

if ($path === '/_statum/entrypoint') {
    $dirs = all_directives();
    $dirs[] = set_dir(snap_live());                       // current broadcast + actions
    $dirs[] = ['type' => 'subscribe', 'key' => 'k_live']; // begin realtime delivery
    statum_respond($dirs);
}

if ($path === '/_statum/slots') {
    // Pre-flight target: refresh each type whose frame the client sent back.
    $body = json_decode(file_get_contents('php://input'), true);
    $dirs = [];
    if (is_array($body)) {
        foreach ($body as $fr) {
            $snap = json_decode($fr['content'] ?? 'null', true);
            $type = $snap['slot']['type'] ?? null;
            if ($type) {
                $s = snap($type);
                if ($s) $dirs[] = set_dir($s);
            }
        }
    }
    statum_respond($dirs);
}

// --- Realtime channel -------------------------------------------------------

// Open the SSE stream. The first event carries the channel id; later events
// are `set`/`clear` for the slots this channel has subscribed to (via PATCH).
if ($path === '/_statum/channel' && $method === 'GET') {
    // Long-lived connection: release the session lock immediately so other
    // requests (served by other workers) for the same session are not blocked.
    session_write_close();
    @set_time_limit(0);
    @ini_set('zlib.output_compression', '0');
    while (ob_get_level() > 0) { @ob_end_flush(); }
    header('Content-Type: text/event-stream');
    header('Cache-Control: no-cache');
    header('X-Accel-Buffering: no');

    $id = gen_channel_id();
    set_channel_subs($id, []);
    @file_put_contents(channel_events_path($id), '');

    echo "event: channel\ndata: {$id}\n\n";
    flush();

    $offset = 0;
    $eventsFile = channel_events_path($id);
    while (!connection_aborted()) {
        clearstatcache(true, $eventsFile);
        $size = is_file($eventsFile) ? filesize($eventsFile) : 0;
        if ($size > $offset) {
            $fp = @fopen($eventsFile, 'rb');
            if ($fp) {
                fseek($fp, $offset);
                $chunk = (string)stream_get_contents($fp);
                $offset = (int)ftell($fp);
                fclose($fp);
                foreach (preg_split('/\r?\n/', trim($chunk)) as $line) {
                    if ($line === '') continue;
                    $ev = json_decode($line, true);
                    if (!is_array($ev)) continue;
                    if (($ev['t'] ?? '') === 'set') {
                        echo 'event: set' . "\ndata: " . json_encode($ev['d']) . "\n\n";
                    } elseif (($ev['t'] ?? '') === 'clear') {
                        echo 'event: clear' . "\ndata: " . $ev['d'] . "\n\n";
                    }
                }
                flush();
            }
        }
        echo ": ping\n\n"; // heartbeat keeps the stream alive
        flush();
        usleep(1000000);
    }
    // Client went away: drop this channel's files.
    @unlink(channel_subs_path($id));
    @unlink($eventsFile);
    exit;
}

// Manage a channel's subscriptions: {subscribe:[keys], unsubscribe:[keys]}.
if (preg_match('#^/_statum/channel/([^/]+)$#', $path, $m) && $method === 'PATCH') {
    $id = $m[1];
    $body = json_decode((string)file_get_contents('php://input'), true);
    if (!is_array($body)) $body = [];
    $add = $body['subscribe'] ?? [];
    $rem = $body['unsubscribe'] ?? [];
    $subs = read_channel_subs($id);
    foreach ($add as $k) {
        if (!in_array($k, $subs, true)) $subs[] = $k;
    }
    $subs = array_values(array_diff($subs, $rem));
    set_channel_subs($id, $subs);
    // Push the current value of any newly-subscribed slot so a fresh client
    // sees the latest state right away.
    foreach ($add as $k) {
        $fr = initial_frame_for_key((string)$k);
        if ($fr !== null) append_channel_event($id, 'set', $fr);
    }
    http_response_code(204);
    exit;
}

// --- Actions ----------------------------------------------------------------

if ($path === '/api/counter/inc' && $method === 'POST') {
    $_SESSION['counter'] = ($_SESSION['counter'] ?? 0) + 1;
    statum_respond([set_dir(snap('counter'))]);
}

if ($path === '/api/counter/dec' && $method === 'POST') {
    $_SESSION['counter'] = ($_SESSION['counter'] ?? 0) - 1;
    statum_respond([set_dir(snap('counter'))]);
}

if ($path === '/api/counter/reset' && $method === 'POST') {
    $_SESSION['counter'] = 0;
    statum_respond([set_dir(snap('counter'))]);
}

if ($path === '/api/counter/add' && $method === 'POST') {
    $amount = intval($_POST['amount'] ?? 1);
    $_SESSION['counter'] = ($_SESSION['counter'] ?? 0) + $amount;
    statum_respond([set_dir(snap('counter'))]);
}

if ($path === '/api/cart/add' && $method === 'POST') {
    $priv = $_SERVER['HTTP_X_STATUM_PRIVATE'] ?? '';
    $raw = base64_decode($priv, true) ?: '';
    if (preg_match('/id=([0-9]+)/', $raw, $m)) {
        $id = $m[1];
        $_SESSION['cart'][$id] = ($_SESSION['cart'][$id] ?? 0) + 1;
    }
    statum_respond([set_dir(snap('cart'))]);
}

if ($path === '/api/cart/clear' && $method === 'POST') {
    $_SESSION['cart'] = [];
    statum_respond([set_dir(snap('cart'))]);
}

if ($path === '/api/cart/checkout' && $method === 'POST') {
    // A `post` directive performs a full-page navigating form POST last.
    statum_respond([['type' => 'post', 'uri' => '/api/checkout-done', 'data' => ['order' => 'ORD-' . strtoupper(bin2hex(random_bytes(3)))]]]);
}

if ($path === '/api/user/toggle' && $method === 'POST') {
    $order = ['guest', 'member', 'admin'];
    $cur = array_search($_SESSION['role'] ?? 'guest', $order);
    $_SESSION['role'] = $order[($cur + 1) % count($order)];
    statum_respond([set_dir(snap('user'))]);
}

if ($path === '/api/notice/set' && $method === 'POST') {
    $_SESSION['notice'] = '<div class="ok"><strong>Notice:</strong> updated at ' . date('H:i:s') . '</div>';
    statum_respond([set_dir(snap('notice'))]);
}

if ($path === '/api/notice/fail' && $method === 'POST') {
    statum_respond([['type' => 'error', 'message' => 'Intentional failure.', 'code' => 'DEMO_FAIL', 'data' => ['at' => date('c')]]]);
}

if ($path === '/api/reset-all' && $method === 'POST') {
    $dirs = [];
    foreach ($TYPES as $t) {
        if (isset($_SESSION['keys'][$t])) $dirs[] = clear_dir($_SESSION['keys'][$t]);
    }
    session_unset();
    $dirs[] = ['type' => 'navigate', 'uri' => '/'];
    statum_respond($dirs);
}

// --- Live channel actions ---------------------------------------------------

if ($path === '/api/live/send' && $method === 'POST') {
    $msg = trim((string)($_POST['msg'] ?? ''));
    if ($msg === '') $msg = '(empty message)';
    @file_put_contents(live_path(), json_encode(['text' => $msg, 'at' => date('H:i:s')]), LOCK_EX);
    $fr = frame(snap_live());
    fanout_set('k_live', $fr);   // push to every subscribed tab via the channel
    statum_respond([
        set_dir_frame($fr),      // immediate feedback (idempotent vs the push)
        ['type' => 'notify', 'kind' => 'success', 'message' => 'Broadcast to all tabs'],
    ]);
}

// Unsubscribing the only slot closes the realtime connection for this client.
if ($path === '/api/live/mute' && $method === 'POST') {
    statum_respond([['type' => 'unsubscribe', 'key' => 'k_live']]);
}

// Re-subscribing reopens the connection; the PATCH initial-push re-sends the
// current broadcast.
if ($path === '/api/live/resume' && $method === 'POST') {
    statum_respond([['type' => 'subscribe', 'key' => 'k_live']]);
}

// Fallback 404.
http_response_code(404);
header('Content-Type: text/plain');
echo "404 Not Found: $path";