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

<?php

namespace App\Services\AdminRealtime;

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

/**
 * 推送网关的拉取入口：按 id 递增读取事件表并解析成列表行，顺带扫描会员上下线。
 *
 * 写入方可能在长事务内插入事件（melbet_sport-api），小 id 可能晚于大 id 提交，
 * 因此网关会把跳过的 id 记为 gaps 在随后一段时间内补拉，本接口按 gaps 原样返回已出现的行。
 */
final class AdminRealtimeFeed
{
    private const ONLINE_SNAPSHOT = 'admin_realtime:online_snapshot';
    private const ONLINE_LOCK = 'admin_realtime:online_scan';

    /**
     * @param int $after 网关已处理的最大事件 id；小于 0 表示网关刚启动，只返回当前最大 id 不回放历史
     * @param list<int> $gaps 网关等待补拉的 id
     * @return array{last_id:int,ids:list<int>,events:list<array>,has_more:bool}
     */
    public static function pull(int $after, array $gaps = []): array
    {
        self::scanOnline();
        if ($after < 0) {
            return ['last_id' => (int) DB::table('admin_realtime_events')->max('id'), 'ids' => [], 'events' => [], 'has_more' => false];
        }

        $limit = max(1, min(1000, (int) config('admin_realtime.batch_size', 200)));
        $rows = DB::table('admin_realtime_events')->where('id', '>', $after)->orderBy('id')->limit($limit)->get();
        $hasMore = $rows->count() >= $limit;
        $gaps = array_values(array_filter(array_map('intval', $gaps), static fn(int $id): bool => $id > 0 && $id <= $after));
        if ($gaps) {
            $rows = DB::table('admin_realtime_events')->whereIn('id', array_slice($gaps, 0, 500))->orderBy('id')->get()->concat($rows);
        }

        return [
            'last_id' => (int) ($rows->where('id', '>', $after)->max('id') ?? $after),
            'ids' => $rows->pluck('id')->map(static fn($id): int => (int) $id)->values()->all(),
            'events' => self::resolve($rows->all()),
            'has_more' => $hasMore,
        ];
    }

    /** 同一记录在一批内多次变更只解析一次，取最后一条事件的 id；先新增后更新仍按 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';
            $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[] = [
                    '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,
                ];
            }
        }
        usort($events, static fn(array $a, array $b): int => $a['id'] <=> $b['id']);

        return $events;
    }

    private static function eventName(string $resource, string $action): string
    {
        if ($resource === AdminRealtime::USER_ONLINE) return 'user.' . $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)->orderBy('id')->limit(5000)->delete();
            $deleted += $batch;
        } while ($batch >= 5000);

        return $deleted;
    }
}