feat(ccgo): 适配用量接口新格式 + 修复 nextCursor null 同步死循环

- 解析迁移到 companion(纯函数可单测),cost 改读 meta.totalCost(美元数值),
  旧格式 creditsTotal 兼容回退
- nextCursor 由服务端直接提供,缺失/null 时回退按末条自编码
- 修复 JSON null 被解析成字符串 "null" 导致 fullSync 死循环(JsonNull 拦截)
- fullSync/incrementalSync 增加游标未前进 + 页数上限双防御
- 新增 8 个解析测试(含 null 游标回归用例)
This commit is contained in:
WaterRain 2026-08-30 05:49:45 +00:00
parent f1ed643b80
commit 6cd54a3a8f
No known key found for this signature in database
3 changed files with 289 additions and 92 deletions

View File

@ -7,6 +7,7 @@ import com.rainy.token.domain.service.ServiceType
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonNull
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.jsonArray
import kotlinx.serialization.json.jsonObject
@ -35,7 +36,6 @@ class CommandCodeUsageRepository(
private val okHttpClient: OkHttpClient,
private val credentialRepository: CredentialRepository
) {
private val json = Json { ignoreUnknownKeys = true }
private val apiBase = "https://api.commandcode.ai"
companion object {
@ -44,6 +44,121 @@ class CommandCodeUsageRepository(
const val COST_DENOM = 100_000_000L
/** CCGO 用量数据在 UsageCache 中的 workspaceId 区分键 */
const val CCGO_WORKSPACE_ID = "commandcode"
private val json = Json { ignoreUnknownKeys = true }
// ===== 纯解析逻辑(无网络依赖,可直接单测)=====
/**
* 解析 JSON 响应。
*
* 后端 2026-08 起改为新格式:
* - 顶层新增 nextCursor / limit / periodBasis / window;
* 分页游标由服务端 nextCursor 直接给出(缺失时回退按末条自编码,兼容旧格式)。
* - 每条记录新增 durationTotal / status / message / type / mode;
* tokensTotal / creditsTotal 已移除。
* - meta 新增 totalCost / inputCost / outputCost / cacheCost / traceId;
* provider / cacheReadInputTokens 已移除。
*/
internal fun parseUsageResponse(body: String): Pair<List<UsageRecord>, String?> {
val root = json.parseToJsonElement(body).jsonObject
val usages = root["usages"]?.jsonArray ?: return emptyList<UsageRecord>() to null
val records = usages.mapNotNull { elem ->
parseUsageObject(elem.jsonObject)
}
// 新格式:优先用服务端游标;缺失或显式 null 时回退按末条自编码(兼容旧格式)
// 注意:JSON null 的 jsonPrimitive.content 是字符串 "null",必须用 is JsonNull 拦截,
// 否则会把 "null" 当游标传给服务端(被忽略→返回第一页),导致 fullSync 死循环。
val serverCursor = root["nextCursor"]?.takeIf { it !is JsonNull }
?.jsonPrimitive?.content?.takeIf { it.isNotEmpty() }
val nextCursor = serverCursor ?: if (records.size >= PAGE_SIZE) {
val last = records.last()
encodeCursor(last.id, last.timeCreated)
} else null
return records to nextCursor
}
/**
* 解析单条 usage 对象。
*
* cost 来源:新格式 meta.totalCost(美元数值);旧格式顶层 creditsTotal(兼容)。
* provider:新格式已移除,缺失时为空字符串;旧格式 meta.provider 兼容读取。
* cacheReadInputTokens:新格式已移除,缺失时按 0(tokensIn 不再拆分缓存命中)。
*/
internal fun parseUsageObject(obj: JsonObject): UsageRecord? {
val id = obj["id"]?.jsonPrimitive?.content ?: return null
val createdAt = obj["createdAt"]?.jsonPrimitive?.content ?: return null
val timeCreated = parseIsoDate(createdAt) ?: return null
val tokensIn = obj["tokensIn"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0L
val tokensOut = obj["tokensOut"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0L
val meta = obj["meta"]?.jsonObject
val creditsTotal = meta?.get("totalCost")?.jsonPrimitive?.content?.toDoubleOrNull()
?: obj["creditsTotal"]?.jsonPrimitive?.content?.toDoubleOrNull()
?: 0.0
val cost = (creditsTotal * COST_DENOM).toLong()
val model = meta?.get("model")?.jsonPrimitive?.content ?: ""
val provider = meta?.get("provider")?.jsonPrimitive?.content ?: ""
val cacheReadInputTokens = meta?.get("cacheReadInputTokens")?.jsonPrimitive?.content?.toLongOrNull() ?: 0L
// CCGO 的 tokensIn 是总输入(缓存命中 + 未命中),按 OCGO 惯例拆分为 inputTokens(未命中)和 cacheReadTokens(命中)
// 新格式无缓存 token 数时无法拆分,inputTokens 直接取 tokensIn
val inputMissTokens = (tokensIn - cacheReadInputTokens).coerceAtLeast(0)
return UsageRecord(
id = id,
workspaceId = CCGO_WORKSPACE_ID,
timeCreated = timeCreated,
timeUpdated = timeCreated,
model = model,
provider = provider,
inputTokens = inputMissTokens,
outputTokens = tokensOut,
reasoningTokens = 0L,
cacheReadTokens = cacheReadInputTokens,
cacheWrite5mTokens = 0L,
cacheWrite1hTokens = 0L,
cost = cost,
keyId = "",
sessionId = "",
enrichmentPlan = ""
)
}
private fun parseIsoDate(iso: String): Long? {
// 处理末尾 Z 和时区偏移
val normalized = iso
.replace("Z", "X")
.replace(Regex("""[+-]\d{2}:\d{2}$"""), "X")
return try {
// SimpleDateFormat 非线程安全,每次创建新实例
val sdf = java.text.SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'X'", Locale.US).apply {
timeZone = TimeZone.getTimeZone("UTC")
}
sdf.parse(normalized)?.time
?: run {
val sdf2 = java.text.SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'X'", Locale.US).apply {
timeZone = TimeZone.getTimeZone("UTC")
}
sdf2.parse(normalized)?.time
}
} catch (_: Exception) { null }
}
/** 从记录信息编码为 base64 cursor */
private fun encodeCursor(id: String, timeCreated: Long): String {
val sdf = SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", Locale.US).apply {
timeZone = TimeZone.getTimeZone("UTC")
}
val createdAt = sdf.format(java.util.Date(timeCreated))
val cursorJson = """{"createdAt":"$createdAt","id":"$id"}"""
return Base64.getUrlEncoder().withoutPadding().encodeToString(cursorJson.toByteArray())
}
}
private suspend fun getCookieHeader(): String {
@ -113,95 +228,4 @@ class CommandCodeUsageRepository(
Result.success(records)
}
}
/**
* 解析 JSON 响应。
*/
private fun parseUsageResponse(body: String): Pair<List<UsageRecord>, String?> {
val root = json.parseToJsonElement(body).jsonObject
val usages = root["usages"]?.jsonArray ?: return emptyList<UsageRecord>() to null
val records = usages.mapNotNull { elem ->
val obj = elem.jsonObject
parseUsageObject(obj)
}
// 从最后一条记录计算下一页 cursor
val nextCursor = if (records.size >= PAGE_SIZE) {
val last = records.last()
encodeCursor(last.id, last.timeCreated)
} else null
return records to nextCursor
}
private fun parseUsageObject(obj: JsonObject): UsageRecord? {
val id = obj["id"]?.jsonPrimitive?.content ?: return null
val createdAt = obj["createdAt"]?.jsonPrimitive?.content ?: return null
val timeCreated = parseIsoDate(createdAt) ?: return null
val tokensIn = obj["tokensIn"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0L
val tokensOut = obj["tokensOut"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0L
val tokensTotal = obj["tokensTotal"]?.jsonPrimitive?.content?.toLongOrNull() ?: 0L
val creditsTotal = obj["creditsTotal"]?.jsonPrimitive?.content?.toDoubleOrNull() ?: 0.0
val cost = (creditsTotal * COST_DENOM).toLong()
val meta = obj["meta"]?.jsonObject
val model = meta?.get("model")?.jsonPrimitive?.content ?: ""
val provider = meta?.get("provider")?.jsonPrimitive?.content ?: ""
val cacheReadInputTokens = meta?.get("cacheReadInputTokens")?.jsonPrimitive?.content?.toLongOrNull() ?: 0L
// CCGO 的 tokensIn 是总输入(缓存命中 + 未命中),按 OCGO 惯例拆分为 inputTokens(未命中)和 cacheReadTokens(命中)
val inputMissTokens = (tokensIn - cacheReadInputTokens).coerceAtLeast(0)
return UsageRecord(
id = id,
workspaceId = CCGO_WORKSPACE_ID,
timeCreated = timeCreated,
timeUpdated = timeCreated,
model = model,
provider = provider,
inputTokens = inputMissTokens,
outputTokens = tokensOut,
reasoningTokens = 0L,
cacheReadTokens = cacheReadInputTokens,
cacheWrite5mTokens = 0L,
cacheWrite1hTokens = 0L,
cost = cost,
keyId = "",
sessionId = "",
enrichmentPlan = ""
)
}
private fun parseIsoDate(iso: String): Long? {
// 处理末尾 Z 和时区偏移
val normalized = iso
.replace("Z", "X")
.replace(Regex("""[+-]\d{2}:\d{2}$"""), "X")
return try {
// SimpleDateFormat 非线程安全,每次创建新实例
val sdf = java.text.SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'X'", Locale.US).apply {
timeZone = TimeZone.getTimeZone("UTC")
}
sdf.parse(normalized)?.time
?: run {
val sdf2 = java.text.SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'X'", Locale.US).apply {
timeZone = TimeZone.getTimeZone("UTC")
}
sdf2.parse(normalized)?.time
}
} catch (_: Exception) { null }
}
/** 从记录信息编码为 base64 cursor */
private fun encodeCursor(id: String, timeCreated: Long): String {
val sdf = SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", Locale.US).apply {
timeZone = TimeZone.getTimeZone("UTC")
}
val createdAt = sdf.format(java.util.Date(timeCreated))
val cursorJson = """{"createdAt":"$createdAt","id":"$id"}"""
return Base64.getUrlEncoder().withoutPadding().encodeToString(cursorJson.toByteArray())
}
}

View File

@ -23,12 +23,18 @@ class SyncCommandCodeUsageUseCase @Inject constructor(
private val usageRepoProvider: Provider<CommandCodeUsageRepository>,
private val cacheProvider: Provider<UsageCache>
) {
/** 防御性页数上限:正常窗口(1 天 ≤ 数十页)远不会触及,防止游标异常导致死循环 */
private companion object {
const val MAX_PAGES = 500
}
suspend fun fullSync(): Result<SyncResult> {
val repo = usageRepoProvider.get()
val cache = cacheProvider.get()
var cursor: String? = null
var totalInserted = 0
val errors = mutableListOf<String>()
var pages = 0
while (true) {
val pageResult = repo.fetchPage(cursor)
@ -38,6 +44,7 @@ class SyncCommandCodeUsageUseCase @Inject constructor(
}
val (records, nextCursor) = pageResult.getOrThrow()
if (records.isEmpty()) break
if (nextCursor == cursor) break // 游标未前进,防死循环
val before = cache.count()
cache.insertAll(records)
@ -45,6 +52,7 @@ class SyncCommandCodeUsageUseCase @Inject constructor(
if (records.size < CommandCodeUsageRepository.PAGE_SIZE) break
cursor = nextCursor
if (++pages >= MAX_PAGES) break // 防御性上限,正常窗口不会触及
}
return if (errors.isEmpty()) Result.success(SyncResult(inserted = totalInserted))
@ -56,6 +64,7 @@ class SyncCommandCodeUsageUseCase @Inject constructor(
val cache = cacheProvider.get()
var cursor: String? = null
var totalInserted = 0
var pages = 0
while (true) {
val pageResult = repo.fetchPage(cursor)
@ -63,6 +72,7 @@ class SyncCommandCodeUsageUseCase @Inject constructor(
val (records, nextCursor) = pageResult.getOrThrow()
if (records.isEmpty()) break
if (nextCursor == cursor) break // 游标未前进,防死循环
// 按 workspace 过滤本地已有 ID,避免跨 workspace 碰撞
val workspaceId = records.firstOrNull()?.workspaceId ?: CommandCodeUsageRepository.CCGO_WORKSPACE_ID
@ -76,6 +86,7 @@ class SyncCommandCodeUsageUseCase @Inject constructor(
if (records.size < CommandCodeUsageRepository.PAGE_SIZE) break
cursor = nextCursor
if (++pages >= MAX_PAGES) break // 防御性上限
}
return Result.success(SyncResult(inserted = totalInserted))

View File

@ -0,0 +1,162 @@
package com.rainy.token.data.repository
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Test
import java.util.Base64
/**
* Unit tests for [CommandCodeUsageRepository] 的纯解析逻辑。
*
* 覆盖 2026-08 后端新格式(cost 改由 meta.totalCost 提供、服务端 nextCursor
* 分页、provider / creditsTotal / cacheReadInputTokens 移除)以及旧格式兼容。
* 全部为纯 JVM 测试,无网络。
*/
class CommandCodeUsageRepositoryTest {
/** 生成一条 usage 记录 JSON(新格式字段)。 */
private fun usageJson(
id: String,
tokensIn: Long,
tokensOut: Long,
totalCost: Number,
provider: String? = null,
cacheReadInputTokens: Long? = null
): String = buildString {
append("{\"id\":\"").append(id).append("\",")
append("\"createdAt\":\"2026-08-28T20:27:30.400Z\",")
append("\"tokensIn\":\"").append(tokensIn).append("\",")
append("\"tokensOut\":\"").append(tokensOut).append("\",")
append("\"durationTotal\":\"2960\",")
append("\"status\":\"completed\",")
append("\"message\":null,")
append("\"meta\":{")
append("\"totalCost\":").append(totalCost).append(",")
append("\"inputCost\":0.001,")
append("\"outputCost\":0.0002,")
append("\"cacheCost\":0.0003,")
append("\"model\":\"deepseek/deepseek-v4-flash\",")
append("\"traceId\":\"trace-1\"")
if (provider != null) append(",\"provider\":\"").append(provider).append("\"")
if (cacheReadInputTokens != null) append(",\"cacheReadInputTokens\":").append(cacheReadInputTokens)
append("},")
append("\"type\":\"api\",")
append("\"mode\":\"agent\"}")
}
private fun newFormatBody(
totalCost: Number = 0.001651536,
tokensIn: Long = 46314,
tokensOut: Long = 118,
provider: String? = null,
cacheReadInputTokens: Long? = null,
nextCursor: String? = null,
count: Int = 1
): String = buildString {
append("""{"usages":[""")
for (i in 0 until count) {
if (i > 0) append(",")
append(usageJson("rec-$i", tokensIn + i, tokensOut, totalCost, provider, cacheReadInputTokens))
}
append("]")
if (nextCursor != null) append(""","nextCursor":"$nextCursor"""")
append("""}""")
}
@Test
fun `new format parses cost from meta totalCost`() {
val (records, next) = CommandCodeUsageRepository.parseUsageResponse(
newFormatBody(totalCost = 0.001651536, tokensIn = 46314, tokensOut = 118)
)
assertEquals(1, records.size)
val r = records.first()
assertEquals((0.001651536 * CommandCodeUsageRepository.COST_DENOM).toLong(), r.cost)
assertEquals(46314L, r.inputTokens)
assertEquals(118L, r.outputTokens)
assertEquals(0L, r.cacheReadTokens)
assertEquals("", r.provider)
assertEquals("deepseek/deepseek-v4-flash", r.model)
assertNull(next)
}
@Test
fun `new format accepts integer totalCost`() {
val (records, _) = CommandCodeUsageRepository.parseUsageResponse(
newFormatBody(totalCost = 0)
)
assertEquals(1, records.size)
assertEquals(0L, records.first().cost)
}
@Test
fun `new format uses server nextCursor`() {
val (_, next) = CommandCodeUsageRepository.parseUsageResponse(
newFormatBody(nextCursor = "eyJjdXJzb3I")
)
assertEquals("eyJjdXJzb3I", next)
}
@Test
fun `falls back to computed cursor on full page without server cursor`() {
val (records, next) = CommandCodeUsageRepository.parseUsageResponse(
newFormatBody(count = CommandCodeUsageRepository.PAGE_SIZE, totalCost = 0.001, nextCursor = null)
)
assertEquals(CommandCodeUsageRepository.PAGE_SIZE, records.size)
assertNotNull(next)
val decoded = String(Base64.getUrlDecoder().decode(next!!))
assertTrue(decoded.contains("\"id\":\"rec-${CommandCodeUsageRepository.PAGE_SIZE - 1}\""))
}
@Test
fun `empty page when usages missing`() {
val (records, next) = CommandCodeUsageRepository.parseUsageResponse("""{"limit":100}""")
assertTrue(records.isEmpty())
assertNull(next)
}
@Test
fun `json null nextCursor is treated as no cursor on small page`() {
// 服务端满页/末页可能返回显式 "nextCursor":null,旧实现会解析成字符串 "null" 导致死循环
val body = """{"usages":[${usageJson("rec-0", 100, 10, 0.001)}],"nextCursor":null}"""
val (records, next) = CommandCodeUsageRepository.parseUsageResponse(body)
assertEquals(1, records.size)
assertNull(next)
}
@Test
fun `json null nextCursor falls back to computed cursor on full page`() {
// 关键回归场景:满页 + nextCursor:null → 必须回退自编码游标,而不是返回 "null" 字符串
val body = buildString {
append("""{"usages":[""")
for (i in 0 until CommandCodeUsageRepository.PAGE_SIZE) {
if (i > 0) append(",")
append(usageJson("rec-$i", 100L + i, 10L, 0.001))
}
append("""],"nextCursor":null}""")
}
val (records, next) = CommandCodeUsageRepository.parseUsageResponse(body)
assertEquals(CommandCodeUsageRepository.PAGE_SIZE, records.size)
assertNotNull(next)
val decoded = String(Base64.getUrlDecoder().decode(next!!))
assertTrue(decoded.contains("\"id\":\"rec-${CommandCodeUsageRepository.PAGE_SIZE - 1}\""))
}
@Test
fun `legacy format stays compatible`() {
val body = """
{"usages":[{"id":"old-1","createdAt":"2026-08-20T10:00:00.000Z",
"tokensIn":"500","tokensOut":"50","tokensTotal":"550",
"creditsTotal":"0.01",
"meta":{"model":"gpt-4o","provider":"morph","cacheReadInputTokens":200}}]}
""".trimIndent()
val (records, _) = CommandCodeUsageRepository.parseUsageResponse(body)
assertEquals(1, records.size)
val r = records.first()
assertEquals((0.01 * CommandCodeUsageRepository.COST_DENOM).toLong(), r.cost)
assertEquals(300L, r.inputTokens) // 500 - 200
assertEquals(200L, r.cacheReadTokens)
assertEquals("morph", r.provider)
}
}