bot-28/app/Console/Commands/RelayAdminRealtimeEvents.php

<?php

namespace App\Console\Commands;

use App\Events\AdminRealtimeBroadcast;
use App\Services\AdminRealtime\AdminRealtimeAccess;
use App\Services\AdminRealtime\AdminRealtimeFeed;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Broadcast;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Str;

/**
 * 常驻进程：把 admin_realtime_events 中的事件解析成列表行并广播到 Reverb，
 * 同时扫描会员上下线、断开 JWT 已过期的后台连接。需由 Supervisor 守护，全局只运行一个实例。
 */
class RelayAdminRealtimeEvents extends Command
{
    private const LOCK = 'admin_realtime:relay';
    private const LOCK_TTL = 30;
    private const SESSION_SWEEP_SECONDS = 5;

    protected $signature = 'admin-realtime:relay
        {--once : 只处理一批后退出，用于排查}
        {--max-seconds=3600 : 运行指定秒数后正常退出，由 Supervisor 拉起新进程释放内存}';
    protected $description = '后台实时推送：读取事件表并广播到 Reverb';

    private bool $running = true;

    public function handle(): int
    {
        if (!config('admin_realtime.enabled')) {
            $this->error('实时推送未启用（ADMIN_WS_ENABLED=false）');
            return self::FAILURE;
        }
        $owner = Str::random(16);
        // 上个进程被强杀时锁要等 LOCK_TTL 才过期，这里等待而不是立即退出，避免 Supervisor 连续重启判定失败
        for ($waited = 0; !Cache::add(self::LOCK, $owner, self::LOCK_TTL); $waited++) {
            if ($waited >= self::LOCK_TTL + 5 || $this->option('once')) {
                $this->error('已有 relay 实例在运行');
                return self::FAILURE;
            }
            sleep(1);
        }
        DB::connection()->disableQueryLog();
        $this->trap(fn() => [SIGTERM, SIGINT, SIGQUIT], function () {
            $this->running = false;
        });

        $deadline = time() + max(60, (int) $this->option('max-seconds'));
        $pollMs = max(200, (int) config('admin_realtime.poll_ms', 1000));
        $cursor = AdminRealtimeFeed::startCursor();
        $nextSweep = 0;
        $this->info("relay 启动，游标 {$cursor}");

        try {
            while ($this->running) {
                if (Cache::get(self::LOCK) !== $owner) {
                    $this->error('relay 锁已失效，退出以免重复广播');
                    return self::FAILURE;
                }
                Cache::put(self::LOCK, $owner, self::LOCK_TTL);

                $hasMore = false;
                try {
                    if (time() >= $nextSweep) {
                        AdminRealtimeAccess::terminateExpiredSessions();
                        $nextSweep = time() + self::SESSION_SWEEP_SECONDS;
                    }
                    $batch = AdminRealtimeFeed::next($cursor);
                    foreach ($batch['events'] as $message) {
                        if (Cache::get(self::LOCK) !== $owner) {
                            throw new \RuntimeException('relay 锁已失效，停止发送当前批次');
                        }
                        Cache::put(self::LOCK, $owner, self::LOCK_TTL);
                        $this->broadcast($message);
                    }
                    AdminRealtimeFeed::acknowledge($batch['ids']);
                    $cursor = $batch['cursor'];
                    $hasMore = $batch['has_more'];
                } catch (\Throwable $e) {
                    // 未确认的批次留在数据库中；稍后或进程重启后先重试，不越过失败事件。
                    Log::channel('realtime')->error('admin_realtime_relay_failed', ['cursor' => $cursor, 'error' => $e->getMessage()]);
                    if ($this->option('once')) return self::FAILURE;
                    sleep(min(10, (int) ceil($pollMs * 5 / 1000)));
                }

                if ($this->option('once') || time() >= $deadline) {
                    break;
                }
                if (!$hasMore) {
                    usleep($pollMs * 1000);
                }
            }
        } finally {
            if (Cache::get(self::LOCK) === $owner) {
                Cache::forget(self::LOCK);
            }
        }

        return self::SUCCESS;
    }

    /** 广播失败必须中断批次，保留其发送状态供重试。 */
    private function broadcast(array $message): void
    {
        try {
            Broadcast::queue(new AdminRealtimeBroadcast($message));
        } catch (\Throwable $e) {
            Log::channel('realtime')->error('admin_realtime_broadcast_failed', [
                'seq' => $message['seq'], 'channel' => $message['channel'], 'error' => $e->getMessage(),
            ]);
            throw $e;
        }
    }
}