Files
photo-live/apps/mobile/lib/features/uploads/task_store.dart
T
浅念 d96e56f653 feat(mobile): drift upload task store with state machine
- upload_tasks 表(accountId 隔离 + 三个复合索引)、状态列枚举转换器、
  datetime 文本落库(UTC 归一,亚秒精度)
- 纯状态机(task_state.dart):合法迁移表 + canTransition/assertTransition,
  completed/cancelled 终态
- TaskStore 仓储:create(clientTaskId 同账号同相册应用侧唯一)、
  updateState(迁移守护+旁路字段)、账号隔离读路径(硬守护)、
  countsByState 徽章、cleanupCandidates/markCleanable(仅终态可清理——
  评审缺口 1 硬守护,未完成源文件永不清理)
- 测试 26 例(状态机表驱动全边 + 守护/隔离/筛选/保留策略);
  sqlite3 3.x build hooks 在 flutter test 宿主与 Android APK 构建均验证
  (libsqlite3.so 已打包全 ABI,运行时验证留 Task 11)

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-19 02:12:39 +08:00

443 lines
16 KiB
Dart
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/// 本地上传任务库(P3 计划 Task 5):drift schema + [TaskStore] 仓储。
///
/// 设计要点(对齐技术实施方案 §5.5 / P3 计划 Task 5/6):
/// - **账号隔离硬守护**:所有任务读路径([TaskStore.get]/
/// [TaskStore.listByAccount]/[TaskStore.countsByState])无条件附加
/// `accountId` 过滤——账号切换后不可能读到他人任务(A15 连拍归属、
/// 队列 UI 均经由这些入口);
/// - **状态机守护**:[TaskStore.updateState] 走 [assertTransition],非法
/// 迁移抛 [StateError](合法表见 task_state.dart);
/// - **clientTaskId 幂等键**:同账号同相册内唯一,**应用侧**在
/// [TaskStore.create] 时校验(不加数据库唯一约束,保留引擎重排/重建的
/// 灵活性,幂等语义由服务端 POST /uploads 409 兜底);
/// - **源文件保留策略硬守护**:[TaskStore.cleanupCandidates] 只返回
/// completed/cancelled 且超过保留期、未清理(cleanedAt IS NULL)的任务,
/// **未完成任务永不进入清理候选**;[TaskStore.markCleanable] 同样拒绝
/// 非终态任务(评审缺口 1,单测覆盖)。
///
/// 字段增补说明:`cleanedAt`(源文件已清理时间)为 P3 计划 Task 5 字段
/// 清单之外的实现增补,用于 [TaskStore.markCleanable] 的幂等(清理执行器
/// 重启后不重复删文件);服务 markCleanable/cleanupCandidates 语义。
library;
import 'dart:io';
import 'package:drift/drift.dart';
import 'package:drift/native.dart';
import 'package:path/path.dart' as p;
import 'package:path_provider/path_provider.dart';
import 'task_state.dart';
part 'task_store.g.dart';
/// `state` 列的枚举转换器(落库取值见 [UploadTaskState.wire])。
class _UploadTaskStateConverter extends TypeConverter<UploadTaskState, String> {
const _UploadTaskStateConverter();
@override
UploadTaskState fromSql(String fromDb) =>
UploadTaskState.fromWire(fromDb);
@override
String toSql(UploadTaskState value) => value.wire;
}
/// 本地上传任务表(表名 upload_tasks)。
///
/// 三个复合索引支撑账号隔离后的高频查询:队列列表(accountId)、状态
/// 徽章(accountId+state)、相册内筛选(accountId+albumId+state,需求 4.3
/// 三维筛选)。
@TableIndex(name: 'upload_tasks_account_idx', columns: {#accountId})
@TableIndex(
name: 'upload_tasks_account_state_idx',
columns: {#accountId, #state},
)
@TableIndex(
name: 'upload_tasks_account_album_state_idx',
columns: {#accountId, #albumId, #state},
)
class UploadTasks extends Table {
/// 任务 ID(客户端生成 uuid,PK;与服务端 uploadId 无关)。
TextColumn get id => text()();
/// 账号隔离键(登录账号,见 Session.account)。
TextColumn get accountId => text()();
/// 目标相册 ID。
TextColumn get albumId => text()();
/// 目标分类 ID;缺省为相册默认分类(契约 UploadInitRequest.categoryId)。
TextColumn get categoryId => text().nullable()();
/// 客户端幂等键(同账号同相册内唯一,应用侧校验,见库级文档)。
TextColumn get clientTaskId => text()();
/// 源文件绝对路径(受管目录副本:app 支持目录/uploads/{albumId}/)。
TextColumn get sourceFilePath => text()();
/// 原始文件名(仅元数据, maxLength 255,契约对齐)。
TextColumn get filename => text()();
/// 申明文件字节数(完成核验时与服务端逐一比对)。
IntColumn get bytes => integer()();
/// MIME 类型(白名单 image/jpeg | image/png | image/heic)。
TextColumn get contentType => text()();
/// 任务状态(状态机见 task_state.dart)。
TextColumn get state =>
text().map(const _UploadTaskStateConverter())();
/// 已尝试传输次数(退避与上限判断依据)。
IntColumn get attempt => integer().withDefault(const Constant(0))();
/// 下次可重试时间(waitingNetwork/retryableFailed 的退避闹钟);null=无
/// 待重试。
DateTimeColumn get nextRetryAt => dateTime().nullable()();
/// 服务端上传会话 ID(initUpload 成功后回填,重试复用)。
TextColumn get serverUploadId => text().nullable()();
/// 服务端预创建照片 ID(initUpload 成功后回填)。
TextColumn get serverPhotoId => text().nullable()();
/// 最近一次失败原因(错误码文本,见 core/error_codes.dart;成功迁移时
/// 清除)。
TextColumn get lastError => text().nullable()();
/// 源文件已清理时间(保留策略执行器写入;null=未清理)。实现增补字段,
/// 见库级文档。
DateTimeColumn get cleanedAt => dateTime().nullable()();
/// 创建时间(入 preparing 时刻)。
DateTimeColumn get createdAt => dateTime()();
/// 最近更新时间(每次状态迁移刷新;保留策略的年龄基准)。
DateTimeColumn get updatedAt => dateTime()();
@override
Set<Column> get primaryKey => {id};
}
/// drift 数据库(生产=app documents dir 文件;测试=[UploadTaskDb.memory])。
@DriftDatabase(tables: [UploadTasks])
class UploadTaskDb extends _$UploadTaskDb {
UploadTaskDb(super.e);
/// 测试用内存库(drift NativeDatabase.memory,随 close 丢弃)。
UploadTaskDb.memory() : super(NativeDatabase.memory());
@override
int get schemaVersion => 1;
}
/// 打开生产库:app documents 目录下 upload_tasks.sqlite,后台 isolate 执行
/// SQL(避免大结果集阻塞 UI)。
LazyDatabase openAppUploadTaskDb() {
return LazyDatabase(() async {
final dir = await getApplicationDocumentsDirectory();
final file = File(p.join(dir.path, 'upload_tasks.sqlite'));
return NativeDatabase.createInBackground(file);
});
}
/// 新任务的创建参数(初始状态固定 preparing,见状态机)。
class UploadTaskDraft {
const UploadTaskDraft({
required this.id,
required this.accountId,
required this.albumId,
this.categoryId,
required this.clientTaskId,
required this.sourceFilePath,
required this.filename,
required this.bytes,
required this.contentType,
});
/// 任务 ID(uuid,调用方生成)。
final String id;
/// 账号隔离键。
final String accountId;
/// 目标相册 ID。
final String albumId;
/// 目标分类 ID(可缺省)。
final String? categoryId;
/// 客户端幂等键。
final String clientTaskId;
/// 源文件路径(受管目录副本)。
final String sourceFilePath;
/// 原始文件名。
final String filename;
/// 字节数。
final int bytes;
/// MIME 类型。
final String contentType;
}
/// 状态迁移时一并写入的旁路字段(未提供的列保持不变;`null` 与"清除"通过
/// clear* 布尔区分)。
class UploadTaskPatch {
const UploadTaskPatch({
this.attempt,
this.nextRetryAt,
this.serverUploadId,
this.serverPhotoId,
this.lastError,
this.clearNextRetryAt = false,
this.clearLastError = false,
});
/// 新的尝试次数(失败分类时 +1 后写入)。
final int? attempt;
/// 新的下一次重试时间(退避计算结果)。
final DateTime? nextRetryAt;
/// 服务端上传会话 ID(initUpload 成功时写入)。
final String? serverUploadId;
/// 服务端照片 ID(initUpload 成功时写入)。
final String? serverPhotoId;
/// 失败原因错误码。
final String? lastError;
/// 显式清除 nextRetryAt(成功出队/完成后)。
final bool clearNextRetryAt;
/// 显式清除 lastError(迁移到成功路径时)。
final bool clearLastError;
}
/// 上传任务仓储:引擎(Task 6)与队列 UI(Task 9)的唯一入口。
///
/// 读路径全部账号隔离(硬守护);按 id 的写路径(updateState/markCleanable/
/// delete)为引擎内部操作,id 是 uuid PK,不构成跨账号读取面。
class TaskStore {
TaskStore(this._db);
/// 生产实例(app documents dir 持久库)。
factory TaskStore.app() => TaskStore(UploadTaskDb(openAppUploadTaskDb()));
/// 测试实例(内存库)。
factory TaskStore.memory() => TaskStore(UploadTaskDb.memory());
final UploadTaskDb _db;
/// 关闭底层数据库(App 退出/测试 tearDown)。
Future<void> close() => _db.close();
/// 时间戳统一 UTC 落库(datetime 为 ISO-8601 文本存储:UTC 值以 `Z`
/// 结尾、本地值带 `+hh:mm` 后缀,混存会让 SQL 文本比较错位一个时区
/// 偏移——全部归一 UTC 后字典序即时间序)。
static DateTime _utcAt(DateTime? at) => (at ?? DateTime.now()).toUtc();
/// 创建任务(初始状态 preparing、attempt=0)。
///
/// 抛 [StateError]:同账号同相册内 clientTaskId 已存在(幂等键不可
/// 复用——应用侧唯一性校验,见库级文档)。
Future<UploadTask> create(UploadTaskDraft draft, {DateTime? now}) {
return _db.transaction(() async {
final conflict = await (_db.select(_db.uploadTasks)
..where(
(t) =>
t.accountId.equals(draft.accountId) &
t.albumId.equals(draft.albumId) &
t.clientTaskId.equals(draft.clientTaskId),
))
.get();
if (conflict.isNotEmpty) {
throw StateError(
'clientTaskId 在同账号同相册内已存在:${draft.clientTaskId}'
'(幂等键不可复用)',
);
}
final timestamp = _utcAt(now);
await _db.into(_db.uploadTasks).insert(
UploadTasksCompanion.insert(
id: draft.id,
accountId: draft.accountId,
albumId: draft.albumId,
categoryId: Value(draft.categoryId),
clientTaskId: draft.clientTaskId,
sourceFilePath: draft.sourceFilePath,
filename: draft.filename,
bytes: draft.bytes,
contentType: draft.contentType,
state: UploadTaskState.preparing,
createdAt: timestamp,
updatedAt: timestamp,
),
);
return await _byId(draft.id);
});
}
/// 按 id 读取(**账号隔离硬守护**:accountId 不匹配视同不存在)。
Future<UploadTask?> get(String accountId, String id) {
return (_db.select(_db.uploadTasks)
..where((t) => t.id.equals(id) & t.accountId.equals(accountId)))
.getSingleOrNull();
}
/// 状态迁移(带合法迁移守护 + updatedAt 刷新),返回迁移后的任务。
///
/// 抛 [StateError]:任务不存在,或 `当前 → to` 不在合法迁移表
/// ([assertTransition])。
Future<UploadTask> updateState(
String id,
UploadTaskState to, {
UploadTaskPatch patch = const UploadTaskPatch(),
DateTime? now,
}) {
return _db.transaction(() async {
final current = await _byId(id);
assertTransition(current.state, to);
await (_db.update(_db.uploadTasks)
..where((t) => t.id.equals(id)))
.write(
UploadTasksCompanion(
state: Value(to),
attempt: patch.attempt == null
? const Value.absent()
: Value(patch.attempt!),
nextRetryAt: patch.clearNextRetryAt
? const Value(null)
: Value(patch.nextRetryAt?.toUtc()),
serverUploadId: Value(patch.serverUploadId),
serverPhotoId: Value(patch.serverPhotoId),
lastError: patch.clearLastError
? const Value(null)
: Value(patch.lastError),
updatedAt: Value(_utcAt(now)),
),
);
return await _byId(id);
});
}
/// 账号任务列表(**accountId 过滤无条件附加——硬守护**)。
///
/// 可选筛选(需求 4.3 三维):albumId 相册、states 状态集合、since 创建
/// 时间下限。按 createdAt 升序(队列 FIFO 语义)。
Future<List<UploadTask>> listByAccount(
String accountId, {
String? albumId,
Set<UploadTaskState>? states,
DateTime? since,
int? limit,
}) {
final query = _db.select(_db.uploadTasks)
..where((t) => t.accountId.equals(accountId));
if (albumId != null) {
query.where((t) => t.albumId.equals(albumId));
}
if (states != null && states.isNotEmpty) {
query.where((t) => t.state.isInValues(states.toList()));
}
if (since != null) {
query
.where((t) => t.createdAt.isBiggerOrEqualValue(since.toUtc()));
}
query.orderBy([(t) => OrderingTerm.asc(t.createdAt)]);
if (limit != null) {
query.limit(limit);
}
return query.get();
}
/// 账号内各状态任务数(队列 UI 徽章;全部状态都会出现,无任务计 0)。
Future<Map<UploadTaskState, int>> countsByState(String accountId) async {
final rows = await (_db.selectOnly(_db.uploadTasks)
..addColumns([_db.uploadTasks.state, _db.uploadTasks.id.count()])
..where(_db.uploadTasks.accountId.equals(accountId))
..groupBy([_db.uploadTasks.state]))
.map((row) => MapEntry(
row.readWithConverter(_db.uploadTasks.state)!,
row.read(_db.uploadTasks.id.count())!,
))
.get();
return {
for (final state in UploadTaskState.values)
state: rows.firstWhere((e) => e.key == state,
orElse: () => MapEntry(state, 0))
.value,
};
}
/// 源文件清理候选:**仅** completed/cancelled、updatedAt 早于
/// `now - keepDays` 且未清理过的任务(硬守护:未完成任务永不出现——
/// 评审缺口 1)。
///
/// 设备级维护例程(清理磁盘上的受管目录副本),跨账号扫描;不影响任务
/// 记录本身(UI/历史仍可见,cleanedAt 仅标记源文件已删)。
Future<List<UploadTask>> cleanupCandidates(
int keepDays, {
DateTime? now,
}) {
final cutoff = (now ?? DateTime.now())
.subtract(Duration(days: keepDays))
// UTC 归一:与落库取值同格式(Z 后缀),文本比较即时间比较。
.toUtc();
return (_db.select(_db.uploadTasks)
..where(
(t) =>
t.state.isInValues(const [
UploadTaskState.completed,
UploadTaskState.cancelled,
]) &
t.cleanedAt.isNull() &
t.updatedAt.isSmallerOrEqualValue(cutoff),
))
.get();
}
/// 标记源文件已清理(cleanedAt=now;幂等——重复标记仅刷新时间)。
///
/// 抛 [StateError]:任务不存在或**非终态**(未完成源文件永不清理——
/// 与 [cleanupCandidates] 同一硬守护的另一面)。
Future<void> markCleanable(String id, {DateTime? now}) {
return _db.transaction(() async {
final task = await _byId(id);
if (!task.state.isTerminal) {
throw StateError(
'未完成任务不可标记清理(状态:${task.state})——源文件硬守护',
);
}
final timestamp = _utcAt(now);
await (_db.update(_db.uploadTasks)
..where((t) => t.id.equals(id)))
.write(
UploadTasksCompanion(cleanedAt: Value(timestamp)),
);
});
}
/// 删除任务记录(连拍未开始取消=直接删;清理策略不依赖删除)。
Future<int> delete(String id) {
return (_db.delete(_db.uploadTasks)..where((t) => t.id.equals(id))).go();
}
/// 事务内按 id 精确读取(不存在即 [StateError]——调用方均持有引擎下发
/// 的合法 id)。
Future<UploadTask> _byId(String id) async {
final row = await (_db.select(_db.uploadTasks)
..where((t) => t.id.equals(id)))
.getSingleOrNull();
if (row == null) {
throw StateError('上传任务不存在:$id');
}
return row;
}
}