where('id', 1)->value('last_event_id'); } /** * 取出一批待广播事件并分配 seq，顺带扫描会员上下线 * * @return array{cursor:int,ids:list,events:list,has_more:bool} */ public static function next(int $cursor): array { self::scanOnline(); AdminRealtimeSummaries::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()) { $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 $channels * @return array{resync:bool,reason?:string,last_seq:int,events:list} */ 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(), ]); throw $e; } 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 (isset(AdminRealtimeSummaries::DEFINITIONS[$resource])) return 'stats.updated'; 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) ->where(fn($q) => $q->whereNotNull('published_at')->orWhereNull('seq')) ->orderBy('id')->limit(5000)->delete(); $deleted += $batch; } while ($batch >= 5000); return $deleted; } }