bot-28/app/Services/AdminRealtime/AdminRealtimeFeed.php

<?php

namespace App\Services\AdminRealtime;

use App\Services\OnlineUserService;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;

/**
 * 事件表的读取端：relay 按 id 递增取出未广播的事件、分配连续 seq 并解析成列表行；前端断线后按 seq 补发。
 *
 * 写入方可能在长事务内插入事件（melbet_sport-api），小 id 可能晚于大 id 提交，
 * 因此 relay 在 gap_seconds 内仍会补发游标之前未广播的行；seq 按广播先后分配，补发不会被前端漏掉。
 */
final class AdminRealtimeFeed
{
    private const ONLINE_SNAPSHOT = 'admin_realtime:online_snapshot';
    private const ONLINE_LOCK = 'admin_realtime:online_scan';
    private const GAP_WINDOW = 10000;

    /** 已发现事件的持久游标；待发送批次另按 seq 优先恢复。 */
    public static function startCursor(): int
    {
        return (int) DB::table('admin_realtime_state')->where('id', 1)->value('last_event_id');
    }

    /**
     * 取出一批待广播事件并分配 seq，顺带扫描会员上下线
     *
     * @return array{cursor:int,ids:list<int>,events:list<array>,has_more:bool}
     */
    public static function next(int $cursor): array
    {
        self::scanOnline();
        AdminRealtimeSummaries::scan();
        AdminRealtimeAgentApplications::scan();
        $limit = max(1, min(1000, (int) config('admin_realtime.batch_size', 200)));
        $gapFrom = now('Asia/Shanghai')->subSeconds(max(0, (int) config('admin_realtime.gap_seconds', 60)))->format('Y-m-d H:i:s');
        [$cursor, $rows] = DB::transaction(function () use ($cursor, $limit, $gapFrom): array {
            $state = DB::table('admin_realtime_state')->where('id', 1)->lockForUpdate()->firstOrFail();
            $cursor = max($cursor, (int) $state->last_event_id);
            // 未确认发送的旧批次优先；不能让后续 seq 越过发送失败的事件。
            $rows = DB::table('admin_realtime_events')->whereNotNull('seq')->whereNull('published_at')
                ->orderBy('seq')->limit($limit)->get();
            if ($rows->isEmpty()) {
                AdminRealtimeExchangeList::collectChanges();
                $rows = DB::table('admin_realtime_events')
                    ->whereNull('seq')
                    ->where('id', '>', max(0, $cursor - self::GAP_WINDOW))
                    ->where(fn($q) => $q->where('id', '>', $cursor)->orWhere('created_at', '>=', $gapFrom))
                    ->orderBy('id')->limit($limit)->get();
                if ($rows->isNotEmpty()) {
                    self::assignSeq($rows, (int) $state->last_seq);
                    $cursor = max($cursor, (int) $rows->max('id'));
                    DB::table('admin_realtime_state')->where('id', 1)->update([
                        'last_seq' => (int) $rows->last()->seq, 'last_event_id' => $cursor,
                    ]);
                }
            }
            return [$cursor, $rows];
        });

        return [
            'cursor' => $cursor,
            'ids' => $rows->pluck('id')->map(static fn($id): int => (int) $id)->all(),
            'events' => self::resolve($rows->all()),
            'has_more' => $rows->count() >= $limit,
        ];
    }

    /** 调用方持有 state 行锁，分配和持久化在同一事务内完成。 */
    private static function assignSeq(Collection $rows, int $seq): void
    {
        $cases = '';
        foreach ($rows as $row) {
            $row->seq = ++$seq;
            $cases .= ' WHEN ' . (int) $row->id . ' THEN ' . $seq;
        }
        DB::table('admin_realtime_events')
            ->whereIn('id', $rows->pluck('id')->all())
            ->update(['seq' => DB::raw('CASE id' . $cases . ' END')]);
    }

    public static function lastSeq(): int
    {
        return (int) DB::table('admin_realtime_state')->where('id', 1)->value('last_seq');
    }

    /** 整批全部发送成功才确认；部分成功后的重试允许重复，由前端去重。 */
    public static function acknowledge(array $ids): void
    {
        if ($ids) {
            DB::table('admin_realtime_events')->whereIn('id', $ids)->whereNotNull('seq')
                ->update(['published_at' => now('Asia/Shanghai')->format('Y-m-d H:i:s')]);
        }
    }

    /**
     * 断线补发：返回 seq 大于 $afterSeq 的事件（按当前数据重新解析），只含请求且有权限的频道。
     * 游标已被清理、无效或积压过多时返回 resync，前端应重新拉取列表。
     *
     * @param list<string> $channels
     * @return array{resync:bool,reason?:string,last_seq:int,events:list<array>}
     */
    public static function replay(int $afterSeq, array $channels): array
    {
        $lastSeq = self::lastSeq();
        $resync = static fn(string $reason): array => ['resync' => true, 'reason' => $reason, 'last_seq' => $lastSeq, 'events' => []];
        if ($afterSeq < 0 || $afterSeq > $lastSeq) {
            return $resync('invalid_cursor');
        }
        if ($afterSeq === $lastSeq) {
            return ['resync' => false, 'last_seq' => $lastSeq, 'events' => []];
        }
        if ($afterSeq > 0 && !DB::table('admin_realtime_events')->where('seq', $afterSeq)->exists()) {
            return $resync('history_unavailable');
        }
        $limit = max(1, (int) config('admin_realtime.replay_limit', 1000));
        $rows = DB::table('admin_realtime_events')->where('seq', '>', $afterSeq)->where('seq', '<=', $lastSeq)
            ->orderBy('seq')->limit($limit + 1)->get();
        if ($rows->count() > $limit) {
            return $resync('too_many_events');
        }
        $wanted = array_flip($channels);

        return [
            'resync' => false,
            'last_seq' => $lastSeq,
            'events' => array_values(array_filter(self::resolve($rows->all()), static fn(array $e): bool => isset($wanted[$e['channel']]))),
        ];
    }

    /** 同一记录在一批内多次变更只解析一次，取最后一条事件的 seq；先新增后更新仍按 created 推送 */
    private static function resolve(array $rows): array
    {
        $groups = [];
        foreach ($rows as $row) {
            $key = $row->resource . ':' . $row->resource_id;
            $created = ($groups[$key]['created'] ?? false) || $row->action === 'created';
            unset($groups[$key]);
            $groups[$key] = ['row' => $row, 'created' => $created];
        }

        $events = [];
        foreach ($groups as $group) {
            $row = $group['row'];
            $action = $row->action === 'updated' && $group['created'] ? 'created' : (string) $row->action;
            try {
                $messages = AdminRealtimeRows::resolve((string) $row->resource, (string) $row->resource_id, $action);
            } catch (\Throwable $e) {
                // 解析失败是该条记录或其列表接口的问题，重试不会恢复；跳过它，不能让一条坏事件卡住之后所有推送。
                // 前端靠后续变更或重新拉列表补上这一行。
                Log::channel('realtime')->error('admin_realtime_resolve_failed', [
                    'event_id' => $row->id, 'resource' => $row->resource, 'resource_id' => $row->resource_id, 'error' => $e->getMessage(),
                ]);
                continue;
            }
            foreach ($messages as $message) {
                $events[] = [
                    'seq' => (int) $row->seq,
                    'id' => (int) $row->id,
                    'channel' => $message['channel'],
                    'event' => self::eventName((string) $row->resource, $action),
                    'action' => $action,
                    'key' => $message['key'],
                    'data' => $message['data'],
                    'source' => (string) $row->source,
                    'created_at' => (string) $row->created_at,
                ];
            }
        }

        return $events;
    }

    private static function eventName(string $resource, string $action): string
    {
        if ($resource === AdminRealtimeExchangeList::RESOURCE) return AdminRealtimeExchangeList::EVENT;
        if (isset(AdminRealtimeSummaries::DEFINITIONS[$resource])) return 'stats.updated';
        if ($resource === AdminRealtime::USER_ONLINE) return 'user.' . $action;
        if (in_array($resource, ['agent_applications', 'agent_plan_applications'], true)) return 'application.' . $action;
        return ($resource === 'users' ? 'user.' : 'order.') . $action;
    }

    /** 在线判定含「最近活跃时间窗口」，下线不一定有写入，只能定期比对在线集合 */
    private static function scanOnline(): void
    {
        $interval = max(5, (int) config('admin_realtime.online_scan_seconds', 15));
        try {
            if (!Cache::add(self::ONLINE_LOCK, 1, $interval)) return;
            $current = OnlineUserService::onlineUserIds();
            $previous = Cache::get(self::ONLINE_SNAPSHOT);
            Cache::forever(self::ONLINE_SNAPSHOT, $current);
            if (!is_array($previous)) return;
            AdminRealtime::record(AdminRealtime::USER_ONLINE, array_diff($current, $previous), 'online');
            AdminRealtime::record(AdminRealtime::USER_ONLINE, array_diff($previous, $current), 'offline');
        } catch (\Throwable $e) {
            Log::channel('realtime')->error('admin_realtime_online_scan_failed', ['error' => $e->getMessage()]);
        }
    }

    public static function prune(): int
    {
        $hours = max(1, (int) config('admin_realtime.retention_hours', 72));
        $before = now('Asia/Shanghai')->subHours($hours)->format('Y-m-d H:i:s');
        $deleted = 0;
        do {
            $batch = DB::table('admin_realtime_events')->where('created_at', '<', $before)
                ->where(fn($q) => $q->whereNotNull('published_at')->orWhereNull('seq'))
                ->orderBy('id')->limit(5000)->delete();
            $deleted += $batch;
        } while ($batch >= 5000);

        return $deleted;
    }
}