git.iko.soy/opencode-android
on 325e359 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/di/AppContainer.kt
406 lines 19 kB history raw
1package soy.iko.opencode.di
3import android.content.Context
4import android.net.ConnectivityManager
5import android.net.Network
6import android.net.NetworkRequest
7import android.util.Log
8import soy.iko.opencode.data.model.BusEvent
9import soy.iko.opencode.data.model.MessagePartUpdated
10import soy.iko.opencode.data.model.MessageUpdated
11import soy.iko.opencode.data.model.ServerProfile
12import soy.iko.opencode.data.model.SessionIdle
13import soy.iko.opencode.data.repo.DraftStore
14import soy.iko.opencode.data.repo.ErrorKind
15import soy.iko.opencode.data.repo.ProfileStore
16import soy.iko.opencode.data.repo.SettingsStore
17import soy.iko.opencode.data.repo.classifyError
18import soy.iko.opencode.data.repo.responseStatusCode
19import soy.iko.opencode.data.network.HttpClientFactory
20import soy.iko.opencode.data.network.NetworkConfig
21import soy.iko.opencode.data.network.OpencodeApiClient
22import soy.iko.opencode.notification.NotificationChannels
23import soy.iko.opencode.util.safeExceptionSummary
24import soy.iko.opencode.notification.SessionNotifications
25import soy.iko.opencode.R
26import soy.iko.opencode.util.runCatchingCancellable
27import kotlinx.coroutines.CoroutineScope
28import kotlinx.coroutines.Dispatchers
29import kotlinx.coroutines.ExperimentalCoroutinesApi
30import kotlinx.coroutines.SupervisorJob
31import kotlinx.coroutines.cancel
32import kotlinx.coroutines.delay
33import kotlinx.coroutines.flow.MutableStateFlow
34import kotlinx.coroutines.flow.StateFlow
35import kotlinx.coroutines.flow.asStateFlow
36import kotlinx.coroutines.flow.emptyFlow
37import kotlinx.coroutines.flow.first
38import kotlinx.coroutines.flow.flatMapLatest
39import kotlinx.coroutines.flow.update
40import kotlinx.coroutines.isActive
41import kotlinx.coroutines.launch
42import kotlinx.coroutines.sync.Mutex
43import kotlinx.coroutines.sync.withLock
44import kotlinx.coroutines.withTimeoutOrNull
45import java.util.concurrent.atomic.AtomicBoolean
47/**
48 * Result of probing a server URL to check connectivity and whether authentication
49 * is required, without persisting a profile or opening a long-lived connection.
50 */
51sealed class ProbeResult {
52 /** Server is reachable and accepts unauthenticated requests. */
53 object Reachable : ProbeResult()
55 /** Server is reachable but rejected the probe with an authentication error. */
56 object NeedsAuth : ProbeResult()
58 /** Server could not be reached (host unresolved, connection refused, timeout, etc.). */
59 data class Unreachable(val error: String) : ProbeResult()
62/**
63 * Hand-written service locator held by [soy.iko.opencode.OpencodeApp]. Owns the
64 * process-wide singletons and the currently active [OpencodeConnection].
65 */
66open class AppContainer private constructor(
67 private val appContext: Context?,
68 private val skipInit: Boolean,
69 @Suppress("unused") private val testMode: Boolean,
70) {
71 constructor(context: Context) : this(context.applicationContext, false, false)
72 protected constructor() : this(null, true, true)
74 private val appScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
75 open val profileStore: ProfileStore by lazy { ProfileStore(appContext!!) }
76 open val settingsStore: SettingsStore by lazy { SettingsStore(appContext!!) }
77 open val draftStore: DraftStore by lazy { DraftStore(appContext!!, appScope) }
79 private val _activeConnection = MutableStateFlow<OpencodeConnection?>(null)
80 open val activeConnection: StateFlow<OpencodeConnection?> = _activeConnection.asStateFlow()
82 /** True while a background auto-reconnect to the most-recent server is in flight. */
83 private val _reconnecting = MutableStateFlow(false)
84 open val reconnecting: StateFlow<Boolean> = _reconnecting.asStateFlow()
86 /**
87 * Set once a startup auto-reconnect succeeds. The server list collects this and
88 * (once) navigates straight to the session list so a returning user skips it.
89 * [consumeAutoConnect] guards against re-firing on screen re-entry.
90 */
91 private val _autoConnectDone = MutableStateFlow(false)
92 open val autoConnectDone: StateFlow<Boolean> = _autoConnectDone.asStateFlow()
93 private val autoConnectConsumed = AtomicBoolean(false)
94 open fun consumeAutoConnect(): Boolean = autoConnectConsumed.compareAndSet(false, true)
96 /** Text shared into the app (ACTION_SEND), prefilled into a session draft. */
97 private val _pendingShare = MutableStateFlow<String?>(null)
98 open val pendingShare: StateFlow<String?> = _pendingShare.asStateFlow()
99 open fun setPendingShare(text: String) { _pendingShare.value = text }
100 open fun consumePendingShare(): String? {
101 val current = _pendingShare.value ?: return null
102 return if (_pendingShare.compareAndSet(current, null)) current else null
103 }
105 /**
106 * A session id to open from an external trigger (a notification tap or a deep link).
107 * The nav host consumes it once a connection is active.
108 */
109 private val _pendingOpenSession = MutableStateFlow<String?>(null)
110 open val pendingOpenSession: StateFlow<String?> = _pendingOpenSession.asStateFlow()
111 open fun requestOpenSession(id: String) { _pendingOpenSession.value = id }
112 open fun consumePendingOpenSession(): String? {
113 val current = _pendingOpenSession.value ?: return null
114 return if (_pendingOpenSession.compareAndSet(current, null)) current else null
115 }
117 /**
118 * The session the user is currently viewing (or null when not in a chat). Drives the
119 * unread tracker: messages arriving for any *other* session mark it unread, and
120 * opening a session clears its unread state.
121 */
122 private val _currentSession = MutableStateFlow<String?>(null)
123 open val currentSession: StateFlow<String?> = _currentSession.asStateFlow()
125 /** Session ids that received activity while not being viewed. */
126 private val _unread = MutableStateFlow<Set<String>>(emptySet())
127 open val unread: StateFlow<Set<String>> = _unread.asStateFlow()
129 open fun setCurrentSession(id: String?) {
130 _currentSession.value = id
131 if (id != null) {
132 _unread.update { it - id }
133 appContext?.let { SessionNotifications.cancel(it, id) }
134 }
135 }
137 /** Restore a session's unread badge after a failed server switch reconnects. */
138 open fun restoreUnread(id: String) {
139 if (id != _currentSession.value) {
140 _unread.update { it + id }
141 }
142 }
144 /** Resolve a localized string — view models reach resources through the container. */
145 open fun string(id: Int, vararg formatArgs: Any): String {
146 if (appContext == null) return ""
147 return if (formatArgs.isEmpty()) appContext.getString(id) else appContext.getString(id, *formatArgs)
148 }
150 /**
151 * Convert a throwable into a user-facing message. Classifies the error by concrete
152 * type (network, timeout, HTTP status) rather than string-matching class names, so
153 * the message reflects what actually went wrong without leaking internal URLs/state.
154 */
155 open fun friendlyError(t: Throwable): String =
156 friendlyErrorFor(t, activeConnection.value?.profile?.baseUrl.orEmpty())
158 /**
159 * Same classification as [friendlyError] but accepts an explicit base URL, so callers
160 * that aren't operating on the active connection (e.g. [probeServer]) can still
161 * produce a user-facing message with the right server address.
162 */
163 open fun friendlyErrorFor(t: Throwable, baseUrl: String): String =
164 when (classifyError(t)) {
165 ErrorKind.NOT_REACHABLE, ErrorKind.NETWORK ->
166 string(R.string.error_not_reachable, baseUrl)
167 ErrorKind.TIMEOUT -> string(R.string.error_timeout)
168 ErrorKind.SERVER -> string(R.string.error_server)
169 // Show only the HTTP status (e.g. 401, 404) — never the request URL,
170 // which a ClientRequestException carries in its message and which we
171 // promised not to leak (it can include auth or internal paths).
172 ErrorKind.CLIENT -> responseStatusCode(t)?.let { string(R.string.error_client_status, it) }
173 ?: string(R.string.error_generic)
174 ErrorKind.UNKNOWN -> string(R.string.error_generic)
175 }
177 init {
178 if (!skipInit) {
179 NotificationChannels.create(appContext!!)
180 observeMessageActivity()
181 appScope.launch { autoConnect() }
182 }
183 }
185 /** Release resources held for the process lifetime (network callback, app scope). */
186 open fun shutdown() {
187 // Cancel the app scope first so any coroutine holding the connection mutex
188 // (e.g. a connect() suspended on profileStore) is cancelled and releases the
189 // mutex. Otherwise the runBlocking below would deadlock waiting for a coroutine
190 // that can't be cancelled until after the runBlocking completes.
191 appScope.cancel()
192 // Bounded runBlocking so a stuck mutex or a slow close() can't ANR the app.
193 // 2 seconds is generous: appScope.cancel() already triggered cancellation of
194 // any coroutine holding the mutex; we're just waiting for it to unwind.
195 kotlinx.coroutines.runBlocking {
196 withTimeoutOrNull(2_000) {
197 connectionMutex.withLock {
198 runCatching { _activeConnection.value?.close() }
199 _activeConnection.value = null
200 }
201 }
202 // If the timeout fired, force-clear the connection so it isn't left dangling.
203 _activeConnection.value = null
204 }
205 val cm = appContext?.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
206 networkCallback?.let { runCatching { cm?.unregisterNetworkCallback(it) } }
207 if (!skipInit) soy.iko.opencode.data.repo.CrashLogger.get(appContext!!).shutdown()
208 draftStore.let { if (!skipInit) it.shutdown() }
209 }
211 /**
212 * Watch the SSE bus for new message activity and badge any session that isn't
213 * currently open. This powers the unread dot on the session list so the user can
214 * tell which conversations got a reply while they were elsewhere.
215 *
216 * Uses [flatMapLatest] so a server switch (which replaces the active connection)
217 * re-subscribes to the new event stream and drops the old one.
218 */
219 @OptIn(ExperimentalCoroutinesApi::class)
220 private fun observeMessageActivity() {
221 appScope.launch {
222 // Retry the observer if it fails for any reason (transient flow exception,
223 // cancellation due to scope issues). Without this, a single failure would
224 // permanently disable unread tracking and completion notifications.
225 while (isActive) {
226 runCatchingCancellable {
227 activeConnection
228 .flatMapLatest { conn -> conn?.events?.events ?: emptyFlow() }
229 .collect { event ->
230 // SessionIdle is not "message activity" (don't badge as unread)
231 // but signals a run finished — fire a completion notification if the
232 // session isn't currently being viewed and was actively streaming.
233 if (event is SessionIdle) {
234 val idleSid = event.properties.sessionID ?: return@collect
235 if (idleSid != _currentSession.value && activeRuns.remove(idleSid)) {
236 notifySessionCompleted(idleSid)
237 }
238 return@collect
239 }
240 val sid = sessionOf(event) ?: return@collect
241 if (sid != _currentSession.value) {
242 _unread.update { it + sid }
243 }
244 // Track sessions actively streaming so we know which idle events
245 // represent a finished run worth notifying about.
246 if (event is MessagePartUpdated || event is MessageUpdated) {
247 synchronized(activeRuns) {
248 // Evict the oldest entry when the cap is reached instead of
249 // clearing everything, so completion notifications for the
250 // remaining active runs still fire on SessionIdle.
251 if (activeRuns.size >= activeRunsLimit) {
252 activeRuns.remove(activeRuns.iterator().next())
253 }
254 activeRuns.add(sid)
255 }
256 }
257 } }.onFailure { Log.w("AppContainer", "Message activity observer failed, will retry: ${safeExceptionSummary(it)}") }
258 if (!isActive) break
259 delay(NetworkConfig.observerRetryDelayMs)
260 }
261 }
262 }
264 /** Session ids currently streaming an assistant run (best-effort, in-process). */
265 private val activeRuns: MutableSet<String> = java.util.Collections.synchronizedSet(mutableSetOf())
267 /** Upper bound on [activeRuns] to prevent unbounded growth if SessionIdle never arrives. */
268 private val activeRunsLimit = NetworkConfig.activeRunsLimit
270 /** Guards connect/disconnect so concurrent callers can't leak an old connection. */
271 private val connectionMutex = Mutex()
273 /** Only persist the `lastUsed` timestamp if it's older than this, to avoid
274 * a DataStore + encrypted-prefs write on every rapid reconnect. */
275 private companion object {
276 const val LAST_USED_SAVE_THRESHOLD_MS = 60_000L
277 }
279 /** Resolve the title for [sessionId] (best-effort) and post a completion notification. */
280 private suspend fun notifySessionCompleted(sessionId: String) {
281 val conn = activeConnection.value ?: return
282 val title = runCatchingCancellable {
283 conn.repository.listSessions().firstOrNull { it.id == sessionId }?.displayTitle
284 }.getOrNull() ?: sessionId
285 appContext?.let { SessionNotifications.postCompleted(it, sessionId, title) }
286 }
288 /** Extract the session id an event pertains to, for message-activity events. */
289 private fun sessionOf(event: BusEvent): String? = sessionOfEvent(event)
291 /**
292 * On cold start, transparently reconnect to the most recently used server so a
293 * returning user lands in their session list instead of the empty server screen.
294 */
295 private suspend fun autoConnect() {
296 val recent = runCatchingCancellable { profileStore.profiles.first() }
297 .getOrDefault(emptyList())
298 .firstOrNull() ?: return
299 _reconnecting.value = true
300 val ok = runCatchingCancellable {
301 val conn = connect(recent)
302 conn.api.ping()
303 }.isSuccess
304 _reconnecting.value = false
305 if (ok) _autoConnectDone.value = true else disconnect()
306 }
308 /**
309 * When the device regains connectivity, nudge the active SSE stream to reconnect
310 * immediately instead of waiting out the exponential backoff.
311 */
312 private val networkCallback: ConnectivityManager.NetworkCallback? = registerNetworkMonitor()
314 private fun registerNetworkMonitor(): ConnectivityManager.NetworkCallback? {
315 if (appContext == null) return null
316 val cm = appContext!!.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
317 ?: return null
318 val callback = object : ConnectivityManager.NetworkCallback() {
319 override fun onAvailable(network: Network) {
320 activeConnection.value?.events?.triggerReconnect()
321 }
322 }
323 runCatching { cm.registerNetworkCallback(NetworkRequest.Builder().build(), callback) }
324 .onFailure { Log.w("AppContainer", "Failed to register network callback", it) }
325 return callback
326 }
328 /** Open (or re-open) a connection to [profile], replacing any current one.
329 * Only persists the updated `lastUsed` timestamp if it's stale (older than a
330 * minute), so rapid reconnect storms don't each trigger a DataStore write. */
331 open suspend fun connect(profile: ServerProfile): OpencodeConnection =
332 connectionMutex.withLock {
333 _activeConnection.value?.close()
334 _activeConnection.value = null
335 activeRuns.clear()
336 _unread.value = emptySet()
337 val now = System.currentTimeMillis()
338 val resolved = profileStore.resolve(profile)
339 val needsSave = (now - resolved.lastUsed) > LAST_USED_SAVE_THRESHOLD_MS
340 val finalProfile = if (needsSave) resolved.copy(lastUsed = now) else resolved
341 if (needsSave) profileStore.save(finalProfile)
342 OpencodeConnection(finalProfile).also { _activeConnection.value = it }
343 }
345 open suspend fun disconnect() {
346 // Acquire the connection mutex so disconnect is serialized with connect().
347 // Previously this was fire-and-forget (launch on appScope), which raced with
348 // a subsequent connect(): connect() could acquire the mutex, create a new
349 // connection, and then the pending disconnect() would close it.
350 connectionMutex.withLock {
351 _activeConnection.value?.close()
352 _activeConnection.value = null
353 }
354 }
356 /**
357 * Probe a server URL (without credentials) to check reachability and detect whether
358 * the server requires authentication. Builds a short-lived HTTP client with the same
359 * base URL normalization as [connect] but no auth, calls the health endpoint, and
360 * classifies the outcome:
361 *
362 * - 2xx → [ProbeResult.Reachable] (no auth needed)
363 * - 401/403 → [ProbeResult.NeedsAuth]
364 * - anything else → [ProbeResult.Unreachable] with a user-facing message
365 *
366 * The probe client is always closed afterwards so no resources linger. This does not
367 * touch the active connection or the profile store.
368 */
369 open suspend fun probeServer(baseUrl: String): ProbeResult {
370 val probeProfile = ServerProfile(
371 id = "probe",
372 label = "",
373 baseUrl = baseUrl.trim(),
374 username = null,
375 password = null,
376 )
377 val client = HttpClientFactory.create(probeProfile)
378 return try {
379 val api = OpencodeApiClient(client)
380 api.ping()
381 ProbeResult.Reachable
382 } catch (e: Exception) {
383 val status = responseStatusCode(e)
384 if (status == 401 || status == 403) {
385 ProbeResult.NeedsAuth
386 } else {
387 ProbeResult.Unreachable(friendlyErrorFor(e, baseUrl))
388 }
389 } finally {
390 runCatching { client.close() }
391 }
392 }
395/**
396 * Extract the session id a message-activity event pertains to, or null for events that
397 * don't describe message activity. This drives the unread badge: any session that
398 * receives activity while it isn't the currently-viewed one gets badged. Extracted as
399 * a top-level `internal` function so the rule is unit-testable without an Android
400 * [Context] (the surrounding [AppContainer] needs one).
401 */
402internal fun sessionOfEvent(event: BusEvent): String? = when (event) {
403 is MessageUpdated -> event.properties.info.sessionID
404 is MessagePartUpdated -> event.properties.part.sessionID ?: event.properties.sessionID
405 else -> null