git.iko.soy/opencode-android
on 06fd41b help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/repo/OutboxStore.kt
168 lines 8.5 kB history raw
1package soy.iko.opencode.data.repo
3import android.content.Context
4import android.util.Log
5import kotlinx.coroutines.Dispatchers
6import kotlinx.coroutines.flow.MutableStateFlow
7import kotlinx.coroutines.flow.StateFlow
8import kotlinx.coroutines.flow.asStateFlow
9import kotlinx.coroutines.sync.Mutex
10import kotlinx.coroutines.sync.withLock
11import kotlinx.coroutines.withContext
12import kotlinx.serialization.Serializable
13import kotlinx.serialization.decodeFromString
14import kotlinx.serialization.encodeToString
15import kotlinx.serialization.json.Json
16import soy.iko.opencode.data.model.ModelRef
17import soy.iko.opencode.util.runCatchingCancellable
18import java.io.File
20/**
21 * One prompt composed while offline (or with no active connection), queued to send when
22 * connectivity returns. Tagged with the [profileId] it was composed for so a reconnect to a
23 * *different* server doesn't misfire it against a session id that only exists on the
24 * original server. [attachments] carry their self-contained base64 data URLs, so a queued
25 * image survives a process restart just like a staged attachment.
26 */
27@Serializable
28data class OutboxMessage(
29 val id: String,
30 val profileId: String,
31 val sessionId: String,
32 val text: String,
33 val attachments: List<PersistedAttachment> = emptyList(),
34 val model: ModelRef? = null,
35 val agent: String? = null,
36 val createdAt: Long,
39/**
40 * A persisted send-queue: prompts the user composed while disconnected, flushed on
41 * reconnect (see [soy.iko.opencode.di.AppContainer.observeOutbox]). Backed by one JSON file
42 * under filesDir (not SharedPreferences) because a queued attachment's base64 payload can be
43 * several megabytes. The in-memory [messages] StateFlow is the source of truth the UI and the
44 * flusher observe; disk is written through under a mutex on every mutation.
45 */
46open class OutboxStore private constructor(
47 private val appContext: Context?,
48 @Suppress("unused") private val testMode: Boolean,
49) {
50 constructor(context: Context) : this(context.applicationContext, false)
51 protected constructor() : this(null, true)
53 private val json = Json { ignoreUnknownKeys = true; encodeDefaults = true }
54 private val ioMutex = Mutex()
56 private val _messages = MutableStateFlow<List<OutboxMessage>>(emptyList())
57 /** All queued messages across sessions/servers, in enqueue order. */
58 open val messages: StateFlow<List<OutboxMessage>> = _messages.asStateFlow()
60 /** Set once the on-disk queue has been read into [_messages] (under [ioMutex]). Until this
61 * is true, a [mutate] must read disk first — otherwise a mutation that beats [load] would
62 * persist from an empty in-memory snapshot and overwrite (or, for a removal, delete) the
63 * persisted queue, losing every message the user composed offline in a previous run. */
64 private var loaded = false
66 private val file: File? by lazy {
67 appContext?.let { File(it.filesDir, "outbox.json") }
68 }
70 /** Read the persisted queue into [messages]. Call once at startup. Idempotent. */
71 open suspend fun load() {
72 ioMutex.withLock { loadLocked() }
73 }
75 /** Read the on-disk queue into memory. MUST be called while holding [ioMutex]. Idempotent
76 * (guarded by [loaded]) so a [mutate] that arrives before [load] can call it to guarantee
77 * disk is read before any write, then a later [load] is a no-op. Returns true once the queue
78 * is in memory (or the file is genuinely absent), false if an existing file couldn't be
79 * read — the caller MUST NOT persist on false or it would overwrite the still-valid queue. */
80 private suspend fun loadLocked(): Boolean {
81 if (loaded) return true
82 val f = file?.takeIf { it.exists() }
83 if (f == null) {
84 // A genuinely absent file is a normal empty queue: mark loaded so mutations proceed.
85 loaded = true
86 return true
87 }
88 // Only mark [loaded] AFTER a successful read. A transient read/decode failure (or a
89 // cancelled read) of an *existing* file must not clobber the still-valid on-disk queue:
90 // leave loaded=false and report failure so the caller skips its write and a later
91 // mutate/load retries the read, instead of persisting an empty snapshot over the user's
92 // offline-composed messages.
93 val diskList = withContext(Dispatchers.IO) {
94 runCatchingCancellable { json.decodeFromString<List<OutboxMessage>>(f.readText()) }
95 }.getOrElse { return false }
96 // Don't clobber items enqueued while the load was in flight: merge by id, keeping
97 // the in-memory version on conflict (it's newer).
98 if (_messages.value.isEmpty()) {
99 _messages.value = diskList.sortedBy { it.createdAt }
100 } else {
101 val known = _messages.value.mapTo(mutableSetOf()) { it.id }
102 val merged = (diskList.filter { it.id !in known } + _messages.value).sortedBy { it.createdAt }
103 _messages.value = merged
104 // The merge reconciled items enqueued while this load was in flight with the
105 // on-disk set; write the merged set back so a later process kill can't drop the
106 // raced enqueue. persist() doesn't take ioMutex, so this won't deadlock.
107 persist(merged)
108 }
109 loaded = true
110 return true
111 }
113 open suspend fun enqueue(message: OutboxMessage) = mutate { current ->
114 current.filterNot { it.id == message.id } + message
115 }
117 open suspend fun remove(id: String) = mutate { current -> current.filterNot { it.id == id } }
119 /** Remove every queued message for [sessionId] (e.g. the user discarded them, or the
120 * session was deleted). */
121 open suspend fun removeForSession(sessionId: String) =
122 mutate { current -> current.filterNot { it.sessionId == sessionId } }
124 private suspend fun mutate(transform: (List<OutboxMessage>) -> List<OutboxMessage>) {
125 // Persist *under* the lock so concurrent mutations (e.g. the flusher removing sent
126 // messages while the UI enqueues a new one) can't race their disk writes: without
127 // this, the older snapshot could land last and drop/resurrect a queued prompt, or two
128 // overlapping writeText/delete calls could tear the file. The lock serializes both the
129 // in-memory update and the write, guaranteeing on-disk order matches memory order.
130 ioMutex.withLock {
131 // Ensure the persisted queue has been read into memory before mutating: a mutation
132 // that races ahead of load() would otherwise transform an empty in-memory snapshot
133 // and persist()/delete() the file, destroying messages queued in a previous run.
134 val loadedOk = loadLocked()
135 val updated = transform(_messages.value).sortedBy { it.createdAt }
136 _messages.value = updated
137 // Only write when the existing on-disk queue was actually read. If the read of an
138 // existing file failed (loadedOk == false), _messages is missing whatever it held, so
139 // persisting `updated` would overwrite the still-valid file with an incomplete
140 // snapshot — losing the user's prior offline-composed messages. Keep the transform in
141 // memory (so the UI reflects the new item) but leave disk untouched; loaded stays
142 // false, so the next mutate/load retries the read and then merges + persists.
143 if (loadedOk) persist(updated)
144 }
145 }
147 private suspend fun persist(list: List<OutboxMessage>) {
148 val f = file ?: return
149 withContext(Dispatchers.IO) {
150 runCatchingCancellable {
151 if (list.isEmpty()) {
152 f.delete()
153 } else {
154 // FIX: write atomically (temp file + rename) like RecentSessionsStore so a
155 // process kill mid-write can't leave a truncated file — a torn file fails to
156 // decode and load() would then drop ALL queued offline messages.
157 val encoded = json.encodeToString(list)
158 val tmp = File(f.parentFile, f.name + ".tmp")
159 tmp.writeText(encoded)
160 if (!tmp.renameTo(f)) {
161 f.writeText(encoded)
162 tmp.delete()
163 }
164 }
165 }.onFailure { Log.w("OutboxStore", "Failed to persist outbox", it) }
166 }
167 }