diff --git a/src/main/kotlin/com/osglab/account/features/content/feed/sources/TopHubHintSource.kt b/src/main/kotlin/com/osglab/account/features/content/feed/sources/TopHubHintSource.kt index 6ea1be6..68b6302 100644 --- a/src/main/kotlin/com/osglab/account/features/content/feed/sources/TopHubHintSource.kt +++ b/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 diff --git a/src/main/kotlin/com/osglab/account/features/gateway/routes/GatewayRoutes.kt b/src/main/kotlin/com/osglab/account/features/gateway/routes/GatewayRoutes.kt index 59dfc76..dd3763b 100644 --- a/src/main/kotlin/com/osglab/account/features/gateway/routes/GatewayRoutes.kt +++ b/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() diff --git a/src/main/kotlin/com/osglab/account/features/gateway/routes/GatewaySseStream.kt b/src/main/kotlin/com/osglab/account/features/gateway/routes/GatewaySseStream.kt new file mode 100644 index 0000000..15dc6c4 --- /dev/null +++ b/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" } diff --git a/src/test/kotlin/com/osglab/account/features/content/feed/sources/HintFeedSourcesTest.kt b/src/test/kotlin/com/osglab/account/features/content/feed/sources/HintFeedSourcesTest.kt index 79c25c1..4a96378 100644 --- a/src/test/kotlin/com/osglab/account/features/content/feed/sources/HintFeedSourcesTest.kt +++ b/src/test/kotlin/com/osglab/account/features/content/feed/sources/HintFeedSourcesTest.kt @@ -20,7 +20,10 @@ class HintFeedSourcesTest : FunSpec({ localDate = LocalDate.parse("2026-08-21"), ) - test("TopHub keeps one daily brief and at most three accurately labelled hot topics") { + test("TopHub keeps one daily brief and at most ten accurately labelled hot topics") { + val eligibleHotItems = (1..11).joinToString(",") { index -> + """{"title":"A useful public topic number $index","sitename":"知乎"}""" + } val client = jsonClient { path -> if (path.endsWith("/daily")) { """ @@ -39,10 +42,7 @@ class HintFeedSourcesTest : FunSpec({ { "data": [ {"title":"震惊！这个标题只是在制造点击","sitename":"知乎"}, - {"title":"A useful public topic one","sitename":"知乎"}, - {"title":"A useful public topic two","sitename":"知乎"}, - {"title":"A useful public topic three","sitename":"知乎"}, - {"title":"A useful public topic four","sitename":"知乎"} + $eligibleHotItems ] } """.trimIndent() @@ -54,16 +54,52 @@ class HintFeedSourcesTest : FunSpec({ val second = source.fetch("zh", context, SETTINGS) first.map { it.id } shouldBe second.map { it.id } - first.size shouldBe 4 + first.size shouldBe 11 first.first().id shouldBe "local-zh-daily-brief" first.map { it.source }.toSet() shouldBe setOf("tophub-daily", "tophub-open-hot") - first.count { it.source == "tophub-open-hot" } shouldBe 3 + first.count { it.source == "tophub-open-hot" } shouldBe 10 first.filter { it.source == "tophub-open-hot" } .all { it.text.orEmpty().startsWith("知乎热议：") } shouldBe true first.none { it.id.startsWith("tophub-history") || it.id.startsWith("tophub-daily-soul") } shouldBe true client.close() } + test("TopHub paid fallback fills the dynamic hot-topic target") { + val paidHotItems = (1..10).joinToString(",") { index -> + """{"title":"A useful paid public topic number $index"}""" + } + val client = HttpClient( + MockEngine { request -> + val content = when { + request.url.encodedPath.endsWith("/daily") -> + """{"data":{"date":"2026-08-21","day":"2026-08-21"}}""" + request.url.host == "open.tophub.today" -> + """ + { + "data": [ + {"title":"A useful open public topic one","sitename":"知乎"}, + {"title":"A useful open public topic two","sitename":"微博"} + ] + } + """.trimIndent() + else -> """{"data":[$paidHotItems]}""" + } + respond( + content = content, + status = HttpStatusCode.OK, + headers = headersOf(HttpHeaders.ContentType, "application/json"), + ) + }, + ) + + val cards = TopHubHintSource(client, "configured-key").fetch("zh", context, SETTINGS) + + cards.size shouldBe 11 + cards.count { it.source == "tophub-open-hot" } shouldBe 2 + cards.count { it.source == "tophub-hot" } shouldBe 8 + client.close() + } + test("Google feeds keep three trends and one safe story per preferred news section") { val client = HttpClient( MockEngine { request -> diff --git a/src/test/kotlin/com/osglab/account/features/gateway/routes/GatewayRequestIdTest.kt b/src/test/kotlin/com/osglab/account/features/gateway/routes/GatewayRequestIdTest.kt index 7c1f1e4..9bfe3f0 100644 --- a/src/test/kotlin/com/osglab/account/features/gateway/routes/GatewayRequestIdTest.kt +++ b/src/test/kotlin/com/osglab/account/features/gateway/routes/GatewayRequestIdTest.kt @@ -28,6 +28,7 @@ import io.ktor.client.request.header import io.ktor.client.request.post import io.ktor.client.request.setBody import io.ktor.http.ContentType +import io.ktor.http.HttpHeaders import io.ktor.http.HttpStatusCode import io.ktor.http.contentType import io.ktor.serialization.kotlinx.json.json @@ -156,6 +157,26 @@ class GatewayRequestIdTest : StringSpec({ GatewayWebSearchMode.REQUIRED } } + + "starts streaming responses with an SSE connection comment" { + val provider = RequestIdProvider() + + testApplication { + application { gatewayTestApplication(provider) } + + val response = client.post("/v1/gateway/llm/ai") { + header("X-Request-ID", "stream-connect-123") + contentType(ContentType.Application.Json) + setBody("""{"input":"hello","stream":true}""") + } + + response.status shouldBe HttpStatusCode.OK + response.headers[HttpHeaders.CacheControl] shouldBe "no-cache" + response.headers["X-Accel-Buffering"] shouldBe "no" + response.bodyAsText() shouldBe ": connected\n\n{\"result\":\"ok\"}" + provider.calls shouldBe 1 + } + } }) private fun io.ktor.server.application.Application.gatewayTestApplication(provider: RequestIdProvider) { diff --git a/src/test/kotlin/com/osglab/account/features/gateway/routes/GatewaySseStreamTest.kt b/src/test/kotlin/com/osglab/account/features/gateway/routes/GatewaySseStreamTest.kt new file mode 100644 index 0000000..fcb1219 --- /dev/null +++ b/src/test/kotlin/com/osglab/account/features/gateway/routes/GatewaySseStreamTest.kt @@ -0,0 +1,79 @@ +package com.osglab.account.features.gateway.routes + +import io.kotest.core.spec.style.StringSpec +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.matchers.collections.shouldContainExactly +import io.kotest.matchers.shouldBe +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.delay +import kotlinx.coroutines.withTimeout + +class GatewaySseStreamTest : StringSpec({ + "writes a connection comment before provider output" { + val writes = mutableListOf() + + GatewaySseStream(heartbeatIntervalMillis = 1_000).execute( + provider = { output -> + output.emit("data: result\n\n".encodeToByteArray()) + }, + write = { writes += it.decodeToString() }, + ) + + writes.shouldContainExactly( + ": connected\n\n", + "data: result\n\n", + ) + } + + "keeps an idle provider connection alive and stops after completion" { + val firstKeepalive = CompletableDeferred() + val writes = mutableListOf() + + withTimeout(1_000) { + GatewaySseStream(heartbeatIntervalMillis = 10).execute( + provider = { output -> + firstKeepalive.await() + output.emit("data: result\n\n".encodeToByteArray()) + }, + write = { + val text = it.decodeToString() + writes += text + if (text == ": keepalive\n\n") { + firstKeepalive.complete(Unit) + } + }, + ) + } + val completedWriteCount = writes.size + delay(30) + + writes.first() shouldBe ": connected\n\n" + writes.contains(": keepalive\n\n") shouldBe true + writes.last() shouldBe "data: result\n\n" + writes.size shouldBe completedWriteCount + } + + "does not start provider execution when the connection comment cannot be written" { + var providerStarted = false + + shouldThrow { + GatewaySseStream(heartbeatIntervalMillis = 1_000).execute( + provider = { + providerStarted = true + }, + write = { throw ClosedWriteChannelException() }, + ) + } + + providerStarted shouldBe false + } + + "recognizes a closed downstream channel through wrapped failures" { + val failure = IllegalStateException("write failed", ClosedWriteChannelException()) + + failure.isDownstreamClosedWrite() shouldBe true + IllegalStateException("provider failed").isDownstreamClosedWrite() shouldBe false + } +}) + +private class ClosedWriteChannelException : RuntimeException()