refactor(server): move the catalog's SQL into repositories

crates/server/src/handlers/platform_pricing.rs

@@ -14,21 +14,13 @@
 use axum::Json;
 use axum::extract::State;
 use rust_decimal::Decimal;
-use serde::{Deserialize, Serialize};
+use serde::Deserialize;
 
 use think_watch_common::errors::AppError;
 
 use crate::app::AppState;
 use crate::middleware::auth_guard::AuthUser;
-
-#[derive(Debug, Serialize, sqlx::FromRow, utoipa::ToSchema)]
-pub struct PlatformPricing {
-    #[schema(value_type = f64)]
-    pub input_price_per_token: Decimal,
-    #[schema(value_type = f64)]
-    pub output_price_per_token: Decimal,
-    pub currency: String,
-}
+use crate::services::pricing_repository::{self as repo, PlatformPricing};
 
 #[derive(Debug, Deserialize, utoipa::ToSchema)]
 pub struct UpdatePlatformPricingRequest {
@@ -54,13 +46,7 @@ pub async fn get_platform_pricing(
     State(state): State<AppState>,
 ) -> Result<Json<PlatformPricing>, AppError> {
     auth_user.require_permission("settings:read")?;
-    let row = sqlx::query_as::<_, PlatformPricing>(
-        "SELECT input_price_per_token, output_price_per_token, currency \
-         FROM platform_pricing WHERE id = 1",
-    )
-    .fetch_one(&state.db)
-    .await?;
-    Ok(Json(row))
+    Ok(Json(repo::get(&state.db).await?))
 }
 
 #[utoipa::path(
@@ -99,19 +85,12 @@ pub async fn update_platform_pricing(
         ));
     }
 
-    let updated = sqlx::query_as::<_, PlatformPricing>(
-        r#"UPDATE platform_pricing SET
-              input_price_per_token  = COALESCE($1, input_price_per_token),
-              output_price_per_token = COALESCE($2, output_price_per_token),
-              currency               = COALESCE($3, currency),
-              updated_at             = now()
-           WHERE id = 1
-           RETURNING input_price_per_token, output_price_per_token, currency"#,
+    let updated = repo::update(
+        &state.db,
+        req.input_price_per_token,
+        req.output_price_per_token,
+        req.currency.as_deref(),
     )
-    .bind(req.input_price_per_token)
-    .bind(req.output_price_per_token)
-    .bind(req.currency.as_ref())
-    .fetch_one(&state.db)
     .await?;
 
     state.audit.log(

crates/server/src/handlers/providers.rs

@@ -8,6 +8,7 @@ use think_watch_common::models::Provider;
 
 use crate::app::AppState;
 use crate::middleware::auth_guard::AuthUser;
+use crate::services::provider_repository as repo;
 
 // ---------------------------------------------------------------------------
 // At-rest encryption for provider secrets stored in `providers.config_json`.
@@ -219,11 +220,7 @@ pub async fn list_providers(
     auth_user
         .require_global_permission(&state.db, "providers:read")
         .await?;
-    let mut providers = sqlx::query_as::<_, Provider>(
-        "SELECT * FROM providers WHERE deleted_at IS NULL ORDER BY created_at DESC",
-    )
-    .fetch_all(&state.db)
-    .await?;
+    let mut providers = repo::list_live(&state.db).await?;
     providers.iter_mut().for_each(redact_provider_secrets);
 
     Ok(Json(providers))
@@ -265,16 +262,14 @@ pub async fn create_provider(
     encrypt_aws_secret_in_config(&mut config, &state.config.encryption_key)?;
     config["headers"] = encrypt_headers_for_storage(&req.headers, &state.config.encryption_key)?;
 
-    let mut provider = sqlx::query_as::<_, Provider>(
-        r#"INSERT INTO providers (name, display_name, provider_type, base_url, config_json)
-           VALUES ($1, $2, $3, $4, $5) RETURNING *"#,
+    let mut provider = repo::insert(
+        &state.db,
+        &req.name,
+        &req.display_name,
+        &req.provider_type,
+        &req.base_url,
+        &config,
     )
-    .bind(&req.name)
-    .bind(&req.display_name)
-    .bind(&req.provider_type)
-    .bind(&req.base_url)
-    .bind(&config)
-    .fetch_one(&state.db)
     .await?;
 
     state.audit.log(
@@ -324,13 +319,9 @@ pub async fn update_provider(
     auth_user
         .require_global_permission(&state.db, "providers:update")
         .await?;
-    let existing = sqlx::query_as::<_, Provider>(
-        "SELECT * FROM providers WHERE id = $1 AND deleted_at IS NULL",
-    )
-    .bind(id)
-    .fetch_optional(&state.db)
-    .await?
-    .ok_or(AppError::NotFound("Provider not found".into()))?;
+    let existing = repo::find_live(&state.db, id)
+        .await?
+        .ok_or(AppError::NotFound("Provider not found".into()))?;
 
     let display_name = req
         .display_name
@@ -362,31 +353,15 @@ pub async fn update_provider(
         config
     };
 
-    let mut updated = sqlx::query_as::<_, Provider>(
-        r#"UPDATE providers SET display_name = $2, base_url = $3, config_json = $4
-           WHERE id = $1 RETURNING *"#,
-    )
-    .bind(id)
-    .bind(display_name)
-    .bind(base_url)
-    .bind(&config_json)
-    .fetch_one(&state.db)
-    .await?;
+    let mut updated = repo::update(&state.db, id, display_name, base_url, &config_json).await?;
 
     // A new base URL or credential can mean an entirely different
     // upstream, so every dialect we learned for this provider's routes
     // is now a guess about a host that may no longer be there. Clear
     // them and let the runtime relearn on first use — stale beats
     // wrong, and the relearn is invisible to the caller.
     if req.base_url.is_some() || req.headers.is_some() {
-        let cleared: u64 = sqlx::query(
-            "UPDATE model_routes SET upstream_protocol = NULL
-             WHERE provider_id = $1 AND upstream_protocol IS NOT NULL",
-        )
-        .bind(id)
-        .execute(&state.db)
-        .await?
-        .rows_affected();
+        let cleared = repo::clear_learned_protocols(&state.db, id).await?;
         // Same reasoning for the probe cache: "this upstream refuses
         // model X" described the old endpoint. Dropping it is also the
         // path back for an operator who fixed access upstream and
@@ -438,13 +413,9 @@ pub async fn get_provider(
     auth_user
         .require_global_permission(&state.db, "providers:read")
         .await?;
-    let mut provider = sqlx::query_as::<_, Provider>(
-        "SELECT * FROM providers WHERE id = $1 AND deleted_at IS NULL",
-    )
-    .bind(id)
-    .fetch_optional(&state.db)
-    .await?
-    .ok_or(AppError::NotFound("Provider not found".into()))?;
+    let mut provider = repo::find_live(&state.db, id)
+        .await?
+        .ok_or(AppError::NotFound("Provider not found".into()))?;
     redact_provider_secrets(&mut provider);
 
     Ok(Json(provider))
@@ -472,27 +443,8 @@ pub async fn delete_provider(
     auth_user
         .require_global_permission(&state.db, "providers:delete")
         .await?;
-    let name: Option<String> = sqlx::query_scalar("SELECT name FROM providers WHERE id = $1")
-        .bind(id)
-        .fetch_optional(&state.db)
-        .await?;
-
-    // Soft-delete + drop routes in one transaction. The `model_routes`
-    // FK is `ON DELETE CASCADE`, but since we only flip `deleted_at`
-    // the cascade doesn't fire — hence the explicit DELETE below.
-    // Orphaned routes would otherwise show up in the Models page with
-    // a raw provider UUID and no way to edit them.
-    let mut tx = state.db.begin().await?;
-    sqlx::query("UPDATE providers SET deleted_at = now() WHERE id = $1 AND deleted_at IS NULL")
-        .bind(id)
-        .execute(&mut *tx)
-        .await?;
-    let routes_deleted = sqlx::query("DELETE FROM model_routes WHERE provider_id = $1")
-        .bind(id)
-        .execute(&mut *tx)
-        .await?
-        .rows_affected();
-    tx.commit().await?;
+    let name = repo::name_of(&state.db, id).await?;
+    let routes_deleted = repo::soft_delete(&state.db, id).await?;
 
     state.audit.log(
         auth_user
@@ -569,13 +521,9 @@ pub async fn test_provider(
 
     let mut req = req;
     if let Some(provider_id) = req.provider_id {
-        let provider = sqlx::query_as::<_, Provider>(
-            "SELECT * FROM providers WHERE id = $1 AND deleted_at IS NULL",
-        )
-        .bind(provider_id)
-        .fetch_optional(&state.db)
-        .await?
-        .ok_or(AppError::NotFound("Provider not found".into()))?;
+        let provider = repo::find_live(&state.db, provider_id)
+            .await?
+            .ok_or(AppError::NotFound("Provider not found".into()))?;
         let stored = decrypt_headers_from_config(
             &provider.config_json,
             &state.config.encryption_key,

crates/server/src/openapi.rs

@@ -29,7 +29,7 @@ use crate::handlers::{
     },
     mcp_tools::{McpToolListResponse, McpToolRow},
     models::{
-        BatchWeightUpdate, BatchWeightsRequest, CreateModelRequest, ModelRow, RouteHistoryBucket,
+        BatchWeightUpdate, BatchWeightsRequest, CreateModelRequest, RouteHistoryBucket,
         RouteHistoryResponse, UpdateModelRequest,
     },
     providers::{TestProviderRequest, TestProviderResponse, UpdateProviderRequest},
@@ -46,6 +46,7 @@ use crate::handlers::{
         ResetCounterResponse, UsageDay,
     },
 };
+use crate::services::model_repository::ModelRow;
 
 /// OpenAPI document covering the ThinkWatch console API (port 3001).
 ///

crates/server/src/services/mod.rs

@@ -26,6 +26,9 @@
 //! a service yet; the migration is iterative.
 
 pub mod auth_lockout;
+pub mod model_repository;
+pub mod pricing_repository;
+pub mod provider_repository;
 pub mod rbac_service;
 pub mod refresh_blacklist;
 pub mod session_service;