Improve dynamic hint availability and streaming

Expand curated hot topics while keeping slow current-information responses alive through reverse proxies and client idle timeouts.

src/main/kotlin/com/osglab/account/features/content/feed/sources/TopHubHintSource.kt

@@ -122,7 +122,7 @@ class TopHubHintSource(
             ?: return emptyList()
         return (payload["data"] as? JsonArray)
             ?.mapNotNull(JsonElement::objectOrNull)
-            ?.take(3)
+            ?.take(MAXIMUM_HOT_CARDS)
             ?.mapNotNull { item ->
                 val title = item.string("title")
                     ?.takeIf(::isEligibleHotTitle)
@@ -210,4 +210,4 @@ private const val OPEN_HOT = "https://open.tophub.today/hot"
 private const val PAID_HOT = "https://api.tophubdata.com/hot"
 private const val USER_AGENT_HEADER = "User-Agent"
 private const val USER_AGENT = "OSGKeyboard-HintFeed/2.0 (+https://account.osglab.com)"
-private const val MAXIMUM_HOT_CARDS = 3
+private const val MAXIMUM_HOT_CARDS = 10

src/main/kotlin/com/osglab/account/features/gateway/routes/GatewayRoutes.kt

@@ -44,6 +44,7 @@ import io.ktor.http.HttpStatusCode
 import io.ktor.server.application.ApplicationCall
 import io.ktor.server.request.receive
 import io.ktor.server.request.receiveChannel
+import io.ktor.server.response.header
 import io.ktor.server.response.respond
 import io.ktor.server.response.respondBytes
 import io.ktor.server.response.respondBytesWriter
@@ -304,19 +305,36 @@ fun Route.configureGatewayRoutes(
                 }
                 var executionStarted = false
                 try {
+                    call.response.header(HttpHeaders.CacheControl, "no-cache")
+                    call.response.header(X_ACCEL_BUFFERING_HEADER, "no")
                     call.respondBytesWriter(ContentType.Text.EventStream) {
                         executionStarted = true
                         var emittedBytes = 0L
+                        var providerExecutionStarted = false
                         try {
-                            service.executePrepared(prepared, ProviderOutput { bytes ->
-                                emittedBytes = Math.addExact(emittedBytes, bytes.size.toLong())
-                                if (emittedBytes > GatewayLimits.MAX_UPSTREAM_RESPONSE_BYTES) {
-                                    throw GatewayOutputLimitException()
-                                }
-                                writeFully(bytes)
-                                flush()
-                            })
+                            GATEWAY_SSE_STREAM.execute(
+                                provider = { output ->
+                                    providerExecutionStarted = true
+                                    service.executePrepared(prepared, ProviderOutput { bytes ->
+                                        emittedBytes = Math.addExact(emittedBytes, bytes.size.toLong())
+                                        if (emittedBytes > GatewayLimits.MAX_UPSTREAM_RESPONSE_BYTES) {
+                                            throw GatewayOutputLimitException()
+                                        }
+                                        output.emit(bytes)
+                                    })
+                                },
+                                write = { bytes ->
+                                    writeFully(bytes)
+                                    flush()
+                                },
+                            )
                         } catch (failure: Throwable) {
+                            if (failure.isDownstreamClosedWrite()) {
+                                if (!providerExecutionStarted) {
+                                    service.releasePrepared(prepared, failure)
+                                }
+                                return@respondBytesWriter
+                            }
                             if (failure is CancellationException &&
                                 failure !is TimeoutCancellationException
                             ) {
@@ -332,8 +350,14 @@ fun Route.configureGatewayRoutes(
                                 ),
                             )
                             val errorEvent = "event: gateway_error\ndata: $payload\n\n"
-                            writeFully(errorEvent.encodeToByteArray())
-                            flush()
+                            try {
+                                writeFully(errorEvent.encodeToByteArray())
+                                flush()
+                            } catch (writeFailure: Throwable) {
+                                if (!writeFailure.isDownstreamClosedWrite()) {
+                                    throw writeFailure
+                                }
+                            }
                         }
                     }
                 } catch (failure: Throwable) {
@@ -668,10 +692,12 @@ private fun String?.toTextCapability(): GatewayCapability? =
 private val REQUEST_ID = Regex("[A-Za-z0-9_-]{8,64}")
 private const val REQUEST_ID_HEADER = "X-Request-ID"
 private const val IDEMPOTENCY_HEADER = "Idempotency-Key"
+private const val X_ACCEL_BUFFERING_HEADER = "X-Accel-Buffering"
 private val ROUTE_JSON = Json {
     ignoreUnknownKeys = false
     explicitNulls = false
 }
+private val GATEWAY_SSE_STREAM = GatewaySseStream()
 
 private class GatewayBodyTooLargeException : IllegalArgumentException()
 private class GatewayRequestTimeoutException : RuntimeException()

src/main/kotlin/com/osglab/account/features/gateway/routes/GatewaySseStream.kt

@@ -0,0 +1,58 @@
+package com.osglab.account.features.gateway.routes
+
+import com.osglab.account.features.gateway.models.ProviderOutput
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+/**
+ * Keeps a downstream SSE connection active while a provider is still
+ * producing its first event. Writes are serialized because provider output
+ * and heartbeat comments can be emitted by different coroutines.
+ */
+internal class GatewaySseStream(
+    private val heartbeatIntervalMillis: Long = DEFAULT_HEARTBEAT_INTERVAL_MILLIS,
+) {
+    init {
+        require(heartbeatIntervalMillis > 0)
+    }
+
+    suspend fun execute(
+        provider: suspend (ProviderOutput) -> Unit,
+        write: suspend (ByteArray) -> Unit,
+    ) = coroutineScope {
+        val writeMutex = Mutex()
+        suspend fun writeSerialized(bytes: ByteArray) {
+            writeMutex.withLock {
+                write(bytes)
+            }
+        }
+
+        writeSerialized(CONNECTED_COMMENT)
+        val heartbeat = launch {
+            while (true) {
+                delay(heartbeatIntervalMillis)
+                writeSerialized(KEEPALIVE_COMMENT)
+            }
+        }
+
+        try {
+            provider(ProviderOutput(::writeSerialized))
+        } finally {
+            heartbeat.cancelAndJoin()
+        }
+    }
+
+    private companion object {
+        const val DEFAULT_HEARTBEAT_INTERVAL_MILLIS = 10_000L
+        val CONNECTED_COMMENT = ": connected\n\n".encodeToByteArray()
+        val KEEPALIVE_COMMENT = ": keepalive\n\n".encodeToByteArray()
+    }
+}
+
+internal fun Throwable.isDownstreamClosedWrite(): Boolean =
+    generateSequence(this) { it.cause }
+        .any { it::class.simpleName == "ClosedWriteChannelException" }