<?php

declare(strict_types=1);

namespace PpPhp\Core\Realtime;

use PpPhp\Core\DB;

/**
 * Real-time message bus — the shared-hosting-friendly backbone behind PPPHP's
 * live event hooks. It is a tiny append-only table that a publisher writes to
 * and subscribers read forward from by id; the SSE endpoint (api/realtime.php)
 * tails it and streams new rows, with a polling fallback for hosts that buffer
 * or time-limit streaming responses.
 *
 * Why not raw WebSockets? A generated app runs as ordinary PHP on ordinary
 * hosting — there is no persistent process to hold sockets. SSE + this bus gives
 * genuine server→client push with zero extra infrastructure. For true
 * bidirectional sockets an external provider (Pusher/Ably/Mercure) can back the
 * same `pp.realtime` client later; the authoring surface never changes.
 *
 * Channels are plain strings. A message may target a single user (`userId`) or
 * everyone on the channel (`userId = null`). Private channels are gated by the
 * app's `onChannelAuth` event, checked in the endpoint before a subscription is
 * honoured.
 */
final class RealtimeBus
{
    private static string $table = 'pp_realtime';

    /** Override the backing table name (build may namespace it). */
    public static function setTable(string $name): void
    {
        $clean = preg_replace('/[^A-Za-z0-9_]/', '', $name);
        if ($clean !== null && $clean !== '') {
            self::$table = $clean;
        }
    }

    public static function table(): string
    {
        return self::$table;
    }

    /** Create the backing table if it is not there yet (id-ordered, GC-friendly). */
    public static function ensureTable(): void
    {
        $t = self::$table;
        DB::Query(
            "CREATE TABLE IF NOT EXISTS `{$t}` (\n" .
            "  id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,\n" .
            "  channel VARCHAR(160) NOT NULL,\n" .
            "  event VARCHAR(120) NOT NULL,\n" .
            "  payload MEDIUMTEXT NULL,\n" .
            "  user_id INT UNSIGNED NULL,\n" .
            "  created_at DATETIME NOT NULL,\n" .
            "  INDEX idx_rt_read (channel, id),\n" .
            "  INDEX idx_rt_gc (created_at)\n" .
            ")"
        );
    }

    /**
     * Publish a message to a channel. Returns the new row id (0 on failure —
     * real-time is best-effort and must never break the action that fired it).
     *
     * @param mixed $payload Any JSON-serialisable value.
     * @param int|null $userId Deliver only to this user; null = everyone on the channel.
     */
    public static function publish(string $channel, string $event, mixed $payload = null, ?int $userId = null): int
    {
        $channel = self::cleanChannel($channel);
        $event = trim($event);
        if ($channel === '' || $event === '') {
            return 0;
        }
        try {
            self::ensureTable();
            DB::Query(
                "INSERT INTO `" . self::$table . "` (channel, event, payload, user_id, created_at) VALUES (?, ?, ?, ?, ?)",
                [$channel, $event, self::encode($payload), $userId, self::now()]
            );
            return (int) DB::LastInsertId();
        } catch (\Throwable $e) {
            return 0;
        }
    }

    /**
     * Read messages newer than $lastId on the given channels, visible to $userId
     * (its own targeted messages + untargeted ones). Ordered oldest→newest.
     *
     * @param array<int,string> $channels
     * @return array<int, array{id:int,channel:string,event:string,data:mixed,at:string}>
     */
    public static function since(array $channels, int $lastId, ?int $userId = null, int $limit = 200): array
    {
        $channels = array_values(array_unique(array_filter(array_map([self::class, 'cleanChannel'], $channels), static fn($c) => $c !== '')));
        if ($channels === []) {
            return [];
        }
        $limit = max(1, min($limit, 500));
        try {
            $place = implode(',', array_fill(0, count($channels), '?'));
            $params = $channels;
            $params[] = $lastId;
            $userClause = '';
            if ($userId !== null) {
                $userClause = ' AND (user_id IS NULL OR user_id = ?)';
                $params[] = $userId;
            } else {
                $userClause = ' AND user_id IS NULL';
            }
            $rows = DB::FetchAll(
                "SELECT id, channel, event, payload FROM `" . self::$table . "`" .
                " WHERE channel IN ($place) AND id > ?{$userClause} ORDER BY id ASC LIMIT {$limit}",
                $params
            );
            $out = [];
            foreach ((array) $rows as $r) {
                $out[] = [
                    'id' => (int) $r['id'],
                    'channel' => (string) $r['channel'],
                    'event' => (string) $r['event'],
                    'data' => self::decode((string) ($r['payload'] ?? '')),
                ];
            }
            return $out;
        } catch (\Throwable $e) {
            return [];
        }
    }

    /** The current head id — a fresh subscriber starts here so it never replays history. */
    public static function latestId(): int
    {
        try {
            self::ensureTable();
            return (int) DB::FetchOne("SELECT COALESCE(MAX(id), 0) FROM `" . self::$table . "`");
        } catch (\Throwable $e) {
            return 0;
        }
    }

    /** Drop messages older than $minutes so the table stays small. Best-effort. */
    public static function gc(int $minutes = 60): void
    {
        try {
            DB::Query(
                "DELETE FROM `" . self::$table . "` WHERE created_at < ?",
                [date('Y-m-d H:i:s', time() - max(1, $minutes) * 60)]
            );
        } catch (\Throwable $e) {
            // ignore
        }
    }

    public static function cleanChannel(string $channel): string
    {
        // Channels are dotted identifiers: "orders", "user.42", "room.general".
        $c = preg_replace('/[^A-Za-z0-9_.:-]/', '', trim($channel)) ?? '';
        return substr($c, 0, 160);
    }

    private static function encode(mixed $payload): string
    {
        if (is_string($payload)) {
            return $payload;
        }
        $json = json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
        return $json === false ? '' : $json;
    }

    private static function decode(string $raw): mixed
    {
        if ($raw === '') {
            return null;
        }
        $v = json_decode($raw, true);
        return $v === null && $raw !== 'null' ? $raw : $v;
    }

    private static function now(): string
    {
        return date('Y-m-d H:i:s');
    }
}
