<?php
/**
 * SSE endpoint: stream events for a specific user's conversation
 * - Usage: /api/stream_user_messages.php?user_id=3180&since=YYYY-MM-DD%20HH:MM:SS
 * - Security: requires session authentication (requireAuthentication)
 * - Behavior: tries to use Redis pub/sub if available, otherwise polls DB every 2s for new messages.
 */
require_once __DIR__ . '/../config/config.php';
requireAuthentication();

header('Content-Type: text/event-stream');
header('Cache-Control: no-cache');
header('Connection: keep-alive');
// Tell proxies not to buffer
if (function_exists('apache_setenv')) @apache_setenv('no-gzip', '1');

try {
    $userId = intval($_GET['user_id'] ?? 0);
    if (!$userId) {
        http_response_code(400);
        echo "data: {\"error\": \"user_id es requerido\"}\n\n";
        exit;
    }

    ignore_user_abort(true);
    set_time_limit(0);

    $db = Database::getInstance();

    // initialize last_ts: if client passed since, use it; else use latest message timestamp or now-1s
    $last = !empty($_GET['since']) ? $_GET['since'] : null;
    if (is_null($last)) {
        try {
            // Database wrapper exposes fetch() for a single row
            $r = $db->fetch('SELECT created_at FROM conversations WHERE user_id = :uid ORDER BY created_at DESC LIMIT 1', ['uid' => $userId]);
            $last = $r && isset($r['created_at']) ? $r['created_at'] : date('Y-m-d H:i:s', time() - 1);
        } catch (Exception $e) {
            $last = date('Y-m-d H:i:s', time() - 1);
        }
    }

    // Try Redis pub/sub first (if available)
    $usedRedis = false;
    if (extension_loaded('redis')) {
        try {
            $redis = new Redis();
            // connect using defaults; expect redis on localhost:6379
            $redis->connect('127.0.0.1');
            // subscribe in a non-blocking manner via pub/sub loop
            $chan = "user_{$userId}";
            // Use raw subscribe loop with callback
            $usedRedis = true;
            $redis->subscribe([$chan], function ($redis, $chanName, $msg) use (&$last) {
                // msg expected JSON string with {type:'new_message', user_id, message_id, created_at}
                $payload = $msg;
                if (is_array($msg)) $payload = json_encode($msg);
                echo "event: new_message\n";
                echo "data: {$payload}\n\n";
                @ob_flush(); @flush();
            });
        } catch (Exception $e) {
            error_log('stream_user_messages.php - redis subscribe failed: ' . $e->getMessage());
            $usedRedis = false;
        }
    }

    // Fallback: polling loop (also used if redis is not available or subscription ended)
    $start = time();
    $timeout = 55; // keep connection alive for ~55s (EventSource will reconnect)
    while (!connection_aborted()) {
        // Query for messages strictly newer than last timestamp
        try {
            $rows = $db->fetchAll(
                'SELECT id, created_at FROM conversations WHERE user_id = :uid AND created_at > :since ORDER BY created_at ASC',
                ['uid' => $userId, 'since' => $last]
            );
        } catch (Exception $e) {
            error_log('stream_user_messages.php - DB poll failed: ' . $e->getMessage());
            $rows = [];
        }

        if (!empty($rows)) {
            foreach ($rows as $r) {
                $d = json_encode(['type' => 'new_message', 'user_id' => $userId, 'message_id' => (int)$r['id'], 'created_at' => $r['created_at']]);
                echo "event: new_message\n";
                echo "data: {$d}\n\n";
                @ob_flush(); @flush();
                $last = $r['created_at'];
            }
        }

        // Periodic ping to keep connection alive
        echo "event: ping\n";
        echo "data: {\"now\": \"" . date('Y-m-d H:i:s') . "\"}\n\n";
        @ob_flush(); @flush();

        // Break/refresh connection after timeout so client reconnects (avoids very long running PHP requests)
        if ((time() - $start) > $timeout) break;

        sleep(2);
    }

    // Final flush
    echo "event: close\n";
    echo "data: {}\n\n";
    @ob_flush(); @flush();
    exit;
} catch (Exception $e) {
    error_log('stream_user_messages.php - fatal: ' . $e->getMessage());
    http_response_code(500);
    echo "data: {\"error\": \"Internal server error\"}\n\n";
    exit;
}