<?php
/**
 * Segment cache for the PHP stream proxy.
 *
 * Strategy (selected at request time, never re-evaluated):
 *   1. APCu       — fastest, in-memory, available on most cPanel hosts
 *                   that have OPcache. ~zero-copy, no syscalls per hit.
 *   2. Redis      — second choice if PhpRedis is loaded AND the env var
 *                   NDP_REDIS_URL points at a reachable instance.
 *   3. File       — universal fallback (works on every shared host),
 *                   stored under sys_get_temp_dir()/ndp-segcache/.
 *                   Slower but correct.
 *
 * Single-fetch semantics:
 *   When two HTTP requests for the same segment hit PHP within the same
 *   ~10 ms window we don't want to send two parallel upstream fetches.
 *   We use an exclusive lock (APCu CAS, Redis SETNX, or flock on file)
 *   on the segment key. The first request acquires the lock, fetches
 *   upstream, writes to the cache, releases. Subsequent requests poll
 *   the cache for up to 12 s before giving up — by that time either
 *   the cache is populated or the original fetch failed.
 *
 * TTL: 60 s (matches the Python SegmentHub).
 *
 * Memory budget: 200 MB — APCu has its own size cap (apcu.shm_size),
 * Redis has its own LRU policy, and we explicitly enforce 200 MB
 * for the file backend by tracking total bytes in a sidecar manifest.
 */

if (!defined('NDP_SEGMENT_CACHE_LOADED')) {
    define('NDP_SEGMENT_CACHE_LOADED', true);

    define('NDP_SEG_TTL_SECONDS',     60);
    define('NDP_SEG_MAX_BYTES',       200 * 1024 * 1024);
    define('NDP_SEG_LOCK_TIMEOUT_S',  12);     // give up waiting after 12 s

    /** Auto-detect backend. Cached for the rest of the process.
     *
     * Defensive design: every backend probe is wrapped so that a broken
     * extension (e.g. APCu .so compiled for the wrong PHP ABI) can NEVER
     * surface as a 500/502 — we always degrade gracefully to the file
     * backend, which works on every host. This is what unblocks an
     * operator whose `extension=apcu.so` line caused PHP-FPM to crash.
     */
    function ndp_seg_backend(): string {
        static $b = null;
        if ($b !== null) return $b;

        // 1) APCu — only if the extension is fully functional.
        try {
            if (function_exists('apcu_enabled') && @apcu_enabled()
                && function_exists('apcu_store') && function_exists('apcu_fetch')) {
                // Smoke test: write and read back a sentinel. If the .so
                // is loaded but the SHM is misconfigured (apcu.shm_size=0
                // or PHP-FPM lacks SysV IPC perms) apcu_store silently
                // returns false — we MUST detect that here, not at first
                // segment-fetch.
                $probe_key = 'ndpseg:_probe_' . getmypid();
                if (@apcu_store($probe_key, 1, 5) === true
                    && @apcu_fetch($probe_key) === 1) {
                    @apcu_delete($probe_key);
                    return $b = 'apcu';
                }
            }
        } catch (Throwable $e) { /* fall through */ }

        // 2) Redis — only if PhpRedis is loaded AND NDP_REDIS_URL env is set.
        try {
            if (extension_loaded('redis') && getenv('NDP_REDIS_URL')) {
                $r = ndp_redis();
                $r->ping();
                return $b = 'redis';
            }
        } catch (Throwable $e) { /* fall through */ }

        // 3) Universal fallback — works on every shared host.
        return $b = 'file';
    }

    function ndp_redis(): \Redis {
        static $r = null;
        if ($r !== null) return $r;
        $url = getenv('NDP_REDIS_URL') ?: 'tcp://127.0.0.1:6379';
        $p = parse_url($url);
        $host = $p['host'] ?? '127.0.0.1';
        $port = (int)($p['port'] ?? 6379);
        $r = new \Redis();
        $r->connect($host, $port, 1.5);
        if (!empty($p['pass'])) $r->auth($p['pass']);
        return $r;
    }

    /**
     * Get cached segment.
     * @return array|null  ['body'=>string, 'ct'=>string] or null
     */
    function ndp_seg_get(string $key): ?array {
        $key = 'ndpseg:' . sha1($key);
        try {
            switch (ndp_seg_backend()) {
                case 'apcu':
                    $v = @apcu_fetch($key, $ok);
                    if (!$ok || !is_array($v)) return null;
                    return $v;
                case 'redis':
                    $v = ndp_redis()->get($key);
                    if (!$v) return null;
                    $arr = json_decode($v, true);
                    if (!is_array($arr)) return null;
                    $arr['body'] = base64_decode($arr['body'] ?? '');
                    return $arr;
                case 'file':
                default:
                    return _ndp_seg_file_get($key);
            }
        } catch (Throwable $e) {
            // Cache fault must NEVER take down the request — we treat it
            // as a miss and let the caller fetch upstream directly.
            @error_log('[NADIPLAY seg cache] get failed: ' . $e->getMessage());
            return null;
        }
    }

    // ── Telemetry counters (hits / misses / inflight) ─────────────
    // Stored alongside the cache so they survive across PHP-FPM
    // workers. APCu is atomic; Redis uses INCR/DECR; the file
    // backend uses an flock-protected counter file.

    function ndp_seg_count_hit():     void { _ndp_seg_counter_inc('hits'); }
    function ndp_seg_count_miss():    void { _ndp_seg_counter_inc('misses'); }
    function ndp_seg_count_inflight_inc(): void { _ndp_seg_counter_inc('inflight'); }
    function ndp_seg_count_inflight_dec(): void { _ndp_seg_counter_inc('inflight', -1); }

    function _ndp_seg_counter_inc(string $name, int $by = 1): void {
        $key = 'ndpseg-stat:' . $name;
        try {
            switch (ndp_seg_backend()) {
                case 'apcu':
                    if ($by >= 0) {
                        if (!@apcu_inc($key, $by)) @apcu_store($key, $by);
                    } else {
                        @apcu_dec($key, -$by);
                    }
                    break;
                case 'redis':
                    $r = ndp_redis();
                    if ($by >= 0) $r->incrBy($key, $by);
                    else          $r->decrBy($key, -$by);
                    break;
                case 'file':
                default:
                    _ndp_seg_file_counter_inc($name, $by);
            }
        } catch (Throwable $e) {
            // Counters are fire-and-forget — they must NEVER throw.
        }
    }

    function _ndp_seg_counter_get(string $name): int {
        $key = 'ndpseg-stat:' . $name;
        switch (ndp_seg_backend()) {
            case 'apcu':
                $v = apcu_fetch($key, $ok);
                return $ok ? (int)$v : 0;
            case 'redis':
                $v = ndp_redis()->get($key);
                return is_numeric($v) ? (int)$v : 0;
            case 'file':
            default:
                $f = _ndp_seg_file_dir() . '/_stat-' . $name;
                return is_file($f) ? max(0, (int)@file_get_contents($f)) : 0;
        }
    }

    function _ndp_seg_file_counter_inc(string $name, int $by): void {
        $f = _ndp_seg_file_dir() . '/_stat-' . $name;
        $fp = @fopen($f, 'c+');
        if (!$fp) return;
        try {
            if (flock($fp, LOCK_EX)) {
                $cur = (int)stream_get_contents($fp);
                $cur = max(0, $cur + $by);
                rewind($fp);
                ftruncate($fp, 0);
                fwrite($fp, (string)$cur);
                fflush($fp);
                flock($fp, LOCK_UN);
            }
        } finally {
            fclose($fp);
        }
    }

    /** Store body+content-type with TTL. */
    function ndp_seg_put(string $key, string $body, string $ct): void {
        $key = 'ndpseg:' . sha1($key);
        $entry = ['body' => $body, 'ct' => $ct];
        try {
            switch (ndp_seg_backend()) {
                case 'apcu':
                    @apcu_store($key, $entry, NDP_SEG_TTL_SECONDS);
                    break;
                case 'redis':
                    $serial = json_encode([
                        'body' => base64_encode($body),
                        'ct'   => $ct,
                    ]);
                    ndp_redis()->setex($key, NDP_SEG_TTL_SECONDS, $serial);
                    break;
                case 'file':
                default:
                    _ndp_seg_file_put($key, $entry);
            }
        } catch (Throwable $e) {
            @error_log('[NADIPLAY seg cache] put failed: ' . $e->getMessage());
        }
    }

    /**
     * Acquire an exclusive lock for `key`. Blocks up to ~12 s waiting
     * for an in-flight fetch to populate the cache. Returns:
     *   - 'creator'  → caller MUST run the upstream fetch + ndp_seg_put + ndp_seg_unlock
     *   - 'cached'   → cache populated while we waited; caller skips fetch
     *   - 'timeout'  → caller should fall through to fetching directly
     */
    function ndp_seg_lock(string $key): string {
        $lockKey = 'ndpseg-lock:' . sha1($key);
        $deadline = microtime(true) + NDP_SEG_LOCK_TIMEOUT_S;
        $backend = ndp_seg_backend();

        while (true) {
            // Try to acquire.
            $got = false;
            if ($backend === 'apcu') {
                $got = apcu_add($lockKey, 1, NDP_SEG_LOCK_TIMEOUT_S + 5);
            } elseif ($backend === 'redis') {
                $got = (bool)ndp_redis()->set($lockKey, '1', ['NX', 'EX' => NDP_SEG_LOCK_TIMEOUT_S + 5]);
            } else {
                $got = _ndp_seg_file_lock_try($lockKey);
            }
            if ($got) return 'creator';
            // Lock contended → wait briefly + check cache.
            usleep(50_000); // 50 ms
            if (ndp_seg_get($key) !== null) return 'cached';
            if (microtime(true) >= $deadline) return 'timeout';
        }
    }

    function ndp_seg_unlock(string $key): void {
        $lockKey = 'ndpseg-lock:' . sha1($key);
        switch (ndp_seg_backend()) {
            case 'apcu':  apcu_delete($lockKey); break;
            case 'redis': ndp_redis()->del($lockKey); break;
            default:      _ndp_seg_file_lock_release($lockKey);
        }
    }

    // ── File backend ──────────────────────────────────────────────
    function _ndp_seg_file_dir(): string {
        $d = sys_get_temp_dir() . '/ndp-segcache';
        if (!is_dir($d)) @mkdir($d, 0700, true);
        return $d;
    }

    function _ndp_seg_file_get(string $key): ?array {
        $f = _ndp_seg_file_dir() . '/' . $key;
        if (!is_file($f)) return null;
        $age = time() - filemtime($f);
        if ($age > NDP_SEG_TTL_SECONDS) {
            @unlink($f);
            return null;
        }
        $raw = @file_get_contents($f);
        if ($raw === false || strlen($raw) < 6) return null;
        // Header: 4 bytes ct length (big-endian), then ct, then body.
        $ctLen = unpack('N', substr($raw, 0, 4))[1] ?? 0;
        if ($ctLen < 0 || $ctLen > 256) return null;
        $ct = substr($raw, 4, $ctLen);
        $body = substr($raw, 4 + $ctLen);
        return ['body' => $body, 'ct' => $ct];
    }

    function _ndp_seg_file_put(string $key, array $entry): void {
        $dir = _ndp_seg_file_dir();
        // Enforce 200 MB cap with a one-shot purge of oldest files when
        // we're over budget (cheap because we run rarely on cache miss).
        _ndp_seg_file_evict($dir);
        $body = (string)($entry['body'] ?? '');
        $ct   = (string)($entry['ct']   ?? 'video/mp2t');
        $blob = pack('N', strlen($ct)) . $ct . $body;
        $tmp  = tempnam($dir, 'seg-');
        if ($tmp === false) return;
        if (file_put_contents($tmp, $blob) !== false) {
            @rename($tmp, $dir . '/' . $key);
        } else {
            @unlink($tmp);
        }
    }

    function _ndp_seg_file_evict(string $dir): void {
        $total = 0;
        $files = [];
        foreach (glob($dir . '/ndpseg:*') ?: [] as $f) {
            $sz = @filesize($f) ?: 0;
            $total += $sz;
            $files[$f] = @filemtime($f) ?: 0;
        }
        if ($total <= NDP_SEG_MAX_BYTES) return;
        asort($files);  // oldest first
        foreach ($files as $f => $_) {
            if ($total <= NDP_SEG_MAX_BYTES) return;
            $sz = @filesize($f) ?: 0;
            if (@unlink($f)) $total -= $sz;
        }
    }

    function _ndp_seg_file_lock_try(string $key): bool {
        $f = _ndp_seg_file_dir() . '/' . $key . '.lock';
        $fp = @fopen($f, 'c');
        if (!$fp) return false;
        if (flock($fp, LOCK_EX | LOCK_NB)) {
            // Hold the fp by stashing it in a global keyed by the lock
            // path — released via _ndp_seg_file_lock_release.
            $GLOBALS['__ndp_lock_fp'][$key] = $fp;
            return true;
        }
        fclose($fp);
        return false;
    }

    function _ndp_seg_file_lock_release(string $key): void {
        if (!empty($GLOBALS['__ndp_lock_fp'][$key])) {
            $fp = $GLOBALS['__ndp_lock_fp'][$key];
            @flock($fp, LOCK_UN);
            @fclose($fp);
            unset($GLOBALS['__ndp_lock_fp'][$key]);
        }
        @unlink(_ndp_seg_file_dir() . '/' . $key . '.lock');
    }

    /** Tiny stats helper for /api/proxy/stats. */
    function ndp_seg_stats(): array {
        $backend = ndp_seg_backend();
        $count = 0; $bytes = 0;
        if ($backend === 'apcu' && function_exists('apcu_cache_info')) {
            $info = @apcu_cache_info();
            foreach ($info['cache_list'] ?? [] as $e) {
                if (isset($e['key']) && strpos($e['key'], 'ndpseg:') === 0) {
                    $count++;
                    $bytes += (int)($e['mem_size'] ?? 0);
                }
            }
        } elseif ($backend === 'redis') {
            // Cheap approximation — KEYS pattern is O(N) over the keyspace
            // but we typically have <1000 segments at a time so it's fine.
            try {
                $keys = ndp_redis()->keys('ndpseg:*');
                $count = is_array($keys) ? count($keys) : 0;
                // Bytes total is hard to compute cheaply; leave at 0.
            } catch (Throwable $e) { /* ignore */ }
        } elseif ($backend === 'file') {
            foreach (glob(_ndp_seg_file_dir() . '/ndpseg:*') ?: [] as $f) {
                $count++; $bytes += @filesize($f) ?: 0;
            }
        }
        $hits     = _ndp_seg_counter_get('hits');
        $misses   = _ndp_seg_counter_get('misses');
        $inflight = _ndp_seg_counter_get('inflight');
        $signs    = _ndp_seg_counter_get('sign_requests');
        $total    = $hits + $misses;
        return [
            'backend'         => $backend,
            'cached_segments' => $count,
            'buffer_bytes'    => $bytes,
            'buffer_max_bytes'=> NDP_SEG_MAX_BYTES,
            'buffer_pct'      => NDP_SEG_MAX_BYTES > 0
                ? round(100.0 * $bytes / NDP_SEG_MAX_BYTES, 2) : 0,
            'ttl_seconds'     => NDP_SEG_TTL_SECONDS,
            'hits'            => $hits,
            'misses'          => $misses,
            'hit_rate_pct'    => $total > 0 ? round(100.0 * $hits / $total, 2) : 0,
            'active_streams'  => max(0, $inflight),
            // NEW: how many times any device has hit /api/proxy/sign.
            // If this stays at 0 while subscribers are watching, the
            // installed APK is older than the proxy code (rebuild).
            'sign_requests'   => max(0, $signs),
        ];
    }
}
