starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: remove SQLite migration support

Browse Source 
Remove the import-sqlite command, bootstrap importer, migration fixtures and tests, and the SQLx SQLite feature.

Keep the runtime and project documentation focused exclusively on PostgreSQL and Redis storage.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-12 21:29:36 +08:00
co-authored by Codex
parent 9b33e60df2
commit d94bca33dd
15 changed files with 5 additions and 1737 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
-6
View File
@@ -16,12 +16,6 @@
# Runtime data
/.runtime/
*.sqlite
*.sqlite3
*.sqlite-*
*.db
*.db-shm
*.db-wal
*.log
# Local configuration and secrets
backend/Cargo.lock
Generated
-1
View File
@@ -1403,7 +1403,6 @@ version = "0.37.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1f111c8c41e7c61a49cd34e44c7619462967221a6443b0ec299e0ac30cfb9b1"
dependencies = [
"cc",
"pkg-config",
"vcpkg",
]
backend/Cargo.toml
+1 -1
View File
@@ -37,7 +37,7 @@ semver = "1.0.27"
serde = { version = "1.0.228", features = ["derive"] }
serde_json = { version = "1.0.150", features = ["preserve_order"] }
sha2 = "0.11.0"
sqlx = { version = "0.9.0", default-features = false, features = ["runtime-tokio", "tls-rustls", "postgres", "sqlite-bundled", "chrono", "json"] }
sqlx = { version = "0.9.0", default-features = false, features = ["runtime-tokio", "tls-rustls", "postgres", "chrono", "json"] }
subtle = "2.6.1"
tar = "0.4.44"
thiserror = "2.0.18"
backend/src/bootstrap/import_sqlite/core.rs
-460
View File
@@ -1,460 +0,0 @@
use futures::TryStreamExt;
use sqlx::{Postgres, Row, SqlitePool, Transaction, types::Json};
use crate::infra::identity::hash_credential;
use super::{
ImportSqliteError, ImportSqliteReport, parse_cookie_expiration, parse_json,
parse_optional_json, parse_optional_timestamp, parse_timestamp,
};
pub(super) async fn import_core_tables(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
import_admin_users(source, target, report).await?;
import_client_api_keys(source, target, report).await?;
import_runtime_settings(source, target, report).await?;
import_accounts(source, target, report).await?;
import_account_usage(source, target, report).await?;
import_account_cookies(source, target, report).await?;
import_fingerprints(source, target, report).await?;
import_fingerprint_history(source, target, report).await
}
async fn import_admin_users(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query(
"select id, password_hash, created_at, updated_at from admin_users order by id",
)
.fetch(source);
while let Some(row) = rows.try_next().await? {
sqlx::query(
"insert into admin_users (id, password_hash, created_at, updated_at)
values ($1, $2, $3, $4)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<String, _>("password_hash")?)
.bind(parse_timestamp(
"admin_users",
"created_at",
row.try_get("created_at")?,
)?)
.bind(parse_timestamp(
"admin_users",
"updated_at",
row.try_get("updated_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("admin_users");
}
Ok(())
}
async fn import_client_api_keys(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query(
"select id, name, prefix, key, label, enabled, created_at, last_used_at
from client_api_keys order by id",
)
.fetch(source);
while let Some(row) = rows.try_next().await? {
sqlx::query(
"insert into client_api_keys
(id, name, prefix, key, label, enabled, created_at, last_used_at)
values ($1, $2, $3, $4, $5, $6, $7, $8)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<String, _>("name")?)
.bind(row.try_get::<String, _>("prefix")?)
.bind(row.try_get::<String, _>("key")?)
.bind(row.try_get::<Option<String>, _>("label")?)
.bind(row.try_get::<i64, _>("enabled")? == 1)
.bind(parse_timestamp(
"client_api_keys",
"created_at",
row.try_get("created_at")?,
)?)
.bind(parse_optional_timestamp(
"client_api_keys",
"last_used_at",
row.try_get("last_used_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("client_api_keys");
}
Ok(())
}
async fn import_runtime_settings(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query(
"select id, model_aliases_json, refresh_margin_seconds, refresh_concurrency,
max_concurrent_per_account, request_interval_ms, rotation_strategy,
admin_api_key, updated_at
from runtime_settings order by id",
)
.fetch(source);
while let Some(row) = rows.try_next().await? {
let aliases: String = row.try_get("model_aliases_json")?;
let admin_key = row
.try_get::<Option<String>, _>("admin_api_key")?
.and_then(|key| (!key.trim().is_empty()).then(|| hash_credential(&key)));
sqlx::query(
"insert into runtime_settings
(id, model_aliases_json, refresh_margin_seconds, refresh_concurrency,
max_concurrent_per_account, request_interval_ms, rotation_strategy,
admin_api_key_hash, usage_retention_days, ops_error_retention_days,
bucket_retention_days, updated_at)
values ($1, $2, $3, $4, $5, $6, $7, $8, 30, 30, 90, $9)",
)
.bind(row.try_get::<i64, _>("id")?)
.bind(Json(parse_json(
"runtime_settings",
"model_aliases_json",
&aliases,
)?))
.bind(row.try_get::<i64, _>("refresh_margin_seconds")?)
.bind(row.try_get::<i64, _>("refresh_concurrency")?)
.bind(row.try_get::<i64, _>("max_concurrent_per_account")?)
.bind(row.try_get::<i64, _>("request_interval_ms")?)
.bind(row.try_get::<String, _>("rotation_strategy")?)
.bind(admin_key)
.bind(parse_timestamp(
"runtime_settings",
"updated_at",
row.try_get("updated_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("runtime_settings");
}
Ok(())
}
async fn import_accounts(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query(
"select id, email, chatgpt_account_id, chatgpt_user_id, label, plan_type,
access_token, refresh_token, access_token_expires_at, next_refresh_at,
status, quota_json, quota_fetched_at, quota_limit_reached,
quota_verify_required, quota_cooldown_until, cloudflare_cooldown_until,
added_at, updated_at
from accounts order by id",
)
.fetch(source);
while let Some(row) = rows.try_next().await? {
let source_status = row.try_get::<String, _>("status")?;
let (status, next_refresh_at) = if source_status == "refreshing" {
report.add_normalized("accounts.refreshing_to_expired");
("expired", None)
} else {
(
source_status.as_str(),
parse_optional_timestamp(
"accounts",
"next_refresh_at",
row.try_get("next_refresh_at")?,
)?,
)
};
let quota = parse_optional_json(
"accounts",
"quota_json",
row.try_get::<Option<String>, _>("quota_json")?,
)?
.map(Json);
sqlx::query(
"insert into accounts
(id, email, chatgpt_account_id, chatgpt_user_id, label, plan_type,
access_token, refresh_token, access_token_expires_at, next_refresh_at,
status, quota_json, quota_fetched_at, quota_limit_reached,
quota_verify_required, quota_cooldown_until, cloudflare_cooldown_until,
added_at, updated_at)
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13,
$14, $15, $16, $17, $18, $19)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<Option<String>, _>("email")?)
.bind(row.try_get::<Option<String>, _>("chatgpt_account_id")?)
.bind(row.try_get::<Option<String>, _>("chatgpt_user_id")?)
.bind(row.try_get::<Option<String>, _>("label")?)
.bind(row.try_get::<Option<String>, _>("plan_type")?)
.bind(row.try_get::<String, _>("access_token")?)
.bind(row.try_get::<Option<String>, _>("refresh_token")?)
.bind(parse_optional_timestamp(
"accounts",
"access_token_expires_at",
row.try_get("access_token_expires_at")?,
)?)
.bind(next_refresh_at)
.bind(status)
.bind(quota)
.bind(parse_optional_timestamp(
"accounts",
"quota_fetched_at",
row.try_get("quota_fetched_at")?,
)?)
.bind(row.try_get::<i64, _>("quota_limit_reached")? == 1)
.bind(row.try_get::<i64, _>("quota_verify_required")? == 1)
.bind(parse_optional_timestamp(
"accounts",
"quota_cooldown_until",
row.try_get("quota_cooldown_until")?,
)?)
.bind(parse_optional_timestamp(
"accounts",
"cloudflare_cooldown_until",
row.try_get("cloudflare_cooldown_until")?,
)?)
.bind(parse_timestamp(
"accounts",
"added_at",
row.try_get("added_at")?,
)?)
.bind(parse_timestamp(
"accounts",
"updated_at",
row.try_get("updated_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("accounts");
}
Ok(())
}
async fn import_account_usage(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query("select * from account_usage order by account_id").fetch(source);
while let Some(row) = rows.try_next().await? {
sqlx::query(
"insert into account_usage
(account_id, request_count, empty_response_count, input_tokens, output_tokens,
cached_tokens, reasoning_tokens, total_tokens, image_input_tokens,
image_output_tokens, image_request_count, image_request_failed_count,
window_request_count, window_input_tokens, window_output_tokens,
window_cached_tokens, window_image_input_tokens, window_image_output_tokens,
window_image_request_count, window_image_request_failed_count,
window_started_at, window_reset_at, limit_window_seconds, last_used_at)
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12,
$13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24)",
)
.bind(row.try_get::<String, _>("account_id")?)
.bind(row.try_get::<i64, _>("request_count")?)
.bind(row.try_get::<i64, _>("empty_response_count")?)
.bind(row.try_get::<i64, _>("input_tokens")?)
.bind(row.try_get::<i64, _>("output_tokens")?)
.bind(row.try_get::<i64, _>("cached_tokens")?)
.bind(row.try_get::<i64, _>("reasoning_tokens")?)
.bind(row.try_get::<i64, _>("total_tokens")?)
.bind(row.try_get::<i64, _>("image_input_tokens")?)
.bind(row.try_get::<i64, _>("image_output_tokens")?)
.bind(row.try_get::<i64, _>("image_request_count")?)
.bind(row.try_get::<i64, _>("image_request_failed_count")?)
.bind(row.try_get::<i64, _>("window_request_count")?)
.bind(row.try_get::<i64, _>("window_input_tokens")?)
.bind(row.try_get::<i64, _>("window_output_tokens")?)
.bind(row.try_get::<i64, _>("window_cached_tokens")?)
.bind(row.try_get::<i64, _>("window_image_input_tokens")?)
.bind(row.try_get::<i64, _>("window_image_output_tokens")?)
.bind(row.try_get::<i64, _>("window_image_request_count")?)
.bind(row.try_get::<i64, _>("window_image_request_failed_count")?)
.bind(parse_optional_timestamp(
"account_usage",
"window_started_at",
row.try_get("window_started_at")?,
)?)
.bind(parse_optional_timestamp(
"account_usage",
"window_reset_at",
row.try_get("window_reset_at")?,
)?)
.bind(row.try_get::<Option<i64>, _>("limit_window_seconds")?)
.bind(parse_optional_timestamp(
"account_usage",
"last_used_at",
row.try_get("last_used_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("account_usage");
}
Ok(())
}
async fn import_account_cookies(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query(
"select id, account_id, domain, name, value, path, expires_at, updated_at
from account_cookies order by id",
)
.fetch(source);
while let Some(row) = rows.try_next().await? {
let expires_at = row
.try_get::<Option<String>, _>("expires_at")?
.and_then(|value| match parse_cookie_expiration(&value) {
Some(timestamp) => Some(timestamp),
None => {
report.add_cookie_expiration_parse_failure();
None
}
});
sqlx::query(
"insert into account_cookies
(id, account_id, domain, name, value, path, expires_at, updated_at)
values ($1, $2, $3, $4, $5, $6, $7, $8)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<String, _>("account_id")?)
.bind(row.try_get::<String, _>("domain")?)
.bind(row.try_get::<String, _>("name")?)
.bind(row.try_get::<String, _>("value")?)
.bind(row.try_get::<String, _>("path")?)
.bind(expires_at)
.bind(parse_timestamp(
"account_cookies",
"updated_at",
row.try_get("updated_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("account_cookies");
}
Ok(())
}
async fn import_fingerprints(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query("select * from fingerprints order by id").fetch(source);
while let Some(row) = rows.try_next().await? {
let default_headers: String = row.try_get("default_headers_json")?;
let header_order: String = row.try_get("header_order_json")?;
sqlx::query(
"insert into fingerprints
(id, originator, app_version, build_number, platform, arch, chromium_version,
user_agent_template, default_headers_json, header_order_json, source,
created_at, updated_at)
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<String, _>("originator")?)
.bind(row.try_get::<String, _>("app_version")?)
.bind(row.try_get::<String, _>("build_number")?)
.bind(row.try_get::<String, _>("platform")?)
.bind(row.try_get::<String, _>("arch")?)
.bind(row.try_get::<String, _>("chromium_version")?)
.bind(row.try_get::<String, _>("user_agent_template")?)
.bind(Json(parse_json(
"fingerprints",
"default_headers_json",
&default_headers,
)?))
.bind(Json(parse_json(
"fingerprints",
"header_order_json",
&header_order,
)?))
.bind(row.try_get::<String, _>("source")?)
.bind(parse_timestamp(
"fingerprints",
"created_at",
row.try_get("created_at")?,
)?)
.bind(parse_timestamp(
"fingerprints",
"updated_at",
row.try_get("updated_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("fingerprints");
}
Ok(())
}
async fn import_fingerprint_history(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows =
sqlx::query("select * from fingerprint_update_history order by id").fetch(source);
while let Some(row) = rows.try_next().await? {
let manifest = parse_optional_json(
"fingerprint_update_history",
"manifest_json",
row.try_get::<Option<String>, _>("manifest_json")?,
)?
.map(Json);
sqlx::query(
"insert into fingerprint_update_history
(id, current_fingerprint_id, app_version, build_number, chromium_version,
source, manifest_json, created_at)
values ($1, $2, $3, $4, $5, $6, $7, $8)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<String, _>("current_fingerprint_id")?)
.bind(row.try_get::<String, _>("app_version")?)
.bind(row.try_get::<String, _>("build_number")?)
.bind(row.try_get::<Option<String>, _>("chromium_version")?)
.bind(row.try_get::<String, _>("source")?)
.bind(manifest)
.bind(parse_timestamp(
"fingerprint_update_history",
"created_at",
row.try_get("created_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("fingerprint_update_history");
}
Ok(())
}
pub(super) async fn count_discarded_runtime_tables(
source: &SqlitePool,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let (sessions, affinities, leases, snapshots, model_usage): (i64, i64, i64, i64, i64) =
sqlx::query_as(
"select
(select count(*) from admin_sessions),
(select count(*) from session_affinities),
(select count(*) from account_refresh_leases),
(select count(*) from model_plan_snapshots),
(select count(*) from account_model_usage)",
)
.fetch_one(source)
.await?;
report.add_discarded("admin_sessions", sessions.max(0) as u64);
report.add_discarded("session_affinities", affinities.max(0) as u64);
report.add_discarded("account_refresh_leases", leases.max(0) as u64);
report.add_discarded("model_plan_snapshots", snapshots.max(0) as u64);
report.add_discarded("account_model_usage", model_usage.max(0) as u64);
Ok(())
}
backend/src/bootstrap/import_sqlite/mod.rs
-232
View File
@@ -1,232 +0,0 @@
//! 一次性 SQLite v3 到 PostgreSQL 终态库导入。
mod core;
mod telemetry;
use std::{collections::BTreeMap, fmt, path::Path};
use chrono::{DateTime, Utc};
use serde_json::Value;
use sqlx::{
PgPool, Postgres, SqlitePool, Transaction,
sqlite::{SqliteConnectOptions, SqlitePoolOptions},
};
use thiserror::Error;
const SQLITE_SCHEMA_VERSION: i64 = 3;
/// SQLite 历史库导入结果。
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct ImportSqliteReport {
imported: BTreeMap<&'static str, u64>,
normalized: BTreeMap<&'static str, u64>,
discarded: BTreeMap<&'static str, u64>,
cookie_expiration_parse_failures: u64,
}
impl ImportSqliteReport {
fn add_imported(&mut self, table: &'static str) {
*self.imported.entry(table).or_default() += 1;
}
fn add_discarded(&mut self, reason: &'static str, count: u64) {
if count > 0 {
*self.discarded.entry(reason).or_default() += count;
}
}
fn add_normalized(&mut self, reason: &'static str) {
*self.normalized.entry(reason).or_default() += 1;
}
fn add_cookie_expiration_parse_failure(&mut self) {
self.cookie_expiration_parse_failures += 1;
}
/// 返回指定目标表导入的行数。
pub fn imported_rows(&self, table: &str) -> u64 {
self.imported.get(table).copied().unwrap_or_default()
}
/// 返回指定原因丢弃的行数。
pub fn discarded_rows(&self, reason: &str) -> u64 {
self.discarded.get(reason).copied().unwrap_or_default()
}
/// 返回指定规则规范化的行数。
pub fn normalized_rows(&self, reason: &str) -> u64 {
self.normalized.get(reason).copied().unwrap_or_default()
}
/// 返回无法解析并降级为会话 Cookie 的过期时间数量。
pub fn cookie_expiration_parse_failures(&self) -> u64 {
self.cookie_expiration_parse_failures
}
}
impl fmt::Display for ImportSqliteReport {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
writeln!(formatter, "SQLite v3 import completed")?;
for (table, count) in &self.imported {
writeln!(formatter, " imported {table}: {count}")?;
}
for (reason, count) in &self.normalized {
writeln!(formatter, " normalized {reason}: {count}")?;
}
for (reason, count) in &self.discarded {
writeln!(formatter, " discarded {reason}: {count}")?;
}
write!(
formatter,
" cookie expiration parse failures: {}",
self.cookie_expiration_parse_failures
)
}
}
/// SQLite 历史库导入错误。
#[derive(Debug, Error)]
pub enum ImportSqliteError {
#[error("database operation failed: {0}")]
Database(#[from] sqlx::Error),
#[error("unsupported SQLite schema version {actual:?}; expected 3")]
UnsupportedSourceVersion { actual: Option<i64> },
#[error("target PostgreSQL database is not empty")]
TargetNotEmpty,
#[error("invalid timestamp in {table}.{column}: {value}")]
InvalidTimestamp {
table: &'static str,
column: &'static str,
value: String,
},
#[error("invalid JSON in {table}.{column}: {source}")]
InvalidJson {
table: &'static str,
column: &'static str,
#[source]
source: serde_json::Error,
},
}
/// 将只读 SQLite v3 历史库一次性导入空的 PostgreSQL 终态库。
pub async fn import_sqlite(
target: &PgPool,
source_path: impl AsRef<Path>,
) -> Result<ImportSqliteReport, ImportSqliteError> {
let options = SqliteConnectOptions::new()
.filename(source_path.as_ref())
.read_only(true)
.create_if_missing(false);
let source = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(options)
.await?;
validate_source_version(&source).await?;
let mut transaction = target.begin().await?;
reject_nonempty_target(&mut transaction).await?;
let mut report = ImportSqliteReport::default();
core::import_core_tables(&source, &mut transaction, &mut report).await?;
telemetry::import_telemetry_tables(&source, &mut transaction, &mut report).await?;
core::count_discarded_runtime_tables(&source, &mut report).await?;
transaction.commit().await?;
source.close().await;
Ok(report)
}
async fn validate_source_version(source: &SqlitePool) -> Result<(), ImportSqliteError> {
let (actual,): (Option<i64>,) = sqlx::query_as("select max(version) from schema_migrations")
.fetch_one(source)
.await?;
if actual != Some(SQLITE_SCHEMA_VERSION) {
return Err(ImportSqliteError::UnsupportedSourceVersion { actual });
}
Ok(())
}
async fn reject_nonempty_target(
transaction: &mut Transaction<'_, Postgres>,
) -> Result<(), ImportSqliteError> {
let (rows,): (i64,) = sqlx::query_as(
"select
(select count(*) from admin_users)
+ (select count(*) from client_api_keys)
+ (select count(*) from runtime_settings)
+ (select count(*) from accounts)
+ (select count(*) from account_usage)
+ (select count(*) from usage_records)
+ (select count(*) from ops_error_logs)
+ (select count(*) from request_time_buckets)
+ (select count(*) from account_cookies)
+ (select count(*) from fingerprints)
+ (select count(*) from fingerprint_update_history)",
)
.fetch_one(&mut **transaction)
.await?;
if rows != 0 {
return Err(ImportSqliteError::TargetNotEmpty);
}
Ok(())
}
fn parse_timestamp(
table: &'static str,
column: &'static str,
value: String,
) -> Result<DateTime<Utc>, ImportSqliteError> {
DateTime::parse_from_rfc3339(&value)
.map(|timestamp| timestamp.with_timezone(&Utc))
.map_err(|_| ImportSqliteError::InvalidTimestamp {
table,
column,
value,
})
}
fn parse_optional_timestamp(
table: &'static str,
column: &'static str,
value: Option<String>,
) -> Result<Option<DateTime<Utc>>, ImportSqliteError> {
value
.map(|value| parse_timestamp(table, column, value))
.transpose()
}
fn parse_cookie_expiration(value: &str) -> Option<DateTime<Utc>> {
DateTime::parse_from_rfc3339(value)
.or_else(|_| DateTime::parse_from_rfc2822(value))
.ok()
.map(|timestamp| timestamp.with_timezone(&Utc))
}
fn parse_json(
table: &'static str,
column: &'static str,
value: &str,
) -> Result<Value, ImportSqliteError> {
serde_json::from_str(value).map_err(|source| ImportSqliteError::InvalidJson {
table,
column,
source,
})
}
fn parse_optional_json(
table: &'static str,
column: &'static str,
value: Option<String>,
) -> Result<Option<Value>, ImportSqliteError> {
value
.map(|value| parse_json(table, column, &value))
.transpose()
}
fn normalized_dimension(value: Option<String>) -> String {
value
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| "__unknown__".to_string())
}
backend/src/bootstrap/import_sqlite/telemetry.rs
-330
View File
@@ -1,330 +0,0 @@
use futures::TryStreamExt;
use serde_json::Value;
use sqlx::{Postgres, Row, SqlitePool, Transaction, types::Json};
use super::{
ImportSqliteError, ImportSqliteReport, normalized_dimension, parse_json, parse_timestamp,
};
pub(super) async fn import_telemetry_tables(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
import_mixed_usage_records(source, target, report).await?;
import_ops_error_logs(source, target, report).await?;
import_request_time_buckets(source, target, report).await
}
async fn import_mixed_usage_records(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows = sqlx::query(
"select id, request_id, kind, level, account_id, route, model, status_code,
transport, attempt_index, upstream_status_code, failure_class,
response_id, upstream_request_id, latency_ms, message, metadata_json,
created_at
from usage_records order by created_at, id",
)
.fetch(source);
while let Some(row) = rows.try_next().await? {
let level: String = row.try_get("level")?;
let account_id: Option<String> = row.try_get("account_id")?;
let model: Option<String> = row.try_get("model")?;
let status_code: Option<i64> = row.try_get("status_code")?;
let metadata_text: String = row.try_get("metadata_json")?;
let metadata = parse_json("usage_records", "metadata_json", &metadata_text)?;
if level == "error" {
insert_usage_error(target, &row, metadata).await?;
report.add_imported("ops_error_logs");
continue;
}
let valid_account = account_id
.as_deref()
.map(str::trim)
.is_some_and(|value| !value.is_empty());
let valid_model = model
.as_deref()
.map(str::trim)
.is_some_and(|value| !value.is_empty());
let valid_status = status_code.is_some_and(|status| (200..=399).contains(&status));
if !valid_account || !valid_model || !valid_status {
report.add_discarded("usage_record_noise", 1);
continue;
}
insert_success_record(
target,
&row,
account_id.expect("validated account id"),
model.expect("validated model"),
status_code.expect("validated status code"),
metadata,
)
.await?;
report.add_imported("usage_records");
}
Ok(())
}
async fn insert_success_record(
target: &mut Transaction<'_, Postgres>,
row: &sqlx::sqlite::SqliteRow,
account_id: String,
model: String,
status_code: i64,
mut metadata: Value,
) -> Result<(), ImportSqliteError> {
let requested_model = metadata_string(&metadata, &["requestedModel", "requested_model"]);
let upstream_model = metadata_string(&metadata, &["upstreamModel", "upstream_model"]);
let service_tier = metadata_string(&metadata, &["serviceTier", "service_tier"]);
let first_token_ms = metadata_nonnegative(&metadata, &["firstTokenMs", "first_token_ms"]);
let input_tokens = usage_number(&metadata, &["inputTokens", "input_tokens"]);
let output_tokens = usage_number(&metadata, &["outputTokens", "output_tokens"]);
let cached_tokens = usage_number(&metadata, &["cachedTokens", "cached_tokens"]);
let reasoning_tokens = usage_number(&metadata, &["reasoningTokens", "reasoning_tokens"]);
strip_promoted_success_fields(&mut metadata);
sqlx::query(
"insert into usage_records
(id, request_id, client_api_key_id, kind, route, provider, account_id, model,
requested_model, upstream_model, service_tier, status_code, transport,
attempt_index, response_id, upstream_request_id, latency_ms, first_token_ms,
input_tokens, output_tokens, cached_tokens, reasoning_tokens, message,
metadata_json, created_at)
values ($1, $2, null, $3, $4, 'openai', $5, $6, $7, $8, $9, $10,
$11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<Option<String>, _>("request_id")?)
.bind(row.try_get::<String, _>("kind")?)
.bind(row.try_get::<Option<String>, _>("route")?)
.bind(account_id)
.bind(model)
.bind(requested_model)
.bind(upstream_model)
.bind(service_tier)
.bind(status_code as i32)
.bind(row.try_get::<Option<String>, _>("transport")?)
.bind(row.try_get::<Option<i64>, _>("attempt_index")?)
.bind(row.try_get::<Option<String>, _>("response_id")?)
.bind(row.try_get::<Option<String>, _>("upstream_request_id")?)
.bind(row.try_get::<Option<i64>, _>("latency_ms")?)
.bind(first_token_ms)
.bind(input_tokens)
.bind(output_tokens)
.bind(cached_tokens)
.bind(reasoning_tokens)
.bind(row.try_get::<String, _>("message")?)
.bind(Json(metadata))
.bind(parse_timestamp(
"usage_records",
"created_at",
row.try_get("created_at")?,
)?)
.execute(&mut **target)
.await?;
Ok(())
}
async fn insert_usage_error(
target: &mut Transaction<'_, Postgres>,
row: &sqlx::sqlite::SqliteRow,
metadata: Value,
) -> Result<(), ImportSqliteError> {
sqlx::query(
"insert into ops_error_logs
(id, request_id, client_api_key_id, kind, provider, account_id, route, model,
status_code, client_status_code, upstream_status_code, transport, attempt_index,
failure_class, response_id, upstream_request_id, latency_ms, message,
metadata_json, created_at)
values ($1, $2, null, $3, 'openai', $4, $5, $6, $7, null, $8, $9,
$10, $11, $12, $13, $14, $15, $16, $17)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<Option<String>, _>("request_id")?)
.bind(row.try_get::<String, _>("kind")?)
.bind(row.try_get::<Option<String>, _>("account_id")?)
.bind(row.try_get::<Option<String>, _>("route")?)
.bind(row.try_get::<Option<String>, _>("model")?)
.bind(optional_status(row.try_get("status_code")?))
.bind(optional_status(row.try_get("upstream_status_code")?))
.bind(row.try_get::<Option<String>, _>("transport")?)
.bind(row.try_get::<Option<i64>, _>("attempt_index")?)
.bind(row.try_get::<Option<String>, _>("failure_class")?)
.bind(row.try_get::<Option<String>, _>("response_id")?)
.bind(row.try_get::<Option<String>, _>("upstream_request_id")?)
.bind(row.try_get::<Option<i64>, _>("latency_ms")?)
.bind(row.try_get::<String, _>("message")?)
.bind(Json(metadata))
.bind(parse_timestamp(
"usage_records",
"created_at",
row.try_get("created_at")?,
)?)
.execute(&mut **target)
.await?;
Ok(())
}
async fn import_ops_error_logs(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows =
sqlx::query("select * from ops_error_logs order by created_at, id").fetch(source);
while let Some(row) = rows.try_next().await? {
let metadata_text: String = row.try_get("metadata_json")?;
sqlx::query(
"insert into ops_error_logs
(id, request_id, client_api_key_id, kind, provider, account_id, route, model,
status_code, client_status_code, upstream_status_code, transport, attempt_index,
failure_class, response_id, upstream_request_id, latency_ms, message,
metadata_json, created_at)
values ($1, $2, null, $3, 'openai', $4, $5, $6, $7, $8, $9, $10,
$11, $12, $13, $14, $15, $16, $17, $18)",
)
.bind(row.try_get::<String, _>("id")?)
.bind(row.try_get::<Option<String>, _>("request_id")?)
.bind(row.try_get::<String, _>("kind")?)
.bind(row.try_get::<Option<String>, _>("account_id")?)
.bind(row.try_get::<Option<String>, _>("route")?)
.bind(row.try_get::<Option<String>, _>("model")?)
.bind(optional_status(row.try_get("status_code")?))
.bind(optional_status(row.try_get("client_status_code")?))
.bind(optional_status(row.try_get("upstream_status_code")?))
.bind(row.try_get::<Option<String>, _>("transport")?)
.bind(row.try_get::<Option<i64>, _>("attempt_index")?)
.bind(row.try_get::<Option<String>, _>("failure_class")?)
.bind(row.try_get::<Option<String>, _>("response_id")?)
.bind(row.try_get::<Option<String>, _>("upstream_request_id")?)
.bind(row.try_get::<Option<i64>, _>("latency_ms")?)
.bind(row.try_get::<String, _>("message")?)
.bind(Json(parse_json(
"ops_error_logs",
"metadata_json",
&metadata_text,
)?))
.bind(parse_timestamp(
"ops_error_logs",
"created_at",
row.try_get("created_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("ops_error_logs");
}
Ok(())
}
async fn import_request_time_buckets(
source: &SqlitePool,
target: &mut Transaction<'_, Postgres>,
report: &mut ImportSqliteReport,
) -> Result<(), ImportSqliteError> {
let mut rows =
sqlx::query("select * from usage_time_buckets order by bucket_start").fetch(source);
while let Some(row) = rows.try_next().await? {
let request_count: i64 = row.try_get("request_count")?;
let error_count: i64 = row.try_get("error_count")?;
let min_latency_ms: i64 = row.try_get("min_latency_ms")?;
sqlx::query(
"insert into request_time_buckets
(bucket_start, provider, account_id, model, service_tier, success_count,
error_count, input_tokens, output_tokens, cached_tokens,
first_token_latency_sum, first_token_latency_count, latency_sum,
latency_count, max_latency_ms, min_latency_ms, updated_at)
values ($1, 'openai', $2, $3, $4, $5, $6, $7, $8, $9, $10, $11,
$12, $13, $14, $15, $16)",
)
.bind(parse_timestamp(
"usage_time_buckets",
"bucket_start",
row.try_get("bucket_start")?,
)?)
.bind(normalized_dimension(row.try_get("account_id")?))
.bind(normalized_dimension(row.try_get("model")?))
.bind(normalized_dimension(row.try_get("service_tier")?))
.bind(request_count.saturating_sub(error_count).max(0))
.bind(error_count)
.bind(row.try_get::<i64, _>("input_tokens")?)
.bind(row.try_get::<i64, _>("output_tokens")?)
.bind(row.try_get::<i64, _>("cached_tokens")?)
.bind(row.try_get::<i64, _>("first_token_latency_sum")?)
.bind(row.try_get::<i64, _>("first_token_latency_count")?)
.bind(row.try_get::<i64, _>("latency_sum")?)
.bind(row.try_get::<i64, _>("latency_count")?)
.bind(row.try_get::<i64, _>("max_latency_ms")?)
.bind((min_latency_ms != 0).then_some(min_latency_ms))
.bind(parse_timestamp(
"usage_time_buckets",
"updated_at",
row.try_get("updated_at")?,
)?)
.execute(&mut **target)
.await?;
report.add_imported("request_time_buckets");
}
Ok(())
}
fn optional_status(status: Option<i64>) -> Option<i32> {
status.and_then(|status| i32::try_from(status).ok())
}
fn metadata_string(metadata: &Value, keys: &[&str]) -> Option<String> {
keys.iter().find_map(|key| {
metadata
.get(*key)
.and_then(Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToString::to_string)
})
}
fn metadata_nonnegative(metadata: &Value, keys: &[&str]) -> Option<i64> {
keys.iter()
.find_map(|key| metadata.get(*key).and_then(Value::as_i64))
.filter(|value| *value >= 0)
}
fn usage_number(metadata: &Value, keys: &[&str]) -> Option<i64> {
metadata
.get("usage")
.and_then(|usage| metadata_nonnegative(usage, keys))
.or_else(|| metadata_nonnegative(metadata, keys))
}
fn strip_promoted_success_fields(metadata: &mut Value) {
let Some(object) = metadata.as_object_mut() else {
return;
};
for key in [
"usage",
"route",
"apiKind",
"requestedModel",
"requested_model",
"upstreamModel",
"upstream_model",
"serviceTier",
"service_tier",
"statusCode",
"transport",
"attemptIndex",
"responseId",
"upstreamRequestId",
"openaiRequestId",
"firstTokenMs",
"first_token_ms",
] {
object.remove(key);
}
}
backend/src/bootstrap/mod.rs
-1
View File
@@ -1,7 +1,6 @@
//! 进程装配、共享状态与后台任务。
pub mod config;
pub mod import_sqlite;
pub mod services;
pub mod shutdown;
pub mod state;
backend/src/main.rs
+2 -15
View File
@@ -1,4 +1,4 @@
use std::{env, error::Error, io, path::PathBuf};
use std::{env, error::Error, io};
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
@@ -8,19 +8,6 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
reject_extra_arguments(arguments)?;
codex_proxy_rs::bootstrap::services::run().await
}
Some("import-sqlite") => {
let source = arguments
.next()
.map(PathBuf::from)
.ok_or_else(|| invalid_cli("usage: codex-proxy-rs import-sqlite <path.sqlite>"))?;
reject_extra_arguments(arguments)?;
let config = codex_proxy_rs::bootstrap::config::AppConfig::load()?;
let pool = codex_proxy_rs::infra::database::connect(&config.database.url).await?;
let report =
codex_proxy_rs::bootstrap::import_sqlite::import_sqlite(&pool, source).await?;
println!("{report}");
Ok(())
}
Some("rebuild-buckets") => {
reject_extra_arguments(arguments)?;
let config = codex_proxy_rs::bootstrap::config::AppConfig::load()?;
@@ -35,7 +22,7 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
Ok(())
}
Some(command) => Err(invalid_cli(&format!(
"unknown command {command:?}; expected serve, import-sqlite, or rebuild-buckets"
"unknown command {command:?}; expected serve or rebuild-buckets"
))
.into()),
}
backend/tests/bootstrap/import_sqlite/mod.rs
-295
View File
@@ -1,295 +0,0 @@
use std::path::PathBuf;
use codex_proxy_rs::{
bootstrap::import_sqlite::{ImportSqliteError, import_sqlite},
keys::{service::KeyVerifier, store::PgClientKeyStore},
};
use sqlx::{Row, sqlite::SqliteConnectOptions, sqlite::SqlitePoolOptions};
use crate::support::storage::init_test_db;
#[tokio::test]
async fn import_sqlite_should_split_facts_preserve_retrievable_keys_and_report_discards() {
let (_source_guard, source_path) = sqlite_v3_source("success").await;
seed_representative_v3_data(&source_path).await;
let (target, _target_guard) = init_test_db("import-sqlite-success").await;
let report = import_sqlite(&target, &source_path).await.unwrap();
assert_eq!(report.imported_rows("client_api_keys"), 1);
assert_eq!(report.imported_rows("accounts"), 2);
assert_eq!(report.normalized_rows("accounts.refreshing_to_expired"), 1);
assert_eq!(report.imported_rows("usage_records"), 1);
assert_eq!(report.imported_rows("ops_error_logs"), 2);
assert_eq!(report.discarded_rows("usage_record_noise"), 1);
assert_eq!(report.discarded_rows("admin_sessions"), 1);
assert_eq!(report.discarded_rows("session_affinities"), 1);
assert_eq!(report.discarded_rows("account_refresh_leases"), 1);
assert_eq!(report.discarded_rows("model_plan_snapshots"), 1);
assert_eq!(report.discarded_rows("account_model_usage"), 1);
assert_eq!(report.cookie_expiration_parse_failures(), 1);
let key: String = sqlx::query_scalar("select key from client_api_keys where id = $1")
.bind("key-1")
.fetch_one(&target)
.await
.unwrap();
assert_eq!(key, "sk_old_secret");
let verifier = KeyVerifier::new(PgClientKeyStore::new(target.clone()));
assert_eq!(
verifier.verify(&key).await.unwrap().as_deref(),
Some("key-1")
);
verifier.flush_pending_last_used().await;
let refreshing_account: (String, Option<chrono::DateTime<chrono::Utc>>) =
sqlx::query_as("select status, next_refresh_at from accounts where id = $1")
.bind("acct-refreshing")
.fetch_one(&target)
.await
.unwrap();
assert_eq!(refreshing_account.0, "expired");
assert!(refreshing_account.1.is_none());
let success = sqlx::query(
"select requested_model, upstream_model, service_tier, first_token_ms,
input_tokens, output_tokens, cached_tokens, reasoning_tokens, metadata_json
from usage_records where id = $1",
)
.bind("usage-success")
.fetch_one(&target)
.await
.unwrap();
assert_eq!(
success
.get::<Option<String>, _>("requested_model")
.as_deref(),
Some("alias")
);
assert_eq!(
success
.get::<Option<String>, _>("upstream_model")
.as_deref(),
Some("gpt-5.5")
);
assert_eq!(
success.get::<Option<String>, _>("service_tier").as_deref(),
Some("priority")
);
assert_eq!(success.get::<Option<i64>, _>("first_token_ms"), Some(0));
assert_eq!(success.get::<Option<i64>, _>("input_tokens"), Some(12));
assert_eq!(success.get::<Option<i64>, _>("output_tokens"), Some(5));
assert_eq!(success.get::<Option<i64>, _>("cached_tokens"), Some(3));
assert_eq!(success.get::<Option<i64>, _>("reasoning_tokens"), Some(2));
let metadata: serde_json::Value = success.get("metadata_json");
assert_eq!(metadata["clientIp"], "127.0.0.1");
assert!(metadata.get("usage").is_none());
let bucket: (String, String, String, i64, i64, Option<i64>) = sqlx::query_as(
"select provider, account_id, model, success_count, error_count, min_latency_ms
from request_time_buckets",
)
.fetch_one(&target)
.await
.unwrap();
assert_eq!(bucket.0, "openai");
assert_eq!(bucket.1, "__unknown__");
assert_eq!(bucket.2, "__unknown__");
assert_eq!(bucket.3, 2);
assert_eq!(bucket.4, 1);
assert_eq!(bucket.5, None);
}
#[tokio::test]
async fn import_sqlite_should_rollback_all_target_writes_on_invalid_source_value() {
let (_source_guard, source_path) = sqlite_v3_source("rollback").await;
let source = open_sqlite(&source_path, false).await;
sqlx::query(
"insert into admin_users (id, password_hash, created_at, updated_at)
values ('admin', 'hash', 'not-a-time', '2026-01-01T00:00:00Z')",
)
.execute(&source)
.await
.unwrap();
source.close().await;
let (target, _target_guard) = init_test_db("import-sqlite-rollback").await;
let error = import_sqlite(&target, &source_path).await.unwrap_err();
std::assert_matches!(error, ImportSqliteError::InvalidTimestamp { .. });
let rows: i64 = sqlx::query_scalar("select count(*) from admin_users")
.fetch_one(&target)
.await
.unwrap();
assert_eq!(rows, 0);
}
#[tokio::test]
async fn import_sqlite_should_reject_unknown_source_version() {
let (_source_guard, source_path) = sqlite_v3_source("version").await;
let source = open_sqlite(&source_path, false).await;
sqlx::query("delete from schema_migrations where version = 3")
.execute(&source)
.await
.unwrap();
source.close().await;
let (target, _target_guard) = init_test_db("import-sqlite-version").await;
let error = import_sqlite(&target, &source_path).await.unwrap_err();
std::assert_matches!(
error,
ImportSqliteError::UnsupportedSourceVersion { actual: Some(2) }
);
}
#[tokio::test]
async fn import_sqlite_should_reject_nonempty_target() {
let (_source_guard, source_path) = sqlite_v3_source("nonempty").await;
let (target, _target_guard) = init_test_db("import-sqlite-nonempty").await;
sqlx::query(
"insert into admin_users (id, password_hash, created_at, updated_at)
values ($1, $2, now(), now())",
)
.bind("existing")
.bind("hash")
.execute(&target)
.await
.unwrap();
let error = import_sqlite(&target, &source_path).await.unwrap_err();
std::assert_matches!(error, ImportSqliteError::TargetNotEmpty);
}
#[tokio::test]
async fn import_sqlite_should_keep_blank_admin_api_key_disabled() {
let (_source_guard, source_path) = sqlite_v3_source("blank-admin-api-key").await;
let source = open_sqlite(&source_path, false).await;
sqlx::query(
"insert into runtime_settings values
(1, '{}', 3600, 2, 3, 50, 'smart', ' ', '2026-01-01T00:00:00Z')",
)
.execute(&source)
.await
.unwrap();
source.close().await;
let (target, _target_guard) = init_test_db("import-sqlite-blank-admin-api-key").await;
import_sqlite(&target, &source_path).await.unwrap();
let admin_api_key_hash: Option<String> =
sqlx::query_scalar("select admin_api_key_hash from runtime_settings where id = 1")
.fetch_one(&target)
.await
.unwrap();
assert!(admin_api_key_hash.is_none());
}
async fn sqlite_v3_source(label: &str) -> (tempfile::TempDir, PathBuf) {
let guard = tempfile::tempdir().unwrap();
let path = guard.path().join(format!("{label}.sqlite"));
let source = open_sqlite(&path, true).await;
sqlx::raw_sql(include_str!("../../fixtures/sqlite_v3.sql"))
.execute(&source)
.await
.unwrap();
source.close().await;
(guard, path)
}
async fn open_sqlite(path: &PathBuf, create: bool) -> sqlx::SqlitePool {
SqlitePoolOptions::new()
.max_connections(1)
.connect_with(
SqliteConnectOptions::new()
.filename(path)
.create_if_missing(create),
)
.await
.unwrap()
}
async fn seed_representative_v3_data(path: &PathBuf) {
let source = open_sqlite(path, false).await;
sqlx::raw_sql(
r#"
insert into admin_users values
('admin', 'argon-hash', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z');
insert into admin_sessions values
('session', 'admin', '2026-01-02T00:00:00Z', '2026-01-01T00:00:00Z');
insert into client_api_keys values
('key-1', 'legacy', 'sk_old_secre', 'sk_old_secret', null, 1,
'2026-01-01T00:00:00Z', null);
insert into runtime_settings values
(1, '{}', 3600, 2, 3, 50, 'smart', 'admin-machine-secret',
'2026-01-01T00:00:00Z');
insert into accounts
(id, email, chatgpt_account_id, chatgpt_user_id, label, plan_type,
access_token, refresh_token, access_token_expires_at, next_refresh_at,
status, quota_json, quota_fetched_at, quota_limit_reached,
quota_verify_required, quota_cooldown_until, cloudflare_cooldown_until,
added_at, updated_at)
values
('acct-1', 'a@example.com', 'chatgpt-1', 'user-1', null, 'plus',
'access', 'refresh', '2026-02-01T00:00:00Z', '2026-01-31T23:00:00Z',
'active', '{"remaining":1}', '2026-01-01T00:00:00Z', 0, 0, null, null,
'2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z'),
('acct-refreshing', 'refreshing@example.com', 'chatgpt-refreshing', 'user-refreshing',
null, 'plus', 'access-refreshing', 'refresh-refreshing', '2026-02-01T00:00:00Z',
'2026-01-31T23:00:00Z', 'refreshing', null, null, 0, 0, null, null,
'2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z');
insert into account_model_usage
(account_id, model, request_count, error_count, input_tokens, output_tokens,
cached_tokens, last_used_at)
values
('acct-1', 'gpt-5.5', 1, 0, 10, 2, 4, '2026-01-01T00:01:00Z');
insert into account_cookies values
('cookie-1', 'acct-1', '.chatgpt.com', 'session', 'secret', '/',
'not-a-cookie-date', '2026-01-01T00:00:00Z');
insert into account_refresh_leases values
('acct-1', 'owner', '2026-01-01T00:05:00Z', '2026-01-01T00:00:00Z');
insert into model_plan_snapshots values
('plus', '[]', '2026-01-01T00:00:00Z');
insert into session_affinities values
('response-1', 'acct-1', 'conversation-1', null, null, null, '[]', null,
'2026-01-02T00:00:00Z', '2026-01-01T00:00:00Z');
insert into usage_records
(id, request_id, kind, level, account_id, route, model, status_code,
transport, attempt_index, upstream_status_code, failure_class, response_id,
upstream_request_id, latency_ms, message, metadata_json, created_at)
values
('usage-success', 'req-1', 'v1.response', 'info', 'acct-1', '/v1/responses',
'gpt-5.5', 200, 'http', 0, 200, null, 'resp-1', 'upstream-1', 0,
'ok',
'{"requestedModel":"alias","upstreamModel":"gpt-5.5","serviceTier":"priority","firstTokenMs":0,"usage":{"inputTokens":12,"outputTokens":5,"cachedTokens":3,"reasoningTokens":2},"clientIp":"127.0.0.1"}',
'2026-01-01T00:01:00Z'),
('usage-error', 'req-2', 'v1.response', 'error', 'acct-1', '/v1/responses',
'gpt-5.5', 502, 'http', 1, 500, 'upstream', null, null, 10,
'failed', '{}', '2026-01-01T00:02:00Z'),
('usage-noise', null, 'diagnostic', 'info', null, null, null, null, null,
null, null, null, null, null, null, 'noise', '{}', '2026-01-01T00:03:00Z');
insert into ops_error_logs
(id, request_id, kind, account_id, route, model, status_code,
client_status_code, upstream_status_code, transport, attempt_index,
failure_class, response_id, upstream_request_id, latency_ms, message,
metadata_json, created_at)
values
('ops-1', 'req-3', 'v1.response', 'acct-1', '/v1/responses', 'gpt-5.5',
429, 429, 429, 'http', 0, 'quota', null, null, 5, 'limited', '{}',
'2026-01-01T00:04:00Z');
insert into usage_time_buckets
(bucket_start, account_id, model, service_tier, request_count, error_count,
input_tokens, output_tokens, cached_tokens, first_token_latency_sum,
first_token_latency_count, latency_sum, latency_count, max_latency_ms,
min_latency_ms, updated_at)
values
('2026-01-01T00:00:00Z', '', '', '', 3, 1, 12, 5, 3, 0, 1, 0, 1, 0, 0,
'2026-01-01T00:15:00Z');
"#,
)
.execute(&source)
.await
.unwrap();
source.close().await;
}
backend/tests/bootstrap/mod.rs
-1
View File
@@ -1,4 +1,3 @@
pub(crate) mod config;
pub(crate) mod import_sqlite;
pub(crate) mod services;
pub(crate) mod tasks;
backend/tests/fixtures/sqlite_v3.sql
Vendored
-382
View File
@@ -1,382 +0,0 @@
create table schema_migrations (
version integer primary key,
name text not null,
applied_at text not null
);
pragma foreign_keys = on;
-- ============================================
-- Admin Users
-- ============================================
create table admin_users (
id text primary key,
password_hash text not null,
created_at text not null,
updated_at text not null
);
-- ============================================
-- Admin Sessions
-- ============================================
create table admin_sessions (
id text primary key,
user_id text not null references admin_users(id) on delete cascade,
expires_at text not null,
created_at text not null
);
create index idx_admin_sessions_expires on admin_sessions(expires_at);
-- ============================================
-- Client API Keys
-- ============================================
create table client_api_keys (
id text primary key,
name text not null,
prefix text not null,
key text not null unique,
label text,
enabled integer not null default 1 check (enabled in (0, 1)),
created_at text not null,
last_used_at text
);
create index idx_client_api_keys_key_enabled on client_api_keys(key) where enabled = 1;
create index idx_client_api_keys_created_id on client_api_keys(created_at desc, id desc);
-- ============================================
-- Runtime Settings
-- ============================================
create table runtime_settings (
id integer primary key check (id = 1),
model_aliases_json text not null default '{}',
refresh_margin_seconds integer not null check (refresh_margin_seconds > 0),
refresh_concurrency integer not null check (refresh_concurrency > 0),
max_concurrent_per_account integer not null check (max_concurrent_per_account > 0),
request_interval_ms integer not null check (request_interval_ms >= 0),
rotation_strategy text not null check (rotation_strategy in ('smart', 'quota_reset_priority', 'round_robin', 'sticky')),
admin_api_key text,
updated_at text not null
);
-- ============================================
-- Accounts
-- ============================================
create table accounts (
id text primary key,
email text,
chatgpt_account_id text,
chatgpt_user_id text,
label text,
plan_type text,
access_token text not null,
refresh_token text,
access_token_expires_at text,
next_refresh_at text,
status text not null check (status in ('active', 'expired', 'quota_exhausted', 'refreshing', 'disabled', 'banned')),
quota_json text,
quota_fetched_at text,
quota_limit_reached integer not null default 0 check (quota_limit_reached in (0, 1)),
quota_verify_required integer not null default 0 check (quota_verify_required in (0, 1)),
quota_cooldown_until text,
cloudflare_cooldown_until text,
added_at text not null,
updated_at text not null
);
create index idx_accounts_status on accounts(status);
create index idx_accounts_added_id on accounts(added_at desc, id desc);
create unique index ux_accounts_chatgpt_identity
on accounts(chatgpt_account_id, coalesce(chatgpt_user_id, ''))
where chatgpt_account_id is not null;
-- ============================================
-- Account Refresh Leases
-- ============================================
create table account_refresh_leases (
account_id text primary key references accounts(id) on delete cascade,
owner text not null,
expires_at text not null,
updated_at text not null
);
create index idx_account_refresh_leases_expires on account_refresh_leases(expires_at);
-- ============================================
-- Account Usage Statistics
-- ============================================
create table account_usage (
account_id text primary key references accounts(id) on delete cascade,
request_count integer not null default 0 check (request_count >= 0),
empty_response_count integer not null default 0 check (empty_response_count >= 0),
input_tokens integer not null default 0 check (input_tokens >= 0),
output_tokens integer not null default 0 check (output_tokens >= 0),
cached_tokens integer not null default 0 check (cached_tokens >= 0),
reasoning_tokens integer not null default 0 check (reasoning_tokens >= 0),
total_tokens integer not null default 0 check (total_tokens >= 0),
image_input_tokens integer not null default 0 check (image_input_tokens >= 0),
image_output_tokens integer not null default 0 check (image_output_tokens >= 0),
image_request_count integer not null default 0 check (image_request_count >= 0),
image_request_failed_count integer not null default 0 check (image_request_failed_count >= 0),
window_request_count integer not null default 0 check (window_request_count >= 0),
window_input_tokens integer not null default 0 check (window_input_tokens >= 0),
window_output_tokens integer not null default 0 check (window_output_tokens >= 0),
window_cached_tokens integer not null default 0 check (window_cached_tokens >= 0),
window_image_input_tokens integer not null default 0 check (window_image_input_tokens >= 0),
window_image_output_tokens integer not null default 0 check (window_image_output_tokens >= 0),
window_image_request_count integer not null default 0 check (window_image_request_count >= 0),
window_image_request_failed_count integer not null default 0 check (window_image_request_failed_count >= 0),
window_started_at text,
window_reset_at text,
limit_window_seconds integer check (limit_window_seconds is null or limit_window_seconds > 0),
last_used_at text
);
create index idx_account_usage_last_used_account
on account_usage(last_used_at desc, account_id desc);
-- ============================================
-- Account Model Usage Statistics
-- ============================================
create table account_model_usage (
account_id text not null references accounts(id) on delete cascade,
model text not null,
request_count integer not null default 0 check (request_count >= 0),
error_count integer not null default 0 check (error_count >= 0),
input_tokens integer not null default 0 check (input_tokens >= 0),
output_tokens integer not null default 0 check (output_tokens >= 0),
cached_tokens integer not null default 0 check (cached_tokens >= 0),
last_used_at text,
primary key (account_id, model)
);
create index idx_account_model_usage_last_used
on account_model_usage(last_used_at desc, account_id, model);
-- ============================================
-- Usage Time Buckets
-- ============================================
create table usage_time_buckets (
bucket_start text not null,
account_id text not null default '',
model text not null default '',
service_tier text not null default '',
request_count integer not null default 0 check (request_count >= 0),
error_count integer not null default 0 check (error_count >= 0),
input_tokens integer not null default 0 check (input_tokens >= 0),
output_tokens integer not null default 0 check (output_tokens >= 0),
cached_tokens integer not null default 0 check (cached_tokens >= 0),
first_token_latency_sum integer not null default 0 check (first_token_latency_sum >= 0),
first_token_latency_count integer not null default 0 check (first_token_latency_count >= 0),
latency_sum integer not null default 0 check (latency_sum >= 0),
latency_count integer not null default 0 check (latency_count >= 0),
max_latency_ms integer not null default 0 check (max_latency_ms >= 0),
min_latency_ms integer not null default 0 check (min_latency_ms >= 0),
updated_at text not null,
primary key (bucket_start, account_id, model, service_tier)
);
create index idx_usage_time_buckets_bucket
on usage_time_buckets(bucket_start);
create index idx_usage_time_buckets_model_bucket
on usage_time_buckets(model, bucket_start);
-- ============================================
-- Account Cookies
-- ============================================
create table account_cookies (
id text primary key,
account_id text not null references accounts(id) on delete cascade,
domain text not null,
name text not null,
value text not null,
path text not null default '/',
expires_at text,
updated_at text not null,
unique(account_id, domain, name, path)
);
create index idx_account_cookies_account on account_cookies(account_id);
create index idx_account_cookies_account_domain on account_cookies(account_id, domain);
create index idx_account_cookies_expires
on account_cookies(expires_at)
where expires_at is not null;
-- ============================================
-- Device Fingerprints
-- ============================================
create table fingerprints (
id text primary key,
originator text not null,
app_version text not null,
build_number text not null,
platform text not null,
arch text not null,
chromium_version text not null,
user_agent_template text not null,
default_headers_json text not null,
header_order_json text not null,
source text not null,
created_at text not null,
updated_at text not null
);
create table fingerprint_update_history (
id text primary key,
current_fingerprint_id text not null references fingerprints(id) on delete cascade,
app_version text not null,
build_number text not null,
chromium_version text,
source text not null,
manifest_json text,
created_at text not null
);
create index idx_fingerprint_update_history_created_id
on fingerprint_update_history(created_at desc, id desc);
-- ============================================
-- Usage Records
-- ============================================
create table usage_records (
id text primary key,
request_id text,
kind text not null,
level text not null check (level in ('debug', 'info', 'warn', 'error')),
account_id text,
route text,
model text,
status_code integer check (status_code is null or (status_code >= 100 and status_code <= 599)),
transport text,
attempt_index integer check (attempt_index is null or attempt_index >= 0),
upstream_status_code integer check (upstream_status_code is null or (upstream_status_code >= 100 and upstream_status_code <= 599)),
failure_class text,
response_id text,
upstream_request_id text,
latency_ms integer check (latency_ms is null or latency_ms >= 0),
message text not null,
metadata_json text not null,
created_at text not null
);
create index idx_usage_records_created_id on usage_records(created_at desc, id desc);
create index idx_usage_records_kind_created on usage_records(kind, created_at desc);
create index idx_usage_records_request_id on usage_records(request_id);
create index idx_usage_records_account on usage_records(account_id, created_at desc) where account_id is not null;
create index idx_usage_records_transport on usage_records(transport, created_at desc) where transport is not null;
create index idx_usage_records_failure_class on usage_records(failure_class, created_at desc) where failure_class is not null;
create index idx_usage_records_response_id on usage_records(response_id) where response_id is not null;
create index idx_usage_records_upstream_request_id on usage_records(upstream_request_id) where upstream_request_id is not null;
create index idx_usage_records_level_created on usage_records(level, created_at desc);
create index idx_usage_records_route_created on usage_records(route, created_at desc) where route is not null;
create index idx_usage_records_model_created on usage_records(model, created_at desc) where model is not null;
create index idx_usage_records_status_created on usage_records(status_code, created_at desc) where status_code is not null;
create index idx_usage_records_upstream_status_created on usage_records(upstream_status_code, created_at desc) where upstream_status_code is not null;
-- ============================================
-- Model Plan Snapshots
-- ============================================
create table model_plan_snapshots (
plan_type text primary key,
models_json text not null,
fetched_at text not null
);
-- ============================================
-- Session Affinities
-- ============================================
create table session_affinities (
response_id text primary key,
account_id text not null references accounts(id) on delete cascade,
conversation_id text not null,
turn_state text,
instructions_hash text,
input_tokens integer check (input_tokens is null or input_tokens >= 0),
function_call_ids_json text not null default '[]',
variant_hash text,
expires_at text not null,
created_at text not null
);
create index idx_session_affinities_conversation on session_affinities(conversation_id, created_at desc);
create index idx_session_affinities_expires on session_affinities(expires_at);
create index idx_session_affinities_active_order on session_affinities(expires_at, created_at, response_id);
create table runtime_settings_new (
id integer primary key check (id = 1),
model_aliases_json text not null default '{}',
refresh_margin_seconds integer not null check (refresh_margin_seconds > 0),
refresh_concurrency integer not null check (refresh_concurrency > 0),
max_concurrent_per_account integer not null check (max_concurrent_per_account > 0),
request_interval_ms integer not null check (request_interval_ms >= 0),
rotation_strategy text not null check (rotation_strategy in ('smart', 'quota_reset_priority', 'round_robin', 'sticky')),
admin_api_key text,
updated_at text not null
);
insert into runtime_settings_new (
id,
model_aliases_json,
refresh_margin_seconds,
refresh_concurrency,
max_concurrent_per_account,
request_interval_ms,
rotation_strategy,
admin_api_key,
updated_at
)
select
id,
model_aliases_json,
refresh_margin_seconds,
refresh_concurrency,
max_concurrent_per_account,
request_interval_ms,
case rotation_strategy
when 'least_used' then 'quota_reset_priority'
else rotation_strategy
end,
admin_api_key,
updated_at
from runtime_settings;
drop table runtime_settings;
alter table runtime_settings_new rename to runtime_settings;
create table ops_error_logs (
id text primary key,
request_id text,
kind text not null,
account_id text,
route text,
model text,
status_code integer check (status_code is null or (status_code >= 100 and status_code <= 599)),
client_status_code integer check (client_status_code is null or (client_status_code >= 100 and client_status_code <= 599)),
upstream_status_code integer check (upstream_status_code is null or (upstream_status_code >= 100 and upstream_status_code <= 599)),
transport text,
attempt_index integer check (attempt_index is null or attempt_index >= 0),
failure_class text,
response_id text,
upstream_request_id text,
latency_ms integer check (latency_ms is null or latency_ms >= 0),
message text not null,
metadata_json text not null,
created_at text not null
);
create index idx_ops_error_logs_created_id on ops_error_logs(created_at desc, id desc);
create index idx_ops_error_logs_request_id on ops_error_logs(request_id) where request_id is not null;
create index idx_ops_error_logs_account on ops_error_logs(account_id, created_at desc) where account_id is not null;
create index idx_ops_error_logs_route_created on ops_error_logs(route, created_at desc) where route is not null;
create index idx_ops_error_logs_model_created on ops_error_logs(model, created_at desc) where model is not null;
create index idx_ops_error_logs_status_created on ops_error_logs(status_code, created_at desc) where status_code is not null;
create index idx_ops_error_logs_transport_created on ops_error_logs(transport, created_at desc) where transport is not null;
create index idx_ops_error_logs_failure_class on ops_error_logs(failure_class, created_at desc) where failure_class is not null;
create index idx_ops_error_logs_response_id on ops_error_logs(response_id) where response_id is not null;
create index idx_ops_error_logs_upstream_request_id on ops_error_logs(upstream_request_id) where upstream_request_id is not null;
insert into schema_migrations (version, name, applied_at) values
(1, 'initial', '2026-01-01T00:00:00Z'),
(2, 'rename_rotation_strategies', '2026-01-01T00:00:00Z'),
(3, 'ops_error_logs', '2026-01-01T00:00:00Z');
deploy/Dockerfile.dockerignore
-6
View File
@@ -13,12 +13,6 @@ data/
logs/
deploy/data/
deploy/logs/
*.sqlite
*.sqlite3
*.db
*.db-shm
*.db-wal
*.sqlite-*
*.log
# Local configuration.
docs/architecture.md
-5
View File
@@ -81,7 +81,6 @@ backend/src/
- `main.rs`：命令行入口。
- `serve` 或无参数：启动 HTTP 服务。
- `import-sqlite <path.sqlite>`：把 SQLite v3 历史库一次性导入空的 PostgreSQL。
- `rebuild-buckets`：从保留期内的成功、失败事实重建请求时间桶。
- `lib.rs`：导出后端顶层模块，供二进制和集成测试使用。
@@ -331,9 +330,6 @@ bootstrap/
├── services.rs
├── state.rs
├── shutdown.rs
├── import_sqlite/
│ ├── core.rs
│ └── telemetry.rs
└── tasks/
├── coordinator.rs
├── periodic.rs
@@ -350,7 +346,6 @@ bootstrap/
- `services.rs`：创建存储、领域服务、上游客户端、路由和后台任务。
- `state.rs`：进程级运行配置和 PostgreSQL/Redis 健康检查。
- `shutdown.rs`：信号、管理端重启和进程替换协调。
- `import_sqlite`：SQLite v3 到空 PostgreSQL 的一次性事务导入。
- `tasks/coordinator.rs`：统一启动和关闭后台任务。
## 启动流程
skills/codex-proxy-rs/SKILL.md
+1 -1
View File
@@ -19,7 +19,7 @@ description: Codex Proxy RS 仓库开发指南。Use when working on its Rust/Ax
## 任务分流
- 后端：遵循 `backend/src` 当前领域边界；测试统一放在 `backend/tests`，禁止写入 `backend/src`。
- 存储：PostgreSQL 是权威持久化，Redis 保存运行态数据，本地数据目录保存身份密钥和更新状态；SQLite 只用于历史库导入。
- 存储：PostgreSQL 是权威持久化，Redis 保存运行态数据，本地数据目录保存身份密钥和更新状态。
- 前端：使用 Vue 3 `<script setup>`、TypeScript、Tailwind v4、已有组件和主题 token。
- Docker / 发布：保留非 root runtime、Compose 密钥注入、命名卷、GitHub Release、GHCR 和在线更新契约。
- 文档：README 面向部署者和使用者；`docs/architecture.md` 面向开发者。不要新增过程型审计或迁移文档。
skills/codex-proxy-rs/references/repo-guide.md
+1 -1
View File
@@ -77,7 +77,7 @@ cd backend
cargo run -- serve
```
`main.rs` 还提供内部维护命令 `import-sqlite` 和 `rebuild-buckets`。SQLite 导入只面向历史 schema v3，不是运行时存储路径。
`main.rs` 还提供内部维护命令 `rebuild-buckets`，用于从保留期内的请求事实重建聚合桶。
## 存储边界