fix(services): move streaming to own managed channel; drop dead connections on app resume

bmc08gt · bmc08gt · commit fa92fc596fda · 2026-02-11T14:56:17.000-05:00
Signed-off-by: Brandon McAnsh &lt;git@bmcreations.dev&gt;
diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/ActivityFeedApi.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/ActivityFeedApi.kt
@@ -15,9 +15,10 @@ import com.getcode.opencode.model.core.ID
 import io.grpc.ManagedChannel
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
-import java.util.concurrent.TimeUnit
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class ActivityFeedApi @Inject constructor(
     @FlipcashManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/EmailVerificationApi.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/EmailVerificationApi.kt
@@ -12,7 +12,9 @@ import io.grpc.ManagedChannel
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class EmailVerificationApi @Inject constructor(
     @FlipcashManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/PhoneVerificationApi.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/PhoneVerificationApi.kt
@@ -13,7 +13,9 @@ import io.grpc.ManagedChannel
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class PhoneVerificationApi @Inject constructor(
     @FlipcashManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/ProfileApi.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/ProfileApi.kt
@@ -13,10 +13,11 @@ import com.getcode.opencode.internal.network.core.GrpcApi
 import com.getcode.opencode.model.core.ID
 import io.grpc.ManagedChannel
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.withContext
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class ProfileApi @Inject constructor(
     @FlipcashManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/PushApi.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/PushApi.kt
@@ -14,7 +14,9 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 import java.util.concurrent.TimeUnit
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class PushApi @Inject constructor(
     @FlipcashManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/ThirdPartyApi.kt b/services/flipcash/src/main/kotlin/com/flipcash/services/internal/network/api/ThirdPartyApi.kt
@@ -12,7 +12,9 @@ import io.grpc.ManagedChannel
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class ThirdPartyApi @Inject constructor(
     @FlipcashManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/RepositoryFactory.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/RepositoryFactory.kt
@@ -137,7 +137,10 @@ object RepositoryFactory {
             OpenCodeModule::class.java,
         )
 
-        val api = CurrencyApi(module.provideManagedChannel(context, config))
+        val api = CurrencyApi(
+            managedChannel = module.provideManagedChannel(context, config),
+            streamingChannel = module.provideManagedStreamingChannel(context, config)
+        )
         val mintMapper = MintMapper(
             vmMetadataMapper = VmMetadataMapper(),
             launchpadMetadataMapper = LaunchpadMetadataMapper(),
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/inject/OpenCodeModule.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/inject/OpenCodeModule.kt
@@ -9,6 +9,7 @@ import com.getcode.opencode.controllers.TokenController
 import com.getcode.opencode.controllers.TransactionController
 import com.getcode.opencode.exchange.Exchange
 import com.getcode.opencode.internal.annotations.OpenCodeManagedChannel
+import com.getcode.opencode.internal.annotations.OpenCodeManagedStreamingChannel
 import com.getcode.opencode.internal.annotations.OpenCodeProtocol
 import com.getcode.opencode.internal.domain.repositories.InternalAccountRepository
 import com.getcode.opencode.internal.domain.repositories.InternalCurrencyRepository
@@ -44,6 +45,8 @@ import io.grpc.android.AndroidChannelBuilder
 import org.kin.sdk.base.network.api.agora.OkHttpChannelBuilderForcedTls12
 import java.util.concurrent.TimeUnit
 import javax.inject.Singleton
+import kotlin.time.Duration
+import kotlin.time.Duration.Companion.seconds
 
 @InstallIn(SingletonComponent::class)
 @Module
@@ -89,8 +92,28 @@ object OpenCodeModule {
             .usingBuilder(OkHttpChannelBuilderForcedTls12.forAddress(config.baseUrl, config.port))
             .context(context)
             .userAgent(config.userAgent)
-            .keepAliveTime(config.keepAlive.inWholeMilliseconds, TimeUnit.MILLISECONDS)
-            .keepAliveTimeout(config.keepAliveTimeout.inWholeMilliseconds, TimeUnit.MILLISECONDS)
+            .keepAliveWithoutCalls(false)
+            .idleTimeout(5, TimeUnit.MINUTES) // drop idle connections
+            .apply {
+                if (BuildConfig.DEBUG) {
+                    this.intercept(LoggingClientInterceptor())
+                }
+            }
+            .build()
+    }
+
+    @Singleton
+    @Provides
+    @OpenCodeManagedStreamingChannel
+    fun provideManagedStreamingChannel(
+        @ApplicationContext context: Context,
+        @OpenCodeProtocol
+        config: ProtocolConfig,
+    ): ManagedChannel {
+        return AndroidChannelBuilder
+            .usingBuilder(OkHttpChannelBuilderForcedTls12.forAddress(config.baseUrl, config.port))
+            .context(context)
+            .userAgent(config.userAgent)
             .keepAliveWithoutCalls(true)
             .apply {
                 if (BuildConfig.DEBUG) {
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/annotations/OpenCodeManagedChannel.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/annotations/OpenCodeManagedChannel.kt
@@ -10,4 +10,15 @@ import javax.inject.Qualifier
     AnnotationTarget.VALUE_PARAMETER,
     AnnotationTarget.FIELD
 )
-internal annotation class OpenCodeManagedChannel
+internal annotation class OpenCodeManagedChannel
+
+
+@Qualifier
+@Target(
+    AnnotationTarget.FUNCTION,
+    AnnotationTarget.PROPERTY_GETTER,
+    AnnotationTarget.PROPERTY_SETTER,
+    AnnotationTarget.VALUE_PARAMETER,
+    AnnotationTarget.FIELD
+)
+internal annotation class OpenCodeManagedStreamingChannel
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/api/CurrencyApi.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/api/CurrencyApi.kt
@@ -3,28 +3,35 @@ package com.getcode.opencode.internal.network.api
 import com.codeinc.opencode.gen.currency.v1.CurrencyGrpcKt
 import com.codeinc.opencode.gen.currency.v1.CurrencyService
 import com.getcode.opencode.internal.annotations.OpenCodeManagedChannel
+import com.getcode.opencode.internal.annotations.OpenCodeManagedStreamingChannel
+import com.getcode.opencode.internal.model.WindowedRange
 import com.getcode.opencode.internal.network.core.GrpcApi
-import com.getcode.opencode.internal.network.extensions.asProtobufTimestamp
 import com.getcode.opencode.internal.network.extensions.asSolanaAccountId
 import com.getcode.opencode.model.financial.CurrencyCode
-import com.getcode.opencode.internal.model.WindowedRange
 import com.getcode.solana.keys.Mint
 import com.getcode.solana.keys.PublicKey
 import io.grpc.ManagedChannel
-import io.grpc.stub.StreamObserver
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.withContext
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class CurrencyApi @Inject constructor(
     @OpenCodeManagedChannel
     managedChannel: ManagedChannel,
-) : GrpcApi(managedChannel) {
+
+    @OpenCodeManagedStreamingChannel
+    streamingChannel: ManagedChannel,
+) : GrpcApi(managedChannel, streamingChannel) {
 
     private val api = CurrencyGrpcKt.CurrencyCoroutineStub(managedChannel)
         .withWaitForReady()
 
+    private val streamingApi = CurrencyGrpcKt.CurrencyCoroutineStub(streamingChannel)
+        .withWaitForReady()
+
     /**
      * Gets mint account metadata by address
      *
@@ -73,6 +80,6 @@ internal class CurrencyApi @Inject constructor(
     fun streamLiveMintData(
         requests: Flow<CurrencyService.StreamLiveMintDataRequest>
     ): Flow<CurrencyService.StreamLiveMintDataResponse> {
-        return api.streamLiveMintData(requests)
+        return streamingApi.streamLiveMintData(requests)
     }
 }
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/api/MessagingApi.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/api/MessagingApi.kt
@@ -16,7 +16,9 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.withContext
 import java.io.ByteArrayOutputStream
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 internal class MessagingApi @Inject constructor(
     @OpenCodeManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/api/TransactionApi.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/api/TransactionApi.kt
@@ -33,7 +33,9 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.withContext
 import javax.inject.Inject
+import javax.inject.Singleton
 
+@Singleton
 class TransactionApi @Inject constructor(
     @OpenCodeManagedChannel
     managedChannel: ManagedChannel,
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/core/GrpcApi.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/core/GrpcApi.kt
@@ -5,7 +5,10 @@ import androidx.lifecycle.LifecycleOwner
 import androidx.lifecycle.ProcessLifecycleOwner
 import io.grpc.ManagedChannel
 
-abstract class GrpcApi(protected val managedChannel: ManagedChannel): DefaultLifecycleObserver {
+abstract class GrpcApi(protected val managedChannels: List<ManagedChannel>): DefaultLifecycleObserver {
+
+    constructor(managedChannel: ManagedChannel): this(listOf(managedChannel))
+    constructor(vararg managedChannels: ManagedChannel): this(managedChannels.toList())
 
     init {
         ProcessLifecycleOwner.get().lifecycle.addObserver(this)
@@ -18,6 +21,6 @@ abstract class GrpcApi(protected val managedChannel: ManagedChannel): DefaultLif
     }
 
     private fun warmUp() {
-        managedChannel.getState(true)
+        managedChannels.onEach { it.enterIdle() }
     }
 }
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/executors/IntentExecutor.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/executors/IntentExecutor.kt
@@ -21,6 +21,7 @@ import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import java.util.concurrent.TimeoutException
 import kotlin.collections.forEach
 import kotlin.coroutines.resume
 
@@ -42,6 +43,20 @@ class IntentExecutor(
         )
         val streamReference = OcpIntentStreamReference(scope, "submitIntent")
 
+        streamReference.timeoutHandler = {
+            trace(
+                tag = "SubmitIntent",
+                message = "Stream timed out, cleaning up.",
+                type = TraceType.Error
+            )
+            streamReference.cancel()
+            if (!cont.isCompleted) {
+                cont.resume(Result.failure(SubmitIntentError.Other(
+                    cause = TimeoutException("Intent stream timed out")
+                )))
+            }
+        }
+
         streamReference.retain()
 
         scope.launch {
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/services/CurrencyService.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/services/CurrencyService.kt
@@ -10,6 +10,7 @@ import com.getcode.opencode.internal.model.WindowedRange
 import com.getcode.opencode.internal.network.api.CurrencyApi
 import com.getcode.opencode.internal.network.extensions.foldWithSuppression
 import com.getcode.opencode.internal.network.streamers.LiveMintDataStreamer
+import com.getcode.opencode.internal.network.streamers.ManagedMintStream
 import com.getcode.opencode.internal.network.streamers.OcpMintStreamingReference
 import com.getcode.opencode.model.core.errors.GetHistoricalMintDataError
 import com.getcode.opencode.model.core.errors.GetMintsError
@@ -92,7 +93,7 @@ internal class CurrencyService @Inject constructor(
         scope: CoroutineScope,
         mints: List<Mint>,
         onUpdate: (LiveMintDataResponse) -> Unit
-    ): OcpMintStreamingReference {
+    ): ManagedMintStream {
         val streamer = LiveMintDataStreamer(api)
         return streamer.stream(scope = scope, mints = mints, tag = tag,) { update ->
             // save protos for later use
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/streamers/LiveMintDataStreamer.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/internal/network/streamers/LiveMintDataStreamer.kt
@@ -7,6 +7,7 @@ import com.getcode.opencode.internal.bidi.openBidirectionalStream
 import com.getcode.opencode.internal.network.api.CurrencyApi
 import com.getcode.opencode.internal.network.extensions.asSolanaAccountId
 import com.getcode.solana.keys.Mint
+import com.getcode.utils.TraceType
 import com.getcode.utils.trace
 import com.google.protobuf.Timestamp
 import kotlinx.coroutines.CoroutineScope
@@ -23,45 +24,78 @@ internal class LiveMintDataStreamer @Inject constructor(
         mints: List<Mint>,
         tag: String? = null,
         onUpdate: (CurrencyService.StreamLiveMintDataResponse.LiveData) -> Unit,
-    ): OcpMintStreamingReference {
+    ): ManagedMintStream {
         trace(
             tag = "Mint Streamer",
             message = "Opening live data stream"
         )
 
-        val streamReference = OcpMintStreamingReference(scope, "live mint data - $tag")
+        val stream = ManagedMintStream(
+            scope = scope,
+            tag = tag,
+            mints = mints,
+            onUpdate = onUpdate,
+            api = currencyApi,
+        )
+
+        stream.start()
+
+        return stream
+    }
+}
+
+class ManagedMintStream internal constructor(
+    private val scope: CoroutineScope,
+    private val tag: String?,
+    private val mints: List<Mint>,
+    private val onUpdate: (CurrencyService.StreamLiveMintDataResponse.LiveData) -> Unit,
+    private val api: CurrencyApi,
+) {
+    private var activeReference: OcpMintStreamingReference? = null
+    private var isDestroyed = false
+
+    fun start() {
+        connect()
+    }
+
+    private fun connect() {
+        if (isDestroyed) return
 
-        streamReference.retain()
+        val ref = OcpMintStreamingReference(scope, "live mint data - $tag")
+        activeReference = ref
 
-        streamReference.timeoutHandler = {
+        ref.retain()
+
+        ref.timeoutHandler = {
             trace(
                 tag = "Mint Streamer",
-                message = "Live data stream timed out"
-            )
-            openStream(
-                reference = streamReference,
-                mints = mints,
-                onUpdate = onUpdate,
+                message = "Live data stream timed out for $tag, reconnecting...",
+                type = TraceType.Error
             )
+            ref.destroy()
+
+            // Reconnect with a fresh reference
+            connect()
         }
 
         scope.launch {
             try {
-                openStream(
-                    streamReference,
-                    mints = mints,
-                    onUpdate = onUpdate,
-                )
+                openStream(ref, mints, onUpdate)
             } catch (e: Exception) {
                 trace(
                     tag = "Mint Streamer",
-                    message = "Failed to open stream.",
+                    message = "Failed to open stream for $tag.",
                     error = e
                 )
             }
         }
+    }
 
-        return streamReference
+    fun cancel() {
+        isDestroyed = true
+        activeReference?.complete()
+        activeReference?.destroy()
+        activeReference = null
     }
 
     private fun openStream(
@@ -70,7 +104,7 @@ internal class LiveMintDataStreamer @Inject constructor(
         onUpdate: (CurrencyService.StreamLiveMintDataResponse.LiveData) -> Unit,
     ) = openBidirectionalStream(
         streamRef = reference,
-        apiCall = currencyApi::streamLiveMintData,
+        apiCall = api::streamLiveMintData,
         reconnectOnCancelled = true,
         reconnectOnUnavailable = true,
         reconnectOnDeadlineExceeded = true,
diff --git a/services/opencode/src/main/kotlin/com/getcode/opencode/repositories/CurrencyRepository.kt b/services/opencode/src/main/kotlin/com/getcode/opencode/repositories/CurrencyRepository.kt
