diff --git a/.gitignore b/.gitignore index 3a54f3979..f5102e4c8 100644 --- a/.gitignore +++ b/.gitignore @@ -16,12 +16,6 @@ # Runtime data /.runtime/ -*.sqlite -*.sqlite3 -*.sqlite-* -*.db -*.db-shm -*.db-wal *.log # Local configuration and secrets diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 371cf1410..8f25c45ad 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -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", ] diff --git a/backend/Cargo.toml b/backend/Cargo.toml index c5a9e08d3..bd40ecfdb 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -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" diff --git a/backend/src/bootstrap/import_sqlite/core.rs b/backend/src/bootstrap/import_sqlite/core.rs deleted file mode 100644 index 33bb98e2b..000000000 --- a/backend/src/bootstrap/import_sqlite/core.rs +++ /dev/null @@ -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::("id")?) - .bind(row.try_get::("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::("id")?) - .bind(row.try_get::("name")?) - .bind(row.try_get::("prefix")?) - .bind(row.try_get::("key")?) - .bind(row.try_get::, _>("label")?) - .bind(row.try_get::("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::, _>("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::("id")?) - .bind(Json(parse_json( - "runtime_settings", - "model_aliases_json", - &aliases, - )?)) - .bind(row.try_get::("refresh_margin_seconds")?) - .bind(row.try_get::("refresh_concurrency")?) - .bind(row.try_get::("max_concurrent_per_account")?) - .bind(row.try_get::("request_interval_ms")?) - .bind(row.try_get::("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::("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::, _>("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::("id")?) - .bind(row.try_get::, _>("email")?) - .bind(row.try_get::, _>("chatgpt_account_id")?) - .bind(row.try_get::, _>("chatgpt_user_id")?) - .bind(row.try_get::, _>("label")?) - .bind(row.try_get::, _>("plan_type")?) - .bind(row.try_get::("access_token")?) - .bind(row.try_get::, _>("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::("quota_limit_reached")? == 1) - .bind(row.try_get::("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::("account_id")?) - .bind(row.try_get::("request_count")?) - .bind(row.try_get::("empty_response_count")?) - .bind(row.try_get::("input_tokens")?) - .bind(row.try_get::("output_tokens")?) - .bind(row.try_get::("cached_tokens")?) - .bind(row.try_get::("reasoning_tokens")?) - .bind(row.try_get::("total_tokens")?) - .bind(row.try_get::("image_input_tokens")?) - .bind(row.try_get::("image_output_tokens")?) - .bind(row.try_get::("image_request_count")?) - .bind(row.try_get::("image_request_failed_count")?) - .bind(row.try_get::("window_request_count")?) - .bind(row.try_get::("window_input_tokens")?) - .bind(row.try_get::("window_output_tokens")?) - .bind(row.try_get::("window_cached_tokens")?) - .bind(row.try_get::("window_image_input_tokens")?) - .bind(row.try_get::("window_image_output_tokens")?) - .bind(row.try_get::("window_image_request_count")?) - .bind(row.try_get::("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::, _>("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::, _>("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::("id")?) - .bind(row.try_get::("account_id")?) - .bind(row.try_get::("domain")?) - .bind(row.try_get::("name")?) - .bind(row.try_get::("value")?) - .bind(row.try_get::("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::("id")?) - .bind(row.try_get::("originator")?) - .bind(row.try_get::("app_version")?) - .bind(row.try_get::("build_number")?) - .bind(row.try_get::("platform")?) - .bind(row.try_get::("arch")?) - .bind(row.try_get::("chromium_version")?) - .bind(row.try_get::("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::("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::, _>("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::("id")?) - .bind(row.try_get::("current_fingerprint_id")?) - .bind(row.try_get::("app_version")?) - .bind(row.try_get::("build_number")?) - .bind(row.try_get::, _>("chromium_version")?) - .bind(row.try_get::("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(()) -} diff --git a/backend/src/bootstrap/import_sqlite/mod.rs b/backend/src/bootstrap/import_sqlite/mod.rs deleted file mode 100644 index 68e15b701..000000000 --- a/backend/src/bootstrap/import_sqlite/mod.rs +++ /dev/null @@ -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 }, - #[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, -) -> Result { - 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,) = 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, 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, -) -> Result>, ImportSqliteError> { - value - .map(|value| parse_timestamp(table, column, value)) - .transpose() -} - -fn parse_cookie_expiration(value: &str) -> Option> { - 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 { - 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, -) -> Result, ImportSqliteError> { - value - .map(|value| parse_json(table, column, &value)) - .transpose() -} - -fn normalized_dimension(value: Option) -> String { - value - .map(|value| value.trim().to_string()) - .filter(|value| !value.is_empty()) - .unwrap_or_else(|| "__unknown__".to_string()) -} diff --git a/backend/src/bootstrap/import_sqlite/telemetry.rs b/backend/src/bootstrap/import_sqlite/telemetry.rs deleted file mode 100644 index a34b28402..000000000 --- a/backend/src/bootstrap/import_sqlite/telemetry.rs +++ /dev/null @@ -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 = row.try_get("account_id")?; - let model: Option = row.try_get("model")?; - let status_code: Option = 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::("id")?) - .bind(row.try_get::, _>("request_id")?) - .bind(row.try_get::("kind")?) - .bind(row.try_get::, _>("route")?) - .bind(account_id) - .bind(model) - .bind(requested_model) - .bind(upstream_model) - .bind(service_tier) - .bind(status_code as i32) - .bind(row.try_get::, _>("transport")?) - .bind(row.try_get::, _>("attempt_index")?) - .bind(row.try_get::, _>("response_id")?) - .bind(row.try_get::, _>("upstream_request_id")?) - .bind(row.try_get::, _>("latency_ms")?) - .bind(first_token_ms) - .bind(input_tokens) - .bind(output_tokens) - .bind(cached_tokens) - .bind(reasoning_tokens) - .bind(row.try_get::("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::("id")?) - .bind(row.try_get::, _>("request_id")?) - .bind(row.try_get::("kind")?) - .bind(row.try_get::, _>("account_id")?) - .bind(row.try_get::, _>("route")?) - .bind(row.try_get::, _>("model")?) - .bind(optional_status(row.try_get("status_code")?)) - .bind(optional_status(row.try_get("upstream_status_code")?)) - .bind(row.try_get::, _>("transport")?) - .bind(row.try_get::, _>("attempt_index")?) - .bind(row.try_get::, _>("failure_class")?) - .bind(row.try_get::, _>("response_id")?) - .bind(row.try_get::, _>("upstream_request_id")?) - .bind(row.try_get::, _>("latency_ms")?) - .bind(row.try_get::("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::("id")?) - .bind(row.try_get::, _>("request_id")?) - .bind(row.try_get::("kind")?) - .bind(row.try_get::, _>("account_id")?) - .bind(row.try_get::, _>("route")?) - .bind(row.try_get::, _>("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::, _>("transport")?) - .bind(row.try_get::, _>("attempt_index")?) - .bind(row.try_get::, _>("failure_class")?) - .bind(row.try_get::, _>("response_id")?) - .bind(row.try_get::, _>("upstream_request_id")?) - .bind(row.try_get::, _>("latency_ms")?) - .bind(row.try_get::("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::("input_tokens")?) - .bind(row.try_get::("output_tokens")?) - .bind(row.try_get::("cached_tokens")?) - .bind(row.try_get::("first_token_latency_sum")?) - .bind(row.try_get::("first_token_latency_count")?) - .bind(row.try_get::("latency_sum")?) - .bind(row.try_get::("latency_count")?) - .bind(row.try_get::("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) -> Option { - status.and_then(|status| i32::try_from(status).ok()) -} - -fn metadata_string(metadata: &Value, keys: &[&str]) -> Option { - 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 { - 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 { - 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); - } -} diff --git a/backend/src/bootstrap/mod.rs b/backend/src/bootstrap/mod.rs index d33502480..c2ea2f814 100644 --- a/backend/src/bootstrap/mod.rs +++ b/backend/src/bootstrap/mod.rs @@ -1,7 +1,6 @@ //! 进程装配、共享状态与后台任务。 pub mod config; -pub mod import_sqlite; pub mod services; pub mod shutdown; pub mod state; diff --git a/backend/src/main.rs b/backend/src/main.rs index 5249dfb69..2960d20f3 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -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> { @@ -8,19 +8,6 @@ async fn main() -> Result<(), Box> { 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 "))?; - 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> { 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()), } diff --git a/backend/tests/bootstrap/import_sqlite/mod.rs b/backend/tests/bootstrap/import_sqlite/mod.rs deleted file mode 100644 index 529a9802f..000000000 --- a/backend/tests/bootstrap/import_sqlite/mod.rs +++ /dev/null @@ -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>) = - 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::, _>("requested_model") - .as_deref(), - Some("alias") - ); - assert_eq!( - success - .get::, _>("upstream_model") - .as_deref(), - Some("gpt-5.5") - ); - assert_eq!( - success.get::, _>("service_tier").as_deref(), - Some("priority") - ); - assert_eq!(success.get::, _>("first_token_ms"), Some(0)); - assert_eq!(success.get::, _>("input_tokens"), Some(12)); - assert_eq!(success.get::, _>("output_tokens"), Some(5)); - assert_eq!(success.get::, _>("cached_tokens"), Some(3)); - assert_eq!(success.get::, _>("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) = 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 = - 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; -} diff --git a/backend/tests/bootstrap/mod.rs b/backend/tests/bootstrap/mod.rs index a9721b8b4..52c2dc1db 100644 --- a/backend/tests/bootstrap/mod.rs +++ b/backend/tests/bootstrap/mod.rs @@ -1,4 +1,3 @@ pub(crate) mod config; -pub(crate) mod import_sqlite; pub(crate) mod services; pub(crate) mod tasks; diff --git a/backend/tests/fixtures/sqlite_v3.sql b/backend/tests/fixtures/sqlite_v3.sql deleted file mode 100644 index 5b104610a..000000000 --- a/backend/tests/fixtures/sqlite_v3.sql +++ /dev/null @@ -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'); diff --git a/deploy/Dockerfile.dockerignore b/deploy/Dockerfile.dockerignore index 8f04c4391..a130a34a4 100644 --- a/deploy/Dockerfile.dockerignore +++ b/deploy/Dockerfile.dockerignore @@ -13,12 +13,6 @@ data/ logs/ deploy/data/ deploy/logs/ -*.sqlite -*.sqlite3 -*.db -*.db-shm -*.db-wal -*.sqlite-* *.log # Local configuration. diff --git a/docs/architecture.md b/docs/architecture.md index fd75cdf5e..f428ab831 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -81,7 +81,6 @@ backend/src/ - `main.rs`:命令行入口。 - `serve` 或无参数:启动 HTTP 服务。 - - `import-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`:统一启动和关闭后台任务。 ## 启动流程 diff --git a/skills/codex-proxy-rs/SKILL.md b/skills/codex-proxy-rs/SKILL.md index 5039caa39..1463fdcfe 100644 --- a/skills/codex-proxy-rs/SKILL.md +++ b/skills/codex-proxy-rs/SKILL.md @@ -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 `