Switch usage sync to Redis-only identities

.env.example

@@ -30,10 +30,6 @@ APP_BASE_PATH=
 # Project business timezone. Affects Today, daily aggregation, daily 03:00 cleanup, and log timestamps. Required: no. Default: Asia/Shanghai.
 TZ=Asia/Shanghai
 
-# usage 同步模式：auto 启动时用 AUTH-only 探测 management data stream；成功则本进程固定使用 redis，失败则本进程固定使用 legacy_export。redis 只使用 data stream；legacy_export 使用旧兼容方式。必填：否。默认值：auto。
-# Usage sync mode: auto runs an AUTH-only startup probe for the management data stream; success fixes this process to redis, failure fixes it to legacy_export. redis uses only the data stream; legacy_export uses the old compatibility path. Required: no. Default: auto.
-USAGE_SYNC_MODE=auto
-
 # CPA management data stream 的 Redis/RESP TCP 地址。留空时默认使用 CPA_BASE_URL 的主机名加 8317 端口；如果通过 nginx stream 暴露到其它端口，请显式填写 host:port。
 # Redis/RESP TCP address for the CPA management data stream. When empty, defaults to the CPA_BASE_URL hostname plus port 8317; set host:port explicitly when exposed through nginx stream on another port.
 REDIS_QUEUE_ADDR=
@@ -46,10 +42,6 @@ REDIS_QUEUE_BATCH_SIZE=1000
 # Idle check interval when the Redis queue is empty. Required: no. Default: 1s.
 REDIS_QUEUE_IDLE_INTERVAL=1s
 
-# legacy_export 的拉取间隔。必填：否。默认值：5m。
-# Pull interval for legacy_export. Required: no. Default: 5m.
-POLL_INTERVAL=5m
-
 # 请求 CPA 接口时的超时时间。必填：否。默认值：30s。
 # Timeout for requests to the CPA service. Required: no. Default: 30s.
 REQUEST_TIMEOUT=30s

README.en.md

@@ -4,7 +4,7 @@
 
 CPA Usage Keeper is a standalone CPA usage persistence and dashboard service.
 
-It relies on [CLIProxyAPI (CPA)](https://github.com/router-for-me/CLIProxyAPI) as the backend CPA data source and adds persistent storage and statistical analysis capabilities on top of CPA. The service periodically pulls CPA data, writes normalized events to SQLite, exposes aggregation APIs, and serves a built-in web dashboard for usage, pricing, request health, and model/API statistics.
+It relies on [CLIProxyAPI (CPA)](https://github.com/router-for-me/CLIProxyAPI) as the backend CPA data source and adds persistent storage and statistical analysis capabilities on top of CPA. The service consumes events from the CPA Redis usage queue into SQLite, periodically pulls CPA metadata, exposes aggregation APIs, and serves a built-in web dashboard for usage, pricing, request health, and model/API statistics.
 
 ![cpa-usage-keeper-screenshot](https://images.bitskyline.com/i/2026/04/h9se9f.png)
 
@@ -52,11 +52,9 @@ cp .env.example .env
 | `APP_PORT` | No | `8080` | HTTP listen port |
 | `APP_BASE_PATH` | No | root path | Subpath prefix such as `/cpa`; empty means `/` |
 | `TZ` | No | `Asia/Shanghai` | Project business timezone; affects Today, daily aggregation, scheduled tasks, and log timestamps |
-| `USAGE_SYNC_MODE` | No | `auto` | Sync mode: `auto` probes at startup and then fixes the process to `redis` or `legacy_export`; can also be set explicitly to `redis` or `legacy_export` |
 | `REDIS_QUEUE_ADDR` | No | `CPA_BASE_URL` hostname + `8317` | CPA Redis/RESP TCP address; set `host:port` for non-default ports |
 | `REDIS_QUEUE_BATCH_SIZE` | No | `1000` | Maximum queue records per pull |
 | `REDIS_QUEUE_IDLE_INTERVAL` | No | `1s` | Empty queue check interval |
-| `POLL_INTERVAL` | No | `5m` | Pull interval for `legacy_export` |
 | `REQUEST_TIMEOUT` | No | `30s` | CPA request timeout |
 | `WORK_DIR` | No | `./data` | Application work directory; database, logs, and backups default to `app.db`, `logs/`, and `backups/` under it |
 | `LOG_LEVEL` | No | `info` | Log level |

README.md

@@ -4,7 +4,7 @@
 
 `CPA Usage Keeper` 是一个独立的 CPA 用量持久化与可视化服务。
 
-它依赖 [CLIProxyAPI（CPA）](https://github.com/router-for-me/CLIProxyAPI) 作为后端 CPA 数据来源，目标是在 CPA 之上补充持久化存储与统计分析能力。服务会定时拉取 CPA 数据，将规范化后的事件写入 SQLite，暴露聚合 API，并提供内置 Web Dashboard 用于查看 usage、pricing、request health 和 model/API 维度的统计信息。
+它依赖 [CLIProxyAPI（CPA）](https://github.com/router-for-me/CLIProxyAPI) 作为后端 CPA 数据来源，目标是在 CPA 之上补充持久化存储与统计分析能力。服务会从 CPA Redis usage 队列消费事件并写入 SQLite，定时拉取 CPA metadata，暴露聚合 API，并提供内置 Web Dashboard 用于查看 usage、pricing、request health 和 model/API 维度的统计信息。
 
 ![cpa-usage-keeper-screenshot](https://images.bitskyline.com/i/2026/04/h9se9f.png)
 
@@ -52,11 +52,9 @@ cp .env.example .env
 | `APP_PORT` | 否 | `8080` | HTTP 监听端口 |
 | `APP_BASE_PATH` | 否 | 根路径 | 子路径部署前缀，例如 `/cpa`；留空表示 `/` |
 | `TZ` | 否 | `Asia/Shanghai` | 项目业务时区，影响 Today、按天聚合、定时任务和日志时间 |
-| `USAGE_SYNC_MODE` | 否 | `auto` | 同步模式：`auto` 启动时探测后固定为 `redis` 或 `legacy_export`；也可显式设置 `redis`、`legacy_export` |
 | `REDIS_QUEUE_ADDR` | 否 | `CPA_BASE_URL` 主机名 + `8317` | CPA Redis/RESP TCP 地址；非默认端口时填写 `host:port` |
 | `REDIS_QUEUE_BATCH_SIZE` | 否 | `1000` | 每次最多拉取的队列记录数 |
 | `REDIS_QUEUE_IDLE_INTERVAL` | 否 | `1s` | 队列为空时的检查间隔 |
-| `POLL_INTERVAL` | 否 | `5m` | `legacy_export` 拉取间隔 |
 | `REQUEST_TIMEOUT` | 否 | `30s` | CPA 请求超时 |
 | `WORK_DIR` | 否 | `./data` | 应用工作目录；数据库、日志和备份默认分别写入 `app.db`、`logs/`、`backups/` |
 | `LOG_LEVEL` | 否 | `info` | 日志级别 |

docker-compose.example.yml

@@ -12,11 +12,9 @@ services:
       APP_PORT: ${APP_PORT:-8080}
       APP_BASE_PATH: ${APP_BASE_PATH:-}
       TZ: ${TZ:-Asia/Shanghai}
-      USAGE_SYNC_MODE: ${USAGE_SYNC_MODE:-auto}
       REDIS_QUEUE_ADDR: ${REDIS_QUEUE_ADDR:-}
       REDIS_QUEUE_BATCH_SIZE: ${REDIS_QUEUE_BATCH_SIZE:-1000}
       REDIS_QUEUE_IDLE_INTERVAL: ${REDIS_QUEUE_IDLE_INTERVAL:-1s}
-      POLL_INTERVAL: ${POLL_INTERVAL:-5m}
       REQUEST_TIMEOUT: ${REQUEST_TIMEOUT:-30s}
       WORK_DIR: ${WORK_DIR:-./data}
       LOG_LEVEL: ${LOG_LEVEL:-info}

internal/api/auth_test.go

@@ -11,7 +11,7 @@ import (
 )
 
 func TestAuthSessionReportsAuthenticatedWhenDisabled(t *testing.T) {
-	router := NewRouter(nil, nil, nil, nil, nil, nil, AuthConfig{Enabled: false}, nil, "")
+	router := NewRouter(nil, nil, nil, nil, AuthConfig{Enabled: false}, nil, "")
 	resp := httptest.NewRecorder()
 	req := httptest.NewRequest(http.MethodGet, "/api/v1/auth/session", nil)
 
@@ -25,7 +25,7 @@ func TestAuthSessionReportsAuthenticatedWhenDisabled(t *testing.T) {
 func TestAuthProtectedRouteRequiresSessionWhenEnabled(t *testing.T) {
 	sessions := auth.NewSessionManager(time.Hour)
 	config := AuthConfig{Enabled: true, LoginPassword: "secret", SessionTTL: time.Hour}
-	router := NewRouter(nil, nil, nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
+	router := NewRouter(nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
 	resp := httptest.NewRecorder()
 	req := httptest.NewRequest(http.MethodGet, "/api/v1/usage/overview", nil)
 
@@ -40,7 +40,7 @@ func TestAuthLoginSetsCookieAndUnlocksProtectedRoute(t *testing.T) {
 	sessions := auth.NewSessionManager(time.Hour)
 	config := AuthConfig{Enabled: true, LoginPassword: "secret", SessionTTL: time.Hour}
 	handler := NewAuthHandler(config, sessions)
-	router := NewRouter(nil, nil, nil, nil, nil, nil, config, handler, "")
+	router := NewRouter(nil, nil, nil, nil, config, handler, "")
 
 	loginResp := httptest.NewRecorder()
 	loginReq := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", strings.NewReader(`{"password":"secret"}`))
@@ -74,7 +74,7 @@ func TestAuthLoginSetsCookieAndUnlocksProtectedRoute(t *testing.T) {
 func TestAuthLoginRejectsWrongPassword(t *testing.T) {
 	sessions := auth.NewSessionManager(time.Hour)
 	config := AuthConfig{Enabled: true, LoginPassword: "secret", SessionTTL: time.Hour}
-	router := NewRouter(nil, nil, nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
+	router := NewRouter(nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
 	resp := httptest.NewRecorder()
 	req := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", strings.NewReader(`{"password":"wrong"}`))
 	req.Header.Set("Content-Type", "application/json")
@@ -89,7 +89,7 @@ func TestAuthLoginRejectsWrongPassword(t *testing.T) {
 func TestAuthLoginRateLimitsRepeatedFailures(t *testing.T) {
 	sessions := auth.NewSessionManager(time.Hour)
 	config := AuthConfig{Enabled: true, LoginPassword: "secret", SessionTTL: time.Hour}
-	router := NewRouter(nil, nil, nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
+	router := NewRouter(nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
 
 	for i := 0; i < 5; i++ {
 		resp := httptest.NewRecorder()
@@ -116,7 +116,7 @@ func TestAuthLoginRateLimitsRepeatedFailures(t *testing.T) {
 func TestAuthLoginAllowsCorrectPasswordAfterRateLimitThreshold(t *testing.T) {
 	sessions := auth.NewSessionManager(time.Hour)
 	config := AuthConfig{Enabled: true, LoginPassword: "secret", SessionTTL: time.Hour}
-	router := NewRouter(nil, nil, nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
+	router := NewRouter(nil, nil, nil, nil, config, NewAuthHandler(config, sessions), "")
 
 	for i := 0; i < 5; i++ {
 		resp := httptest.NewRecorder()
@@ -144,7 +144,7 @@ func TestAuthLogoutDeletesSessionCookie(t *testing.T) {
 	sessions := auth.NewSessionManager(time.Hour)
 	config := AuthConfig{Enabled: true, LoginPassword: "secret", SessionTTL: time.Hour}
 	handler := NewAuthHandler(config, sessions)
-	router := NewRouter(nil, nil, nil, nil, nil, nil, config, handler, "")
+	router := NewRouter(nil, nil, nil, nil, config, handler, "")
 
 	loginResp := httptest.NewRecorder()
 	loginReq := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", strings.NewReader(`{"password":"secret"}`))
@@ -183,7 +183,7 @@ func TestSubpathAuthUsesPrefixedRoutesAndCookiePath(t *testing.T) {
 	sessions := auth.NewSessionManager(time.Hour)
 	config := AuthConfig{Enabled: true, LoginPassword: "secret", SessionTTL: time.Hour, BasePath: "/cpa"}
 	handler := NewAuthHandler(config, sessions)
-	router := NewRouter(nil, nil, nil, nil, nil, nil, config, handler, "/cpa")
+	router := NewRouter(nil, nil, nil, nil, config, handler, "/cpa")
 
 	sessionResp := httptest.NewRecorder()
 	sessionReq := httptest.NewRequest(http.MethodGet, "/cpa/api/v1/auth/session", nil)

internal/api/pricing_test.go

@@ -39,7 +39,7 @@ func (s *pricingStub) DeletePricing(_ context.Context, model string) error {
 }
 
 func TestPricingRoutesReturnEmptyResponsesWithoutProvider(t *testing.T) {
-	router := NewRouter(nil, nil, nil, nil, nil, nil, AuthConfig{}, nil, "")
+	router := NewRouter(nil, nil, nil, nil, AuthConfig{}, nil, "")
 
 	usedReq := httptest.NewRequest(http.MethodGet, "/api/v1/models/used", nil)
 	usedResp := httptest.NewRecorder()
@@ -57,7 +57,7 @@ func TestPricingRoutesReturnEmptyResponsesWithoutProvider(t *testing.T) {
 }
 
 func TestPricingRoutesReturnConfiguredData(t *testing.T) {
-	router := NewRouter(nil, nil, nil, nil, nil, &pricingStub{
+	router := NewRouter(nil, nil, nil, &pricingStub{
 		usedModels: []string{"claude-sonnet"},
 		pricing: []models.ModelPriceSetting{{
 			Model:                "claude-sonnet",
@@ -91,7 +91,7 @@ func TestUpdatePricingRoute(t *testing.T) {
 			CachePricePer1M:      0.3,
 		},
 	}
-	router := NewRouter(nil, nil, nil, nil, nil, provider, AuthConfig{}, nil, "")
+	router := NewRouter(nil, nil, nil, provider, AuthConfig{}, nil, "")
 
 	req := httptest.NewRequest(http.MethodPut, "/api/v1/pricing/claude-sonnet", strings.NewReader(`{"prompt_price_per_1m":3,"completion_price_per_1m":15,"cache_price_per_1m":0.3}`))
 	req.Header.Set("Content-Type", "application/json")
@@ -112,7 +112,7 @@ func TestUpdatePricingRouteAcceptsModelInBody(t *testing.T) {
 			CachePricePer1M:      0.3,
 		},
 	}
-	router := NewRouter(nil, nil, nil, nil, nil, provider, AuthConfig{}, nil, "")
+	router := NewRouter(nil, nil, nil, provider, AuthConfig{}, nil, "")
 
 	req := httptest.NewRequest(http.MethodPut, "/api/v1/pricing", strings.NewReader(`{"model":"openai/gpt-4.1","prompt_price_per_1m":3,"completion_price_per_1m":15,"cache_price_per_1m":0.3}`))
 	req.Header.Set("Content-Type", "application/json")
@@ -129,7 +129,7 @@ func TestUpdatePricingRouteAcceptsModelInBody(t *testing.T) {
 
 func TestDeletePricingRoute(t *testing.T) {
 	provider := &pricingStub{}
-	router := NewRouter(nil, nil, nil, nil, nil, provider, AuthConfig{}, nil, "")
+	router := NewRouter(nil, nil, nil, provider, AuthConfig{}, nil, "")
 
 	req := httptest.NewRequest(http.MethodDelete, "/api/v1/pricing?model=openai%2Fgpt-4.1", nil)
 	resp := httptest.NewRecorder()