diff --git a/watch/wearos/app/src/main/AndroidManifest.xml b/watch/wearos/app/src/main/AndroidManifest.xml index c40b166..d92bc61 100644 --- a/watch/wearos/app/src/main/AndroidManifest.xml +++ b/watch/wearos/app/src/main/AndroidManifest.xml @@ -4,6 +4,9 @@ + + + @@ -59,6 +62,11 @@ android:resource="@drawable/tile_preview" /> + + (Channel.UNLIMITED) + + override fun onBind(intent: Intent?): IBinder? = null + + override fun onCreate() { + super.onCreate() + registerNotificationChannel() + } + + override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int { + val notification = buildNotification() + ServiceCompat.startForeground( + this, + NOTIFICATION_ID, + notification, + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.UPSIDE_DOWN_CAKE) { + ServiceInfo.FOREGROUND_SERVICE_TYPE_DATA_SYNC + } else 0, + ) + + // Forward the incoming kick (if any) to the loop. Multiple onStartCommand + // calls just enqueue more kicks — the loop coalesces. + intent?.let { i -> + val kindOrdinal = i.getIntExtra(EXTRA_KICK_KIND, KICK_KIND_NEW_PENDING) + val kick = when (kindOrdinal) { + KICK_KIND_BURST_POLL -> { + val caseId = i.getStringExtra(EXTRA_BURST_CASE_ID).orEmpty() + val deadline = i.getLongExtra(EXTRA_BURST_DEADLINE_MS, 0L) + SyncKick.BurstPoll(caseId, deadline) + } + else -> SyncKick.NewPending + } + kickChannel.trySend(kick) + } + + if (loopJob?.isActive != true) { + loopJob = startLoop() + } + return START_STICKY + } + + private fun startLoop(): Job { + val app = applicationContext as DoctateApp + val loop = SyncWorkerLoop( + pendingStore = app.pendingStore, + uploader = SyncWorkerLoop.Uploader { caseId, recordedAt, audio -> + app.uploadClient.upload(caseId, recordedAt, audio) + }, + caseStore = app.caseStore, + state = app.syncStateProvider, + kickChannel = kickChannel, + ) + return scope.launch { + try { + loop.run() + } catch (e: Throwable) { + Log.w(TAG, "loop crashed: ${e.message}", e) + } + } + } + + override fun onDestroy() { + super.onDestroy() + loopJob?.cancel() + scope.cancel() + kickChannel.close() + } + + private fun registerNotificationChannel() { + if (Build.VERSION.SDK_INT < Build.VERSION_CODES.O) return + val nm = getSystemService(NotificationManager::class.java) ?: return + if (nm.getNotificationChannel(CHANNEL_ID) != null) return + val channel = NotificationChannel( + CHANNEL_ID, + "Doctate sync", + NotificationManager.IMPORTANCE_LOW, + ).apply { + description = "Background uploads of recordings" + setShowBadge(false) + } + nm.createNotificationChannel(channel) + } + + private fun buildNotification(): Notification = + NotificationCompat.Builder(this, CHANNEL_ID) + .setContentTitle("Doctate") + .setContentText("Synchronisiere Aufnahmen") + .setSmallIcon(R.drawable.ic_complication) + .setOngoing(true) + .setPriority(NotificationCompat.PRIORITY_LOW) + .build() + + companion object { + private const val TAG = "SyncService" + private const val NOTIFICATION_ID = 1 + private const val CHANNEL_ID = "doctate_sync" + + const val ACTION_KICK = "com.doctate.watch.sync.KICK" + const val EXTRA_KICK_KIND = "kick_kind" + const val EXTRA_BURST_CASE_ID = "burst_case_id" + const val EXTRA_BURST_DEADLINE_MS = "burst_deadline_ms" + + const val KICK_KIND_NEW_PENDING = 0 + const val KICK_KIND_BURST_POLL = 1 + + fun startIntent(context: Context, kick: SyncKick): Intent { + val intent = Intent(context, SyncService::class.java).apply { + action = ACTION_KICK + when (kick) { + SyncKick.NewPending -> putExtra(EXTRA_KICK_KIND, KICK_KIND_NEW_PENDING) + is SyncKick.BurstPoll -> { + putExtra(EXTRA_KICK_KIND, KICK_KIND_BURST_POLL) + putExtra(EXTRA_BURST_CASE_ID, kick.caseId) + putExtra(EXTRA_BURST_DEADLINE_MS, kick.deadlineMs) + } + } + } + return intent + } + } +} + +private fun CoroutineScope.cancel() { + (coroutineContext[Job] ?: return).cancel() +} diff --git a/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncServiceLauncher.kt b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncServiceLauncher.kt new file mode 100644 index 0000000..ee2b653 --- /dev/null +++ b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncServiceLauncher.kt @@ -0,0 +1,20 @@ +package com.doctate.watch.sync + +import android.content.Context +import android.os.Build + +/** + * [SyncTrigger] that routes kicks into [SyncService]. Wired in + * [com.doctate.watch.DoctateApp.onCreate]; replaces [NoopSyncTrigger] + * once the service is registered in the manifest. + */ +object SyncServiceLauncher : SyncTrigger { + override fun kick(context: Context, kick: SyncKick) { + val intent = SyncService.startIntent(context, kick) + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) { + context.startForegroundService(intent) + } else { + context.startService(intent) + } + } +} diff --git a/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncState.kt b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncState.kt new file mode 100644 index 0000000..6540a64 --- /dev/null +++ b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncState.kt @@ -0,0 +1,21 @@ +package com.doctate.watch.sync + +/** + * Snapshot of what the sync worker is currently doing, surfaced to the UI + * (Tile / List / Detail) so it can render the cloud indicator. Mirrors + * `WorkerSnapshot` in `doctate-client-core::server_sync`. + * + * This is the authoritative view — never trust the queue length you'd + * compute by listing the directory yourself, because the worker can be + * mid-delete-pair and the disk count is briefly off-by-one. + */ +enum class WorkerPhase { Idle, Uploading, Polling } + +data class SyncState( + val phase: WorkerPhase = WorkerPhase.Idle, + val queueLen: Int = 0, + val pendingCaseIds: Set = emptySet(), + val lastSuccessAt: Long? = null, + val lastFailureAt: Long? = null, + val lastFailureReason: String? = null, +) diff --git a/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncStateProvider.kt b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncStateProvider.kt new file mode 100644 index 0000000..b79c3b0 --- /dev/null +++ b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncStateProvider.kt @@ -0,0 +1,36 @@ +package com.doctate.watch.sync + +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.update + +/** + * Owns the [SyncState] StateFlow. Written by [SyncWorkerLoop], read by + * the UI. Decoupled from the Service / Application class so JVM tests + * can drive the worker against a fresh provider. + */ +class SyncStateProvider { + private val _state = MutableStateFlow(SyncState()) + val state: StateFlow = _state.asStateFlow() + + fun setIdle(queueLen: Int, pendingCaseIds: Set) { + _state.update { it.copy(phase = WorkerPhase.Idle, queueLen = queueLen, pendingCaseIds = pendingCaseIds) } + } + + fun setUploading(queueLen: Int, pendingCaseIds: Set) { + _state.update { it.copy(phase = WorkerPhase.Uploading, queueLen = queueLen, pendingCaseIds = pendingCaseIds) } + } + + fun setPolling(queueLen: Int, pendingCaseIds: Set) { + _state.update { it.copy(phase = WorkerPhase.Polling, queueLen = queueLen, pendingCaseIds = pendingCaseIds) } + } + + fun recordSuccess(at: Long) { + _state.update { it.copy(lastSuccessAt = at, lastFailureReason = null) } + } + + fun recordFailure(at: Long, reason: String) { + _state.update { it.copy(lastFailureAt = at, lastFailureReason = reason) } + } +} diff --git a/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncWorkerLoop.kt b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncWorkerLoop.kt new file mode 100644 index 0000000..8a5b66e --- /dev/null +++ b/watch/wearos/app/src/main/java/com/doctate/watch/sync/SyncWorkerLoop.kt @@ -0,0 +1,147 @@ +package com.doctate.watch.sync + +import android.util.Log +import com.doctate.watch.audio.PendingStore +import com.doctate.watch.audio.PendingUpload +import com.doctate.watch.domain.CaseStore +import com.doctate.watch.net.UploadResult +import java.io.File +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.selects.onTimeout +import kotlinx.coroutines.selects.select + +/** + * Pure upload + poll worker, testable on the JVM. Mirrors the loop in + * `doctate-client-core::server_sync::run` — uploads have priority over + * polling (no point asking the server about state if we still have the + * doctor's audio in our pocket). + * + * The poll branch is wired in Phase 6; for now the empty-queue arm just + * idles on the kick channel + interval. + * + * Backoff is purely in-process state. After a process restart it drops + * back to [initialBackoffMs] and tries again immediately — which is the + * right behaviour: "the watch came back to life, give the server one + * fresh chance" is friendlier than "wait 60 s because we left the loop + * mid-backoff". + * + * Architecturally split into [runOnce] (single iteration, suspends only + * on real I/O — testable directly) and [run] (infinite loop, suspends + * indefinitely on the idle branch — used by the production service). + */ +class SyncWorkerLoop( + private val pendingStore: PendingStore, + private val uploader: Uploader, + private val caseStore: CaseStore, + private val state: SyncStateProvider, + private val kickChannel: Channel, + private val initialBackoffMs: Long = INITIAL_BACKOFF_MS, + private val maxBackoffMs: Long = MAX_BACKOFF_MS, + private val pollIntervalMsProvider: () -> Long = { DEFAULT_POLL_INTERVAL_MS }, + private val nowMs: () -> Long = { System.currentTimeMillis() }, +) { + /** Functional seam: takes the same args as `UploadClient.upload`. */ + fun interface Uploader { + suspend fun upload(caseId: String, recordedAt: String, audio: File): UploadResult + } + + /** Outcome of a single iteration. */ + sealed interface Step { + /** A pending upload was processed, regardless of HTTP outcome. */ + data class Uploaded(val result: UploadResult) : Step + /** Queue was empty when this iteration started. */ + data object Idle : Step + } + + private var backoffMs: Long = initialBackoffMs + + /** + * Run a single iteration. Uploads have priority over polling. + * + * - Pending non-empty: do exactly one upload and apply its outcome + * (delete pair + mark synced on success, delete on terminal, keep + * for retry on transient with backoff already incremented). + * - Pending empty: publish Idle and return — caller decides how to + * wait for the next wake-up (kick channel + poll timeout in + * production; immediate return in tests). + */ + suspend fun runOnce(): Step { + val pending = pendingStore.scan() + val next = pending.firstOrNull() + if (next != null) { + publish(WorkerPhase.Uploading, pending) + val result = uploader.upload(next.caseId, next.recordedAt, next.file) + applyResult(next, result) + return Step.Uploaded(result) + } + publish(WorkerPhase.Idle, emptyList()) + return Step.Idle + } + + /** Infinite loop variant for production. Idle branch waits on kick or poll interval. */ + @OptIn(ExperimentalCoroutinesApi::class) + suspend fun run() { + while (currentCoroutineContext().isActive) { + when (runOnce()) { + is Step.Uploaded -> Unit // proceed to next iteration immediately + Step.Idle -> waitForKickOrTimeout() + } + } + } + + @OptIn(ExperimentalCoroutinesApi::class) + private suspend fun waitForKickOrTimeout() { + select { + onTimeout(pollIntervalMsProvider()) {} + kickChannel.onReceiveCatching { /* drained, just wake up */ } + } + } + + private suspend fun applyResult(next: PendingUpload, result: UploadResult) { + when (result) { + is UploadResult.Success -> { + pendingStore.deletePair(next.file) + caseStore.markSynced(next.caseId) + state.recordSuccess(nowMs()) + backoffMs = initialBackoffMs + Log.i(TAG, "upload ok caseId=${next.caseId.take(8)} status=${result.status}") + } + is UploadResult.Transient -> { + state.recordFailure(nowMs(), result.reason) + Log.w(TAG, "transient: ${result.reason} → backoff ${backoffMs}ms") + delay(backoffMs) + backoffMs = (backoffMs * 2).coerceAtMost(maxBackoffMs) + } + is UploadResult.Terminal -> { + // Server says "do not retry" (4xx) — drop the pair so the queue can drain. + Log.w(TAG, "terminal: ${result.reason} → dropping caseId=${next.caseId.take(8)}") + pendingStore.deletePair(next.file) + state.recordFailure(nowMs(), result.reason) + backoffMs = initialBackoffMs + } + } + } + + private fun publish(phase: WorkerPhase, pending: List) { + val ids = pending.map { it.caseId }.toSet() + when (phase) { + WorkerPhase.Idle -> state.setIdle(pending.size, ids) + WorkerPhase.Uploading -> state.setUploading(pending.size, ids) + WorkerPhase.Polling -> state.setPolling(pending.size, ids) + } + } + + /** Test-only accessor. */ + internal val currentBackoffMs: Long get() = backoffMs + + companion object { + private const val TAG = "SyncWorkerLoop" + const val INITIAL_BACKOFF_MS: Long = 2_000L + const val MAX_BACKOFF_MS: Long = 60_000L + const val DEFAULT_POLL_INTERVAL_MS: Long = 30_000L + } +} diff --git a/watch/wearos/app/src/test/java/com/doctate/watch/sync/SyncWorkerLoopTest.kt b/watch/wearos/app/src/test/java/com/doctate/watch/sync/SyncWorkerLoopTest.kt new file mode 100644 index 0000000..858c4f9 --- /dev/null +++ b/watch/wearos/app/src/test/java/com/doctate/watch/sync/SyncWorkerLoopTest.kt @@ -0,0 +1,147 @@ +package com.doctate.watch.sync + +import com.doctate.watch.audio.PendingStore +import com.doctate.watch.audio.PendingUpload +import com.doctate.watch.net.UploadResult +import com.doctate.watch.testfixtures.InMemoryCaseStore +import com.google.common.truth.Truth.assertThat +import java.io.File +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.test.runTest +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder + +@OptIn(ExperimentalCoroutinesApi::class) +class SyncWorkerLoopTest { + @get:Rule val tempDir = TemporaryFolder() + + private fun newPendingStore(): PendingStore = PendingStore(File(tempDir.root, "unsynced")) + + private fun PendingStore.seed(caseId: String, recordedAt: String): File { + val audio = audioPathFor(caseId, recordedAt) + audio.writeBytes(byteArrayOf(0x66, 0x74, 0x79, 0x70)) + writeSidecar(PendingUpload(caseId, recordedAt, audio)) + return audio + } + + @Test fun success_deletes_pair_marks_synced_and_resets_backoff() = runTest { + val store = InMemoryCaseStore() + val caseId = store.createLocal() + val pending = newPendingStore() + val audio = pending.seed(caseId, "2026-04-26T10:00:00Z") + val state = SyncStateProvider() + val uploader = SyncWorkerLoop.Uploader { c, t, _ -> UploadResult.Success(c, t, "received") } + val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED)) + + val step = loop.runOnce() + assertThat(step).isInstanceOf(SyncWorkerLoop.Step.Uploaded::class.java) + assertThat(audio.exists()).isFalse() + assertThat(pending.sidecarPathFor(audio).exists()).isFalse() + assertThat(store.casesFlow.value.first { it.caseId == caseId }.syncedToServer).isTrue() + assertThat(state.state.value.lastSuccessAt).isNotNull() + } + + @Test fun fifo_order_processes_oldest_recordedAt_first() = runTest { + val store = InMemoryCaseStore() + val pending = newPendingStore() + pending.seed("case-newer", "2026-04-26T12:00:00Z") + pending.seed("case-older", "2026-04-26T10:00:00Z") + + val processed = mutableListOf() + val state = SyncStateProvider() + val uploader = SyncWorkerLoop.Uploader { c, t, _ -> + processed += c + UploadResult.Success(c, t, "received") + } + val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED)) + + loop.runOnce() // first + loop.runOnce() // second + assertThat(processed).containsExactly("case-older", "case-newer").inOrder() + } + + @Test fun transient_keeps_pair_and_increments_backoff_capped() = runTest { + val store = InMemoryCaseStore() + val pending = newPendingStore() + val audio = pending.seed("case-x", "2026-04-26T10:00:00Z") + + val state = SyncStateProvider() + val uploader = SyncWorkerLoop.Uploader { _, _, _ -> UploadResult.Transient("HTTP 503") } + val loop = SyncWorkerLoop( + pending, uploader, store, state, Channel(Channel.UNLIMITED), + initialBackoffMs = 100L, + maxBackoffMs = 800L, + ) + + // Each runOnce in a transient state delays internally — runTest uses + // virtual time, so this completes instantly under the test scheduler. + loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(200L) + loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(400L) + loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(800L) + loop.runOnce(); assertThat(loop.currentBackoffMs).isEqualTo(800L) // capped + + assertThat(audio.exists()).isTrue() + assertThat(pending.sidecarPathFor(audio).exists()).isTrue() + assertThat(state.state.value.lastFailureReason).contains("HTTP 503") + } + + @Test fun terminal_drops_pair_no_retry() = runTest { + val store = InMemoryCaseStore() + val pending = newPendingStore() + val audio = pending.seed("case-terminal", "2026-04-26T10:00:00Z") + + var calls = 0 + val state = SyncStateProvider() + val uploader = SyncWorkerLoop.Uploader { _, _, _ -> + calls++ + UploadResult.Terminal("HTTP 401") + } + val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED)) + + loop.runOnce() + // Second runOnce sees an empty queue — terminal must have dropped the pair. + val nextStep = loop.runOnce() + assertThat(nextStep).isEqualTo(SyncWorkerLoop.Step.Idle) + assertThat(audio.exists()).isFalse() + assertThat(pending.sidecarPathFor(audio).exists()).isFalse() + assertThat(calls).isEqualTo(1) + } + + @Test fun idle_branch_publishes_zero_queue() = runTest { + val store = InMemoryCaseStore() + val pending = newPendingStore() + val state = SyncStateProvider() + val uploader = SyncWorkerLoop.Uploader { _, _, _ -> error("must not be called") } + val loop = SyncWorkerLoop(pending, uploader, store, state, Channel(Channel.UNLIMITED)) + + val step = loop.runOnce() + assertThat(step).isEqualTo(SyncWorkerLoop.Step.Idle) + assertThat(state.state.value.phase).isEqualTo(WorkerPhase.Idle) + assertThat(state.state.value.queueLen).isEqualTo(0) + } + + @Test fun success_after_transient_resets_backoff() = runTest { + val store = InMemoryCaseStore() + val pending = newPendingStore() + pending.seed("case-y", "2026-04-26T10:00:00Z") + + var attempt = 0 + val state = SyncStateProvider() + val uploader = SyncWorkerLoop.Uploader { c, t, _ -> + attempt++ + if (attempt == 1) UploadResult.Transient("server warming up") + else UploadResult.Success(c, t, "received") + } + val loop = SyncWorkerLoop( + pending, uploader, store, state, Channel(Channel.UNLIMITED), + initialBackoffMs = 100L, maxBackoffMs = 800L, + ) + + loop.runOnce() // transient → backoff doubles to 200 + assertThat(loop.currentBackoffMs).isEqualTo(200L) + loop.runOnce() // success → backoff resets + assertThat(loop.currentBackoffMs).isEqualTo(100L) + } +}