- AnalyticsEventStore：shared_preferences 分段存储（每段 ≤20 条、 总上限 500 超限丢最旧整段）、冷启动恢复、损坏段/损坏索引容错、 droppedCount 丢弃诊断计数 - AnalyticsService 接入持久化队列：上传拿到终态（202/4xx）才删段 实现 at-least-once；冲刷按段拼批 ≤50 条循环上传（契约单批上限）； 取批即封段，冲刷在途新事件写入新开放段不丢 - app.dart 冷启动 restore() 恢复离线积压并冲刷（13 号 §3.4 触发点） - 保留第一波语义：flushNow()、4xx 毒丸丢弃、满 20 条冲刷触发 - 新增 13 个单测（恢复/上限淘汰/损坏容错/202 清段/flushNow 协同/ 分批上传），全套 64 测试全绿 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,229 @@
|
||||
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<Map<String, dynamic>> get events =>
|
||||
List.unmodifiable([for (final seg in _segments) ...seg.events]);
|
||||
|
||||
/// 冷启动恢复：读回未上传段。损坏段（JSON 解析失败）删 key 丢弃、
|
||||
/// 计入丢弃数，不崩溃；索引本身损坏则按 key 前缀清扫孤儿段后重建。
|
||||
/// 恢复的事件排在本实例已入队事件之前（更旧优先上传/淘汰）。
|
||||
Future<void> 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 <String>[];
|
||||
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<String, dynamic>.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<void> add(Map<String, dynamic> 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<String, dynamic>.from),
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
/// 上传终态后删段：202 受理或 4xx 永久拒绝（后者 [countAsDropped]
|
||||
/// 计入丢弃诊断数）。段可能已被溢出淘汰，缺失即忽略。
|
||||
Future<void> removeSegments(
|
||||
List<String> 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<void> _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<void> _persistIndex() => _guard(
|
||||
() => _prefs?.setString(
|
||||
segIndexKey,
|
||||
jsonEncode([for (final seg in _segments) seg.id]),
|
||||
),
|
||||
);
|
||||
|
||||
Future<void> _persistSegment(_Segment seg) => _guard(
|
||||
() => _prefs?.setString('$segKeyPrefix${seg.id}', jsonEncode(seg.events)),
|
||||
);
|
||||
|
||||
Future<void> _removeSegmentKey(String id) =>
|
||||
_guard(() => _prefs?.remove('$segKeyPrefix$id'));
|
||||
|
||||
Future<void> _persistDroppedCount() =>
|
||||
_guard(() => _prefs?.setInt(droppedCountKey, _droppedCount));
|
||||
|
||||
/// 持久化是尽力而为的镜像，失败只打日志，绝不影响内存队列。
|
||||
Future<void> _guard(Future<void>? 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<String> segmentIds;
|
||||
final List<Map<String, dynamic>> events;
|
||||
|
||||
bool get isEmpty => events.isEmpty;
|
||||
}
|
||||
|
||||
class _Segment {
|
||||
_Segment(this.id, this.events, {this.sealed = false});
|
||||
|
||||
final String id;
|
||||
final List<Map<String, dynamic>> events;
|
||||
|
||||
/// 封段后不再接受追加（满 20 条或已被取入上传批次）。
|
||||
bool sealed;
|
||||
}
|
||||
Reference in New Issue
Block a user