import 'dart:convert'; import 'package:flutter/foundation.dart'; import 'package:shared_preferences/shared_preferences.dart'; /// 13 号规范 §3.3 分段持久化事件队列(shared_preferences 版)。 /// /// 事件按段(每段最多 [segmentCapacity] 条)存储,避免单 key 整队列 /// 重写的 O(n) 放大;总量超过 [maxEvents] 时丢最旧的整段。内存中的 /// 段列表是唯一事实来源,shared_preferences 是尽力而为的镜像—— /// 持久化不可用(如未初始化插件的测试环境)时降级为纯内存队列, /// 任何存取失败都不向调用方抛出。 /// /// at-least-once 语义:段只在上传拿到终态(202 受理 / 4xx 永久拒绝) /// 后由调用方删除;应用在响应前被杀,事件仍在本地,冷启动 [restore] /// 恢复后重发,服务端靠 eventId 幂等去重。 class AnalyticsEventStore { AnalyticsEventStore({this.maxEvents = 500, this.segmentCapacity = 20}); /// 段 ID 有序列表(旧 → 新),JSON 数组。 static const segIndexKey = 'pb.analytics.segIndex'; /// 单段序列化事件 JSON 数组,key 为前缀 + 段 ID。 static const segKeyPrefix = 'pb.analytics.seg.'; /// 本地累计丢弃计数(溢出淘汰 + 4xx 丢批 + 损坏段),诊断用。 static const droppedCountKey = 'pb.analytics.droppedCount'; final int maxEvents; final int segmentCapacity; SharedPreferences? _prefs; final List<_Segment> _segments = []; int _seq = 0; int _droppedCount = 0; int get length => _segments.fold(0, (sum, seg) => sum + seg.events.length); int get droppedCount => _droppedCount; /// 全部待上传事件快照(旧 → 新)。 List> get events => List.unmodifiable([for (final seg in _segments) ...seg.events]); /// 冷启动恢复:读回未上传段。损坏段(JSON 解析失败)删 key 丢弃、 /// 计入丢弃数,不崩溃;索引本身损坏则按 key 前缀清扫孤儿段后重建。 /// 恢复的事件排在本实例已入队事件之前(更旧优先上传/淘汰)。 Future restore() async { try { _prefs = await SharedPreferences.getInstance(); } catch (error) { debugPrint('Analytics store: prefs unavailable, memory-only: $error'); return; } final prefs = _prefs!; _droppedCount = prefs.getInt(droppedCountKey) ?? 0; final restored = <_Segment>[]; final rawIndex = prefs.getString(segIndexKey); if (rawIndex != null) { var ids = const []; try { ids = [for (final id in jsonDecode(rawIndex) as List) id.toString()]; } catch (error) { debugPrint('Analytics store: corrupted index, sweeping: $error'); final orphanKeys = prefs .getKeys() .where((key) => key.startsWith(segKeyPrefix)) .toList(); for (final key in orphanKeys) { await _guard(() => prefs.remove(key)); } _droppedCount += 1; } for (final id in ids) { final raw = prefs.getString('$segKeyPrefix$id'); if (raw == null) continue; try { final events = [ for (final event in jsonDecode(raw) as List) Map.from(event as Map), ]; if (events.isNotEmpty) { restored.add(_Segment(id, events, sealed: true)); } } catch (error) { debugPrint('Analytics store: dropping corrupted seg $id: $error'); await _guard(() => prefs.remove('$segKeyPrefix$id')); _droppedCount += 1; } } } // restore 前入队的事件此前无处可写,恢复后连同清理过的索引一并落盘。 final preExisting = List<_Segment>.of(_segments); _segments ..clear() ..addAll(restored) ..addAll(preExisting); await _evictOverflow(); for (final seg in preExisting.where(_segments.contains)) { await _persistSegment(seg); } await _persistIndex(); await _persistDroppedCount(); } /// 追加事件到当前开放段并持久化该段(重写 ≤ [segmentCapacity] 条); /// 段满即封段,下次写入开新段;超总量上限丢最旧整段。 Future add(Map event) async { var open = _segments.isEmpty ? null : _segments.last; if (open == null || open.sealed || open.events.length >= segmentCapacity) { open = _Segment(_newSegId(), []); _segments.add(open); await _persistIndex(); } open.events.add(event); if (open.events.length >= segmentCapacity) open.sealed = true; await _persistSegment(open); await _evictOverflow(); } /// 从最旧段起取整段组一批(≤ [maxBatchEvents] 条)。入选段即封段, /// 上传在途期间新事件只会写入新的开放段,批内容与对应段不再变化, /// 上传成功后可安全整段删除(冲刷中新事件不丢)。 UploadBatch takeBatch(int maxBatchEvents) { final selected = <_Segment>[]; var count = 0; for (final seg in _segments) { if (count + seg.events.length > maxBatchEvents) break; selected.add(seg); count += seg.events.length; } for (final seg in selected) { seg.sealed = true; } return UploadBatch( segmentIds: [for (final seg in selected) seg.id], events: [ for (final seg in selected) ...seg.events.map(Map.from), ], ); } /// 上传终态后删段:202 受理或 4xx 永久拒绝(后者 [countAsDropped] /// 计入丢弃诊断数)。段可能已被溢出淘汰,缺失即忽略。 Future removeSegments( List ids, { bool countAsDropped = false, }) async { final idSet = ids.toSet(); var removed = 0; _segments.removeWhere((seg) { if (!idSet.contains(seg.id)) return false; removed += seg.events.length; return true; }); if (countAsDropped) _droppedCount += removed; for (final id in ids) { await _removeSegmentKey(id); } await _persistIndex(); if (countAsDropped) await _persistDroppedCount(); } Future _evictOverflow() async { var indexDirty = false; while (length > maxEvents && _segments.length > 1) { final victim = _segments.removeAt(0); _droppedCount += victim.events.length; await _removeSegmentKey(victim.id); indexDirty = true; } if (indexDirty) { await _persistIndex(); await _persistDroppedCount(); } } Future _persistIndex() => _guard( () => _prefs?.setString( segIndexKey, jsonEncode([for (final seg in _segments) seg.id]), ), ); Future _persistSegment(_Segment seg) => _guard( () => _prefs?.setString('$segKeyPrefix${seg.id}', jsonEncode(seg.events)), ); Future _removeSegmentKey(String id) => _guard(() => _prefs?.remove('$segKeyPrefix$id')); Future _persistDroppedCount() => _guard(() => _prefs?.setInt(droppedCountKey, _droppedCount)); /// 持久化是尽力而为的镜像,失败只打日志,绝不影响内存队列。 Future _guard(Future? Function() write) async { try { await write(); } catch (error) { debugPrint('Analytics store: persistence write failed: $error'); } } String _newSegId() => '${DateTime.now().microsecondsSinceEpoch.toRadixString(36)}-${_seq++}'; } /// 一次上传的段快照:事件按旧 → 新排列,段 ID 用于成功后删段。 class UploadBatch { const UploadBatch({required this.segmentIds, required this.events}); final List segmentIds; final List> events; bool get isEmpty => events.isEmpty; } class _Segment { _Segment(this.id, this.events, {this.sealed = false}); final String id; final List> events; /// 封段后不再接受追加(满 20 条或已被取入上传批次)。 bool sealed; }