diff --git a/backend/src/api/admin/accounts_routes/mod.rs b/backend/src/api/admin/accounts_routes/mod.rs index f6b08e4a8..3b7f846f0 100644 --- a/backend/src/api/admin/accounts_routes/mod.rs +++ b/backend/src/api/admin/accounts_routes/mod.rs @@ -20,8 +20,7 @@ use crate::{ api::admin::{ response::{ parse_editable_update, AdminEnvelope, AdminError, AdminResponse, BatchDeleteData, - CursorPageMeta, EditableUpdateMessages, NumberedPageMeta, PageMeta, ADMIN_OK_CODE, - ADMIN_OK_MESSAGE, + EditableUpdateMessages, PageMeta, ADMIN_OK_CODE, ADMIN_OK_MESSAGE, }, session::AdminAuth, }, @@ -41,7 +40,7 @@ use crate::{ format::{ format_cost, format_percent, format_plain_number, format_tokens, nonnegative_i64_to_u64, }, - json::{clamp_limit, clamp_page, total_pages, Page}, + json::{clamp_limit, clamp_page, total_pages}, time::{china_datetime, china_relative_time, china_rfc3339}, }, telemetry::{ @@ -57,8 +56,6 @@ const ACCOUNT_EXPORT_CONFIRMATION: &str = "export_sensitive_accounts"; #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct AccountsQuery { - cursor: Option, - limit: Option, page: Option, page_size: Option, search: Option, @@ -361,22 +358,7 @@ struct AdminAccountPageEnvelope { } impl AdminAccountPageEnvelope { - fn cursor(page: Page, limit: u32, summary: AdminAccountSummaryData) -> Self { - Self { - code: ADMIN_OK_CODE, - message: ADMIN_OK_MESSAGE.into(), - data: AdminAccountPageData { - items: page.items, - page: PageMeta::Cursor(CursorPageMeta { - limit, - next_cursor: page.next_cursor, - }), - summary, - }, - } - } - - fn numbered( + fn ok( page: crate::infra::json::NumberedPage, summary: AdminAccountSummaryData, ) -> Self { @@ -384,12 +366,12 @@ impl AdminAccountPageEnvelope { code: ADMIN_OK_CODE, message: ADMIN_OK_MESSAGE.into(), data: AdminAccountPageData { - page: PageMeta::Numbered(NumberedPageMeta { + page: PageMeta { page: page.page, page_size: page.page_size, total: page.total, total_pages: total_pages(page.total, page.page_size), - }), + }, items: page.items, summary, }, @@ -521,58 +503,32 @@ pub(crate) async fn accounts( _auth: AdminAuth, Query(params): Query, ) -> Result { - let limit = clamp_limit(params.page_size.or(params.limit).unwrap_or(50)); - let use_numbered_page = params.page.is_some() || params.page_size.is_some(); + let page = clamp_page(params.page.unwrap_or(1)); + let page_size = clamp_limit(params.page_size.unwrap_or(50)); let quota_by_account = quota_snapshots_by_account(&state).await; let summary = account_summary_data(&state, "a_by_account).await; - if use_numbered_page { - return match state - .services - .admin_accounts - .list_page(clamp_page(params.page.unwrap_or(1)), limit, params.search) - .await - { - Ok(page) => { - let stats = account_list_stats(&state, &page.items, "a_by_account).await?; - let page = crate::infra::json::NumberedPage { - items: page - .items - .into_iter() - .map(|item| stats.data_for(item)) - .collect(), - total: page.total, - page: page.page, - page_size: page.page_size, - }; - Ok(AdminResponse::new( - StatusCode::OK, - AdminAccountPageEnvelope::numbered(page, summary), - )) - } - Err(error) => Err(account_error(&error)), - }; - } - match state .services .admin_accounts - .list(params.cursor, limit) + .list_page(page, page_size, params.search) .await { Ok(page) => { let stats = account_list_stats(&state, &page.items, "a_by_account).await?; - let page = Page { + let page = crate::infra::json::NumberedPage { items: page .items .into_iter() .map(|item| stats.data_for(item)) .collect(), - next_cursor: page.next_cursor, + total: page.total, + page: page.page, + page_size: page.page_size, }; Ok(AdminResponse::new( StatusCode::OK, - AdminAccountPageEnvelope::cursor(page, limit, summary), + AdminAccountPageEnvelope::ok(page, summary), )) } Err(error) => Err(account_error(&error)), diff --git a/backend/src/api/admin/dashboard_routes.rs b/backend/src/api/admin/dashboard_routes.rs index f34c570d3..a68248e50 100644 --- a/backend/src/api/admin/dashboard_routes.rs +++ b/backend/src/api/admin/dashboard_routes.rs @@ -211,9 +211,8 @@ pub(crate) async fn dashboard_summary( let recent_events = state .services .usage_records - .list(None, DASHBOARD_USAGE_RECORD_LIMIT, today_filter) + .list_recent(DASHBOARD_USAGE_RECORD_LIMIT, today_filter) .await - .map(|page| page.items) .unwrap_or_default(); let time_buckets = dashboard_time_buckets(&state, now).await; diff --git a/backend/src/api/admin/keys_routes.rs b/backend/src/api/admin/keys_routes.rs index 5a3f39c40..b15dddd55 100644 --- a/backend/src/api/admin/keys_routes.rs +++ b/backend/src/api/admin/keys_routes.rs @@ -17,7 +17,7 @@ use crate::{ api::admin::session::AdminAuth, api::AppState, infra::{ - json::{clamp_limit, Page}, + json::{clamp_limit, clamp_page, NumberedPage}, time::{china_relative_time_str, china_rfc3339_str}, }, keys::types::{KeyManageError, ManagedClientApiKey}, @@ -26,8 +26,9 @@ use crate::{ #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct ApiKeysQuery { - cursor: Option, - limit: Option, + page: Option, + page_size: Option, + search: Option, } #[derive(Debug, Deserialize)] @@ -117,21 +118,24 @@ pub(crate) async fn api_keys( _auth: AdminAuth, Query(query): Query, ) -> Result { - let limit = clamp_limit(query.limit.unwrap_or(50)); + let page = clamp_page(query.page.unwrap_or(1)); + let page_size = clamp_limit(query.page_size.unwrap_or(50)); match state .services .admin_client_keys - .list(query.cursor, limit) + .list_page(page, page_size, query.search) .await { Ok(page) => { - let page = Page { + let page = NumberedPage { items: page.items.into_iter().map(ClientApiKeyData::from).collect(), - next_cursor: page.next_cursor, + total: page.total, + page: page.page, + page_size: page.page_size, }; Ok(AdminResponse::new( StatusCode::OK, - AdminPageEnvelope::ok(page, limit), + AdminPageEnvelope::ok(page), )) } Err(error) => Err(client_key_error(&error)), diff --git a/backend/src/api/admin/ops_routes.rs b/backend/src/api/admin/ops_routes.rs index 75792648d..80df4cbc7 100644 --- a/backend/src/api/admin/ops_routes.rs +++ b/backend/src/api/admin/ops_routes.rs @@ -14,7 +14,7 @@ use crate::{ }, api::AppState, infra::{ - json::{clamp_limit, clamp_page, NumberedPage, Page}, + json::{clamp_limit, clamp_page, NumberedPage}, time::china_datetime, }, telemetry::{ops::query::OpsQueryError, ops::store::OpsErrorFilter, ops::types::OpsErrorLog}, @@ -23,8 +23,6 @@ use crate::{ #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct OpsErrorsQuery { - cursor: Option, - limit: Option, page: Option, page_size: Option, kind: Option, @@ -61,41 +59,23 @@ pub(crate) async fn ops_errors( _auth: AdminAuth, Query(query): Query, ) -> Result { - let limit = clamp_limit(query.page_size.or(query.limit).unwrap_or(50)); - let page = query.page; - let numbered = page.is_some() || query.page_size.is_some(); - let cursor = query.cursor.clone(); + let page = clamp_page(query.page.unwrap_or(1)); + let page_size = clamp_limit(query.page_size.unwrap_or(50)); let filter = filter_from_query(query)?; - if numbered { - return state - .services - .ops_errors - .list_page(clamp_page(page.unwrap_or(1)), limit, filter) - .await - .map(|page| { - let page = NumberedPage { - items: ops_error_items(page.items), - total: page.total, - page: page.page, - page_size: page.page_size, - }; - AdminResponse::new(StatusCode::OK, AdminPageEnvelope::numbered(page)) - }) - .map_err(log_error); - } - state .services .ops_errors - .list(cursor, limit, filter) + .list_page(page, page_size, filter) .await .map(|page| { - let page = Page { + let page = NumberedPage { items: ops_error_items(page.items), - next_cursor: page.next_cursor, + total: page.total, + page: page.page, + page_size: page.page_size, }; - AdminResponse::new(StatusCode::OK, AdminPageEnvelope::ok(page, limit)) + AdminResponse::new(StatusCode::OK, AdminPageEnvelope::ok(page)) }) .map_err(log_error) } diff --git a/backend/src/api/admin/response.rs b/backend/src/api/admin/response.rs index 86cb13b30..42deef828 100644 --- a/backend/src/api/admin/response.rs +++ b/backend/src/api/admin/response.rs @@ -1,6 +1,6 @@ //! 管理端响应封装与 From 转换。 -use crate::infra::json::{total_pages, NumberedPage, Page}; +use crate::infra::json::{total_pages, NumberedPage}; use axum::{ http::StatusCode, response::{IntoResponse, Response}, @@ -187,33 +187,16 @@ impl AdminEnvelope { } } -/// 分页元数据。 -#[derive(Debug, Clone, Serialize)] -#[serde(rename_all = "camelCase")] -pub(crate) struct CursorPageMeta { - pub(crate) limit: u32, - #[serde(skip_serializing_if = "Option::is_none")] - pub(crate) next_cursor: Option, -} - /// 页码分页元数据。 #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] -pub(crate) struct NumberedPageMeta { +pub(crate) struct PageMeta { pub(crate) page: u32, pub(crate) page_size: u32, pub(crate) total: u64, pub(crate) total_pages: u32, } -/// 分页元数据。 -#[derive(Debug, Clone, Serialize)] -#[serde(untagged)] -pub(crate) enum PageMeta { - Cursor(CursorPageMeta), - Numbered(NumberedPageMeta), -} - /// 分页响应信封。 #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -240,19 +223,7 @@ pub(crate) struct BatchDeleteData { } impl AdminPageEnvelope { - pub fn ok(page: Page, limit: u32) -> Self { - let Page { items, next_cursor } = page; - Self { - code: ADMIN_OK_CODE, - message: ADMIN_OK_MESSAGE.into(), - data: PageData { - items, - page: PageMeta::Cursor(CursorPageMeta { limit, next_cursor }), - }, - } - } - - pub fn numbered(page: NumberedPage) -> Self { + pub fn ok(page: NumberedPage) -> Self { let NumberedPage { items, total, @@ -264,12 +235,12 @@ impl AdminPageEnvelope { message: ADMIN_OK_MESSAGE.into(), data: PageData { items, - page: PageMeta::Numbered(NumberedPageMeta { + page: PageMeta { page, page_size, total, total_pages: total_pages(total, page_size), - }), + }, }, } } diff --git a/backend/src/api/admin/usage_routes.rs b/backend/src/api/admin/usage_routes.rs index 029450a7c..e6f1b9e07 100644 --- a/backend/src/api/admin/usage_routes.rs +++ b/backend/src/api/admin/usage_routes.rs @@ -20,7 +20,7 @@ use crate::{ format_compact_number, format_cost, format_duration_ms, format_duration_ms_f64, format_multiplier, format_number, format_percent, format_token_price, format_tokens, }, - json::{clamp_limit, clamp_page, NumberedPage, Page}, + json::{clamp_limit, clamp_page, NumberedPage}, time::china_datetime, }, telemetry::{ @@ -36,8 +36,6 @@ use crate::{ #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub(crate) struct UsageRecordsQuery { - cursor: Option, - limit: Option, page: Option, page_size: Option, kind: Option, @@ -192,45 +190,14 @@ pub(crate) async fn usage_records( _auth: AdminAuth, Query(query): Query, ) -> Result { - let limit = clamp_limit(query.page_size.or(query.limit).unwrap_or(50)); - let page = query.page; - let use_numbered_page = page.is_some() || query.page_size.is_some(); - let cursor = query.cursor.clone(); + let page = clamp_page(query.page.unwrap_or(1)); + let page_size = clamp_limit(query.page_size.unwrap_or(50)); let filter = filter_from_query(query)?; - if use_numbered_page { - return match state - .services - .usage_records - .list_page(clamp_page(page.unwrap_or(1)), limit, filter) - .await - { - Ok(page) => { - let account_emails = state - .services - .usage_records - .account_email_map(&page.items) - .await - .map_err(|error| log_error(&error))?; - let page = NumberedPage { - items: usage_record_items(page.items, &account_emails), - total: page.total, - page: page.page, - page_size: page.page_size, - }; - Ok(AdminResponse::new( - StatusCode::OK, - AdminPageEnvelope::numbered(page), - )) - } - Err(error) => Err(log_error(&error)), - }; - } - match state .services .usage_records - .list(cursor, limit, filter) + .list_page(page, page_size, filter) .await { Ok(page) => { @@ -240,13 +207,15 @@ pub(crate) async fn usage_records( .account_email_map(&page.items) .await .map_err(|error| log_error(&error))?; - let page = Page { + let page = NumberedPage { items: usage_record_items(page.items, &account_emails), - next_cursor: page.next_cursor, + total: page.total, + page: page.page, + page_size: page.page_size, }; Ok(AdminResponse::new( StatusCode::OK, - AdminPageEnvelope::ok(page, limit), + AdminPageEnvelope::ok(page), )) } Err(error) => Err(log_error(&error)), diff --git a/backend/src/api/client/errors.rs b/backend/src/api/client/errors.rs index fe49fcbf7..c7fd49933 100644 --- a/backend/src/api/client/errors.rs +++ b/backend/src/api/client/errors.rs @@ -21,27 +21,21 @@ const UPSTREAM_CODEX_RESPONSE_FAILED_MESSAGE: &str = "Upstream Codex response fa const HISTORY_UNAVAILABLE_MESSAGE: &str = "Previous response context is unavailable. Start a new conversation or resend the complete input without previous_response_id."; #[derive(Clone, Copy)] -pub enum ResponseDispatchMessageStyle { +enum ResponseDispatchMessageStyle { Standard, ResponsesStream, } -#[derive(Clone, Copy)] -pub enum ResponseDispatchStatusMode { - UpstreamFailureStatus, - Client, +struct DispatchHttpError { + status: StatusCode, + message: String, } -pub struct DispatchHttpError { - pub status: StatusCode, - pub message: String, -} - -pub struct OpenAiErrorDetails { - pub status: StatusCode, - pub message: String, - pub error_type: &'static str, - pub code: &'static str, +struct OpenAiErrorDetails { + status: StatusCode, + message: String, + error_type: &'static str, + code: &'static str, } #[derive(Debug, Clone, Copy)] @@ -50,9 +44,9 @@ struct OpenAiErrorKind { code: &'static str, } -const UPSTREAM_UNAVAILABLE_ERROR: OpenAiErrorKind = OpenAiErrorKind { +const NO_AVAILABLE_ACCOUNTS_ERROR: OpenAiErrorKind = OpenAiErrorKind { error_type: "server_error", - code: "upstream_unavailable", + code: "no_available_accounts", }; const UPSTREAM_ERROR: OpenAiErrorKind = OpenAiErrorKind { error_type: "server_error", @@ -86,6 +80,10 @@ const CODEX_API_ERROR: OpenAiErrorKind = OpenAiErrorKind { error_type: "server_error", code: "codex_api_error", }; +const CODEX_CLIENT_ERROR: OpenAiErrorKind = OpenAiErrorKind { + error_type: "invalid_request_error", + code: "codex_client_error", +}; /// OpenAI 兼容错误响应。 pub fn openai_error_response( @@ -115,54 +113,22 @@ pub fn responses_dispatch_error_response(error: ResponseDispatchError) -> Respon } } -pub fn responses_compact_dispatch_error_response(error: ResponseDispatchError) -> Response { - match error { - ResponseDispatchError::NoActiveAccount => { - responses_no_available_accounts_response().into_response() - } - error => { - let error = response_dispatch_http_error( - &error, - ResponseDispatchStatusMode::Client, - ResponseDispatchMessageStyle::Standard, - ); - responses_error_response(error.status, &error.message).into_response() - } - } -} - pub fn responses_stream_dispatch_failed_sse_event(error: &ResponseDispatchError) -> String { - let http_error = response_dispatch_http_error( - error, - ResponseDispatchStatusMode::UpstreamFailureStatus, - ResponseDispatchMessageStyle::ResponsesStream, - ); + let http_error = + response_dispatch_http_error(error, ResponseDispatchMessageStyle::ResponsesStream); let kind = response_dispatch_error_kind(error); responses_failed_sse_event(kind, &http_error.message) } fn responses_no_available_accounts_response() -> (StatusCode, Json) { + let kind = NO_AVAILABLE_ACCOUNTS_ERROR; ( StatusCode::SERVICE_UNAVAILABLE, - Json(json!({ - "error": { - "type": "server_error", - "code": "no_available_accounts", - "message": NO_AVAILABLE_RESPONSES_ACCOUNTS_MESSAGE, - } - })), - ) -} - -fn responses_error_response(status: StatusCode, message: &str) -> (StatusCode, Json) { - let kind = responses_error_kind_for_status(status); - ( - status, Json(json!({ "error": { "type": kind.error_type, "code": kind.code, - "message": message, + "message": NO_AVAILABLE_RESPONSES_ACCOUNTS_MESSAGE, } })), ) @@ -213,11 +179,8 @@ fn invalid_openai_request_response(message: &str) -> (StatusCode, Json) { } fn response_dispatch_openai_error(error: &ResponseDispatchError) -> OpenAiErrorDetails { - let DispatchHttpError { status, message } = response_dispatch_http_error( - error, - ResponseDispatchStatusMode::UpstreamFailureStatus, - ResponseDispatchMessageStyle::Standard, - ); + let DispatchHttpError { status, message } = + response_dispatch_http_error(error, ResponseDispatchMessageStyle::Standard); let kind = response_dispatch_error_kind(error); OpenAiErrorDetails { @@ -231,17 +194,16 @@ fn response_dispatch_openai_error(error: &ResponseDispatchError) -> OpenAiErrorD fn response_dispatch_error_kind(error: &ResponseDispatchError) -> OpenAiErrorKind { dispatch_failure_openai_error_kind( error.metadata().failure_class, - status_from_u16(error.http_status_code(), StatusCode::BAD_GATEWAY), + status_from_u16(error.client_http_status_code(), StatusCode::BAD_GATEWAY), ) } -pub fn response_dispatch_http_error( +fn response_dispatch_http_error( error: &ResponseDispatchError, - status_mode: ResponseDispatchStatusMode, message_style: ResponseDispatchMessageStyle, ) -> DispatchHttpError { DispatchHttpError { - status: response_dispatch_status(error, status_mode), + status: status_from_u16(error.client_http_status_code(), StatusCode::BAD_GATEWAY), message: response_dispatch_message(error, message_style), } } @@ -254,27 +216,12 @@ fn responses_error_kind_for_status(status: StatusCode) -> OpenAiErrorKind { } else if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { AUTHENTICATION_ERROR } else if status.is_client_error() { - OpenAiErrorKind { - error_type: "invalid_request_error", - code: "codex_api_error", - } + CODEX_CLIENT_ERROR } else { CODEX_API_ERROR } } -fn response_dispatch_status( - error: &ResponseDispatchError, - mode: ResponseDispatchStatusMode, -) -> StatusCode { - match (mode, error) { - (ResponseDispatchStatusMode::UpstreamFailureStatus, ResponseDispatchError::Upstream(_)) => { - StatusCode::BAD_GATEWAY - } - _ => status_from_u16(error.http_status_code(), StatusCode::BAD_GATEWAY), - } -} - fn response_dispatch_message( error: &ResponseDispatchError, style: ResponseDispatchMessageStyle, @@ -335,7 +282,7 @@ fn dispatch_failure_openai_error_kind( status: StatusCode, ) -> OpenAiErrorKind { match failure_class { - DispatchFailureClass::NoAvailableAccounts => UPSTREAM_UNAVAILABLE_ERROR, + DispatchFailureClass::NoAvailableAccounts => NO_AVAILABLE_ACCOUNTS_ERROR, DispatchFailureClass::QuotaExhausted => INSUFFICIENT_QUOTA_ERROR, DispatchFailureClass::RateLimited => RATE_LIMIT_ERROR, DispatchFailureClass::Expired @@ -347,9 +294,13 @@ fn dispatch_failure_openai_error_kind( | DispatchFailureClass::MissingCompleted | DispatchFailureClass::EmptyUpstreamResponse => INVALID_UPSTREAM_RESPONSE_ERROR, DispatchFailureClass::ResponseFailed => responses_error_kind_for_status(status), - DispatchFailureClass::Upstream - | DispatchFailureClass::CloudflareChallenge - | DispatchFailureClass::CloudflarePathBlocked => UPSTREAM_ERROR, + DispatchFailureClass::Upstream if status.is_client_error() => { + responses_error_kind_for_status(status) + } + DispatchFailureClass::Upstream => UPSTREAM_ERROR, + DispatchFailureClass::CloudflareChallenge | DispatchFailureClass::CloudflarePathBlocked => { + UPSTREAM_ERROR + } } } diff --git a/backend/src/api/client/responses.rs b/backend/src/api/client/responses.rs index ac6e58b38..e3e0f881e 100644 --- a/backend/src/api/client/responses.rs +++ b/backend/src/api/client/responses.rs @@ -23,8 +23,8 @@ use crate::{ use super::{ errors::{ invalid_responses_request_response, missing_client_api_key_response, - model_not_found_response, responses_compact_dispatch_error_response, - responses_dispatch_error_response, responses_stream_dispatch_failed_sse_event, + model_not_found_response, responses_dispatch_error_response, + responses_stream_dispatch_failed_sse_event, }, models::model_catalog_for_state, sse::{done_sse_frame, event_stream_response as sse_event_stream_response, SseResponseOptions}, @@ -353,7 +353,7 @@ async fn handle_compact_responses( .await { Ok(body) => (StatusCode::OK, Json(body)).into_response(), - Err(error) => responses_compact_dispatch_error_response(error), + Err(error) => responses_dispatch_error_response(error), } } diff --git a/backend/src/dispatch/errors.rs b/backend/src/dispatch/errors.rs index eef5a1758..054d6b9ee 100644 --- a/backend/src/dispatch/errors.rs +++ b/backend/src/dispatch/errors.rs @@ -420,6 +420,13 @@ impl ResponseDispatchError { } } + pub fn client_http_status_code(&self) -> u16 { + match self { + Self::Upstream(_) => client_upstream_http_status_code(self.http_status_code()), + _ => self.http_status_code(), + } + } + pub(crate) fn metadata(&self) -> DispatchErrorMetadata { match self { Self::NoActiveAccount => DispatchErrorMetadata::no_available_accounts(), @@ -583,6 +590,13 @@ impl ResponseDispatchError { } } +pub(crate) fn client_upstream_http_status_code(status: u16) -> u16 { + match status { + 400..=499 | 503 => status, + _ => 502, + } +} + /// Responses live SSE body stream error. #[derive(Debug, Error)] pub enum ResponseDispatchStreamError { diff --git a/backend/src/dispatch/recording.rs b/backend/src/dispatch/recording.rs index 0c9621840..d9fd5c44f 100644 --- a/backend/src/dispatch/recording.rs +++ b/backend/src/dispatch/recording.rs @@ -107,7 +107,7 @@ pub(super) async fn record_response_dispatch_error_event( insert_response_status_metadata_object( object, i64::from(record.error.http_status_code()), - dispatch_error_client_status_code(record.error, record.stream, record.compact), + dispatch_error_client_status_code(record.error, record.stream), dispatch_error_upstream_status_code(record.error), ); } @@ -152,7 +152,7 @@ pub(super) async fn record_response_upstream_error_event( insert_response_status_metadata_object( object, event_status_code, - upstream_failure_client_status_code(record.stream), + upstream_failure_client_status_code(event_status_code, record.stream), Some(event_status_code), ); insert_response_request_summary_object(object, record.request, record.transport); @@ -453,25 +453,20 @@ fn dispatch_error_upstream_status_code(error: &ResponseDispatchError) -> Option< } } -fn dispatch_error_client_status_code( - error: &ResponseDispatchError, - stream: bool, - compact: bool, -) -> i64 { +fn dispatch_error_client_status_code(error: &ResponseDispatchError, stream: bool) -> i64 { if stream { return 200; } - if matches!(error, ResponseDispatchError::Upstream(_)) && !compact { - return 502; - } - i64::from(error.http_status_code()) + i64::from(error.client_http_status_code()) } -fn upstream_failure_client_status_code(stream: bool) -> i64 { +fn upstream_failure_client_status_code(upstream_status: i64, stream: bool) -> i64 { if stream { 200 } else { - 502 + i64::from(crate::dispatch::errors::client_upstream_http_status_code( + u16::try_from(upstream_status).unwrap_or(502), + )) } } diff --git a/backend/src/fleet/manage/lifecycle.rs b/backend/src/fleet/manage/lifecycle.rs index a253bf831..6360d194d 100644 --- a/backend/src/fleet/manage/lifecycle.rs +++ b/backend/src/fleet/manage/lifecycle.rs @@ -11,10 +11,7 @@ use uuid::Uuid; use crate::{ fleet::{account::AccountStatus, store::AccountStore}, - infra::{ - json::{NumberedPage, Page}, - time::elapsed_millis_i64, - }, + infra::{json::NumberedPage, time::elapsed_millis_i64}, upstream::openai::token_client::RefreshFailure, }; @@ -30,22 +27,6 @@ use super::{ const ADMIN_REFRESH_LEASE_TTL_SECONDS: i64 = 5 * 60; impl AccountManageService { - pub async fn list( - &self, - cursor: Option, - limit: u32, - ) -> Result, AccountManageError> { - let page = self - .store - .list_metadata(cursor, limit) - .await - .map_err(|_| AccountManageError::List)?; - Ok(Page { - items: page.items.into_iter().map(ManagedAccount::from).collect(), - next_cursor: page.next_cursor, - }) - } - pub async fn list_page( &self, page: u32, diff --git a/backend/src/fleet/store/mod.rs b/backend/src/fleet/store/mod.rs index 6ab50f331..74d9994d8 100644 --- a/backend/src/fleet/store/mod.rs +++ b/backend/src/fleet/store/mod.rs @@ -8,10 +8,7 @@ use sqlx::{PgPool, Postgres, QueryBuilder, Row}; use thiserror::Error; use crate::fleet::account::{Account, AccountStatus}; -use crate::infra::{ - json::{decode_cursor, page_offset, NumberedPage, Page}, - time::parse_rfc3339_utc, -}; +use crate::infra::json::{page_offset, NumberedPage}; mod queries; mod rows; @@ -20,7 +17,7 @@ use queries::*; use rows::{ count_account_metadata, get_pool_account, list_pool_accounts, map_account_store_error, metadata_from_row, optional_update_value, push_account_metadata_search, quota_plan_type, - quota_snapshot_from_row, status_to_db, stored_account_from_row, to_page, + quota_snapshot_from_row, status_to_db, stored_account_from_row, }; // ============================================================================ @@ -33,17 +30,11 @@ pub enum PgAccountStoreError { /// 数据库错误。 #[error("PostgreSQL account store database error: {0}")] Database(#[from] sqlx::Error), - /// 时间格式错误。 - #[error("account store timestamp error: {0}")] - Timestamp(#[from] chrono::ParseError), #[error("account store JSON error: {0}")] Json(#[from] serde_json::Error), /// 账号状态非法。 #[error("PostgreSQL account store status error: {0}")] InvalidStatus(String), - /// 分页游标非法。 - #[error("invalid account pagination cursor")] - InvalidCursor, } /// PostgreSQL 账号仓储结果。 diff --git a/backend/src/fleet/store/queries.rs b/backend/src/fleet/store/queries.rs index 4d50974fd..08792a212 100644 --- a/backend/src/fleet/store/queries.rs +++ b/backend/src/fleet/store/queries.rs @@ -113,82 +113,6 @@ set updated_at = $14 where id = $15"; -pub(super) const LIST_STORED_ACCOUNTS_AFTER_CURSOR_SQL: &str = r" -select - id, - email, - chatgpt_account_id as account_id, - chatgpt_user_id as user_id, - label, - plan_type, - access_token, - refresh_token, - access_token_expires_at, - next_refresh_at, - status, - added_at, - updated_at -from accounts -where added_at < $1 - or (added_at = $2 and id < $3) -order by added_at desc, id desc -limit $4"; - -pub(super) const LIST_STORED_ACCOUNTS_SQL: &str = r" -select - id, - email, - chatgpt_account_id as account_id, - chatgpt_user_id as user_id, - label, - plan_type, - access_token, - refresh_token, - access_token_expires_at, - next_refresh_at, - status, - added_at, - updated_at -from accounts -order by added_at desc, id desc -limit $1"; - -pub(super) const LIST_ACCOUNT_METADATA_AFTER_CURSOR_SQL: &str = r" -select - id, - email, - chatgpt_account_id as account_id, - chatgpt_user_id as user_id, - label, - plan_type, - (refresh_token is not null and refresh_token <> '') as has_refresh_token, - access_token_expires_at, - status, - added_at, - updated_at -from accounts -where added_at < $1 - or (added_at = $2 and id < $3) -order by added_at desc, id desc -limit $4"; - -pub(super) const LIST_ACCOUNT_METADATA_SQL: &str = r" -select - id, - email, - chatgpt_account_id as account_id, - chatgpt_user_id as user_id, - label, - plan_type, - (refresh_token is not null and refresh_token <> '') as has_refresh_token, - access_token_expires_at, - status, - added_at, - updated_at -from accounts -order by added_at desc, id desc -limit $1"; - pub(super) const LIST_ACCOUNT_METADATA_SELECT_SQL: &str = r" select id, @@ -415,74 +339,6 @@ impl PgAccountStore { row.as_ref().map(stored_account_from_row).transpose() } - /// 分页列出所有账号（含 token）。 - pub async fn list( - &self, - cursor: Option, - limit: u32, - ) -> PgAccountStoreResult> { - let limit = limit.clamp(1, 200); - if let Some(cursor) = cursor { - let (added_at, id) = - decode_cursor(&cursor).ok_or(PgAccountStoreError::InvalidCursor)?; - let added_at = - parse_rfc3339_utc(&added_at).map_err(|_| PgAccountStoreError::InvalidCursor)?; - let rows = sqlx::query(LIST_STORED_ACCOUNTS_AFTER_CURSOR_SQL) - .bind(added_at) - .bind(added_at) - .bind(&id) - .bind(i64::from(limit) + 1) - .fetch_all(&self.pool) - .await?; - Ok(to_page( - &rows, - limit, - stored_account_from_row, - ("added_at", "id"), - )) - } else { - let rows = sqlx::query(LIST_STORED_ACCOUNTS_SQL) - .bind(i64::from(limit) + 1) - .fetch_all(&self.pool) - .await?; - Ok(to_page( - &rows, - limit, - stored_account_from_row, - ("added_at", "id"), - )) - } - } - - /// 分页列出账号元数据（不含 token）。 - pub async fn list_metadata( - &self, - cursor: Option, - limit: u32, - ) -> PgAccountStoreResult> { - let limit = limit.clamp(1, 200); - if let Some(cursor) = cursor { - let (added_at, id) = - decode_cursor(&cursor).ok_or(PgAccountStoreError::InvalidCursor)?; - let added_at = - parse_rfc3339_utc(&added_at).map_err(|_| PgAccountStoreError::InvalidCursor)?; - let rows = sqlx::query(LIST_ACCOUNT_METADATA_AFTER_CURSOR_SQL) - .bind(added_at) - .bind(added_at) - .bind(&id) - .bind(i64::from(limit) + 1) - .fetch_all(&self.pool) - .await?; - Ok(to_page(&rows, limit, metadata_from_row, ("added_at", "id"))) - } else { - let rows = sqlx::query(LIST_ACCOUNT_METADATA_SQL) - .bind(i64::from(limit) + 1) - .fetch_all(&self.pool) - .await?; - Ok(to_page(&rows, limit, metadata_from_row, ("added_at", "id"))) - } - } - /// 按页码列出账号元数据（不含 token）。 pub async fn list_metadata_page( &self, diff --git a/backend/src/fleet/store/rows.rs b/backend/src/fleet/store/rows.rs index 33d285e97..32315af0d 100644 --- a/backend/src/fleet/store/rows.rs +++ b/backend/src/fleet/store/rows.rs @@ -9,7 +9,7 @@ use crate::{ account::{Account, AccountStatus}, quota::{quota_snapshot_limit_window_seconds, quota_snapshot_reset_at}, }, - infra::{format::nonnegative_i64_to_u64, json::Page}, + infra::format::nonnegative_i64_to_u64, }; use super::{ @@ -218,36 +218,3 @@ pub(super) fn push_account_metadata_search( builder.push(" or chatgpt_user_id ilike "); builder.push_bind(pattern); } - -pub(super) fn to_page( - rows: &[PgRow], - limit: u32, - mapper: impl Fn(&PgRow) -> PgAccountStoreResult, - cursor_fields: (&str, &str), -) -> Page { - let has_more = rows.len() > limit as usize; - let mut items: Vec = Vec::with_capacity(limit as usize); - let mut last_row: Option<&PgRow> = None; - for (i, row) in rows.iter().enumerate() { - if i >= limit as usize { - break; - } - if let Ok(item) = mapper(row) { - items.push(item); - last_row = Some(row); - } - } - let next_cursor = if has_more { - last_row.map(|row| { - use sqlx::Row; - let ts = row - .get::, _>(cursor_fields.0) - .to_rfc3339(); - let id: String = row.get(cursor_fields.1); - crate::infra::json::encode_cursor(&ts, &id) - }) - } else { - None - }; - Page { items, next_cursor } -} diff --git a/backend/src/infra/json.rs b/backend/src/infra/json.rs index 4be7a0f56..34a7390c8 100644 --- a/backend/src/infra/json.rs +++ b/backend/src/infra/json.rs @@ -1,23 +1,11 @@ //! 无业务含义的 JSON 与分页辅助。 -use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; use serde::Serialize; // --------------------------------------------------------------------------- // 分页 // --------------------------------------------------------------------------- -/// 通用分页结果。 -#[derive(Debug, Clone, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct Page { - /// 当前页条目。 - pub items: Vec, - /// 下一页游标。 - #[serde(skip_serializing_if = "Option::is_none")] - pub next_cursor: Option, -} - /// 页码分页结果。 #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -32,19 +20,6 @@ pub struct NumberedPage { pub page_size: u32, } -/// 编码游标。 -pub fn encode_cursor(created_at: &str, id: &str) -> String { - URL_SAFE_NO_PAD.encode(format!("{created_at}|{id}")) -} - -/// 解码游标。 -pub fn decode_cursor(cursor: &str) -> Option<(String, String)> { - let raw = URL_SAFE_NO_PAD.decode(cursor).ok()?; - let text = String::from_utf8(raw).ok()?; - let (created_at, id) = text.split_once('|')?; - Some((created_at.to_string(), id.to_string())) -} - /// 限制分页大小。 pub fn clamp_limit(limit: u32) -> u32 { limit.clamp(1, 200) diff --git a/backend/src/keys/manage.rs b/backend/src/keys/manage.rs index 823fb7fbf..532d806db 100644 --- a/backend/src/keys/manage.rs +++ b/backend/src/keys/manage.rs @@ -1,6 +1,6 @@ //! 客户端 API Key 管理服务。 -use crate::infra::json::Page; +use crate::infra::json::NumberedPage; use super::{ store::PgClientKeyStore, @@ -31,19 +31,22 @@ impl KeyManageService { .map_err(|_| KeyManageError::Create) } - pub async fn list( + pub async fn list_page( &self, - cursor: Option, - limit: u32, - ) -> Result, KeyManageError> { + page: u32, + page_size: u32, + search: Option, + ) -> Result, KeyManageError> { let page = self .store - .list(cursor, limit) + .list_page(page, page_size, search.as_deref()) .await .map_err(|_| KeyManageError::List)?; - Ok(Page { + Ok(NumberedPage { items: page.items.into_iter().map(Into::into).collect(), - next_cursor: page.next_cursor, + total: page.total, + page: page.page, + page_size: page.page_size, }) } diff --git a/backend/src/keys/store.rs b/backend/src/keys/store.rs index e2a0ac7f9..495f50e41 100644 --- a/backend/src/keys/store.rs +++ b/backend/src/keys/store.rs @@ -3,22 +3,23 @@ use std::collections::BTreeMap; use chrono::{DateTime, Utc}; -use sqlx::{PgPool, Row}; +use sqlx::{PgPool, Postgres, QueryBuilder, Row}; use thiserror::Error; use uuid::Uuid; use crate::infra::{ identity::generate_client_api_key, - json::{decode_cursor, encode_cursor, Page}, - time::parse_rfc3339_utc, + json::{page_offset, NumberedPage}, }; +const CLIENT_KEY_SELECT_SQL: &str = r"select + id, name, label, prefix, key, enabled, created_at, last_used_at +from client_api_keys"; + #[derive(Debug, Error)] pub enum PgClientKeyStoreError { #[error("PostgreSQL client key operation failed: {0}")] Database(#[from] sqlx::Error), - #[error("invalid client key pagination cursor")] - InvalidCursor, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -90,53 +91,39 @@ values ($1, $2, null, $3, $4, true, $5, null)", Ok(stored) } - pub async fn list( + pub async fn list_page( &self, - cursor: Option, - limit: u32, - ) -> Result, PgClientKeyStoreError> { - let fetch_limit = i64::from(limit) + 1; - let rows = if let Some(cursor) = cursor { - let (created_at, id) = - decode_cursor(&cursor).ok_or(PgClientKeyStoreError::InvalidCursor)?; - let created_at = - parse_rfc3339_utc(&created_at).map_err(|_| PgClientKeyStoreError::InvalidCursor)?; - sqlx::query( - r" -select id, name, label, prefix, key, enabled, created_at, last_used_at -from client_api_keys -where created_at < $1 or (created_at = $1 and id < $2) -order by created_at desc, id desc -limit $3", - ) - .bind(created_at) - .bind(id) - .bind(fetch_limit) - .fetch_all(&self.pool) + page: u32, + page_size: u32, + search: Option<&str>, + ) -> Result, PgClientKeyStoreError> { + let page = page.max(1); + let page_size = page_size.clamp(1, 200); + let search = search.map(str::trim).filter(|value| !value.is_empty()); + + let mut count_builder = + QueryBuilder::::new("select count(*) from client_api_keys"); + push_search(&mut count_builder, search); + let total = count_builder + .build_query_scalar::() + .fetch_one(&self.pool) .await? - } else { - sqlx::query( - r" -select id, name, label, prefix, key, enabled, created_at, last_used_at -from client_api_keys -order by created_at desc, id desc -limit $1", - ) - .bind(fetch_limit) - .fetch_all(&self.pool) - .await? - }; - let has_next = rows.len() > limit as usize; - let items = rows - .into_iter() - .take(limit as usize) - .map(|row| Self::key_from_row(&row)) - .collect::>(); - let next_cursor = has_next.then(|| { - let key = items.last().expect("next page requires at least one key"); - encode_cursor(&key.created_at.to_rfc3339(), &key.id) - }); - Ok(Page { items, next_cursor }) + .max(0) as u64; + + let mut builder = QueryBuilder::::new(CLIENT_KEY_SELECT_SQL); + push_search(&mut builder, search); + builder.push(" order by created_at desc, id desc limit "); + builder.push_bind(i64::from(page_size)); + builder.push(" offset "); + builder.push_bind(page_offset(page, page_size).min(i64::MAX as u64) as i64); + let rows = builder.build().fetch_all(&self.pool).await?; + let items = rows.iter().map(Self::key_from_row).collect::>(); + Ok(NumberedPage { + items, + total, + page, + page_size, + }) } pub async fn get(&self, id: &str) -> Result, PgClientKeyStoreError> { @@ -225,3 +212,17 @@ where keys.id = touched.id", } } } + +fn push_search(builder: &mut QueryBuilder, search: Option<&str>) { + let Some(search) = search else { + return; + }; + let pattern = format!("%{search}%"); + builder.push(" where (name ilike "); + builder.push_bind(pattern.clone()); + builder.push(" or label ilike "); + builder.push_bind(pattern.clone()); + builder.push(" or id ilike "); + builder.push_bind(pattern); + builder.push(")"); +} diff --git a/backend/src/telemetry/account_usage/query.rs b/backend/src/telemetry/account_usage/query.rs index 8c6bdeda4..b63bdcee6 100644 --- a/backend/src/telemetry/account_usage/query.rs +++ b/backend/src/telemetry/account_usage/query.rs @@ -4,7 +4,7 @@ use std::collections::HashMap; use chrono::{DateTime, Utc}; use thiserror::Error; -use crate::infra::{format::nonnegative_i64_to_u64, json::Page}; +use crate::infra::format::nonnegative_i64_to_u64; use crate::telemetry::{ account_usage::store::{PgAccountUsageStore, UsageListRecord, UsageSummary}, billing, @@ -30,27 +30,6 @@ impl AccountUsageQueryService { } } - /// 分页列出账号用量。 - pub async fn list( - &self, - cursor: Option, - limit: u32, - ) -> Result, AccountUsageQueryError> { - let page = self - .store - .list_usage(cursor, limit) - .await - .map_err(|_| AccountUsageQueryError::List)?; - Ok(Page { - items: page - .items - .into_iter() - .map(AccountUsageRecord::from) - .collect(), - next_cursor: page.next_cursor, - }) - } - /// 按账号 ID 批量读取账号用量。 pub async fn list_by_account_ids( &self, diff --git a/backend/src/telemetry/account_usage/store.rs b/backend/src/telemetry/account_usage/store.rs index 635274a04..811dadfe7 100644 --- a/backend/src/telemetry/account_usage/store.rs +++ b/backend/src/telemetry/account_usage/store.rs @@ -7,11 +7,7 @@ use chrono::{DateTime, Utc}; use sqlx::{postgres::PgRow, PgPool, Postgres, QueryBuilder, Row}; use thiserror::Error; -use crate::infra::{ - format::nonnegative_i64_to_u64, - json::{decode_cursor, encode_cursor, Page}, - time::parse_rfc3339_utc as parse_rfc3339, -}; +use crate::infra::format::nonnegative_i64_to_u64; use super::should_reset_usage_window; @@ -135,8 +131,7 @@ select au.window_started_at, au.window_reset_at, au.limit_window_seconds, - au.last_used_at, - coalesce(au.last_used_at, 'epoch'::timestamptz) as sort_last_used_at + au.last_used_at from account_usage au left join accounts a on a.id = au.account_id"; @@ -162,12 +157,6 @@ pub enum PgAccountUsageStoreError { /// 数据库错误。 #[error("PostgreSQL usage store database error: {0}")] Database(#[from] sqlx::Error), - /// 时间格式错误。 - #[error("PostgreSQL usage store timestamp error: {0}")] - Timestamp(#[from] chrono::ParseError), - /// 分页游标非法。 - #[error("invalid usage pagination cursor")] - InvalidCursor, } /// PostgreSQL 用量存储结果。 @@ -582,37 +571,6 @@ where account_id in (", Ok(()) } - /// 分页列出账号用量。 - pub async fn list_usage( - &self, - cursor: Option, - limit: u32, - ) -> PgAccountUsageStoreResult> { - let limit = limit.clamp(1, 200); - let fetch_limit = i64::from(limit) + 1; - let mut builder = QueryBuilder::::new(LIST_USAGE_SELECT_SQL); - if let Some(cursor) = cursor { - let (last_used_at, account_id) = - decode_cursor(&cursor).ok_or(PgAccountUsageStoreError::InvalidCursor)?; - let last_used_at = parse_rfc3339(&last_used_at) - .map_err(|_| PgAccountUsageStoreError::InvalidCursor)?; - builder.push("\nwhere coalesce(au.last_used_at, 'epoch'::timestamptz) < "); - builder.push_bind(last_used_at); - builder.push("\n or (coalesce(au.last_used_at, 'epoch'::timestamptz) = "); - builder.push_bind(last_used_at); - builder.push(" and au.account_id < "); - builder.push_bind(account_id); - builder.push(")"); - } - builder.push( - "\norder by coalesce(au.last_used_at, 'epoch'::timestamptz) desc, au.account_id desc\nlimit ", - ); - builder.push_bind(fetch_limit); - - let rows = builder.build().fetch_all(&self.pool).await?; - Ok(to_page(&rows, limit)) - } - /// 按账号 ID 批量读取账号用量。 pub async fn list_usage_by_account_ids( &self, @@ -770,30 +728,3 @@ fn optional_positive_i64_to_u64(value: Option) -> Option { .and_then(|value| u64::try_from(value).ok()) .filter(|value| *value > 0) } - -fn to_page(rows: &[PgRow], limit: u32) -> Page { - let has_more = rows.len() > limit as usize; - let mut items = Vec::with_capacity(limit as usize); - let mut last_row: Option<&PgRow> = None; - for (i, row) in rows.iter().enumerate() { - if i >= limit as usize { - break; - } - if let Ok(item) = usage_list_from_row(row) { - items.push(item); - last_row = Some(row); - } - } - let next_cursor = if has_more { - last_row.map(|row| { - let ts = row - .get::, _>("sort_last_used_at") - .to_rfc3339(); - let id: String = row.get("account_id"); - encode_cursor(&ts, &id) - }) - } else { - None - }; - Page { items, next_cursor } -} diff --git a/backend/src/telemetry/ops/query.rs b/backend/src/telemetry/ops/query.rs index ff53fd165..f13efe028 100644 --- a/backend/src/telemetry/ops/query.rs +++ b/backend/src/telemetry/ops/query.rs @@ -27,18 +27,6 @@ impl OpsQueryService { Self { store } } - pub async fn list( - &self, - cursor: Option, - limit: u32, - filter: OpsErrorFilter, - ) -> Result, OpsQueryError> { - self.store - .list(filter, cursor, limit) - .await - .map_err(|_| OpsQueryError::List) - } - pub async fn list_page( &self, page: u32, diff --git a/backend/src/telemetry/ops/store.rs b/backend/src/telemetry/ops/store.rs index 63e18cbdf..5f59d883a 100644 --- a/backend/src/telemetry/ops/store.rs +++ b/backend/src/telemetry/ops/store.rs @@ -5,10 +5,7 @@ use sqlx::{PgPool, Postgres, QueryBuilder, Row}; use thiserror::Error; use crate::{ - infra::{ - json::{decode_cursor, encode_cursor, page_offset, NumberedPage, Page}, - time::parse_rfc3339_utc, - }, + infra::json::{page_offset, NumberedPage}, telemetry::{buckets::store::PgRequestBucketStore, ops::types::OpsErrorLog}, }; @@ -25,8 +22,6 @@ pub enum PgOpsErrorLogStoreError { Database(#[from] sqlx::Error), #[error("invalid ops error retention days: {0}")] InvalidRetention(i64), - #[error("invalid ops error pagination cursor")] - InvalidCursor, } pub type PgOpsErrorLogStoreResult = Result; @@ -86,31 +81,6 @@ insert into ops_error_logs ( Ok(()) } - pub async fn list( - &self, - filter: OpsErrorFilter, - cursor: Option, - limit: u32, - ) -> PgOpsErrorLogStoreResult> { - let limit = limit.clamp(1, 200); - let mut builder = QueryBuilder::::new(OPS_ERROR_SELECT_SQL); - push_filter(&mut builder, &filter, cursor.as_deref())?; - builder.push(" order by created_at desc, id desc limit "); - builder.push_bind(i64::from(limit) + 1); - let rows = builder.build().fetch_all(&self.pool).await?; - let has_next = rows.len() > limit as usize; - let items = rows - .into_iter() - .take(limit as usize) - .map(|row| ops_error_from_row(&row)) - .collect::>(); - let next_cursor = has_next.then(|| { - let item = items.last().expect("next page requires an ops error"); - encode_cursor(&item.created_at.to_rfc3339(), &item.id) - }); - Ok(Page { items, next_cursor }) - } - pub async fn list_page( &self, filter: OpsErrorFilter, @@ -120,7 +90,7 @@ insert into ops_error_logs ( let page_size = page_size.clamp(1, 200); let total = count_ops_errors(&self.pool, &filter).await?; let mut builder = QueryBuilder::::new(OPS_ERROR_SELECT_SQL); - push_filter(&mut builder, &filter, None)?; + push_filter(&mut builder, &filter); builder.push(" order by created_at desc, id desc limit "); builder.push_bind(i64::from(page_size)); builder.push(" offset "); @@ -195,11 +165,7 @@ pub struct OpsErrorFilter { pub end_time: Option>, } -fn push_filter( - builder: &mut QueryBuilder, - filter: &OpsErrorFilter, - cursor: Option<&str>, -) -> PgOpsErrorLogStoreResult<()> { +fn push_filter(builder: &mut QueryBuilder, filter: &OpsErrorFilter) { builder.push(" where true"); push_optional_text(builder, "kind", filter.kind.as_deref()); push_optional_text( @@ -248,20 +214,6 @@ fn push_filter( builder.push(" and created_at <= "); builder.push_bind(end_time); } - if let Some(cursor) = cursor { - let (created_at, id) = - decode_cursor(cursor).ok_or(PgOpsErrorLogStoreError::InvalidCursor)?; - let created_at = - parse_rfc3339_utc(&created_at).map_err(|_| PgOpsErrorLogStoreError::InvalidCursor)?; - builder.push(" and (created_at < "); - builder.push_bind(created_at); - builder.push(" or (created_at = "); - builder.push_bind(created_at); - builder.push(" and id < "); - builder.push_bind(id); - builder.push("))"); - } - Ok(()) } fn push_optional_text(builder: &mut QueryBuilder, column: &str, value: Option<&str>) { @@ -284,7 +236,7 @@ fn push_optional_i64(builder: &mut QueryBuilder, column: &str, value: async fn count_ops_errors(pool: &PgPool, filter: &OpsErrorFilter) -> PgOpsErrorLogStoreResult { let mut builder = QueryBuilder::::new("select count(*) from ops_error_logs"); - push_filter(&mut builder, filter, None)?; + push_filter(&mut builder, filter); let total: i64 = builder.build_query_scalar().fetch_one(pool).await?; Ok(total.max(0) as u64) } diff --git a/backend/src/telemetry/usage/query.rs b/backend/src/telemetry/usage/query.rs index c11205fdb..529744baf 100644 --- a/backend/src/telemetry/usage/query.rs +++ b/backend/src/telemetry/usage/query.rs @@ -7,10 +7,7 @@ use sqlx::{PgPool, Postgres, QueryBuilder, Row}; use thiserror::Error; use crate::{ - infra::{ - format::optional_nonnegative_i64_to_u64, - json::{NumberedPage, Page}, - }, + infra::{format::optional_nonnegative_i64_to_u64, json::NumberedPage}, telemetry::{ billing, usage::store::{ @@ -129,15 +126,14 @@ impl UsageQueryService { Self { store } } - /// 分页查询日志。 - pub async fn list( + /// 查询最近的日志。 + pub async fn list_recent( &self, - cursor: Option, limit: u32, filter: UsageQueryFilter, - ) -> Result, UsageQueryError> { + ) -> Result, UsageQueryError> { self.store - .list(filter.into(), cursor, limit) + .list_recent(filter.into(), limit) .await .map_err(|_| UsageQueryError::List) } @@ -298,7 +294,7 @@ pub(super) async fn count_usage_records( filter: &UsageRecordFilter, ) -> PgUsageRecordStoreResult { let mut builder = QueryBuilder::::new("select count(*) from usage_records"); - push_filter(&mut builder, filter, None)?; + push_filter(&mut builder, filter); let total: i64 = builder.build_query_scalar().fetch_one(pool).await?; Ok(total.max(0) as u64) } @@ -317,7 +313,7 @@ select avg(latency_ms::double precision) as average_latency_ms from usage_records", ); - push_filter(&mut builder, filter, None)?; + push_filter(&mut builder, filter); let row = builder.build().fetch_one(pool).await?; let input_tokens = nonnegative(row.get("input_tokens")); let output_tokens = nonnegative(row.get("output_tokens")); @@ -346,7 +342,7 @@ select max(created_at) as last_used_at from usage_records", ); - push_filter(&mut builder, filter, None)?; + push_filter(&mut builder, filter); builder.push(" group by account_id order by last_used_at desc, account_id limit "); builder.push_bind(i64::from(limit.clamp(1, 50))); Ok(builder @@ -420,7 +416,7 @@ pub(super) async fn usage_breakdown( count(latency_ms) as latency_count from usage_records", ); - push_filter(&mut builder, filter, None)?; + push_filter(&mut builder, filter); builder.push(" group by name, billing_model, service_tier order by request_count desc limit "); builder.push_bind(i64::from(limit.clamp(1, 50) * 8)); @@ -484,7 +480,7 @@ select count(latency_ms) as latency_count from usage_records", ); - push_filter(&mut builder, filter, None)?; + push_filter(&mut builder, filter); builder.push(" group by date, billing_model, service_tier order by date asc"); let mut days = BTreeMap::::new(); diff --git a/backend/src/telemetry/usage/store.rs b/backend/src/telemetry/usage/store.rs index 39c0ee0f9..987660d11 100644 --- a/backend/src/telemetry/usage/store.rs +++ b/backend/src/telemetry/usage/store.rs @@ -14,10 +14,7 @@ use sqlx::{PgPool, Postgres, QueryBuilder}; use thiserror::Error; use crate::{ - infra::{ - json::{decode_cursor, encode_cursor, page_offset, NumberedPage, Page}, - time::parse_rfc3339_utc, - }, + infra::json::{page_offset, NumberedPage}, telemetry::{buckets::store::PgRequestBucketStore, usage::types::UsageRecord}, }; @@ -36,8 +33,6 @@ from usage_records"; pub enum PgUsageRecordStoreError { #[error("PostgreSQL usage record operation failed: {0}")] Database(#[from] sqlx::Error), - #[error("invalid usage record pagination cursor")] - InvalidCursor, #[error("invalid usage retention days: {0}")] InvalidRetention(i64), } @@ -144,29 +139,18 @@ insert into usage_records ( Ok(result.rows_affected()) } - pub async fn list( + pub async fn list_recent( &self, filter: UsageRecordFilter, - cursor: Option, limit: u32, - ) -> PgUsageRecordStoreResult> { + ) -> PgUsageRecordStoreResult> { let limit = limit.clamp(1, 200); let mut builder = QueryBuilder::::new(USAGE_RECORD_SELECT_SQL); - push_filter(&mut builder, &filter, cursor.as_deref())?; + push_filter(&mut builder, &filter); builder.push(" order by created_at desc, id desc limit "); - builder.push_bind(i64::from(limit) + 1); + builder.push_bind(i64::from(limit)); let rows = builder.build().fetch_all(&self.pool).await?; - let has_next = rows.len() > limit as usize; - let items = rows - .into_iter() - .take(limit as usize) - .map(|row| usage_record_from_row(&row)) - .collect::>(); - let next_cursor = has_next.then(|| { - let item = items.last().expect("next page requires a usage record"); - encode_cursor(&item.created_at.to_rfc3339(), &item.id) - }); - Ok(Page { items, next_cursor }) + Ok(rows.iter().map(usage_record_from_row).collect()) } pub async fn list_page( @@ -178,7 +162,7 @@ insert into usage_records ( let page_size = page_size.clamp(1, 200); let total = count_usage_records(&self.pool, &filter).await?; let mut builder = QueryBuilder::::new(USAGE_RECORD_SELECT_SQL); - push_filter(&mut builder, &filter, None)?; + push_filter(&mut builder, &filter); builder.push(" order by created_at desc, id desc limit "); builder.push_bind(i64::from(page_size)); builder.push(" offset "); @@ -281,11 +265,7 @@ insert into usage_records ( } } -pub(super) fn push_filter( - builder: &mut QueryBuilder, - filter: &UsageRecordFilter, - cursor: Option<&str>, -) -> PgUsageRecordStoreResult<()> { +pub(super) fn push_filter(builder: &mut QueryBuilder, filter: &UsageRecordFilter) { builder.push(" where true"); push_optional_text(builder, "kind", filter.kind.as_deref()); push_optional_text( @@ -329,20 +309,6 @@ pub(super) fn push_filter( builder.push(" and created_at <= "); builder.push_bind(end_time); } - if let Some(cursor) = cursor { - let (created_at, id) = - decode_cursor(cursor).ok_or(PgUsageRecordStoreError::InvalidCursor)?; - let created_at = - parse_rfc3339_utc(&created_at).map_err(|_| PgUsageRecordStoreError::InvalidCursor)?; - builder.push(" and (created_at < "); - builder.push_bind(created_at); - builder.push(" or (created_at = "); - builder.push_bind(created_at); - builder.push(" and id < "); - builder.push_bind(id); - builder.push("))"); - } - Ok(()) } fn push_optional_text(builder: &mut QueryBuilder, column: &str, value: Option<&str>) { diff --git a/backend/tests/api/admin/accounts_routes/import_routes.rs b/backend/tests/api/admin/accounts_routes/import_routes.rs index 9916192d7..2503c5dd6 100644 --- a/backend/tests/api/admin/accounts_routes/import_routes.rs +++ b/backend/tests/api/admin/accounts_routes/import_routes.rs @@ -898,7 +898,7 @@ async fn admin_accounts_import_should_store_plain_tokens_and_list_sanitized_acco .oneshot( Request::builder() .method("GET") - .uri("/api/admin/accounts?limit=10") + .uri("/api/admin/accounts?page=1&pageSize=10") .header("cookie", "cpr_admin_session=session_1") .header("x-request-id", "req_accounts_import_list") .body(Body::empty()) diff --git a/backend/tests/api/admin/accounts_routes/list.rs b/backend/tests/api/admin/accounts_routes/list.rs index a294e8fa8..24ebcf2b2 100644 --- a/backend/tests/api/admin/accounts_routes/list.rs +++ b/backend/tests/api/admin/accounts_routes/list.rs @@ -28,7 +28,7 @@ async fn admin_accounts_list_should_not_expose_account_tokens() { .oneshot( Request::builder() .method("GET") - .uri("/api/admin/accounts?limit=10") + .uri("/api/admin/accounts") .header("cookie", "cpr_admin_session=session_1") .header("x-request-id", "req_accounts_list") .body(Body::empty()) @@ -39,6 +39,10 @@ async fn admin_accounts_list_should_not_expose_account_tokens() { assert_eq!(response.status(), StatusCode::OK); let body = response_json(response).await; + assert_eq!(body["data"]["page"]["page"], 1); + assert_eq!(body["data"]["page"]["pageSize"], 50); + assert_eq!(body["data"]["page"]["total"], 2); + assert!(body["data"]["page"].get("nextCursor").is_none()); let items = body["data"]["items"].as_array().unwrap(); let plain = items .iter() diff --git a/backend/tests/api/admin/keys_routes/authorization.rs b/backend/tests/api/admin/keys_routes/authorization.rs index 45690bf7c..8879c67bc 100644 --- a/backend/tests/api/admin/keys_routes/authorization.rs +++ b/backend/tests/api/admin/keys_routes/authorization.rs @@ -2,7 +2,7 @@ use super::*; #[tokio::test] async fn admin_client_keys_route_should_create_list_and_authorize_v1_requests() { - let (app, _dir) = admin_client_key_test_app("admin-client-key-create").await; + let (app, _pool, _dir) = admin_client_key_test_app("admin-client-key-create").await; let rejected = app .clone() @@ -27,7 +27,7 @@ async fn admin_client_keys_route_should_create_list_and_authorize_v1_requests() .header("content-type", "application/json") .header("cookie", "cpr_admin_session=session_1") .header("x-request-id", "req_api_key") - .body(Body::from(r#"{"name":"cursor"}"#)) + .body(Body::from(r#"{"name":"primary"}"#)) .unwrap(), ) .await @@ -45,7 +45,7 @@ async fn admin_client_keys_route_should_create_list_and_authorize_v1_requests() .oneshot( Request::builder() .method("GET") - .uri("/api/admin/keys?limit=10") + .uri("/api/admin/keys?page=1&pageSize=10") .header("cookie", "cpr_admin_session=session_1") .body(Body::empty()) .unwrap(), @@ -53,8 +53,11 @@ async fn admin_client_keys_route_should_create_list_and_authorize_v1_requests() .await .unwrap(); let list_body = response_json(list_response).await; - assert_eq!(list_body["data"]["items"][0]["name"], "cursor"); + assert_eq!(list_body["data"]["items"][0]["name"], "primary"); assert_eq!(list_body["data"]["items"][0]["key"], key); + assert_eq!(list_body["data"]["page"]["page"], 1); + assert_eq!(list_body["data"]["page"]["pageSize"], 10); + assert_eq!(list_body["data"]["page"]["total"], 1); assert!(key.starts_with(list_body["data"]["items"][0]["prefix"].as_str().unwrap())); let models_response = app @@ -73,7 +76,7 @@ async fn admin_client_keys_route_should_create_list_and_authorize_v1_requests() #[tokio::test] async fn admin_client_keys_status_should_disable_and_enable_authorization() { - let (app, _dir) = admin_client_key_test_app("admin-client-key-status").await; + let (app, _pool, _dir) = admin_client_key_test_app("admin-client-key-status").await; let (key_id, key) = create_admin_client_key(&app, "status-key").await; let disabled = app @@ -142,7 +145,7 @@ async fn admin_client_keys_status_should_disable_and_enable_authorization() { #[tokio::test] async fn admin_client_keys_delete_should_remove_authorization() { - let (app, _dir) = admin_client_key_test_app("admin-client-key-delete").await; + let (app, _pool, _dir) = admin_client_key_test_app("admin-client-key-delete").await; let (key_id, key) = create_admin_client_key(&app, "delete-key").await; let deleted = app @@ -174,3 +177,57 @@ async fn admin_client_keys_delete_should_remove_authorization() { .unwrap(); assert_eq!(rejected.status(), StatusCode::UNAUTHORIZED); } + +#[tokio::test] +async fn admin_client_keys_list_should_page_and_search_the_complete_key_set() { + let (app, pool, _dir) = admin_client_key_test_app("admin-client-key-pagination").await; + sqlx::query( + r" +insert into client_api_keys (id, name, label, prefix, key, enabled, created_at) +select + 'key_' || lpad(sequence::text, 3, '0'), + 'name-' || sequence, + case when sequence = 7 then 'nightly-needle' end, + 'sk_test_' || sequence, + 'sk_test_key_' || sequence, + true, + timestamptz '2026-07-01T00:00:00Z' + sequence * interval '1 second' +from generate_series(1, 55) as seeded(sequence)", + ) + .execute(&pool) + .await + .unwrap(); + + let page_response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/keys?page=3&pageSize=20") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let page_body = response_json(page_response).await; + assert_eq!(page_body["data"]["page"]["total"], 55); + assert_eq!(page_body["data"]["page"]["totalPages"], 3); + assert_eq!(page_body["data"]["items"].as_array().unwrap().len(), 15); + assert_eq!(page_body["data"]["items"][0]["id"], "key_015"); + + let search_response = app + .oneshot( + Request::builder() + .method("GET") + .uri("/api/admin/keys?page=1&pageSize=20&search=nightly-needle") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let search_body = response_json(search_response).await; + assert_eq!(search_body["data"]["page"]["total"], 1); + assert_eq!(search_body["data"]["items"][0]["id"], "key_007"); +} diff --git a/backend/tests/api/admin/keys_routes/lifecycle.rs b/backend/tests/api/admin/keys_routes/lifecycle.rs index 2647cf35b..0b594d983 100644 --- a/backend/tests/api/admin/keys_routes/lifecycle.rs +++ b/backend/tests/api/admin/keys_routes/lifecycle.rs @@ -2,7 +2,7 @@ use super::*; #[tokio::test] async fn admin_client_keys_label_should_update_clear_and_validate_label() { - let (app, _dir) = admin_client_key_test_app("admin-client-key-label").await; + let (app, _pool, _dir) = admin_client_key_test_app("admin-client-key-label").await; let (key_id, _key) = create_admin_client_key(&app, "label-key").await; let renamed = app @@ -59,7 +59,7 @@ async fn admin_client_keys_label_should_update_clear_and_validate_label() { #[tokio::test] async fn admin_client_keys_batch_delete_should_remove_found_keys_and_report_missing_ids() { - let (app, _dir) = admin_client_key_test_app("admin-client-key-delete").await; + let (app, _pool, _dir) = admin_client_key_test_app("admin-client-key-delete").await; let (key_a, api_key_a) = create_admin_client_key(&app, "batch-a").await; let (key_b, api_key_b) = create_admin_client_key(&app, "batch-b").await; diff --git a/backend/tests/api/admin/keys_routes/mod.rs b/backend/tests/api/admin/keys_routes/mod.rs index 8329d995c..8b8b67c35 100644 --- a/backend/tests/api/admin/keys_routes/mod.rs +++ b/backend/tests/api/admin/keys_routes/mod.rs @@ -4,6 +4,7 @@ use axum::{ }; use codex_proxy_rs::{api::AppState, bootstrap::services::Services}; use serde_json::json; +use sqlx::PgPool; use tower::util::ServiceExt; use crate::support::{ @@ -19,7 +20,11 @@ mod lifecycle; async fn admin_client_key_test_app( db_name: &str, -) -> (axum::Router, crate::support::storage::TestDatabaseGuard) { +) -> ( + axum::Router, + PgPool, + crate::support::storage::TestDatabaseGuard, +) { let (pool, dir) = init_test_db(db_name).await; let redis = create_test_redis(db_name).await; seed_admin_session(&pool, &redis, "session_1").await; @@ -32,7 +37,11 @@ async fn admin_client_key_test_app( runtime_fingerprint(fingerprint), )); let state = AppState::from(services.as_ref()); - (codex_proxy_rs::api::router::router().with_state(state), dir) + ( + codex_proxy_rs::api::router::router().with_state(state), + pool, + dir, + ) } async fn create_admin_client_key(app: &axum::Router, name: &str) -> (String, String) { diff --git a/backend/tests/api/admin/response.rs b/backend/tests/api/admin/response.rs index b33f987f3..e483f185f 100644 --- a/backend/tests/api/admin/response.rs +++ b/backend/tests/api/admin/response.rs @@ -6,7 +6,7 @@ use axum::{ use codex_proxy_rs::api::admin::response::{ AdminEnvelope, AdminError, AdminPageEnvelope, AdminResponse, }; -use codex_proxy_rs::infra::json::{NumberedPage, Page}; +use codex_proxy_rs::infra::json::NumberedPage; use serde_json::{json, Value}; #[test] @@ -26,65 +26,14 @@ fn admin_envelope_should_not_duplicate_request_id_in_body() { } #[test] -fn admin_page_envelope_should_expose_items_and_page_metadata_inside_data() { - let page = Page { - items: vec![json!({ "id": "evt_1" })], - next_cursor: Some("cursor_1".to_string()), - }; - let body = AdminPageEnvelope::ok(page, 50); - - let value = serde_json::to_value(body).unwrap(); - - assert_eq!( - value, - json!({ - "code": 200, - "message": "OK", - "data": { - "items": [{ "id": "evt_1" }], - "page": { - "limit": 50, - "nextCursor": "cursor_1" - } - } - }) - ); -} - -#[test] -fn admin_page_envelope_should_skip_empty_next_cursor() { - let page = Page { - items: vec![json!({ "id": "evt_1" })], - next_cursor: None, - }; - let body = AdminPageEnvelope::ok(page, 50); - - let value = serde_json::to_value(body).unwrap(); - - assert_eq!( - value, - json!({ - "code": 200, - "message": "OK", - "data": { - "items": [{ "id": "evt_1" }], - "page": { - "limit": 50 - } - } - }) - ); -} - -#[test] -fn admin_page_envelope_should_expose_numbered_page_metadata() { +fn admin_page_envelope_should_expose_stable_page_metadata() { let page = NumberedPage { items: vec![json!({ "id": "acct_1" })], total: 21, page: 2, page_size: 10, }; - let body = AdminPageEnvelope::numbered(page); + let body = AdminPageEnvelope::ok(page); let value = serde_json::to_value(body).unwrap(); diff --git a/backend/tests/api/admin/usage_routes.rs b/backend/tests/api/admin/usage_routes.rs index 63968f3d1..9438a8547 100644 --- a/backend/tests/api/admin/usage_routes.rs +++ b/backend/tests/api/admin/usage_routes.rs @@ -48,7 +48,7 @@ async fn usage_route_returns_success_facts_without_legacy_level() { let response = app .oneshot( Request::builder() - .uri("/api/admin/usage/records?page=1&pageSize=20&clientApiKeyId=key_42") + .uri("/api/admin/usage/records?clientApiKeyId=key_42") .header("cookie", "cpr_admin_session=session_1") .body(Body::empty()) .unwrap(), @@ -57,6 +57,10 @@ async fn usage_route_returns_success_facts_without_legacy_level() { .unwrap(); assert_eq!(response.status(), StatusCode::OK); let body = response_json(response).await; + assert_eq!(body["data"]["page"]["page"], 1); + assert_eq!(body["data"]["page"]["pageSize"], 50); + assert_eq!(body["data"]["page"]["total"], 1); + assert!(body["data"]["page"].get("nextCursor").is_none()); let item = &body["data"]["items"][0]; assert_eq!(item["id"], "usage_route_success"); assert_eq!(item["clientApiKeyId"], "key_42"); @@ -102,10 +106,12 @@ async fn ops_errors_route_filters_failure_facts_independently() { let body = admin_get( &app, - "/api/admin/ops/errors?page=1&pageSize=20&failureClass=rate_limited&search=req_ops", + "/api/admin/ops/errors?failureClass=rate_limited&search=req_ops", ) .await; assert_eq!(body["data"]["page"]["total"], 1); + assert_eq!(body["data"]["page"]["page"], 1); + assert_eq!(body["data"]["page"]["pageSize"], 50); assert_eq!(body["data"]["items"][0]["id"], "ops_route_error"); assert_eq!(body["data"]["items"][0]["upstreamStatusCode"], 429); diff --git a/backend/tests/api/client/responses.rs b/backend/tests/api/client/responses.rs index 96c222ada..7d8b83fba 100644 --- a/backend/tests/api/client/responses.rs +++ b/backend/tests/api/client/responses.rs @@ -97,6 +97,7 @@ async fn responses_route_should_terminate_generated_stream_errors_with_done_mark let body = String::from_utf8(bytes.to_vec()).unwrap(); assert_eq!(status, StatusCode::OK); assert!(body.contains("event: response.failed")); + assert!(body.contains(r#""code":"no_available_accounts""#)); assert!(body.ends_with("data: [DONE]\n\n")); } diff --git a/backend/tests/dispatch/service.rs b/backend/tests/dispatch/service.rs index 577ef4283..0cf3928a3 100644 --- a/backend/tests/dispatch/service.rs +++ b/backend/tests/dispatch/service.rs @@ -1530,19 +1530,17 @@ async fn latest_response_ops_error_log(pool: &PgPool) -> ResponseUsageRecordSnap } async fn latest_usage_record(pool: &PgPool, kind: &str) -> ResponseUsageRecordSnapshot { - let page = PgUsageRecordStore::new(pool.clone()) - .list( + let events = PgUsageRecordStore::new(pool.clone()) + .list_recent( UsageRecordFilter { kind: Some(kind.to_string()), ..UsageRecordFilter::default() }, - None, 1, ) .await .unwrap(); - let event = page - .items + let event = events .into_iter() .next() .unwrap_or_else(|| panic!("expected a {kind} usage record")); diff --git a/backend/tests/dispatch/service/compact_routes.rs b/backend/tests/dispatch/service/compact_routes.rs index c516b073a..9d5b08aa7 100644 --- a/backend/tests/dispatch/service/compact_routes.rs +++ b/backend/tests/dispatch/service/compact_routes.rs @@ -334,7 +334,7 @@ async fn responses_compact_should_preserve_upstream_client_error_status() { assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"]["type"], "invalid_request_error"); - assert_eq!(body["error"]["code"], "codex_api_error"); + assert_eq!(body["error"]["code"], "codex_client_error"); let event = latest_response_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); assert_eq!( @@ -353,6 +353,84 @@ async fn responses_compact_should_preserve_upstream_client_error_status() { .is_some_and(|value| value.contains("invalid_encrypted_content"))); } +#[tokio::test] +async fn responses_compact_should_map_upstream_internal_error_to_bad_gateway() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses/compact")) + .respond_with(ResponseTemplate::new(500).set_body_json(json!({ + "error": {"message": "upstream internal error"} + }))) + .expect(3) + .mount(&server) + .await; + let (app, api_key, _dir) = test_app_with_account(server.uri()).await; + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses/compact") + .header("authorization", format!("Bearer {api_key}")) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "input": [{"role": "user", "content": "hello"}] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = response_json(response).await; + + assert_eq!(status, StatusCode::BAD_GATEWAY); + assert_eq!(body["error"]["type"], "server_error"); + assert_eq!(body["error"]["code"], "upstream_error"); +} + +#[tokio::test] +async fn responses_compact_should_preserve_upstream_service_unavailable_status() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses/compact")) + .respond_with(ResponseTemplate::new(503).set_body_json(json!({ + "error": {"message": "upstream unavailable"} + }))) + .expect(3) + .mount(&server) + .await; + let (app, api_key, _dir) = test_app_with_account(server.uri()).await; + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses/compact") + .header("authorization", format!("Bearer {api_key}")) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "input": [{"role": "user", "content": "hello"}] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = response_json(response).await; + + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(body["error"]["type"], "server_error"); + assert_eq!(body["error"]["code"], "upstream_error"); +} + #[tokio::test] async fn responses_compact_should_return_responses_error_when_no_accounts_are_available() { let server = MockServer::start().await; diff --git a/backend/tests/dispatch/service/responses_recovery.rs b/backend/tests/dispatch/service/responses_recovery.rs index 570cc2b7c..c45d1d2d2 100644 --- a/backend/tests/dispatch/service/responses_recovery.rs +++ b/backend/tests/dispatch/service/responses_recovery.rs @@ -87,7 +87,51 @@ async fn responses_should_classify_sse_cyber_policy_failure_as_bad_request() { assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"]["type"], "invalid_request_error"); - assert_eq!(body["error"]["code"], "codex_api_error"); + assert_eq!(body["error"]["code"], "codex_client_error"); +} + +#[tokio::test] +async fn responses_should_preserve_upstream_client_error_status() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "error": { + "code": "invalid_input", + "message": "A required parameter is invalid." + } + }))) + .expect(1) + .mount(&server) + .await; + let (app, api_key, _dir) = test_app_with_account(server.uri()).await; + + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header("authorization", format!("Bearer {api_key}")) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "input": [{"role": "user", "content": "hello"}], + "stream": false, + "use_websocket": false + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = response_json(response).await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["type"], "invalid_request_error"); + assert_eq!(body["error"]["code"], "codex_client_error"); } #[tokio::test] diff --git a/backend/tests/fleet/store/mod.rs b/backend/tests/fleet/store/mod.rs index 2be3b041c..e9572c56f 100644 --- a/backend/tests/fleet/store/mod.rs +++ b/backend/tests/fleet/store/mod.rs @@ -54,7 +54,7 @@ async fn account_store_should_store_plain_tokens_and_load_secret_wrappers() { } #[tokio::test] -async fn account_store_should_cursor_page_accounts_by_added_at_desc() { +async fn account_store_should_page_account_metadata_by_added_at_desc() { let (pool, _dir) = account_store_parts("accounts", 5).await; let repo = PgAccountStore::new(pool.clone()); @@ -62,7 +62,7 @@ async fn account_store_should_cursor_page_accounts_by_added_at_desc() { seed_repo_account(&pool, "acct_b", "2026-06-11T00:01:00Z").await; seed_repo_account(&pool, "acct_c", "2026-06-11T00:02:00Z").await; - let first_page = repo.list(None, 2).await.unwrap(); + let first_page = repo.list_metadata_page(1, 2, None).await.unwrap(); assert_eq!( first_page .items @@ -72,7 +72,7 @@ async fn account_store_should_cursor_page_accounts_by_added_at_desc() { ["acct_c", "acct_b"] ); - let second_page = repo.list(first_page.next_cursor, 2).await.unwrap(); + let second_page = repo.list_metadata_page(2, 2, None).await.unwrap(); assert_eq!( second_page .items @@ -100,7 +100,7 @@ async fn account_store_should_list_metadata_without_exposing_tokens() { .await .unwrap(); - let page = repo.list_metadata(None, 10).await.unwrap(); + let page = repo.list_metadata_page(1, 10, None).await.unwrap(); assert_eq!(page.items[0].id, "acct_plain"); assert_eq!(page.items[0].email.as_deref(), Some("user@example.com")); diff --git a/backend/tests/keys/store.rs b/backend/tests/keys/store.rs index 185a74f38..b9dc13a78 100644 --- a/backend/tests/keys/store.rs +++ b/backend/tests/keys/store.rs @@ -9,10 +9,10 @@ async fn client_key_store_should_create_list_disable_enable_and_delete_keys() { let (store, _dir) = client_key_store("client-keys").await; let created = KeyManageService::new(store.clone()) - .create("cursor") + .create("primary") .await .unwrap(); - assert_eq!(created.name, "cursor"); + assert_eq!(created.name, "primary"); assert!(created.key.starts_with("sk_")); let stored_key: (String,) = sqlx::query_as("select key from client_api_keys where id = $1") @@ -22,9 +22,9 @@ async fn client_key_store_should_create_list_disable_enable_and_delete_keys() { .unwrap(); assert_eq!(stored_key.0, created.key); - let first_page = store.list(None, 10).await.unwrap(); + let first_page = store.list_page(1, 10, None).await.unwrap(); assert_eq!(first_page.items.len(), 1); - assert_eq!(first_page.items[0].name, "cursor"); + assert_eq!(first_page.items[0].name, "primary"); assert_eq!(first_page.items[0].prefix, created.prefix); assert_eq!(first_page.items[0].key, created.key); assert!(first_page.items[0].last_used_at.is_none()); @@ -97,7 +97,8 @@ async fn client_key_store_should_page_keys_by_created_at_desc() { .await .unwrap(); - let first_page = store.list(None, 2).await.unwrap(); + let first_page = store.list_page(1, 2, None).await.unwrap(); + assert_eq!(first_page.total, 3); assert_eq!( first_page .items @@ -107,7 +108,7 @@ async fn client_key_store_should_page_keys_by_created_at_desc() { ["c", "b"] ); - let second_page = store.list(first_page.next_cursor, 2).await.unwrap(); + let second_page = store.list_page(2, 2, None).await.unwrap(); assert_eq!( second_page .items @@ -118,6 +119,32 @@ async fn client_key_store_should_page_keys_by_created_at_desc() { ); } +#[tokio::test] +async fn client_key_store_should_search_name_label_and_id_before_paging() { + let (store, _dir) = client_key_store("client-keys-search").await; + let admin = KeyManageService::new(store.clone()); + let matching = admin.create("automation").await.unwrap(); + let named = admin.create("interactive").await.unwrap(); + store + .set_label(&matching.id, Some("nightly-needle".to_string())) + .await + .unwrap(); + + let page = store + .list_page(1, 10, Some("nightly-needle")) + .await + .unwrap(); + + assert_eq!(page.total, 1); + assert_eq!(page.items[0].id, matching.id); + + let name_page = store.list_page(1, 10, Some("interact")).await.unwrap(); + assert_eq!(name_page.items[0].id, named.id); + + let id_page = store.list_page(1, 10, Some(&matching.id)).await.unwrap(); + assert_eq!(id_page.items[0].id, matching.id); +} + async fn client_key_store( db_name: &str, ) -> (PgClientKeyStore, crate::support::storage::TestDatabaseGuard) { diff --git a/backend/tests/telemetry/usage/store.rs b/backend/tests/telemetry/usage/store.rs index 96fe81c6a..7562a4933 100644 --- a/backend/tests/telemetry/usage/store.rs +++ b/backend/tests/telemetry/usage/store.rs @@ -7,7 +7,7 @@ use codex_proxy_rs::telemetry::{ use crate::support::storage::init_test_db; #[tokio::test] -async fn usage_store_filters_and_cursor_pages_success_facts() { +async fn usage_store_should_filter_numbered_pages_of_success_facts() { let (pool, _guard) = init_test_db("usage-record-filter").await; let store = PgUsageRecordStore::new(pool); let mut matching = success_record("matching"); @@ -19,24 +19,50 @@ async fn usage_store_filters_and_cursor_pages_success_facts() { store.append(&success_record("other")).await.unwrap(); let page = store - .list( + .list_page( UsageRecordFilter { kind: Some("request".to_string()), client_api_key_id: Some("key_1".to_string()), search: Some("matching".to_string()), ..UsageRecordFilter::default() }, - None, 1, + 20, ) .await .unwrap(); - assert_eq!(page.items.len(), 1); + assert_eq!(page.total, 1); assert_eq!(page.items[0].id, "usage_matching"); assert_eq!(page.items[0].request_id.as_deref(), Some("req_1")); } +#[tokio::test] +async fn usage_store_should_list_recent_records_in_descending_order() { + let (pool, _guard) = init_test_db("usage-record-recent").await; + let store = PgUsageRecordStore::new(pool); + let now = Utc::now(); + for (id, seconds) in [("usage_old", 0), ("usage_middle", 1), ("usage_new", 2)] { + let mut record = success_record(id); + record.id = id.to_string(); + record.created_at = now + Duration::seconds(seconds); + store.append(&record).await.unwrap(); + } + + let records = store + .list_recent(UsageRecordFilter::default(), 2) + .await + .unwrap(); + + assert_eq!( + records + .iter() + .map(|record| record.id.as_str()) + .collect::>(), + ["usage_new", "usage_middle"] + ); +} + #[tokio::test] async fn usage_record_and_bucket_are_one_transaction_and_keep_zero_ms_samples() { let (pool, _guard) = init_test_db("usage-record-transaction").await; diff --git a/docs/api-contract-review.md b/docs/api-contract-review.md new file mode 100644 index 000000000..bffe9d350 --- /dev/null +++ b/docs/api-contract-review.md @@ -0,0 +1,107 @@ +# API 契约一致性审查(backend HTTP API) + +> 审查于 `feat/postgres-redis-migration` 工作树(fleet 改名之后)。覆盖 `api/client/`(OpenAI 兼容面)、`api/admin/`(管理台面)、`api/router.rs` 路由组织、`infra/json.rs` 分页原语、以及前端 `frontend/src/api/` 消费侧的类型对齐。方法:codegraph + rg 定位,逐个 endpoint 对照 serde struct 与 axum 路由,前后端字段名两侧核对。 +> +> 状态：2026-07-11 已完成 P0 第 1 条、P1 第 2/3 条和 P2 第 4/7/8 条的代码实施与全量质量门禁。下文“现状”均指修复前审查基线。 + +先说结论:字段命名极其干净(全局 camelCase,95 处 `rename_all = "camelCase"`,无一处 snake_case 泄漏),两套错误信封各自服务两类客户端且都不泄露内部错误细节,OpenAI 兼容面贴合官方契约。**真问题集中在分页**:`infra/json.rs` 里 cursor 与 page/offset 两套并存,导致三种矛盾——(1)keys endpoint 只有 cursor 一套,前端不翻页 → **超过 50 个 key 被静默截断**(唯一的真数据丢失 bug);(2)同一 endpoint 的 `data.page` 形状随请求参数在两种 schema 间漂移,且用 untagged enum 无判别字段;(3)cursor 分支在 accounts/usage/ops 上已是前端不走的死代码。其余是 REST 方法语义(`POST /delete` vs `DELETE`)、版本化(admin 无 `/v1`)、compact 与 responses 的上游错误 status 映射分叉等风格/边角不一致。 + +## P0 — 会导致客户端出错的真问题 + +### 1. `/api/admin/keys` 只有 cursor 分页,前端不跟游标 → 超过 50 个 key 静默丢失 +- 现状:keys endpoint 是**唯一只支持 cursor 一套**的列表接口。`api_keys`(`api/admin/keys_routes.rs:114-139`)只接收 `ApiKeysQuery { cursor, limit }`(`keys_routes.rs:26-31`),不认 `page`/`pageSize`;`limit = clamp_limit(query.limit.unwrap_or(50))` 默认 50、上限 200。 +- 前端 `getApiKeys()`(`frontend/src/api/modules/api-keys.ts:5-10`)**不传任何参数**,`loadApiKeys` 直接 `apiKeys.value = data.items`(`frontend/src/views/api-keys/composables/useApiKeyMutations.ts:38-39`),`next_cursor` 被丢弃;`useApiKeyFilters` 再对这已被截断的数组做**纯客户端分页**(`frontend/src/views/api-keys/composables/useApiKeyFilters.ts:21-22`)。 +- 后果:一旦 key 总数 > 50,前端永远只看得到**最新创建的 50 个**,较旧密钥永久不可见,分页器还显示得像是全量。这是本次审查唯一的**真数据可见性 bug**,不是洁癖。 +- 修复:keys 已统一实现 `page`/`pageSize`/`search` + `count(*)`,前端直接消费服务端页和 `data.page.total`,搜索覆盖名称、标签、ID 的完整数据集,不再对单页结果做本地搜索或二次分页。 + +## P1 — 契约稳定性 + +### 2. 同一 endpoint 的 `data.page` 形状随请求参数在两套 schema 间漂移 +- 现状:accounts / usage / ops 三个列表同时支持两种模式,按"请求里有没有 `page`/`pageSize`"运行时二选一: + - `usage_records`(`api/admin/usage_routes.rs:197`)`use_numbered_page = page.is_some() || page_size.is_some()`; + - `ops_errors`(`api/admin/ops_routes.rs:66`)同款判定; + - `accounts`(`api/admin/accounts_routes/mod.rs:525`)同款判定。 +- 响应用 untagged enum `PageMeta`(`api/admin/response.rs:210-215`)承载两种截然不同的 `data.page`: + - Cursor 分支 → `{ limit, nextCursor }`(`response.rs:193-197`); + - Numbered 分支 → `{ page, pageSize, total, totalPages }`(`response.rs:200-207`)。 +- 问题:同一个 URL,传 `?cursor=` 与传 `?page=` 拿到的 `data.page` 是两种 schema;`#[serde(untagged)]` 又不带判别字段(没有 `"mode":"cursor"` 之类),第三方 API 消费者只能靠"猜哪些字段在"来分辨。前端因为**永远只传 numbered 参数**(`useUsageRecordsTable` 传 `page/pageSize`、`useAccountMutations` 传 `page/pageSize`、`useOpsErrorsTable` 读 `result.page.total/page/pageSize/totalPages`)侥幸无感,但契约面是脆的。 +- 建议:统一收敛到 numbered 一套(前端事实上只用它),响应 `data.page` 定为单一稳定 schema;若要保留 cursor 供未来无限滚动,至少给 `PageMeta` 加显式判别字段,别用 untagged。 +- 修复:accounts / keys / usage / ops 已统一为 numbered 分页,`PageMeta` 已收敛为单一结构 `{ page, pageSize, total, totalPages }`,默认请求也不再产生另一种 schema。 + +### 3. cursor 分页在 accounts/usage/ops 上已是前端不走的死代码 +- 现状:`infra/json.rs` 同时导出 cursor 原语(`Page`/`encode_cursor`/`decode_cursor`,`json.rs:11-19`、`36-46`)与 page/offset 原语(`NumberedPage`/`page_offset`,`json.rs:22-33`、`59-61`)。每个双模 handler 都完整保留了 cursor 分支(如 `usage_routes.rs:230-253`、`ops_routes.rs:88-100`、`accounts_routes/mod.rs:557-579`),但前端全部只走 numbered 分支。 +- 唯一还真用 cursor 的调用方是 `dashboard_summary` 内部 `list(None, LIMIT, ...)` 取最近 N 条(`api/admin/dashboard_routes.rs:214`)——它只要"第一页",本质是 `limit` 而非翻页。 +- 额外死代码:`AccountUsageQueryService::list` 与 `PgAccountUsageStore::list_usage` 没有 HTTP 或其他生产调用方,同样只保留了无消费者的 cursor SQL。 +- 问题:两套原语 + 每 handler 双分支,是这组契约不一致(第 1、2 条)的根因,也是持续的维护税与测试盲区。 +- 建议:与第 2 条一起做——对外列表统一 numbered;cursor 若仅服务 dashboard 的"取前 N 条",可降级成一个不暴露翻页语义的 `list_recent(limit)`,把 `Page`/`encode_cursor` 从公共分页原语里摘出去。 +- 修复:Dashboard 已改用显式 `list_recent(limit)`,`Page`/`encode_cursor`/`decode_cursor`、各列表 cursor 分支和账号用量 cursor 死路径均已删除。 + +## P2 — 风格不统一 / 边角契约,视精力而定 + +### 4. `/v1/responses` 与 `/v1/responses/compact` 对同一上游失败映射出不同 HTTP status +- 现状:非流式 `/v1/responses` 走 `responses_dispatch_error_response`(`api/client/responses.rs:316`),内部用 `ResponseDispatchStatusMode::UpstreamFailureStatus`,对 `Upstream(_)` **强制 502**(`api/client/errors.rs:266-276`);compact 走 `responses_compact_dispatch_error_response`(`responses.rs:356`),用 `ResponseDispatchStatusMode::Client`,**透传上游真实 status**(如 500/503,`errors.rs:118-132`)。 +- 问题:上游同一个 5xx,responses 返回 502、compact 返回 500——两个姊妹 endpoint 对"同类错误"给客户端不同 status。二者都自洽(代理返 502 更规范;compact 透传更贴上游),但彼此分叉。 +- 建议:统一策略(建议都按代理语义规整为 502/503),或在文档里显式标注 compact 为"透传上游 status"。 +- 修复:Responses 与 Compact 共用同一客户端状态策略。可由客户端修复的上游 4xx 保留原状态;上游 503 保留为 503;其他通用上游 5xx 统一为 502。PostgreSQL 运维日志的 `client_status_code` 使用同一计算。 + +### 5. admin 写操作用 `POST /xxx/delete`、`POST /xxx/update`,与 settings 的 `DELETE` 语义自相矛盾 +- 现状:同一 admin 面里,删除/更新混用两种风格: + - 动词塞进路径 + POST:`POST /api/admin/accounts/delete`、`/accounts/update`、`/keys/delete`、`/keys/update`、`/usage/records/delete`(`api/admin/router.rs:72-73`、`101`、`105-106`); + - 又存在规范的 `DELETE /api/admin/settings/admin-api-key`(`router.rs:44`)。 +- 问题:纯 REST 语义不一致(POST 造资源 vs DELETE 删资源被 `POST /delete` 绕过)。因前端硬编码 method,不会真出错,属洁癖级。 +- 建议:要么全面 `DELETE`/`PATCH`,要么承认"动作式 POST"约定并统一到底、连 settings 那个 DELETE 一起归拢;不必强改,但应二选一写进规范。 + +### 6. client 面 `/v1` 版本化,admin 面 `/api/admin` 无版本 +- 现状:OpenAI 面全部 `/v1/*`(`api/client/router.rs:17-27`);admin 面全部 `/api/admin/*` 无版本段(`api/admin/router.rs`)。 +- 问题:两面版本化策略不一致。admin 是内部管理台、前后端同仓齐发,无版本可接受;仅记录以免误会成疏漏。 +- 建议:维持现状即可;若未来 admin 要对外开放,再引入 `/api/admin/v1`。（这个不做） + +### 7. `code: "codex_api_error"` 在两种 `type` 下复用 +- 现状:`responses_error_kind_for_status`(`api/client/errors.rs:249-264`)对 4xx 上游返回 `{ type: "invalid_request_error", code: "codex_api_error" }`,对 5xx 返回常量 `CODEX_API_ERROR = { type: "server_error", code: "codex_api_error" }`(`errors.rs:85-88`)。同一 `code` 字符串挂在两种 `type` 下。 +- 问题:按 `(type, code)` 对做分支的客户端会看到同 code 不同 type。非阻断。 +- 建议:给 4xx 场景换一个独立 code(如 `codex_client_error`),让 code 与 type 一一对应。 +- 修复:通用上游 4xx 使用 `type: "invalid_request_error"` + `code: "codex_client_error"`;`codex_api_error` 只用于服务端错误类型。 + +### 8. 流式与非流式对"无可用账号"给出不同 error code +- 现状:非流式 `NoActiveAccount` → HTTP 503 + `code: "no_available_accounts"`(`api/client/errors.rs:144-155`);流式则以 SSE(HTTP 200)内嵌 `response.failed` 事件,kind 走 `UPSTREAM_UNAVAILABLE_ERROR` → `code: "upstream_unavailable"`(`errors.rs:53-56`、`338`)。 +- 问题:同一"无账号"条件,流式/非流式 code 命名不同(`upstream_unavailable` vs `no_available_accounts`)。HTTP status 差异是流式固有(200 后无法改 status),但 code 字符串可以对齐。 +- 建议:两条路径统一 code 命名。 +- 修复:流式 `response.failed` 与非流式响应均使用 `code: "no_available_accounts"`。 + +### 9. 前端仅 `system` 模块有 TS 类型,其余全 `any` —— 无编译期漂移防护 +- 现状:`system.ts` 定义了 `SystemVersion`/`SystemUpdateInfo` 等接口(`frontend/src/api/modules/system.ts:7-33`),经核对与后端 `VersionData`(`update/service.rs:43-54`)、`UpdateInfoData`(`update/release.rs:25-39`)**字段逐一吻合**;但 accounts/usage/dashboard/keys/settings 模块的入参与返回全是 `any`(如 `api/modules/accounts.ts:5`、`usage.ts:3`)。 +- 问题:高频实体(账号/key/用量)没有一处 TS 契约,后端改字段名前端零编译告警,只会运行时 `undefined`。当前无实际错配(camelCase 两侧一致,抽查 `OpsErrorDetailModal` 读 `failureClass/clientApiKeyId/upstreamRequestId` 与 `OpsErrorLog` flatten 序列化对得上),但缺防护网。 +- 建议:为账号/key/用量/设置补最小 TS interface(可从后端 struct 半自动生成),把 `request` 的泛型用起来。 + +## 做得好的(记录一下,免得反复怀疑) + +- **字段命名全局统一 camelCase**:95 处 `rename_all = "camelCase"`,仅 3 处 `lowercase` 且都用在枚举变体序列化(`telemetry/usage/types.rs:10`、`telemetry/dashboard.rs:28`、`update/service.rs:67`,产出 `"info"/"success"` 这类小写值),用法正确,不算不一致。 +- **内部错误细节不泄露**(抽查确认):admin 各错误类型(`UsageQueryError`/`OpsQueryError`/`AccountManageError`/`KeyManageError`)的 `Display` 全是静态文案(如 `"failed to list usage records"`),`AdminError::internal(error.to_string())` 落地的是这些静态串而非裸 DB 错误;settings 更是刻意把 `Database/Json/StoredField` 统一收敛成 `"Failed to persist settings"`(`api/admin/settings_routes.rs:155-164`)。少数带 detail 的(`FetchQuota`/`OAuthCodeExchange`,`accounts_routes/quota_routes.rs:38`)是上游 Codex 报文且仅面向已鉴权 admin,可接受。 +- **两套错误信封各服务其众且都自洽**:admin 用 `{ code:u32, message, data:null }` + 业务码段(400xx/401xx/409xx/500xx,`api/admin/response.rs:13-29`),OpenAI 面用 `{ error:{ message, type, code } }`(`api/client/errors.rs:91-107`)贴官方结构。这是按受众分层,不是分裂。 +- **OpenAI 兼容度到位**:`/v1/models` 返回 `{ object:"list", data:[{ id, object:"model", created, owned_by:"openai" }] }`(`api/client/models.rs:35-42`、`92-99`),`/v1/models/{id}` 返回单 model 对象,SSE 用 `text/event-stream` + `data: [DONE]\n\n` 收尾(`api/client/sse.rs`、`upstream/openai/protocol/sse.rs:23`),均合官方契约;`/v1/responses` 是透明代理原样透传。 +- **admin 鉴权入口统一**:`AdminAuth` 提取器一处实现(会话 Cookie 或 `x-api-key`,`api/admin/session.rs:45-58`),所有 admin handler 走同一门。 +- **SPA fallback 对 API 路径正确回 404 JSON**:`/api`、`/v1` 前缀命中时返回 `{ code:40401, message, data:null }` 而非把 index.html 当接口喂回(`api/assets.rs:29-39`),与 admin 错误信封同构。 +- **request_id 契约闭环**:入站校验/生成、注入 extension、回写响应头 `x-request-id`(`api/middleware/request_id.rs`),前端 `request.ts` 从 `response.headers['x-request-id']` 取回挂到 `ApiError`(`frontend/src/api/request.ts:71`)。 + +## 核实过的误报(排查后判定无问题) + +- **`system` 前端接口 vs 后端 struct 疑似漂移**:逐字段核对 `SystemVersion`/`SystemUpdateInfo` 与 `VersionData`/`UpdateInfoData`,camelCase 名称全部一致,无漂移。 +- **query 参数疑似 snake_case**:各 `*Query` struct 字段名虽写作 `client_api_key_id`/`page_size`/`start_time`,但 struct 上都有 `rename_all = "camelCase"`,实际接收的是 `clientApiKeyId`/`pageSize`/`startTime`,与前端发送一致。 +- **`AdminAccountPageEnvelope` 是不是又一套分页 schema**:它(`accounts_routes/mod.rs:355-398`)确实手写重复了 `AdminPageEnvelope` 的 cursor/numbered 逻辑,但**产出的 `data.page` 形状与共享 `PageMeta` 完全一致**,只是多挂了一个 `data.summary`。属代码冗余(可复用 `AdminPageEnvelope`),非契约不一致,故不计入 P 级问题。 +- **client 面 401 消息"Missing client API key"用词**:key 存在但非法时也回这句(`api/client/auth.rs` 只对外暴露 bool,失败原因仅进日志),消息略含糊但 status 正确、不泄露 key,属文案洁癖非契约错。 + +## 建议动手顺序 + +1. **P0 第 1 条**:已完成。keys 使用服务端 numbered 分页、全量搜索和总数统计。 +2. **P1 第 2、3 条**:已完成。对外列表统一 numbered,Dashboard 使用 `list_recent(limit)`,cursor 原语与死路径已删除。 +3. **P2 第 4、7、8 条**:已完成。Responses/Compact/流式错误状态与错误码已统一。 +4. **P2 第 9 条**(前端补 TS 类型)可作为独立收尾,给账号/key/用量/设置建最小 interface。 +5. **P2 第 5、6 条**(REST 方法语义 / 版本化)属规范决策,择期定调即可,不必强改。 + +## 实施验收 + +- Rust 格式检查通过。 +- `cargo clippy --all-targets --all-features --locked -- -D warnings` 通过,0 警告。 +- 后端全量集成测试 623 项通过,0 失败。 +- 前端 Prettier 检查、`vue-tsc` 和 Vite 生产构建通过,0 警告。 +- Docker runtime 镜像 `codex-proxy-rs:api-contract-review` 构建成功;容器健康检查为 healthy,`GET /healthz` 返回 204,启动日志 WARN/ERROR 为 0。 diff --git a/frontend/src/api/modules/api-keys.ts b/frontend/src/api/modules/api-keys.ts index e19d67c4a..1cf176d2c 100644 --- a/frontend/src/api/modules/api-keys.ts +++ b/frontend/src/api/modules/api-keys.ts @@ -2,15 +2,47 @@ import request, { type ApiPayload } from '../request' // ==================== API Keys 管理 ==================== -export function getApiKeys() { - return request({ +export interface ApiKeyListParams { + page: number + pageSize: number + search?: string +} + +export interface ApiKeyPageMeta { + page: number + pageSize: number + total: number + totalPages: number +} + +export interface ClientApiKey { + id: string + name: string + label: string | null + prefix: string + key: string + enabled: boolean + createdAt: string + createdAtDisplay: string + lastUsedAt: string | null + lastUsedAtDisplay: string +} + +export interface ApiKeyListResult { + items: ClientApiKey[] + page: ApiKeyPageMeta +} + +export function getApiKeys(params: ApiKeyListParams) { + return request({ url: '/api/admin/keys', method: 'GET', + params, }) } export function createApiKey(data: ApiPayload) { - return request({ + return request({ url: '/api/admin/keys', method: 'POST', data, @@ -26,7 +58,7 @@ export function deleteApiKeys(data: ApiPayload) { } export function updateApiKey(data: ApiPayload) { - return request({ + return request({ url: '/api/admin/keys/update', method: 'POST', data, diff --git a/frontend/src/views/api-keys/composables/useApiKeyFilters.ts b/frontend/src/views/api-keys/composables/useApiKeyFilters.ts index 1cd345e7a..af0e9abbf 100644 --- a/frontend/src/views/api-keys/composables/useApiKeyFilters.ts +++ b/frontend/src/views/api-keys/composables/useApiKeyFilters.ts @@ -1,64 +1,56 @@ -import { clamp } from 'es-toolkit' -import { computed, ref, watch, type Ref } from 'vue' +// @env browser +import { watchDebounced } from '@vueuse/core' +import { computed, shallowRef, type Ref } from 'vue' -export function useApiKeyFilters(apiKeys: Ref) { - const page = ref(1) - const pageSize = ref(20) - const searchQuery = ref('') - - const filteredKeys = computed(() => { - if (!searchQuery.value) return apiKeys.value - const query = searchQuery.value.toLowerCase() - return apiKeys.value.filter( - (key) => - key.name.toLowerCase().includes(query) || - key.label?.toLowerCase().includes(query) || - key.id.toLowerCase().includes(query), - ) - }) - - const pagedKeys = computed(() => { - const start = (page.value - 1) * pageSize.value - return filteredKeys.value.slice(start, start + pageSize.value) - }) +export function useApiKeyFilters(totalApiKeys: Ref) { + const page = shallowRef(1) + const pageSize = shallowRef(20) + const searchQuery = shallowRef('') + let loadApiKeys: (() => Promise | void) | undefined const apiKeyPagination = computed(() => ({ page: page.value, pageSize: pageSize.value, - total: filteredKeys.value.length, + total: totalApiKeys.value, pageSizes: [10, 20, 50, 100], })) + function bindApiKeyLoader(loader: () => Promise | void) { + loadApiKeys = loader + } + + function requestLoad() { + if (loadApiKeys) { + void loadApiKeys() + } + } + function handlePageChange(nextPage: number) { page.value = nextPage + requestLoad() } function handlePageSizeChange(nextPageSize: number) { pageSize.value = nextPageSize page.value = 1 + requestLoad() } - watch(searchQuery, () => { - page.value = 1 - }) - - watch(filteredKeys, () => { - const totalPages = clamp( - Math.ceil(filteredKeys.value.length / pageSize.value), - 1, - Number.POSITIVE_INFINITY, - ) - if (page.value > totalPages) { - page.value = totalPages - } - }) + watchDebounced( + searchQuery, + () => { + page.value = 1 + requestLoad() + }, + { debounce: 250 }, + ) return { page, pageSize, searchQuery, - pagedKeys, apiKeyPagination, + bindApiKeyLoader, handlePageChange, handlePageSizeChange, } diff --git a/frontend/src/views/api-keys/composables/useApiKeyMutations.ts b/frontend/src/views/api-keys/composables/useApiKeyMutations.ts index 3d3d460e5..8adcbb746 100644 --- a/frontend/src/views/api-keys/composables/useApiKeyMutations.ts +++ b/frontend/src/views/api-keys/composables/useApiKeyMutations.ts @@ -1,23 +1,31 @@ // @env browser import { useClipboard } from '@vueuse/core' +import { clamp } from 'es-toolkit' import { onMounted, ref, type Ref } from 'vue' import { createApiKey, deleteApiKeys, getApiKeys, updateApiKey } from '@/api' +import type { ClientApiKey } from '@/api/modules/api-keys' import { toast } from '@/components/base/BaseToast' import { useAsyncAction } from '@/composables/useAsyncAction' import { useIdSet } from '@/composables/useIdSet' -export function useApiKeyMutations(selectedIds: Ref>) { +export function useApiKeyMutations(options: { + page: Ref + pageSize: Ref + searchQuery: Ref + selectedIds: Ref> + totalApiKeys: Ref +}) { const { copy } = useClipboard() const loading = ref(true) - const apiKeys = ref([]) + const apiKeys = ref([]) const showCreateModal = ref(false) const showDeleteModal = ref(false) const showSingleDeleteModal = ref(false) const showKeyModal = ref(false) const createdKey = ref('') const createdKeyName = ref('') - const pendingDeleteKey = ref(null) + const pendingDeleteKey = ref(null) const creatingKeyAction = useAsyncAction() const deletingKeyAction = useAsyncAction() const batchDeletingAction = useAsyncAction() @@ -35,8 +43,20 @@ export function useApiKeyMutations(selectedIds: Ref>) { async function loadApiKeys() { try { loading.value = true - const data = await getApiKeys() - apiKeys.value = data.items + const result = await getApiKeys({ + page: options.page.value, + pageSize: options.pageSize.value, + search: options.searchQuery.value.trim() || undefined, + }) + apiKeys.value = result.items + options.page.value = result.page.page + options.pageSize.value = result.page.pageSize + options.totalApiKeys.value = result.page.total + + if (apiKeys.value.length === 0 && result.page.total > 0 && result.page.page > 1) { + options.page.value = clamp(result.page.totalPages, 1, Number.POSITIVE_INFINITY) + await loadApiKeys() + } } catch (error: any) { toast.error(error.message || '加载失败') } finally { @@ -71,7 +91,7 @@ export function useApiKeyMutations(selectedIds: Ref>) { ) } - function requestDeleteKey(key: any) { + function requestDeleteKey(key: ClientApiKey) { pendingDeleteKey.value = key showSingleDeleteModal.value = true } @@ -96,13 +116,13 @@ export function useApiKeyMutations(selectedIds: Ref>) { async function handleBatchDelete() { if (batchDeleting.value) return - if (selectedIds.value.size === 0) return + if (options.selectedIds.value.size === 0) return await batchDeletingAction.run( async () => { - const deleteCount = selectedIds.value.size - await deleteApiKeys({ ids: [...selectedIds.value] }) - selectedIds.value = new Set() + const deleteCount = options.selectedIds.value.size + await deleteApiKeys({ ids: [...options.selectedIds.value] }) + options.selectedIds.value = new Set() showDeleteModal.value = false await loadApiKeys() toast.success(`已删除 ${deleteCount} 个 API Key`) @@ -111,7 +131,7 @@ export function useApiKeyMutations(selectedIds: Ref>) { ) } - async function handleToggleStatus(key: any) { + async function handleToggleStatus(key: ClientApiKey) { await updatingStatusKeys.run(key.id, async () => { try { await updateApiKey({ id: key.id, status: key.enabled ? 'disabled' : 'active' }) @@ -138,7 +158,7 @@ export function useApiKeyMutations(selectedIds: Ref>) { } onMounted(() => { - loadApiKeys() + void loadApiKeys() }) return { @@ -156,6 +176,7 @@ export function useApiKeyMutations(selectedIds: Ref>) { batchDeleting, updatingStatusKeyIds, createForm, + loadApiKeys, handleCreate, requestDeleteKey, handleDelete, diff --git a/frontend/src/views/api-keys/composables/useApiKeysTable.ts b/frontend/src/views/api-keys/composables/useApiKeysTable.ts index 54df43946..aeba812a3 100644 --- a/frontend/src/views/api-keys/composables/useApiKeysTable.ts +++ b/frontend/src/views/api-keys/composables/useApiKeysTable.ts @@ -1,13 +1,17 @@ import { computed, ref, type Ref } from 'vue' -export function useApiKeysTable(pagedKeys: Ref, selectedIds = ref>(new Set())) { +import type { ClientApiKey } from '@/api/modules/api-keys' + +export function useApiKeysTable( + pageKeys: Ref, + selectedIds = ref>(new Set()), +) { const allSelected = computed( - () => - pagedKeys.value.length > 0 && pagedKeys.value.every((key) => selectedIds.value.has(key.id)), + () => pageKeys.value.length > 0 && pageKeys.value.every((key) => selectedIds.value.has(key.id)), ) const indeterminate = computed( - () => pagedKeys.value.some((key) => selectedIds.value.has(key.id)) && !allSelected.value, + () => pageKeys.value.some((key) => selectedIds.value.has(key.id)) && !allSelected.value, ) const selectedRowKeys = computed(() => [...selectedIds.value]) @@ -25,9 +29,9 @@ export function useApiKeysTable(pagedKeys: Ref, selectedIds = ref next.delete(key.id)) + pageKeys.value.forEach((key) => next.delete(key.id)) } else { - pagedKeys.value.forEach((key) => next.add(key.id)) + pageKeys.value.forEach((key) => next.add(key.id)) } selectedIds.value = next } diff --git a/frontend/src/views/api-keys/index.vue b/frontend/src/views/api-keys/index.vue index e8eb17c5f..c8d2fb294 100644 --- a/frontend/src/views/api-keys/index.vue +++ b/frontend/src/views/api-keys/index.vue @@ -2,6 +2,7 @@ import { computed, ref, shallowRef } from 'vue' import { API_BASE_URL } from '@/api/constants' +import type { ClientApiKey } from '@/api/modules/api-keys' import BaseCard from '@/components/base/BaseCard.vue' import BaseCheckbox from '@/components/base/BaseCheckbox.vue' import BaseConfirmModal from '@/components/base/BaseConfirmModal.vue' @@ -20,8 +21,20 @@ import ApiKeyStatusBadge from './components/ApiKeyStatusBadge.vue' import ApiKeyUseModal from './components/ApiKeyUseModal.vue' const selectedIds = ref>(new Set()) +const totalApiKeys = ref(0) const showUseKeyModal = shallowRef(false) -const selectedUseKey = shallowRef(null) +const selectedUseKey = shallowRef(null) + +const { + page, + pageSize, + searchQuery, + apiKeyPagination, + bindApiKeyLoader, + handlePageChange, + handlePageSizeChange, +} = useApiKeyFilters(totalApiKeys) + const { loading, apiKeys, @@ -37,22 +50,22 @@ const { batchDeleting, updatingStatusKeyIds, createForm, + loadApiKeys, handleCreate, requestDeleteKey, handleDelete, handleBatchDelete, handleToggleStatus, copyToClipboard, -} = useApiKeyMutations(selectedIds) - -const { searchQuery, pagedKeys, apiKeyPagination, handlePageChange, handlePageSizeChange } = - useApiKeyFilters(apiKeys) +} = useApiKeyMutations({ page, pageSize, searchQuery, selectedIds, totalApiKeys }) const { allSelected, indeterminate, selectedRowKeys, toggleSelection, toggleAll } = useApiKeysTable( - pagedKeys, + apiKeys, selectedIds, ) +bindApiKeyLoader(loadApiKeys) + const serviceRootUrl = computed(() => resolveServiceRootUrl()) const openAiBaseUrl = computed(() => `${serviceRootUrl.value}/v1`) @@ -85,12 +98,12 @@ function importCreatedKeyToCcs() { }) } -function openUseKeyModal(apiKey: any) { +function openUseKeyModal(apiKey: ClientApiKey) { selectedUseKey.value = apiKey showUseKeyModal.value = true } -function importToCcs(apiKey: any) { +function importToCcs(apiKey: ClientApiKey) { if (!apiKey.key) return window.location.href = buildCodexCcSwitchImportDeeplink({ @@ -134,7 +147,7 @@ function importToCcs(apiKey: any) {