From 4e35bdc0b2113cb6047e3fb21a9e9252e82ee9d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Sat, 19 Sep 2026 03:22:42 +0800 Subject: [PATCH] feat(mobile): android background upload bridge via workmanager MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kotlin 侧（MethodChannel photolive/uploads + EventChannel photolive/uploads/events）： - UploadPlugin：registerTask 入队 WorkManager OneTimeWorkRequest（unique work=upload-task-{taskId}，expedited + OutOfQuotaPolicy 回退、CONNECTED 约束、指数退避 10s 起）、cancelTask / queryTask / ackEvents - UploadWorker（CoroutineWorker）：dataSync 前台服务 + 进度通知， OkHttp PUT（ETag 捕获、64KB 分片进度），结局分类：2xx→completed、 403→needsRenew、4xx→failed{HTTP_4XX}、网络/5xx→Result.retry（上限 5 次后 failed{NETWORK|SERVER_ERROR}）、缺文件→failed{SOURCE_MISSING} - UploadEventBus：终态事件单调 seq + 未 ack 缓存重放（上限 200）， 重注册/取消作废旧缓存——进程重启/切后台恢复不丢完成信号 - Manifest：FOREGROUND_SERVICE(_DATA_SYNC)/POST_NOTIFICATIONS 权限 + SystemForegroundService tools:node=merge 补 dataSync 类型（已验证 合并产物） Dart 侧：Transport 接口（init/complete/renew 留引擎，仅 PUT 委托）—— DartHttpTransport（Task 6 逻辑抽出，403→renew 异常归一）+ NativeTransport （registerTask→事件流终态/进度映射、终态 ack、取消联动）；引擎传输选择： Android 缺省原生，dart-define FORCE_DART_TRANSPORT=true 强制 Dart 路径； 失败分类与 dio 路径语义对齐（401 短退避/429 网络桶/SOURCE_MISSING 不 自动重试等） 测试：gradle testDebugUnitTest 29 例（executor MockWebServer 分类×10、 协议解析×11、事件总线×8）；flutter test 219 全绿（新增 NativeTransport 事件映射/取消/ack/并发路由、引擎×Transport 委托与 needsRenew 重注册、 传输选择矩阵）；flutter analyze 0；debug apk 构建通过 Co-Authored-By: Claude Code --- apps/mobile/android/app/build.gradle.kts | 10 + .../android/app/src/main/AndroidManifest.xml | 17 +- .../com/photolive/photo_live/MainActivity.kt | 9 +- .../photolive/photo_live/UploadEventBus.kt | 132 +++++ .../com/photolive/photo_live/UploadPlugin.kt | 151 ++++++ .../photolive/photo_live/UploadProtocol.kt | 102 ++++ .../photolive/photo_live/UploadPutExecutor.kt | 121 +++++ .../com/photolive/photo_live/UploadWorker.kt | 250 ++++++++++ .../src/main/res/drawable/ic_stat_upload.xml | 11 + .../app/src/main/res/values-zh/strings.xml | 4 + .../app/src/main/res/values/strings.xml | 5 + .../photo_live/UploadEventBusTest.kt | 112 +++++ .../photo_live/UploadProtocolTest.kt | 108 ++++ .../photo_live/UploadPutExecutorTest.kt | 175 +++++++ .../features/uploads/native_transport.dart | 421 ++++++++++++++++ .../lib/features/uploads/upload_engine.dart | 175 +++++-- .../uploads/native_transport_test.dart | 472 ++++++++++++++++++ .../features/uploads/upload_engine_test.dart | 227 ++++++++- 18 files changed, 2454 insertions(+), 48 deletions(-) create mode 100644 apps/mobile/android/app/src/main/kotlin/com/photolive/photo_live/UploadEventBus.kt create mode 100644 apps/mobile/android/app/src/main/kotlin/com/photolive/photo_live/UploadPlugin.kt create mode 100644 apps/mobile/android/app/src/main/kotlin/com/photolive/photo_live/UploadProtocol.kt create mode 100644 apps/mobile/android/app/src/main/kotlin/com/photolive/photo_live/UploadPutExecutor.kt create mode 100644 apps/mobile/android/app/src/main/kotlin/com/photolive/photo_live/UploadWorker.kt create mode 100644 apps/mobile/android/app/src/main/res/drawable/ic_stat_upload.xml create mode 100644 apps/mobile/android/app/src/main/res/values-zh/strings.xml create mode 100644 apps/mobile/android/app/src/main/res/values/strings.xml create mode 100644 apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadEventBusTest.kt create mode 100644 apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadProtocolTest.kt create mode 100644 apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadPutExecutorTest.kt create mode 100644 apps/mobile/lib/features/uploads/native_transport.dart create mode 100644 apps/mobile/test/features/uploads/native_transport_test.dart diff --git a/apps/mobile/android/app/build.gradle.kts b/apps/mobile/android/app/build.gradle.kts index 32c0ba0..c0a4a4f 100644 --- a/apps/mobile/android/app/build.gradle.kts +++ b/apps/mobile/android/app/build.gradle.kts @@ -40,6 +40,16 @@ kotlin { } } +dependencies { + // P3 Task 7 后台上传桥接：WorkManager（expedited + dataSync 前台服务） + // + OkHttp 直传 presign PUT（独立 4.12，与 Flutter 引擎无耦合）。 + implementation("androidx.work:work-runtime-ktx:2.10.0") + implementation("androidx.core:core-ktx:1.15.0") + implementation("com.squareup.okhttp3:okhttp:4.12.0") + testImplementation("junit:junit:4.13.2") + testImplementation("com.squareup.okhttp3:mockwebserver:4.12.0") +} + flutter { source = "../.." } diff --git a/apps/mobile/android/app/src/main/AndroidManifest.xml b/apps/mobile/android/app/src/main/AndroidManifest.xml index 40435a7..7566d7f 100644 --- a/apps/mobile/android/app/src/main/AndroidManifest.xml +++ b/apps/mobile/android/app/src/main/AndroidManifest.xml @@ -1,6 +1,14 @@ - + + + + + + + + + + diff --git a/apps/mobile/android/app/src/main/res/values-zh/strings.xml b/apps/mobile/android/app/src/main/res/values-zh/strings.xml new file mode 100644 index 0000000..24ea5dd --- /dev/null +++ b/apps/mobile/android/app/src/main/res/values-zh/strings.xml @@ -0,0 +1,4 @@ + + 上传 + 正在上传照片 + diff --git a/apps/mobile/android/app/src/main/res/values/strings.xml b/apps/mobile/android/app/src/main/res/values/strings.xml new file mode 100644 index 0000000..f3b3e58 --- /dev/null +++ b/apps/mobile/android/app/src/main/res/values/strings.xml @@ -0,0 +1,5 @@ + + + Uploads + Uploading photo + diff --git a/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadEventBusTest.kt b/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadEventBusTest.kt new file mode 100644 index 0000000..ef6ef81 --- /dev/null +++ b/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadEventBusTest.kt @@ -0,0 +1,112 @@ +package com.photolive.photo_live + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +/** + * 上传事件总线测试（P3 计划 Task 7）：终态缓存重放（生命周期安全）、 + * ack 回收、purgeTask、进度瞬态、缓冲上限。 + */ +class UploadEventBusTest { + + @Before + fun reset() { + UploadEventBus.resetForTest() + } + + @Test + fun `terminal events replayed to late subscriber in seq order`() { + // 无 sink 时终态缓存；attach 后按 seq 顺序重放。 + UploadEventBus.emitCompleted("t1", "\"etag-1\"") + UploadEventBus.emitFailed("t2", UploadBridge.FAIL_HTTP_4XX, 404) + val received = mutableListOf>() + UploadEventBus.attach { received.add(it) } + assertEquals(2, received.size) + assertEquals(1L, received[0]["seq"]) + assertEquals("completed", received[0]["event"]) + assertEquals("\"etag-1\"", received[0]["etag"]) + assertEquals(2L, received[1]["seq"]) + assertEquals("failed", received[1]["event"]) + assertEquals("HTTP_4XX", received[1]["code"]) + assertEquals(404, received[1]["status"]) + } + + @Test + fun `live events delivered to attached sink with seq`() { + val received = mutableListOf>() + UploadEventBus.attach { received.add(it) } + UploadEventBus.emitProgress("t1", 10, 100) + UploadEventBus.emitCompleted("t1", null) + assertEquals(2, received.size) + assertEquals("progress", received[0]["event"]) + assertEquals(10L, received[0]["sent"]) + assertEquals(100L, received[0]["total"]) + assertEquals("completed", received[1]["event"]) + assertTrue((received[1]["seq"] as Long) > (received[0]["seq"] as Long)) + } + + @Test + fun `progress events not buffered or replayed`() { + UploadEventBus.emitProgress("t1", 1, 100) + val received = mutableListOf>() + UploadEventBus.attach { received.add(it) } + assertTrue(received.isEmpty()) + assertEquals(0, UploadEventBus.bufferedSnapshots().size) + } + + @Test + fun `ack drops events up to seq and keeps later ones`() { + UploadEventBus.emitCompleted("t1", null) // seq 1 + UploadEventBus.emitCompleted("t2", null) // seq 2 + UploadEventBus.emitFailed("t3", UploadBridge.FAIL_NETWORK) // seq 3 + UploadEventBus.ack(2) + val remaining = UploadEventBus.bufferedSnapshots() + assertEquals(1, remaining.size) + assertEquals("t3", remaining[0]["taskId"]) + assertEquals(3L, remaining[0]["seq"]) + } + + @Test + fun `purgeTask removes only that task's buffered events`() { + UploadEventBus.emitCompleted("t1", null) + UploadEventBus.emitFailed("t2", UploadBridge.FAIL_NETWORK) + UploadEventBus.purgeTask("t1") + val remaining = UploadEventBus.bufferedSnapshots() + assertEquals(1, remaining.size) + assertEquals("t2", remaining[0]["taskId"]) + } + + @Test + fun `buffer capped at MAX_BUFFERED dropping oldest`() { + repeat(UploadEventBus.MAX_BUFFERED + 5) { n -> + UploadEventBus.emitCompleted("t$n", null) + } + val buffered = UploadEventBus.bufferedSnapshots() + assertEquals(UploadEventBus.MAX_BUFFERED, buffered.size) + // 最老的 5 个被丢弃。 + assertEquals(6L, buffered.first()["seq"]) + assertEquals((UploadEventBus.MAX_BUFFERED + 5).toLong(), buffered.last()["seq"]) + } + + @Test + fun `needsRenew is terminal and replayed`() { + UploadEventBus.emitNeedsRenew("t9") + val received = mutableListOf>() + UploadEventBus.attach { received.add(it) } + assertEquals(listOf("needsRenew"), received.map { it["event"] }) + assertEquals(1, UploadEventBus.bufferedSnapshots().size) + } + + @Test + fun `detach stops delivery but keeps buffer`() { + val received = mutableListOf>() + UploadEventBus.attach { received.add(it) } + UploadEventBus.emitCompleted("t1", null) + UploadEventBus.detach() + UploadEventBus.emitCompleted("t2", null) + assertEquals(1, received.size) + assertEquals(2, UploadEventBus.bufferedSnapshots().size) + } +} diff --git a/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadProtocolTest.kt b/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadProtocolTest.kt new file mode 100644 index 0000000..8f48d76 --- /dev/null +++ b/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadProtocolTest.kt @@ -0,0 +1,108 @@ +package com.photolive.photo_live + +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertThrows +import org.junit.Test + +/** + * registerTask 参数解析测试（P3 计划 Task 7）。 + */ +class UploadProtocolTest { + + private fun baseArgs( + metadata: Map? = mapOf("uploadId" to "upl-1", "attempt" to 2), + headers: Any? = mapOf("Content-Type" to "image/jpeg"), + ): Map = mapOf( + "taskId" to "task-9", + "url" to "http://object.test.local/staging/a.jpg", + "method" to "PUT", + "headers" to headers, + "filePath" to "/data/uploads/a.jpg", + "metadata" to metadata, + ) + + @Test + fun `parses full args`() { + val spec = parseRegisterSpec(baseArgs()) + assertEquals("task-9", spec.taskId) + assertEquals("http://object.test.local/staging/a.jpg", spec.url) + assertEquals("PUT", spec.method) + assertEquals(mapOf("Content-Type" to "image/jpeg"), spec.headers) + assertEquals("/data/uploads/a.jpg", spec.filePath) + assertEquals("upl-1", spec.uploadId) + assertEquals(2, spec.attempt) + } + + @Test + fun `blank method defaults to PUT`() { + val spec = parseRegisterSpec(baseArgs().toMutableMap().apply { put("method", "") }) + assertEquals("PUT", spec.method) + } + + @Test + fun `null metadata defaults to attempt 0 and null uploadId`() { + val spec = parseRegisterSpec(baseArgs(metadata = null)) + assertEquals(0, spec.attempt) + assertNull(spec.uploadId) + } + + @Test + fun `missing headers defaults to empty map`() { + val spec = parseRegisterSpec( + baseArgs(headers = null).toMutableMap().apply { remove("headers") }, + ) + assertEquals(emptyMap(), spec.headers) + } + + @Test + fun `long attempt coerced to int`() { + val spec = parseRegisterSpec( + baseArgs(metadata = mapOf("uploadId" to null, "attempt" to 3L)), + ) + assertEquals(3, spec.attempt) + } + + @Test + fun `missing taskId throws`() { + assertThrows(IllegalArgumentException::class.java) { + parseRegisterSpec(baseArgs().toMutableMap().apply { remove("taskId") }) + } + } + + @Test + fun `non-string url throws`() { + assertThrows(IllegalArgumentException::class.java) { + parseRegisterSpec(baseArgs().toMutableMap().apply { put("url", 123) }) + } + } + + @Test + fun `non-string header value throws`() { + assertThrows(IllegalArgumentException::class.java) { + parseRegisterSpec(baseArgs(headers = mapOf("x" to 1))) + } + } + + @Test + fun `non-map headers throws`() { + assertThrows(IllegalArgumentException::class.java) { + parseRegisterSpec(baseArgs(headers = "Content-Type: image/jpeg")) + } + } + + @Test + fun `non-number attempt throws`() { + assertThrows(IllegalArgumentException::class.java) { + parseRegisterSpec( + baseArgs(metadata = mapOf("uploadId" to null, "attempt" to "two")), + ) + } + } + + @Test + fun `work tag prefixes task id`() { + assertEquals("upload-task-abc", UploadBridge.workTag("abc")) + assertEquals("upload-task-", UploadBridge.WORK_TAG_PREFIX) + } +} diff --git a/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadPutExecutorTest.kt b/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadPutExecutorTest.kt new file mode 100644 index 0000000..6358bb5 --- /dev/null +++ b/apps/mobile/android/app/src/test/java/com/photolive/photo_live/UploadPutExecutorTest.kt @@ -0,0 +1,175 @@ +package com.photolive.photo_live + +import java.nio.file.Files +import okhttp3.OkHttpClient +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +/** + * UploadPutExecutor 结局分类测试（P3 计划 Task 7；MockWebServer 驱动）： + * 200→Success(etag)、500→Retryable(server)、403→RenewRequired、 + * 4xx→HttpFail、连接失败→Retryable(network)、缺文件→SourceMissing、 + * 请求头透传与进度回调。 + */ +class UploadPutExecutorTest { + + private lateinit var server: MockWebServer + private lateinit var file: java.io.File + private val client = OkHttpClient() + + @Before + fun setUp() { + server = MockWebServer() + server.start() + val tmp = Files.createTempFile("upload-put", ".bin") + tmp.toFile().deleteOnExit() + file = tmp.toFile() + Files.write(tmp, ByteArray(3000) { (it % 251).toByte() }) + } + + @After + fun tearDown() { + server.shutdown() + } + + private fun execute(status: Int = 200): PutOutcome = + UploadPutExecutor.execute( + url = server.url("/staging/x.jpg").toString(), + method = "PUT", + headers = mapOf("Content-Type" to "image/jpeg", "x-amz-a" to "sig-1"), + filePath = file.absolutePath, + client = client, + ) + + @Test + fun `2xx maps to Success with etag captured`() { + server.enqueue( + MockResponse().setResponseCode(200).addHeader("ETag", "\"etag-abc\""), + ) + val outcome = execute() + assertTrue(outcome is PutOutcome.Success) + assertEquals("\"etag-abc\"", (outcome as PutOutcome.Success).etag) + + val recorded = server.takeRequest() + assertEquals("PUT", recorded.method) + assertEquals("/staging/x.jpg", recorded.path) + assertEquals("image/jpeg", recorded.getHeader("Content-Type")) + assertEquals("sig-1", recorded.getHeader("x-amz-a")) + assertEquals(file.length(), recorded.bodySize) + // 请求体逐字节一致。 + val sent = recorded.body.readByteArray() + assertEquals(file.length().toInt(), sent.size) + } + + @Test + fun `403 maps to RenewRequired`() { + server.enqueue(MockResponse().setResponseCode(403)) + assertTrue(execute() is PutOutcome.RenewRequired) + } + + @Test + fun `404 maps to HttpFail with status`() { + server.enqueue(MockResponse().setResponseCode(404)) + val outcome = execute() + assertTrue(outcome is PutOutcome.HttpFail) + assertEquals(404, (outcome as PutOutcome.HttpFail).status) + } + + @Test + fun `400 maps to HttpFail`() { + server.enqueue(MockResponse().setResponseCode(400)) + assertTrue(execute() is PutOutcome.HttpFail) + } + + @Test + fun `500 maps to Retryable server error`() { + server.enqueue(MockResponse().setResponseCode(500)) + val outcome = execute() + assertTrue(outcome is PutOutcome.Retryable) + outcome as PutOutcome.Retryable + assertTrue(outcome.serverError) + assertEquals(500, outcome.status) + } + + @Test + fun `connection failure maps to Retryable network`() { + // 关停服务后直连 → IOException（连接层失败）→ network 重试分类。 + val url = server.url("/staging/x.jpg").toString() + server.shutdown() + val outcome = UploadPutExecutor.execute( + url = url, + method = "PUT", + headers = emptyMap(), + filePath = file.absolutePath, + client = client, + ) + assertTrue(outcome is PutOutcome.Retryable) + outcome as PutOutcome.Retryable + assertTrue(!outcome.serverError) + assertEquals(null, outcome.status) + } + + @Test + fun `missing file maps to SourceMissing`() { + val outcome = UploadPutExecutor.execute( + url = server.url("/staging/x.jpg").toString(), + method = "PUT", + headers = emptyMap(), + filePath = java.io.File(file.parentFile, "no-such-file.bin").absolutePath, + client = client, + ) + assertTrue(outcome is PutOutcome.SourceMissing) + assertEquals(0, server.requestCount) + } + + @Test + fun `progress callback reports monotonic bytes up to total`() { + server.enqueue(MockResponse().setResponseCode(200)) + val samples = mutableListOf>() + UploadPutExecutor.execute( + url = server.url("/staging/x.jpg").toString(), + method = "PUT", + headers = mapOf("Content-Type" to "image/jpeg"), + filePath = file.absolutePath, + client = client, + ) { sent, total -> samples.add(sent to total) } + assertTrue(samples.isNotEmpty()) + assertEquals(file.length(), samples.last().first) + assertEquals(file.length(), samples.last().second) + samples.zipWithNext().forEach { (prev, next) -> + assertTrue(next.first >= prev.first) + } + samples.forEach { assertEquals(file.length(), it.second) } + } + + @Test + fun `blank method defaults to PUT and uppercases`() { + assertEquals("PUT", UploadPutExecutor.normalizeMethod("")) + assertEquals("PUT", UploadPutExecutor.normalizeMethod("put")) + assertEquals("PUT", UploadPutExecutor.normalizeMethod("PUT")) + } + + @Test + fun `content length header from credential is not duplicated`() { + // Content-Length 由 RequestBody.contentLength 声明，重复注入会破坏 + // okhttp 请求；显式传入同名头应被忽略。 + server.enqueue(MockResponse().setResponseCode(200)) + UploadPutExecutor.execute( + url = server.url("/staging/x.jpg").toString(), + method = "PUT", + headers = mapOf( + "Content-Type" to "image/jpeg", + "Content-Length" to "999999", + ), + filePath = file.absolutePath, + client = client, + ) + val recorded = server.takeRequest() + assertEquals(file.length(), recorded.getHeader("Content-Length")?.toLong()) + } +} diff --git a/apps/mobile/lib/features/uploads/native_transport.dart b/apps/mobile/lib/features/uploads/native_transport.dart new file mode 100644 index 0000000..df0ed1e --- /dev/null +++ b/apps/mobile/lib/features/uploads/native_transport.dart @@ -0,0 +1,421 @@ +/// 上传传输抽象与双实现（P3 计划 Task 7）。 +/// +/// 引擎的传输步骤（presign PUT）经 [Transport] 接口委托： +/// - **[DartHttpTransport]**（Task 6 逻辑原样抽出）：裸 dio 流式 PUT + +/// 进度计数流 + 取消令牌——iOS / 桌面 / 调试路径（dart-define +/// `FORCE_DART_TRANSPORT=true` 强制启用）； +/// - **[NativeTransport]**（Task 7 新增，Android）：MethodChannel +/// `photolive/uploads` 注册 WorkManager 后台工作（expedited + dataSync +/// 前台服务），EventChannel `photolive/uploads/events` 回流 +/// progress/completed/failed/needsRenew 事件——App 切后台/锁屏后续传 +/// 由系统调度承担。 +/// +/// 失败语义（两实现归一为 [TransportException] 族，引擎分类见 +/// upload_engine.dart `_classifyFailure`）： +/// | 情形 | 异常 | +/// |---|---| +/// | 网络不可达/超时/取消令牌触发前的连接层失败 | `TransportException(network)` | +/// | HTTP 403（presign 凭证过期） | [TransportRenewRequiredException]（引擎 renew 后重发） | +/// | 其余 HTTP 非 2xx | `TransportException(http, statusCode)` | +/// | 源文件缺失（原生 worker 发现） | `TransportException(sourceMissing)` | +/// | 平台通道调用失败 | `TransportException(channel)` | +/// | 取消令牌触发 | `TransportException(cancelled)`（引擎静默收敛） | +library; + +import 'dart:async'; +import 'dart:io'; + +import 'package:dio/dio.dart'; +import 'package:flutter/foundation.dart' show kIsWeb; +import 'package:flutter/services.dart'; + +/// presign PUT 传输接口（引擎只委托文件 PUT；init/complete/renew 仍在 +/// Dart 引擎内，保持服务端交互语义单源）。 +abstract interface class Transport { + /// 单次文件 PUT。 + /// + /// - 成功返回 [TransportResult]（ETag 供诊断；完成核验以服务端为准）； + /// - 失败抛 [TransportException] 族（见库级注释分类表）； + /// - [cancelToken] 取消时抛 `TransportException(cancelled)`。 + Future send({ + required String taskId, + required String url, + required String method, + required Map headers, + required String filePath, + Map? metadata, + void Function(int sent, int total)? onProgress, + CancelToken? cancelToken, + }); + + /// 释放底层资源（引擎 dispose 调用；注入方自持传输时自行管理）。 + void dispose(); +} + +/// PUT 成功结果。 +class TransportResult { + const TransportResult({this.etag}); + + /// 服务端返回的 ETag 响应头（可空；仅诊断用途）。 + final String? etag; +} + +/// 传输失败码（原生侧事件码与本侧异常码的归一层）。 +abstract final class TransportErrorCodes { + /// 网络不可达/超时/连接层失败。 + static const String network = 'NETWORK'; + + /// HTTP 层失败（statusCode 保留在异常上）。 + static const String http = 'HTTP'; + + /// 取消令牌触发（非失败分类；引擎静默收敛）。 + static const String cancelled = 'CANCELLED'; + + /// 源文件缺失（受管副本被清理，不可恢复）。 + static const String sourceMissing = 'SOURCE_MISSING'; + + /// 平台通道调用失败（桥接层异常）。 + static const String channel = 'CHANNEL'; +} + +/// 传输失败（两实现统一的异常类型）。 +class TransportException implements Exception { + const TransportException(this.code, {this.statusCode, this.cause}); + + /// 失败分类（[TransportErrorCodes]）。 + final String code; + + /// HTTP 状态码（http 类失败携带）。 + final int? statusCode; + + /// 底层异常（诊断）。 + final Object? cause; + + @override + String toString() => + 'TransportException($code${statusCode != null ? ', status=$statusCode' : ''}' + '${cause != null ? ', cause=$cause' : ''})'; +} + +/// HTTP 403（presign 凭证过期）：引擎 renewUpload 换新凭证后重发。 +class TransportRenewRequiredException extends TransportException { + const TransportRenewRequiredException({super.cause}) + : super(TransportErrorCodes.http, statusCode: 403); +} + +// ── Dart（dio）实现 ────────────────────────────────────────────────────── + +/// Task 6 的裸 dio 直传实现（iOS / 桌面 / FORCE_DART_TRANSPORT 调试）。 +/// +/// 严格携带 credential.headers（+ Content-Length——流式发送 dio 无法自动 +/// 推断，显式声明）；进度经计数流上报（不依赖适配器对 onSendProgress 的 +/// 实现）。 +class DartHttpTransport implements Transport { + DartHttpTransport({Dio? dio}) + : _dio = dio, + _ownedDio = dio == null; + + final Dio? _dio; + final bool _ownedDio; + + /// 自建实例惰性创建（注入 dio 时不建）。 + Dio? _created; + Dio get _client => _dio ?? (_created ??= Dio()); + + @override + Future send({ + required String taskId, + required String url, + required String method, + required Map headers, + required String filePath, + Map? metadata, + void Function(int sent, int total)? onProgress, + CancelToken? cancelToken, + }) async { + final file = File(filePath); + final length = await file.length(); + var sent = 0; + final progressStream = file.openRead().map((List chunk) { + sent += chunk.length; + onProgress?.call(sent, length); + return chunk; + }); + try { + final resp = await _client.put( + url, + data: progressStream, + cancelToken: cancelToken, + options: Options( + method: method, + headers: { + ...headers, + Headers.contentLengthHeader: length, + }, + // 响应体非 JSON（MinIO XML/空），防 transformer 解析报错。 + responseType: ResponseType.plain, + receiveDataWhenStatusError: true, + ), + ); + final etag = resp.headers.value(HttpHeaders.etagHeader); + return TransportResult(etag: etag); + } on DioException catch (e) { + throw _mapDioException(e); + } + } + + /// dio 异常 → 传输异常归一（403 → renew；连接层 → network；取消 → + /// cancelled；其余 badResponse → http）。 + static TransportException _mapDioException(DioException e) { + if (e.type == DioExceptionType.cancel) { + return const TransportException(TransportErrorCodes.cancelled); + } + final status = e.response?.statusCode; + if (status != null) { + if (status == 403) return const TransportRenewRequiredException(); + return TransportException(TransportErrorCodes.http, + statusCode: status, cause: e); + } + if (e.type == DioExceptionType.connectionTimeout || + e.type == DioExceptionType.sendTimeout || + e.type == DioExceptionType.receiveTimeout || + e.type == DioExceptionType.transformTimeout || + e.type == DioExceptionType.connectionError || + e.type == DioExceptionType.badCertificate || + e.type == DioExceptionType.unknown) { + return TransportException(TransportErrorCodes.network, cause: e); + } + // badResponse 无响应必经上方分支，防御性兜底为网络类。 + return TransportException(TransportErrorCodes.network, cause: e); + } + + @override + void dispose() { + if (_ownedDio) { + _client.close(); + } + } +} + +// ── 原生（WorkManager）实现 ────────────────────────────────────────────── + +/// 平台通道名（与 Android UploadPlugin 一致；Kotlin 侧常量见 +/// UploadProtocol.kt）。 +abstract final class UploadBridgeChannels { + /// 命令通道：registerTask / cancelTask / queryTask / ackEvents。 + static const String methods = 'photolive/uploads'; + + /// 事件流：progress / completed / failed / needsRenew（带 seq， + /// 终态事件未 ack 前原生侧缓存重放——生命周期安全）。 + static const String events = 'photolive/uploads/events'; +} + +/// 平台通道绑定薄层（默认真实通道；测试注入替身）。 +class NativeUploadChannel { + const NativeUploadChannel(); + + static const MethodChannel _methods = MethodChannel( + UploadBridgeChannels.methods, + ); + static const EventChannel _events = EventChannel( + UploadBridgeChannels.events, + ); + + Future invoke(String method, Map args) => + _methods.invokeMethod(method, args); + + Stream get events => _events.receiveBroadcastStream(); +} + +/// 原生事件等待句柄。 +class _NativeWaiter { + _NativeWaiter(this.onProgress); + + final void Function(int sent, int total)? onProgress; + final Completer completer = Completer(); + + void complete(TransportResult result) { + if (!completer.isCompleted) completer.complete(result); + } + + void completeError(Object error) { + if (!completer.isCompleted) completer.completeError(error); + } +} + +/// Android 原生传输：registerTask 入队 WorkManager（expedited + dataSync +/// 前台服务），事件流回流终态/进度。 +/// +/// 生命周期安全：事件带单调 seq；终态事件在 Dart ack 前由原生侧缓存， +/// 重新 listen 时重放——进程重启/切后台恢复后不丢完成信号。进度事件不 +/// 重放（瞬态）。 +class NativeTransport implements Transport { + NativeTransport({NativeUploadChannel? channel}) + : _channel = channel ?? const NativeUploadChannel(); + + final NativeUploadChannel _channel; + + StreamSubscription? _eventsSub; + final Map _waiters = {}; + + void _ensureSubscribed() { + if (_eventsSub != null) return; + _eventsSub = _channel.events.listen( + _onEvent, + onError: (Object _) { + // 事件流出错：丢弃订阅，下次 send 重建（注册重试语义兜底）。 + _eventsSub?.cancel(); + _eventsSub = null; + }, + onDone: () => _eventsSub = null, + ); + } + + void _onEvent(dynamic raw) { + if (raw is! Map) return; + final event = Map.from(raw); + final taskId = event['taskId'] as String?; + final waiter = taskId == null ? null : _waiters[taskId]; + switch (event['event']) { + case 'progress': + final sent = (event['sent'] as num?)?.toInt(); + final total = (event['total'] as num?)?.toInt(); + if (waiter != null && sent != null && total != null) { + waiter.onProgress?.call(sent, total); + } + case 'completed': + waiter?.complete( + TransportResult(etag: event['etag'] as String?), + ); + _ackTerminal(event); + case 'needsRenew': + waiter?.completeError(const TransportRenewRequiredException()); + _ackTerminal(event); + case 'failed': + waiter?.completeError(_mapFailedEvent(event)); + _ackTerminal(event); + default: + break; // 未知事件：前向兼容忽略。 + } + } + + /// 原生失败码 → 传输异常。 + static TransportException _mapFailedEvent(Map event) { + final status = (event['status'] as num?)?.toInt(); + switch (event['code']) { + case 'NETWORK': + return const TransportException(TransportErrorCodes.network); + case 'SOURCE_MISSING': + return const TransportException(TransportErrorCodes.sourceMissing); + case 'HTTP_4XX': + case 'SERVER_ERROR': + return TransportException( + TransportErrorCodes.http, + statusCode: status, + ); + default: + return TransportException( + event['code'] as String? ?? TransportErrorCodes.channel, + statusCode: status, + ); + } + } + + void _ackTerminal(Map event) { + final seq = (event['seq'] as num?)?.toInt(); + if (seq == null) return; + unawaited( + _channel + .invoke('ackEvents', {'seq': seq}) + .catchError((Object _) => null), + ); + } + + @override + Future send({ + required String taskId, + required String url, + required String method, + required Map headers, + required String filePath, + Map? metadata, + void Function(int sent, int total)? onProgress, + CancelToken? cancelToken, + }) async { + _ensureSubscribed(); + final waiter = _NativeWaiter(onProgress); + _waiters[taskId] = waiter; + // 取消令牌联动：cancelTask（原生尽力停止 worker）+ 等待句柄立即失败。 + cancelToken?.whenCancel.then((_) { + // 已终结则 no-op（isCompleted 守护）。 + waiter.completeError( + const TransportException(TransportErrorCodes.cancelled), + ); + unawaited( + _channel + .invoke('cancelTask', {'taskId': taskId}) + .catchError((Object _) => null), + ); + }); + try { + await _channel.invoke('registerTask', { + 'taskId': taskId, + 'url': url, + 'method': method, + 'headers': headers, + 'filePath': filePath, + 'metadata': metadata ?? const {}, + }); + } on PlatformException catch (e) { + // 注册失败：摘除等待句柄后归一为通道异常。 + if (_waiters[taskId] == waiter) _waiters.remove(taskId); + throw TransportException(TransportErrorCodes.channel, cause: e); + } + try { + return await waiter.completer.future; + } finally { + if (_waiters[taskId] == waiter) _waiters.remove(taskId); + } + } + + /// 查询原生侧工作状态（WorkInfo 概览；Task 9/11 校准用）。 + Future?> queryTask(String taskId) async { + try { + final result = await _channel + .invoke('queryTask', {'taskId': taskId}); + return result == null + ? null + : Map.from(result as Map); + } on PlatformException { + return null; + } + } + + @override + void dispose() { + _eventsSub?.cancel(); + _eventsSub = null; + } +} + +// ── 传输选择 ───────────────────────────────────────────────────────────── + +/// dart-define FORCE_DART_TRANSPORT=true：Android 也强制 Dart dio 路径 +/// （桥接层问题排查用）。 +const bool kForceDartTransport = bool.fromEnvironment('FORCE_DART_TRANSPORT'); + +/// 传输选择纯函数（可测）：Android 且未强制 Dart → 原生。 +bool preferNativeTransport({ + required bool isAndroid, + bool forceDartTransport = kForceDartTransport, +}) => + isAndroid && !forceDartTransport; + +/// 默认传输解析：Android → [NativeTransport]（WorkManager 后台续传）； +/// 其余平台 / 强制覆盖 → [DartHttpTransport]。 +Transport defaultUploadTransport() { + if (preferNativeTransport(isAndroid: !kIsWeb && Platform.isAndroid)) { + return NativeTransport(); + } + return DartHttpTransport(); +} diff --git a/apps/mobile/lib/features/uploads/upload_engine.dart b/apps/mobile/lib/features/uploads/upload_engine.dart index eb13a14..79b3798 100644 --- a/apps/mobile/lib/features/uploads/upload_engine.dart +++ b/apps/mobile/lib/features/uploads/upload_engine.dart @@ -10,8 +10,8 @@ /// 介入，永不自动出队）；网络恢复（[onNetworkRestored]）对 /// waitingNetwork 立即绕过退避； /// - **传输三步**：initUpload（幂等，clientTaskId + sourceDeviceId）→ -/// presign PUT（dio 流式 + 进度回调 + 取消令牌；**裸 Dio**——不挂 -/// ApiClient 拦截器，避免 Bearer 头污染 MinIO 预签名签名）→ +/// presign PUT（**经 [Transport] 委托**——Android 走原生 WorkManager +/// 后台桥接（Task 7），其余平台裸 dio 流式；进度回调 + 取消令牌）→ /// completeUpload（幂等核验）→ completed； /// - **失败分类**（[UploadEngineSourceMissing] 之外见 [_classifyFailure]）： /// | 错误 | 目标状态 | 自动重试 | @@ -46,6 +46,7 @@ import '../../core/app_exception.dart'; import '../../core/device_id.dart'; import '../../core/error_codes.dart'; import 'models.dart'; +import 'native_transport.dart'; import 'task_state.dart'; import 'task_store.dart'; @@ -75,27 +76,41 @@ const int kMaxCredentialRenewals = 2; /// 传输引擎。 /// /// 事件驱动 + 周期兜底；不 own isolate（传输多为 IO 等待，UI isolate -/// 足够；Android 后台延续由 Task 7 原生桥接承担）。 +/// 足够；Android 后台延续由 [Transport] 原生实现承担——Task 7）。 class UploadEngine { UploadEngine( this._store, this._apiClient, this._deviceId, { Dio? objectDio, + Transport? transport, DateTime Function()? now, this.maxParallel = kMaxParallelUploads, this.keepCompletedDays = kKeepCompletedDays, this.onProgress, this.onError, - }) : _objectDio = objectDio ?? Dio(), + }) : _objectDio = objectDio, + _transport = transport ?? + (objectDio != null + ? DartHttpTransport(dio: objectDio) + : defaultUploadTransport()), + _ownsTransport = transport == null, _now = now ?? DateTime.now; final TaskStore _store; final ApiClient _apiClient; final DeviceId _deviceId; - /// presign PUT 专用裸 Dio（无鉴权拦截器，见库级注释）。 - final Dio _objectDio; + /// presign PUT 专用裸 Dio（无鉴权拦截器，见库级注释；[objectDio] 注入 + /// 时持有引用供 dispose 关闭——传输经 [DartHttpTransport] 使用）。 + final Dio? _objectDio; + + /// PUT 传输委托（Android 原生 WorkManager 桥接 / 其余平台 Dart dio； + /// 选择逻辑见 native_transport.dart）。 + final Transport _transport; + + /// dispose 时是否释放 [_transport]（注入传输由注入方管理生命周期）。 + final bool _ownsTransport; final DateTime Function() _now; /// 并行传输上限（运行时 <1 钳位为 1，见 [_effectiveParallel]）。 @@ -294,9 +309,13 @@ class UploadEngine { now: _now(), ); await _transferSteps(current, handle); + } on TransportException catch (e) { + // 传输层失败（两实现归一；取消已在 _putObject 内短路，到达这里的 + // 均为待分类失败）。 + await _classifyFailure(task, e); } on DioException catch (e) { - // 仅来自 PUT（api 调用已归一 AppException）；取消已在 - // _transferSteps 内短路，到达这里的均为待分类失败。 + // 防御性兜底（api 调用已归一 AppException；dio 异常仅可能自传输 + // 层漏网），按原语义分类。 await _classifyFailure(task, e); } on AppException catch (e) { await _classifyFailure(task, e); @@ -333,20 +352,17 @@ class UploadEngine { now: _now(), ); - // b. presign PUT（流式 + 进度 + 取消令牌；403 → renew 换凭证重传）。 + // b. presign PUT（transport 委托 + 进度 + 取消令牌；403 → renew 换 + // 凭证重传——原生路径 re-register 新 URL，Dart 路径重发 dio PUT）。 var credential = init.credential; var renewals = 0; while (true) { try { - await _putObject(task, credential, handle.cancelToken); + await _putObject(task, credential, handle.cancelToken, + serverUploadId: init.uploadId); break; - } on DioException catch (e) { - if (handle.cancelToken.isCancelled) { - // 用户取消：状态已置 cancelled（cancel() 内），终止协程。 - return; - } - if (e.response?.statusCode == 403 && - renewals < kMaxCredentialRenewals) { + } on TransportRenewRequiredException { + if (renewals < kMaxCredentialRenewals) { renewals++; final renewed = await _apiClient.renewUpload(init.uploadId); credential = renewed.credential; @@ -375,14 +391,15 @@ class UploadEngine { ); } - /// presign PUT：裸 Dio 直传 credential.url，严格携带 credential.headers - /// （+ Content-Length——流式发送 dio 无法自动推断，显式声明）。进度经 - /// 计数流上报（引擎自持，不依赖适配器对 onSendProgress 的实现）。 + /// presign PUT：委托 [_transport]（Android 原生 WorkManager 桥接 / + /// 其余 Dart dio 流式），严格携带 credential.headers。源文件存在性前置 + /// 检查保留在引擎侧（两路径共用的 sourceMissing 分类依据）。 Future _putObject( UploadTask task, UploadCredential credential, - CancelToken cancelToken, - ) async { + CancelToken cancelToken, { + String? serverUploadId, + }) async { final file = File(task.sourceFilePath); if (!await file.exists()) { // 源文件缺失：不可恢复，分类按「不自动重试」处理。 @@ -391,37 +408,43 @@ class UploadEngine { message: 'source file missing', ); } - final length = await file.length(); - var sent = 0; - final progressStream = - file.openRead().map((List chunk) { - sent += chunk.length; - onProgress?.call(task.id, sent, length); - return chunk; - }); - await _objectDio.put( - credential.url, - data: progressStream, - cancelToken: cancelToken, - options: Options( + try { + await _transport.send( + taskId: task.id, + url: credential.url, method: credential.method, - headers: { - ...credential.headers, - Headers.contentLengthHeader: length, + headers: credential.headers, + filePath: task.sourceFilePath, + metadata: { + 'uploadId': serverUploadId, + 'attempt': task.attempt, }, - // 响应体非 JSON（MinIO XML/空），防 transformer 解析报错。 - responseType: ResponseType.plain, - receiveDataWhenStatusError: true, - ), - ); + onProgress: (sent, total) => onProgress?.call(task.id, sent, total), + cancelToken: cancelToken, + ); + } on TransportException catch (e) { + if (_isCancelledTransfer(e, cancelToken)) { + // 用户取消：状态已置 cancelled（cancel() 内），终止协程。 + return; + } + rethrow; + } } + /// 取消判定：取消令牌已触发或传输层报告 cancelled。 + static bool _isCancelledTransfer(Object error, CancelToken cancelToken) => + cancelToken.isCancelled || + (error is TransportException && + error.code == TransportErrorCodes.cancelled); + /// 失败分类与状态落库（分类表见库级注释；attempt 每次 +1，目标状态 /// 自 uploading/uploadedConfirming 均合法迁移）。 Future _classifyFailure(UploadTask task, Object error) async { final AppException app; if (error is AppException) { app = error; + } else if (error is TransportException) { + app = _normalizeTransportFailure(error); } else if (error is DioException) { app = ApiClient.mapDioException(error); } else { @@ -488,6 +511,64 @@ class UploadEngine { ); } + /// 传输层异常 → AppException 归一（与 dio 路径 mapDioException 语义 + /// 对齐：网络类/429 → 网络桶；401 → 短退避桶；sourceMissing → 不自动 + /// 重试桶；其余按 HTTP 状态归桶 4）。 + static AppException _normalizeTransportFailure(TransportException e) { + switch (e.code) { + case TransportErrorCodes.network: + return const AppException( + code: ErrorCodes.networkError, + message: '传输失败（网络）', + ); + case TransportErrorCodes.sourceMissing: + return const AppException( + code: UploadEngine.sourceMissingCode, + message: 'source file missing', + ); + case TransportErrorCodes.cancelled: + return const AppException( + code: ErrorCodes.cancelled, + message: '传输已取消', + ); + case TransportErrorCodes.channel: + return AppException( + code: ErrorCodes.internalError, + message: '原生桥接通道错误: ${e.cause}', + ); + case TransportErrorCodes.http: + default: + final status = e.statusCode; + if (status == 401) { + return const AppException( + code: ErrorCodes.authTokenExpired, + statusCode: 401, + message: '未认证或会话已过期', + ); + } + if (status == 429 || status == 408) { + return const AppException( + code: ErrorCodes.rateLimited, + statusCode: 429, + message: '传输限速', + ); + } + if (status != null) { + return AppException( + code: status >= 500 + ? ErrorCodes.internalError + : ErrorCodes.httpError, + statusCode: status, + message: '传输 HTTP $status', + ); + } + return AppException( + code: ErrorCodes.internalError, + message: '传输失败: ${e.cause ?? e}', + ); + } + } + /// 分类落地（并发取消致 StateError 时静默——cancelled 已是终态）。 Future _fail( UploadTask task, @@ -583,13 +664,17 @@ class UploadEngine { return cleaned; } - /// 关闭（App 退出）：停 timer、尽力断在途、释放 object Dio。 + /// 关闭（App 退出）：停 timer、尽力断在途、释放注入的 object Dio 与 + /// 自建传输。 void dispose() { stop(); for (final handle in _inFlight.values) { handle.cancelToken.cancel(); } - _objectDio.close(); + _objectDio?.close(); + if (_ownsTransport) { + _transport.dispose(); + } } } diff --git a/apps/mobile/test/features/uploads/native_transport_test.dart b/apps/mobile/test/features/uploads/native_transport_test.dart new file mode 100644 index 0000000..c10c215 --- /dev/null +++ b/apps/mobile/test/features/uploads/native_transport_test.dart @@ -0,0 +1,472 @@ +/// 传输层测试（P3 计划 Task 7）。 +/// +/// 覆盖： +/// - NativeTransport：真实通道名 + 测试 messenger（TestDefaultBinaryMessenger +/// 的 mock method/stream handler）——registerTask 参数形状、事件映射 +/// （progress/completed/needsRenew/failed×码表）、终态 ack、取消联动、 +/// 注册失败归一、多任务路由； +/// - DartHttpTransport：dio 异常 → 传输异常归一（403→renew、取消、网络）； +/// - 传输选择：Android 默认原生、FORCE_DART_TRANSPORT 覆盖、VM 缺省 +/// DartHttp。 +library; + +import 'dart:async'; +import 'dart:io'; + +import 'package:dio/dio.dart'; +import 'package:flutter/services.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:photo_live/features/uploads/native_transport.dart'; + +/// 捕获 Future 的异常对象（完成则失败测试）。 +Future captureError(Future future) async { + try { + await future; + } on Object catch (e) { + return e; + } + throw StateError('预期失败但 Future 成功完成'); +} + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + final messenger = + TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + const methodChannel = MethodChannel(UploadBridgeChannels.methods); + const eventChannel = EventChannel(UploadBridgeChannels.events); + + late NativeTransport transport; + late List methodCalls; + MockStreamHandlerEventSink? sink; + + setUp(() { + methodCalls = []; + sink = null; + messenger.setMockMethodCallHandler(methodChannel, (call) async { + methodCalls.add(call); + switch (call.method) { + case 'registerTask': + case 'cancelTask': + case 'ackEvents': + return true; + case 'queryTask': + return { + 'state': 'RUNNING', + 'attempt': 1, + 'progress': {'sent': 5, 'total': 10}, + }; + default: + return null; + } + }); + messenger.setMockStreamHandler( + eventChannel, + MockStreamHandler.inline( + onListen: (arguments, events) { + sink = events; + }, + onCancel: (arguments) {}, + ), + ); + transport = NativeTransport(); + }); + + tearDown(() async { + transport.dispose(); + messenger.setMockMethodCallHandler(methodChannel, null); + messenger.setMockStreamHandler(eventChannel, null); + }); + + /// 泵事件循环直到 registerTask 落账（事件由用例经 [sink] 注入）。 + Future untilRegistered() async { + for (var i = 0; + i < 100 && !methodCalls.any((c) => c.method == 'registerTask'); + i++) { + await pumpEventQueue(); + } + expect(methodCalls.any((c) => c.method == 'registerTask'), isTrue, + reason: 'registerTask 已发出'); + } + + void emitEvent(Map event) => sink!.success(event); + + group('NativeTransport 命令与事件映射', () { + test('registerTask 参数形状：taskId/url/method/headers/filePath/metadata', + () async { + final future = transport.send( + taskId: 'task-1', + url: 'http://obj.test/staging/a.jpg', + method: 'PUT', + headers: const {'Content-Type': 'image/jpeg'}, + filePath: '/data/a.jpg', + metadata: const {'uploadId': 'upl-1', 'attempt': 2}, + ); + await untilRegistered(); + emitEvent({ + 'event': 'completed', + 'taskId': 'task-1', + 'etag': '"e-1"', + 'seq': 7, + }); + final result = await future; + expect(result.etag, '"e-1"'); + + final call = methodCalls.firstWhere((c) => c.method == 'registerTask'); + final args = Map.from(call.arguments as Map); + expect(args['taskId'], 'task-1'); + expect(args['url'], 'http://obj.test/staging/a.jpg'); + expect(args['method'], 'PUT'); + expect(args['headers'], {'Content-Type': 'image/jpeg'}); + expect(args['filePath'], '/data/a.jpg'); + expect(args['metadata'], {'uploadId': 'upl-1', 'attempt': 2}); + + // 终态事件消费后 ack 回传 seq。 + await pumpEventQueue(); + expect( + methodCalls.any( + (c) => c.method == 'ackEvents' && (c.arguments as Map)['seq'] == 7, + ), + isTrue, + reason: '终态 ack 已回传', + ); + }); + + test('progress 事件映射到 onProgress；completed 前不终结', () async { + final samples = <(int, int)>[]; + final future = transport.send( + taskId: 'task-2', + url: 'http://obj.test/s', + method: 'PUT', + headers: const {}, + filePath: '/data/b.jpg', + onProgress: (sent, total) => samples.add((sent, total)), + ); + await untilRegistered(); + emitEvent({ + 'event': 'progress', + 'taskId': 'task-2', + 'sent': 10, + 'total': 100, + 'seq': 1, + }); + emitEvent({ + 'event': 'progress', + 'taskId': 'task-2', + 'sent': 90, + 'total': 100, + 'seq': 2, + }); + await pumpEventQueue(); + expect(samples, [(10, 100), (90, 100)]); + emitEvent({ + 'event': 'completed', + 'taskId': 'task-2', + 'etag': null, + 'seq': 3, + }); + expect((await future).etag, isNull); + }); + + test('needsRenew 事件 → TransportRenewRequiredException（403 语义）', + () async { + final future = transport.send( + taskId: 'task-3', + url: 'http://obj.test/s', + method: 'PUT', + headers: const {}, + filePath: '/data/c.jpg', + ); + await untilRegistered(); + emitEvent({'event': 'needsRenew', 'taskId': 'task-3', 'seq': 1}); + final err = await captureError(future); + expect(err, isA()); + expect((err as TransportException).statusCode, 403); + }); + + test('failed 事件码表：NETWORK / HTTP_4XX / SERVER_ERROR / SOURCE_MISSING', + () async { + Future drive(Map failEvent) async { + final taskId = failEvent['taskId']! as String; + final future = transport.send( + taskId: taskId, + url: 'http://obj.test/s', + method: 'PUT', + headers: const {}, + filePath: '/data/x.jpg', + ); + await untilRegistered(); + emitEvent(failEvent); + return captureError(future); + } + + var err = await drive({ + 'event': 'failed', + 'taskId': 't-net', + 'code': 'NETWORK', + 'seq': 1, + }); + expect((err as TransportException).code, TransportErrorCodes.network); + expect(err.statusCode, isNull); + + err = await drive({ + 'event': 'failed', + 'taskId': 't-4xx', + 'code': 'HTTP_4XX', + 'status': 400, + 'seq': 2, + }); + expect((err as TransportException).code, TransportErrorCodes.http); + expect(err.statusCode, 400); + + err = await drive({ + 'event': 'failed', + 'taskId': 't-5xx', + 'code': 'SERVER_ERROR', + 'status': 502, + 'seq': 3, + }); + expect((err as TransportException).code, TransportErrorCodes.http); + expect(err.statusCode, 502); + + err = await drive({ + 'event': 'failed', + 'taskId': 't-src', + 'code': 'SOURCE_MISSING', + 'seq': 4, + }); + expect( + (err as TransportException).code, + TransportErrorCodes.sourceMissing, + ); + }); + + test('取消令牌：cancelTask 下发 + cancelled 异常收敛', () async { + final cancelToken = CancelToken(); + final future = transport.send( + taskId: 'task-4', + url: 'http://obj.test/s', + method: 'PUT', + headers: const {}, + filePath: '/data/d.jpg', + cancelToken: cancelToken, + ); + await untilRegistered(); + cancelToken.cancel('用户取消'); + final err = await captureError(future); + expect((err as TransportException).code, TransportErrorCodes.cancelled); + await pumpEventQueue(); + expect( + methodCalls.any( + (c) => + c.method == 'cancelTask' && + (c.arguments as Map)['taskId'] == 'task-4', + ), + isTrue, + reason: 'cancelTask 已下发原生', + ); + }); + + test('registerTask 通道异常 → channel 归一（注册失败不悬挂）', () async { + messenger.setMockMethodCallHandler(methodChannel, (call) async { + throw PlatformException(code: 'bad_args', message: 'missing url'); + }); + final err = await captureError( + transport.send( + taskId: 'task-5', + url: 'http://obj.test/s', + method: 'PUT', + headers: const {}, + filePath: '/data/e.jpg', + ), + ); + expect((err as TransportException).code, TransportErrorCodes.channel); + }); + + test('并发多任务：事件按 taskId 路由互不串扰', () async { + final f1 = transport.send( + taskId: 'a', + url: 'http://obj.test/1', + method: 'PUT', + headers: const {}, + filePath: '/1.jpg', + ); + final f2 = transport.send( + taskId: 'b', + url: 'http://obj.test/2', + method: 'PUT', + headers: const {}, + filePath: '/2.jpg', + ); + await untilRegistered(); + emitEvent({'event': 'completed', 'taskId': 'b', 'etag': '"b"', 'seq': 1}); + emitEvent({ + 'event': 'failed', + 'taskId': 'a', + 'code': 'NETWORK', + 'seq': 2, + }); + expect((await f2).etag, '"b"'); + final err = await captureError(f1); + expect((err as TransportException).code, TransportErrorCodes.network); + }); + + test('queryTask 透传原生 WorkInfo 概览', () async { + final info = await transport.queryTask('task-9'); + expect(info!['state'], 'RUNNING'); + expect((info['progress'] as Map)['sent'], 5); + }); + }); + + group('DartHttpTransport dio 异常归一', () { + Dio stubDio(Future Function(RequestOptions) router) { + return Dio()..httpClientAdapter = _StubAdapter(router); + } + + test('2xx 返回 TransportResult（etag 头捕获）', () async { + final dio = stubDio((o) async => const (200, '"etag-9"')); + final dartTransport = DartHttpTransport(dio: dio); + final file = _tempFile(); + + final result = await dartTransport.send( + taskId: 't', + url: 'http://obj.test/f.jpg', + method: 'PUT', + headers: const {'Content-Type': 'image/jpeg'}, + filePath: file.path, + ); + expect(result.etag, '"etag-9"'); + dartTransport.dispose(); + }); + + test('403 → TransportRenewRequiredException', () async { + final dio = stubDio((o) async => const (403, null)); + final dartTransport = DartHttpTransport(dio: dio); + final err = await captureError( + dartTransport.send( + taskId: 't', + url: 'http://obj.test/f.jpg', + method: 'PUT', + headers: const {}, + filePath: _tempFile().path, + ), + ); + expect(err, isA()); + dartTransport.dispose(); + }); + + test('连接失败 → network；取消令牌 → cancelled', () async { + final file = _tempFile(); + final dio = stubDio((o) async { + throw DioException( + requestOptions: o, + type: DioExceptionType.connectionError, + ); + }); + final dartTransport = DartHttpTransport(dio: dio); + final err = await captureError( + dartTransport.send( + taskId: 't', + url: 'http://obj.test/f.jpg', + method: 'PUT', + headers: const {}, + filePath: file.path, + ), + ); + expect( + (err as TransportException).code, + TransportErrorCodes.network, + ); + dartTransport.dispose(); + + final cancelDio = stubDio((o) async { + throw DioException( + requestOptions: o, + type: DioExceptionType.cancel, + ); + }); + final cancelTransport = DartHttpTransport(dio: cancelDio); + final token = CancelToken()..cancel(); + final cerr = await captureError( + cancelTransport.send( + taskId: 't', + url: 'http://obj.test/f.jpg', + method: 'PUT', + headers: const {}, + filePath: file.path, + cancelToken: token, + ), + ); + expect( + (cerr as TransportException).code, + TransportErrorCodes.cancelled, + ); + cancelTransport.dispose(); + }); + }); + + group('传输选择（Android 原生 / FORCE_DART_TRANSPORT）', () { + test('选择矩阵：Android→native；强制 Dart / 非 Android→dart', () { + expect(preferNativeTransport(isAndroid: true), isTrue); + expect(preferNativeTransport(isAndroid: false), isFalse); + expect( + preferNativeTransport(isAndroid: true, forceDartTransport: true), + isFalse, + reason: 'FORCE_DART_TRANSPORT 覆盖', + ); + expect( + preferNativeTransport(isAndroid: true, forceDartTransport: false), + isTrue, + ); + }); + + test('缺省解析：VM 测试环境（非 Android）→ DartHttpTransport', () { + final resolved = defaultUploadTransport(); + expect(resolved, isA()); + resolved.dispose(); + }); + + test('kForceDartTransport 常量缺省 false（未传 dart-define 时）', () { + expect(kForceDartTransport, isFalse); + }); + }); +} + +File _tempFile() { + final dir = Directory.systemTemp.createTempSync('transport-'); + return File('${dir.path}/f.bin')..writeAsBytesSync(List.filled(16, 1)); +} + +/// 打桩适配器：router 返回 (status, etag) 记录或抛异常；请求流消费后按 +/// 状态构造响应（headers 携带 ETag）。 +class _StubAdapter implements HttpClientAdapter { + _StubAdapter(this.router); + + final Future Function(RequestOptions options) router; + + @override + Future fetch( + RequestOptions options, + Stream? requestStream, + Future? cancelFuture, + ) async { + if (requestStream != null) { + await requestStream.drain(); + } + final result = await router(options); + if (result is (int, String?)) { + final (status, etag) = result; + return ResponseBody.fromString( + '', + status, + headers: { + if (etag != null) 'etag': [etag], + }, + ); + } + throw result; + } + + @override + void close({bool force = false}) {} +} diff --git a/apps/mobile/test/features/uploads/upload_engine_test.dart b/apps/mobile/test/features/uploads/upload_engine_test.dart index 1540fe5..ac03f51 100644 --- a/apps/mobile/test/features/uploads/upload_engine_test.dart +++ b/apps/mobile/test/features/uploads/upload_engine_test.dart @@ -7,6 +7,7 @@ library; import 'dart:async'; +import 'dart:collection'; import 'dart:io'; import 'package:dio/dio.dart'; @@ -15,6 +16,7 @@ import 'package:photo_live/core/api_client.dart'; import 'package:photo_live/core/device_id.dart'; import 'package:photo_live/core/error_codes.dart'; import 'package:photo_live/core/session_store.dart'; +import 'package:photo_live/features/uploads/native_transport.dart'; import 'package:photo_live/features/uploads/preparer.dart'; import 'package:photo_live/features/uploads/task_state.dart'; import 'package:photo_live/features/uploads/task_store.dart'; @@ -81,12 +83,14 @@ class Harness { int maxParallel = 2, int keepCompletedDays = 14, void Function(String taskId, int sent, int total)? onProgress, + Transport? transport, }) => UploadEngine( store, api, deviceId, - objectDio: objectDio, + objectDio: transport == null ? objectDio : null, + transport: transport, now: () => clock, maxParallel: maxParallel, keepCompletedDays: keepCompletedDays, @@ -246,6 +250,68 @@ class _FixedDeviceIdStore implements DeviceIdStore { Future write(String value) async {} } +/// 一次 send 调用的快照（Task 7 传输委托断言）。 +class ScriptedTransportCall { + const ScriptedTransportCall( + this.taskId, + this.url, + this.method, + this.headers, + this.filePath, + this.metadata, + ); + final String taskId; + final String url; + final String method; + final Map headers; + final String filePath; + final Map? metadata; +} + +/// 脚本化传输替身（P3 计划 Task 7：引擎 × Transport 接口集成）。 +/// +/// 每任务按队列依次弹出手本：[TransportResult] 即成功返回；其余视为 +/// 异常抛出（TransportException 族）。无手本任务默认成功并补发一次 +/// 全量进度回调。 +class ScriptedTransport implements Transport { + final Map> script = {}; + final List calls = []; + bool progressBeforeComplete = true; + + void enqueue(String taskId, List outcomes) => + script.putIfAbsent(taskId, ListQueue.new).addAll(outcomes); + + @override + Future send({ + required String taskId, + required String url, + required String method, + required Map headers, + required String filePath, + Map? metadata, + void Function(int sent, int total)? onProgress, + CancelToken? cancelToken, + }) async { + calls.add( + ScriptedTransportCall(taskId, url, method, headers, filePath, metadata), + ); + final queue = script[taskId]; + final outcome = queue == null || queue.isEmpty + ? const TransportResult() + : queue.removeFirst(); + if (outcome is TransportResult) { + if (progressBeforeComplete) { + onProgress?.call(1024, 1024); + } + return outcome; + } + throw outcome; + } + + @override + void dispose() {} +} + void main() { late Harness h; late Directory tmp; @@ -872,4 +938,163 @@ void main() { // 文件检查在 PUT 阶段（init 已发生、无 PUT/complete）。 expect(h.events, ['init:ct-q']); }); + + group('传输委托（Task 7 Transport 接口）', () { + late ScriptedTransport transport; + + setUp(() { + transport = ScriptedTransport(); + }); + + test('引擎 PUT 委托 Transport：凭证透传 + init/complete 仍在 Dart 引擎', + () async { + final file = photoFile('t1.jpg'); + final progress = >{}; + engine = h.buildEngine( + transport: transport, + onProgress: (id, sent, total) => + progress.putIfAbsent(id, () => []).add(sent), + ); + engine.start('alice', period: const Duration(hours: 1)); + + final seeded = await h.seedTask(file, clientTaskId: 'ct-t1'); + await cycle(); + + final t = await task(seeded.id); + expect(t.state, UploadTaskState.completed); + expect(t.serverUploadId, 'upl1'); + // 原生路径绕过 dio：对象适配器无请求，API 序列只含 init/complete。 + expect(h.objectAdapter.requests, isEmpty); + expect(h.events, ['init:ct-t1', 'complete:upl1']); + + // 委托参数：init 返回的凭证原样透传（URL/方法/头/文件路径/元数据）。 + final call = transport.calls.single; + expect(call.taskId, seeded.id); + expect(call.url, 'http://object.test.local/staging/upl1/pho1.jpg'); + expect(call.method, 'PUT'); + expect(call.headers, {'Content-Type': 'image/jpeg'}); + expect(call.filePath, file.path); + expect(call.metadata?['uploadId'], 'upl1'); + expect(call.metadata?['attempt'], 0); + // 进度回调由 Transport 上报（队列 UI 挂接点不变）。 + expect(progress[seeded.id], [1024]); + }); + + test('原生 needsRenew：引擎 renew 后携带新 URL 重新委托', () async { + final file = photoFile('t2.jpg'); + engine = h.buildEngine(transport: transport); + engine.start('alice', period: const Duration(hours: 1)); + + final seeded = await h.seedTask(file, clientTaskId: 'ct-t2'); + transport.enqueue(seeded.id, [ + const TransportRenewRequiredException(), + const TransportResult(), + ]); + await cycle(); + + final t = await task(seeded.id); + expect(t.state, UploadTaskState.completed); + expect(h.renewCalls, 1); + expect(transport.calls, hasLength(2)); + expect(transport.calls.first.url, endsWith('upl1/pho1.jpg')); + expect(transport.calls.last.url, endsWith('upl1/pho1.jpg?renewed=1')); + expect(h.events, ['init:ct-t2', 'renew:upl1', 'complete:upl1']); + }); + + test('原生 network 失败 → waitingNetwork 退避；恢复后重新委托完成', + () async { + final file = photoFile('t3.jpg'); + engine = h.buildEngine(transport: transport); + engine.start('alice', period: const Duration(hours: 1)); + + final seeded = await h.seedTask(file, clientTaskId: 'ct-t3'); + transport.enqueue(seeded.id, [ + const TransportException(TransportErrorCodes.network), + ]); + await cycle(); + + var t = await task(seeded.id); + expect(t.state, UploadTaskState.waitingNetwork); + expect(t.lastError, ErrorCodes.networkError); + expect(t.attempt, 1); + expect(t.nextRetryAt, h.clock.add(const Duration(seconds: 5))); + + // 到期重试：无手本 → 默认成功。 + h.clock = h.clock.add(const Duration(seconds: 10)); + await cycle(); + t = await task(seeded.id); + expect(t.state, UploadTaskState.completed); + expect(transport.calls, hasLength(2)); + }); + + test('原生 HTTP_4XX(400) → retryableFailed（自动重试退避）', () async { + final file = photoFile('t4.jpg'); + engine = h.buildEngine(transport: transport); + engine.start('alice', period: const Duration(hours: 1)); + + final seeded = await h.seedTask(file, clientTaskId: 'ct-t4'); + transport.enqueue(seeded.id, [ + const TransportException(TransportErrorCodes.http, statusCode: 400), + ]); + await cycle(); + + final t = await task(seeded.id); + expect(t.state, UploadTaskState.retryableFailed); + expect(t.lastError, ErrorCodes.httpError); + expect(t.attempt, 1); + expect(t.nextRetryAt, isNotNull, reason: 'attempt 1 < 3：仍自动重试'); + }); + + test('原生 HTTP 401 → waitingNetwork 短退避（会话语义）', () async { + final file = photoFile('t5.jpg'); + engine = h.buildEngine(transport: transport); + engine.start('alice', period: const Duration(hours: 1)); + + final seeded = await h.seedTask(file, clientTaskId: 'ct-t5'); + transport.enqueue(seeded.id, [ + const TransportException(TransportErrorCodes.http, statusCode: 401), + ]); + await cycle(); + + final t = await task(seeded.id); + expect(t.state, UploadTaskState.waitingNetwork); + expect(t.lastError, ErrorCodes.authTokenExpired); + expect(t.attempt, 0, reason: '会话过期不烧次数'); + expect(t.nextRetryAt, h.clock.add(const Duration(seconds: 30))); + }); + + test('原生 SOURCE_MISSING → 不自动重试的 retryableFailed', () async { + final file = photoFile('t6.jpg'); + engine = h.buildEngine(transport: transport); + engine.start('alice', period: const Duration(hours: 1)); + + final seeded = await h.seedTask(file, clientTaskId: 'ct-t6'); + transport.enqueue(seeded.id, [ + const TransportException(TransportErrorCodes.sourceMissing), + ]); + await cycle(); + + final t = await task(seeded.id); + expect(t.state, UploadTaskState.retryableFailed); + expect(t.lastError, UploadEngine.sourceMissingCode); + expect(t.nextRetryAt, isNull); + }); + + test('原生 SERVER_ERROR(500) → retryableFailed（5xx 语义桶）', () async { + final file = photoFile('t7.jpg'); + engine = h.buildEngine(transport: transport); + engine.start('alice', period: const Duration(hours: 1)); + + final seeded = await h.seedTask(file, clientTaskId: 'ct-t7'); + transport.enqueue(seeded.id, [ + const TransportException(TransportErrorCodes.http, statusCode: 500), + ]); + await cycle(); + + final t = await task(seeded.id); + expect(t.state, UploadTaskState.retryableFailed); + expect(t.lastError, ErrorCodes.internalError); + expect(t.attempt, 1); + }); + }); }