i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-10-07 01:47:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(data): 解耦 API Key 历史统计身份

Browse Source
This commit is contained in:
MMEXA
2026-07-18 04:42:45 +08:00
parent 4d9a648202
commit 98fbf029fc
4 changed files with 222 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aether-data/runtime/src/lifecycle/bootstrap/postgres.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use tracing::info;
// Generated by build.rs from schema/bootstrap/postgres.
pub(crate) static EMPTY_DATABASE_SNAPSHOT_SQL: &str =
include_str!(concat!(env!("OUT_DIR"), "/empty_database_snapshot.sql"));
pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260718000000;
pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260718010000;
const PUBLIC_BASE_TABLE_COUNT_SQL: &str = r#"
SELECT COUNT(*)::BIGINT
crates/aether-data/runtime/src/lifecycle/migrate/tests.rs
+216 -22
View File
@@ -237,6 +237,48 @@ SELECT EXISTS (
.await
}
async fn foreign_key_exists(
pool: &PgPool,
table_name: &str,
constraint_name: &str,
) -> Result<bool, sqlx::Error> {
query_scalar::<_, bool>(
r#"
SELECT EXISTS (
SELECT 1
FROM pg_catalog.pg_constraint
WHERE conname = $1
AND conrelid = to_regclass($2)
AND contype = 'f'
)
"#,
)
.bind(constraint_name)
.bind(format!("public.{table_name}"))
.fetch_one(pool)
.await
}
fn historical_stats_day() -> chrono::DateTime<chrono::Utc> {
chrono::DateTime::parse_from_rfc3339("2026-07-17T00:00:00Z")
.expect("stats day should parse")
.with_timezone(&chrono::Utc)
}
fn postgres_backend(database_url: &str) -> crate::PostgresBackend {
crate::PostgresBackend::from_config(crate::driver::postgres::PostgresPoolConfig {
database_url: database_url.to_string(),
min_connections: 1,
max_connections: 4,
acquire_timeout_ms: 1_000,
idle_timeout_ms: 5_000,
max_lifetime_ms: 30_000,
statement_cache_capacity: 64,
require_ssl: false,
})
.expect("postgres backend should build")
}
#[test]
fn baseline_migration_restores_search_path_for_sqlx_bookkeeping() {
let baseline = POSTGRES_MIGRATOR
@@ -330,6 +372,7 @@ fn empty_database_snapshot_covers_current_cutoff_versions() {
20260715130100,
20260716000000,
20260718000000,
20260718010000,
]
);
}
@@ -519,6 +562,25 @@ fn request_candidate_api_key_identity_is_decoupled_for_historical_ingestion() {
);
}
#[test]
fn stats_daily_api_key_identity_is_decoupled_for_historical_reaggregation() {
let migration = POSTGRES_MIGRATOR
.iter()
.find(|migration| migration.version == 20260718010000)
.expect("daily API key stats identity decoupling migration should be embedded");
assert!(
migration
.sql
.contains("DROP CONSTRAINT IF EXISTS stats_daily_api_key_api_key_id_fkey"),
"migration should drop stats_daily_api_key_api_key_id_fkey"
);
assert!(
!EMPTY_DATABASE_SNAPSHOT_SQL.contains("ADD CONSTRAINT stats_daily_api_key_api_key_id_fkey"),
"fresh bootstrap snapshot should not recreate stats_daily_api_key_api_key_id_fkey"
);
}
#[test]
fn empty_database_snapshot_sql_includes_payment_gateway_and_plans() {
assert!(EMPTY_DATABASE_SNAPSHOT_SQL.contains("payment_provider character varying(64)"));
@@ -1603,6 +1665,7 @@ fn pending_migrations_from_applied_skips_versions_already_applied() {
20260715130100,
20260716000000,
20260718000000,
20260718010000,
]
);
}
@@ -2209,14 +2272,18 @@ INSERT INTO public.usage (
id,
request_id,
api_key_id,
api_key_name,
provider_name,
model
model,
created_at
) VALUES (
'usage-before-api-key-delete',
'usage-request-before-api-key-delete',
'deleted-api-key',
'Deleted API Key Snapshot',
'historical-provider',
'historical-model'
'historical-model',
'2026-07-17 07:03:43+00'
)
"#,
)
@@ -2246,6 +2313,38 @@ INSERT INTO public.usage (
.expect("usage API key identity should be readable");
assert_eq!(usage_api_key_id.as_deref(), Some("deleted-api-key"));
let stats_day = historical_stats_day();
let stats_backend = postgres_backend(server.database_url());
let stats_summary = stats_backend
.aggregate_stats_daily(&crate::StatsDailyAggregationInput {
target_day_utc: stats_day,
aggregated_at: stats_day + chrono::Duration::days(1),
})
.await
.expect("daily aggregation should accept a deleted API Key identity")
.expect("daily aggregation should find the historical usage day");
assert_eq!(stats_summary.day_start_utc, stats_day);
assert_eq!(stats_summary.total_requests, 1);
assert_eq!(stats_summary.api_key_rows, 1);
let stats_api_key_id: Option<String> = query_scalar(
"SELECT api_key_id FROM public.stats_daily_api_key WHERE api_key_id = 'deleted-api-key'",
)
.fetch_one(&pool)
.await
.expect("daily stats API key identity should be readable");
assert_eq!(stats_api_key_id.as_deref(), Some("deleted-api-key"));
let stats_api_key_name: Option<String> = query_scalar(
"SELECT api_key_name FROM public.stats_daily_api_key WHERE api_key_id = 'deleted-api-key'",
)
.fetch_one(&pool)
.await
.expect("daily stats API key name snapshot should be readable");
assert_eq!(
stats_api_key_name.as_deref(),
Some("Deleted API Key Snapshot")
);
query(
r#"
INSERT INTO public.request_candidates (
@@ -2357,17 +2456,11 @@ async fn postgres_request_candidate_migration_decouples_legacy_api_key_foreign_k
let pool = PgPool::connect(server.database_url())
.await
.expect("pool should connect");
let legacy_constraint_exists: bool = query_scalar(
r#"
SELECT EXISTS (
SELECT 1
FROM pg_catalog.pg_constraint
WHERE conname = 'request_candidates_api_key_id_fkey'
AND conrelid = 'public.request_candidates'::regclass
)
"#,
let legacy_constraint_exists = foreign_key_exists(
&pool,
"request_candidates",
"request_candidates_api_key_id_fkey",
)
.fetch_one(&pool)
.await
.expect("legacy request candidate constraint should be readable");
assert!(legacy_constraint_exists);
@@ -2376,17 +2469,11 @@ SELECT EXISTS (
.await
.expect("request candidate API key decoupling migration should apply");
let upgraded_constraint_exists: bool = query_scalar(
r#"
SELECT EXISTS (
SELECT 1
FROM pg_catalog.pg_constraint
WHERE conname = 'request_candidates_api_key_id_fkey'
AND conrelid = 'public.request_candidates'::regclass
)
"#,
let upgraded_constraint_exists = foreign_key_exists(
&pool,
"request_candidates",
"request_candidates_api_key_id_fkey",
)
.fetch_one(&pool)
.await
.expect("upgraded request candidate constraint should be readable");
assert!(!upgraded_constraint_exists);
@@ -2421,6 +2508,113 @@ INSERT INTO public.request_candidates (
assert_eq!(api_key_id.as_deref(), Some("legacy-deleted-api-key"));
}
#[tokio::test]
async fn postgres_stats_daily_api_key_migration_decouples_legacy_foreign_key() {
const PREVIOUS_SNAPSHOT_CUTOFF_VERSION: i64 = 20260718000000;
let Some(server) = ManagedPostgresServer::try_start()
.await
.expect("postgres daily API key stats migration test should start or skip")
else {
return;
};
let mut conn = PgConnection::connect(server.database_url())
.await
.expect("postgres migration connection should open");
conn.ensure_migrations_table()
.await
.expect("migration table should be created");
for migration in POSTGRES_MIGRATOR
.iter()
.filter(|migration| migration.version <= PREVIOUS_SNAPSHOT_CUTOFF_VERSION)
{
conn.apply(migration)
.await
.expect("previous postgres migration should apply");
}
drop(conn);
let pool = PgPool::connect(server.database_url())
.await
.expect("pool should connect");
let legacy_constraint_exists = foreign_key_exists(
&pool,
"stats_daily_api_key",
"stats_daily_api_key_api_key_id_fkey",
)
.await
.expect("legacy daily API key stats constraint should be readable");
assert!(legacy_constraint_exists);
super::run_migrations(&pool)
.await
.expect("daily API key stats identity decoupling migration should apply");
let upgraded_constraint_exists = foreign_key_exists(
&pool,
"stats_daily_api_key",
"stats_daily_api_key_api_key_id_fkey",
)
.await
.expect("upgraded daily API key stats constraint should be readable");
assert!(!upgraded_constraint_exists);
query(
r#"
INSERT INTO public.usage (
id,
request_id,
api_key_id,
api_key_name,
provider_name,
model,
created_at
) VALUES (
'legacy-deleted-api-key-usage',
'legacy-deleted-api-key-request',
'legacy-deleted-api-key',
'Deleted API Key Snapshot',
'historical-provider',
'historical-model',
'2026-07-17 07:03:43+00'
)
"#,
)
.execute(&pool)
.await
.expect("upgraded schema should accept a deleted API Key usage snapshot");
let stats_day = historical_stats_day();
let stats_summary = postgres_backend(server.database_url())
.aggregate_stats_daily(&crate::StatsDailyAggregationInput {
target_day_utc: stats_day,
aggregated_at: stats_day + chrono::Duration::days(1),
})
.await
.expect("upgraded schema should reaggregate a deleted API Key identity")
.expect("upgraded schema should find the historical usage day");
assert_eq!(stats_summary.day_start_utc, stats_day);
assert_eq!(stats_summary.total_requests, 1);
assert_eq!(stats_summary.api_key_rows, 1);
let api_key_id: Option<String> = query_scalar(
"SELECT api_key_id FROM public.stats_daily_api_key WHERE api_key_id = 'legacy-deleted-api-key'",
)
.fetch_one(&pool)
.await
.expect("upgraded daily API key stats identity should be readable");
assert_eq!(api_key_id.as_deref(), Some("legacy-deleted-api-key"));
let api_key_name: Option<String> = query_scalar(
"SELECT api_key_name FROM public.stats_daily_api_key WHERE api_key_id = 'legacy-deleted-api-key'",
)
.fetch_one(&pool)
.await
.expect("upgraded daily API key stats name snapshot should be readable");
assert_eq!(api_key_name.as_deref(), Some("Deleted API Key Snapshot"));
}
#[tokio::test]
async fn postgres_usage_billing_facts_total_tokens_counts_cached_input_once() {
const LEGACY_VIEW_MIGRATION_VERSION: i64 = 20260505130000;