feat: add data source ingestion audit api

apps/api/src/handlers.rs

@@ -4,9 +4,10 @@ use serde::Serialize;
 
 use crate::error::{ApiError, ApiResult};
 use crate::models::{
-    AreaScore, AreaScoreSummary, CreateWatchlistItem, MarketOverview, MonthQuery, Neighborhood,
-    NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery, UpdateWatchlistItem,
-    WatchlistItem,
+    AreaScore, AreaScoreSummary, CreateDataSource, CreateIngestionRun, CreateWatchlistItem,
+    DataSource, FinishIngestionRun, IngestionRun, IngestionRunQuery, MarketOverview, MonthQuery,
+    Neighborhood, NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery,
+    UpdateWatchlistItem, WatchlistItem,
 };
 use crate::state::AppState;
 
@@ -176,6 +177,195 @@ pub async fn list_neighborhood_scores(
     Ok(Json(scores))
 }
 
+pub async fn list_data_sources(State(state): State<AppState>) -> ApiResult<Json<Vec<DataSource>>> {
+    let sources = sqlx::query_as::<_, DataSource>(
+        r#"
+        SELECT
+            source_id,
+            name,
+            source_type,
+            url,
+            cadence,
+            reliability,
+            notes,
+            created_at::text AS created_at
+        FROM audit.data_sources
+        ORDER BY name ASC
+        "#,
+    )
+    .fetch_all(&state.pool)
+    .await?;
+
+    Ok(Json(sources))
+}
+
+pub async fn create_data_source(
+    State(state): State<AppState>,
+    Json(payload): Json<CreateDataSource>,
+) -> ApiResult<Json<DataSource>> {
+    payload
+        .validate()
+        .map_err(|message| ApiError::BadRequest(message.to_string()))?;
+
+    let source = sqlx::query_as::<_, DataSource>(
+        r#"
+        INSERT INTO audit.data_sources (
+            name,
+            source_type,
+            url,
+            cadence,
+            reliability,
+            notes
+        )
+        VALUES ($1, $2, $3, $4, COALESCE($5, 'unknown'), COALESCE($6, ''))
+        ON CONFLICT (name) DO UPDATE
+        SET source_type = EXCLUDED.source_type,
+            url = EXCLUDED.url,
+            cadence = EXCLUDED.cadence,
+            reliability = EXCLUDED.reliability,
+            notes = EXCLUDED.notes
+        RETURNING
+            source_id,
+            name,
+            source_type,
+            url,
+            cadence,
+            reliability,
+            notes,
+            created_at::text AS created_at
+        "#,
+    )
+    .bind(payload.name.trim().to_string())
+    .bind(payload.source_type.trim().to_string())
+    .bind(payload.url)
+    .bind(payload.cadence)
+    .bind(payload.reliability)
+    .bind(payload.notes)
+    .fetch_one(&state.pool)
+    .await?;
+
+    Ok(Json(source))
+}
+
+pub async fn list_ingestion_runs(
+    State(state): State<AppState>,
+    Query(query): Query<IngestionRunQuery>,
+) -> ApiResult<Json<Vec<IngestionRun>>> {
+    let runs = sqlx::query_as::<_, IngestionRun>(
+        r#"
+        SELECT
+            r.run_id,
+            r.source_id,
+            s.name AS source_name,
+            r.status,
+            r.started_at::text AS started_at,
+            r.finished_at::text AS finished_at,
+            r.raw_uri,
+            r.row_count,
+            r.error_message
+        FROM audit.ingestion_runs r
+        LEFT JOIN audit.data_sources s ON s.source_id = r.source_id
+        WHERE ($1::bigint IS NULL OR r.source_id = $1)
+        ORDER BY r.started_at DESC, r.run_id DESC
+        LIMIT 100
+        "#,
+    )
+    .bind(query.source_id)
+    .fetch_all(&state.pool)
+    .await?;
+
+    Ok(Json(runs))
+}
+
+pub async fn create_ingestion_run(
+    State(state): State<AppState>,
+    Json(payload): Json<CreateIngestionRun>,
+) -> ApiResult<Json<IngestionRun>> {
+    payload
+        .validate()
+        .map_err(|message| ApiError::BadRequest(message.to_string()))?;
+
+    let run = sqlx::query_as::<_, IngestionRun>(
+        r#"
+        INSERT INTO audit.ingestion_runs (
+            source_id,
+            status,
+            raw_uri,
+            row_count,
+            error_message
+        )
+        VALUES ($1, COALESCE($2, 'running'), $3, $4, $5)
+        RETURNING
+            run_id,
+            source_id,
+            (
+                SELECT name
+                FROM audit.data_sources
+                WHERE source_id = audit.ingestion_runs.source_id
+            ) AS source_name,
+            status,
+            started_at::text AS started_at,
+            finished_at::text AS finished_at,
+            raw_uri,
+            row_count,
+            error_message
+        "#,
+    )
+    .bind(payload.source_id)
+    .bind(payload.status)
+    .bind(payload.raw_uri)
+    .bind(payload.row_count)
+    .bind(payload.error_message)
+    .fetch_one(&state.pool)
+    .await?;
+
+    Ok(Json(run))
+}
+
+pub async fn finish_ingestion_run(
+    State(state): State<AppState>,
+    Path(run_id): Path<i64>,
+    Json(payload): Json<FinishIngestionRun>,
+) -> ApiResult<Json<IngestionRun>> {
+    payload
+        .validate()
+        .map_err(|message| ApiError::BadRequest(message.to_string()))?;
+
+    let run = sqlx::query_as::<_, IngestionRun>(
+        r#"
+        UPDATE audit.ingestion_runs
+        SET status = $2,
+            finished_at = now(),
+            row_count = COALESCE($3, row_count),
+            error_message = $4
+        WHERE run_id = $1
+        RETURNING
+            run_id,
+            source_id,
+            (
+                SELECT name
+                FROM audit.data_sources
+                WHERE source_id = audit.ingestion_runs.source_id
+            ) AS source_name,
+            status,
+            started_at::text AS started_at,
+            finished_at::text AS finished_at,
+            raw_uri,
+            row_count,
+            error_message
+        "#,
+    )
+    .bind(run_id)
+    .bind(payload.status)
+    .bind(payload.row_count)
+    .bind(payload.error_message)
+    .fetch_optional(&state.pool)
+    .await?
+    .ok_or_else(|| ApiError::BadRequest("ingestion run not found".to_string()))?;
+
+    Ok(Json(run))
+}
+
 pub async fn list_watchlist_items(
     State(state): State<AppState>,
 ) -> ApiResult<Json<Vec<WatchlistItem>>> {
@@ -367,7 +557,10 @@ fn validate_month(month: &str) -> ApiResult<()> {
 
 #[cfg(test)]
 mod tests {
-    use crate::models::{CreateWatchlistItem, UpdateWatchlistItem};
+    use crate::models::{
+        CreateDataSource, CreateIngestionRun, CreateWatchlistItem, FinishIngestionRun,
+        UpdateWatchlistItem,
+    };
 
     use super::validate_month;
 
@@ -430,4 +623,79 @@ mod tests {
         .validate()
         .is_ok());
     }
+
+    #[test]
+    fn validates_data_source_payload() {
+        assert!(CreateDataSource {
+            name: "".to_string(),
+            source_type: "official".to_string(),
+            url: None,
+            cadence: None,
+            reliability: None,
+            notes: None,
+        }
+        .validate()
+        .is_err());
+
+        assert!(CreateDataSource {
+            name: "上海市统计局".to_string(),
+            source_type: "official".to_string(),
+            url: Some("https://tjj.sh.gov.cn/".to_string()),
+            cadence: Some("monthly".to_string()),
+            reliability: Some("high".to_string()),
+            notes: None,
+        }
+        .validate()
+        .is_ok());
+
+        assert!(CreateDataSource {
+            name: "unsafe".to_string(),
+            source_type: "manual".to_string(),
+            url: None,
+            cadence: None,
+            reliability: None,
+            notes: Some("contains password marker".to_string()),
+        }
+        .validate()
+        .is_err());
+    }
+
+    #[test]
+    fn validates_ingestion_run_payloads() {
+        assert!(CreateIngestionRun {
+            source_id: Some(1),
+            status: Some("running".to_string()),
+            raw_uri: Some("raw/official/2026-05.csv".to_string()),
+            row_count: Some(10),
+            error_message: None,
+        }
+        .validate()
+        .is_ok());
+
+        assert!(CreateIngestionRun {
+            source_id: Some(1),
+            status: Some("unknown".to_string()),
+            raw_uri: None,
+            row_count: None,
+            error_message: None,
+        }
+        .validate()
+        .is_err());
+
+        assert!(FinishIngestionRun {
+            status: "succeeded".to_string(),
+            row_count: Some(10),
+            error_message: None,
+        }
+        .validate()
+        .is_ok());
+
+        assert!(FinishIngestionRun {
+            status: "running".to_string(),
+            row_count: None,
+            error_message: None,
+        }
+        .validate()
+        .is_err());
+    }
 }

apps/api/src/models.rs

@@ -17,6 +17,11 @@ pub struct NeighborhoodScoreQuery {
     pub area_id: Option<String>,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct IngestionRunQuery {
+    pub source_id: Option<i64>,
+}
+
 #[derive(Debug, Clone, Serialize, FromRow)]
 pub struct AreaScore {
     pub area_id: String,
@@ -118,6 +123,123 @@ pub struct NeighborhoodScore {
     pub available_units: i32,
 }
 
+#[derive(Debug, Clone, Serialize, FromRow)]
+pub struct DataSource {
+    pub source_id: i64,
+    pub name: String,
+    pub source_type: String,
+    pub url: Option<String>,
+    pub cadence: Option<String>,
+    pub reliability: String,
+    pub notes: String,
+    pub created_at: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct CreateDataSource {
+    pub name: String,
+    pub source_type: String,
+    pub url: Option<String>,
+    pub cadence: Option<String>,
+    pub reliability: Option<String>,
+    pub notes: Option<String>,
+}
+
+impl CreateDataSource {
+    pub fn validate(&self) -> Result<(), &'static str> {
+        if self.name.trim().is_empty() {
+            return Err("name is required");
+        }
+        if self.source_type.trim().is_empty() {
+            return Err("source_type is required");
+        }
+        for value in [
+            Some(self.name.as_str()),
+            Some(self.source_type.as_str()),
+            self.url.as_deref(),
+            self.cadence.as_deref(),
+            self.reliability.as_deref(),
+            self.notes.as_deref(),
+        ]
+        .into_iter()
+        .flatten()
+        {
+            if has_sensitive_text(value) {
+                return Err("payload contains sensitive text");
+            }
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, Serialize, FromRow)]
+pub struct IngestionRun {
+    pub run_id: i64,
+    pub source_id: Option<i64>,
+    pub source_name: Option<String>,
+    pub status: String,
+    pub started_at: String,
+    pub finished_at: Option<String>,
+    pub raw_uri: Option<String>,
+    pub row_count: Option<i32>,
+    pub error_message: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct CreateIngestionRun {
+    pub source_id: Option<i64>,
+    pub status: Option<String>,
+    pub raw_uri: Option<String>,
+    pub row_count: Option<i32>,
+    pub error_message: Option<String>,
+}
+
+impl CreateIngestionRun {
+    pub fn validate(&self) -> Result<(), &'static str> {
+        if let Some(status) = &self.status {
+            if !is_ingestion_status(status) {
+                return Err("status must be running, succeeded, or failed");
+            }
+        }
+        if matches!(self.row_count, Some(value) if value < 0) {
+            return Err("row_count must be non-negative");
+        }
+        for value in [self.raw_uri.as_deref(), self.error_message.as_deref()]
+            .into_iter()
+            .flatten()
+        {
+            if has_sensitive_text(value) {
+                return Err("payload contains sensitive text");
+            }
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Deserialize)]
+pub struct FinishIngestionRun {
+    pub status: String,
+    pub row_count: Option<i32>,
+    pub error_message: Option<String>,
+}
+
+impl FinishIngestionRun {
+    pub fn validate(&self) -> Result<(), &'static str> {
+        if !matches!(self.status.as_str(), "succeeded" | "failed") {
+            return Err("status must be succeeded or failed");
+        }
+        if matches!(self.row_count, Some(value) if value < 0) {
+            return Err("row_count must be non-negative");
+        }
+        if let Some(value) = self.error_message.as_deref() {
+            if has_sensitive_text(value) {
+                return Err("payload contains sensitive text");
+            }
+        }
+        Ok(())
+    }
+}
+
 #[derive(Debug, Clone, Serialize, FromRow)]
 pub struct WatchlistItem {
     pub watchlist_item_id: i64,
@@ -148,6 +270,21 @@ impl CreateWatchlistItem {
     }
 }
 
+fn is_ingestion_status(status: &str) -> bool {
+    matches!(status, "running" | "succeeded" | "failed")
+}
+
+fn has_sensitive_text(value: &str) -> bool {
+    let lower = value.to_ascii_lowercase();
+    lower.contains("password")
+        || lower.contains("secret")
+        || lower.contains("postgres://")
+        || lower.contains("root_")
+        || lower.contains("database_url")
+        || value.starts_with("/Users/")
+        || value.starts_with("/private/")
+}
+
 #[derive(Debug, Deserialize)]
 pub struct UpdateWatchlistItem {
     pub target_price_psm: Option<f64>,

apps/api/src/routes.rs

@@ -4,9 +4,10 @@ use tower_http::cors::CorsLayer;
 use tower_http::trace::TraceLayer;
 
 use crate::handlers::{
-    archive_watchlist_item, create_watchlist_item, get_neighborhood, health, list_area_scores,
-    list_neighborhood_scores, list_neighborhoods, list_watchlist_items, market_overview, ready,
-    update_watchlist_item,
+    archive_watchlist_item, create_data_source, create_ingestion_run, create_watchlist_item,
+    finish_ingestion_run, get_neighborhood, health, list_area_scores, list_data_sources,
+    list_ingestion_runs, list_neighborhood_scores, list_neighborhoods, list_watchlist_items,
+    market_overview, ready, update_watchlist_item,
 };
 use crate::state::AppState;
 
@@ -18,6 +19,18 @@ pub fn build_router(state: AppState) -> Router {
             "/api/v1",
             Router::new()
                 .route("/areas/scores", get(list_area_scores))
+                .route(
+                    "/data-sources",
+                    get(list_data_sources).post(create_data_source),
+                )
+                .route(
+                    "/ingestion-runs",
+                    get(list_ingestion_runs).post(create_ingestion_run),
+                )
+                .route(
+                    "/ingestion-runs/{run_id}/finish",
+                    patch(finish_ingestion_run),
+                )
                 .route("/market/overview", get(market_overview))
                 .route("/neighborhoods", get(list_neighborhoods))
                 .route("/neighborhoods/scores", get(list_neighborhood_scores))