patbond-flutter/lib/analytics/analytics_event_store.dart

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;
}