Skip to main content

max / makenotwork

4.9 KB · 154 lines History Blame Raw
1 //! Import API endpoints: start import, check status, list jobs.
2
3 use axum::{
4 Json,
5 extract::{Path, State},
6 response::IntoResponse,
7 };
8 use base64::Engine;
9 use serde::Deserialize;
10 use serde_json::json;
11
12 use crate::background::BackgroundTx;
13 use sqlx::PgPool;
14
15 use crate::{
16 auth::AuthUser,
17 db::{self, ImportJobId, ProjectId},
18 error::{AppError, Result},
19 import::{self, ColumnMapping, ImportSource},
20 };
21
22 /// Maximum CSV size: 10 MB (base64-encoded).
23 const MAX_CSV_SIZE: usize = 10 * 1024 * 1024;
24
25 /// Request body for starting an import.
26 #[derive(Debug, Deserialize)]
27 pub(super) struct StartImportRequest {
28 pub project_id: ProjectId,
29 pub source: ImportSource,
30 pub csv_data: String,
31 pub column_mapping: ColumnMapping,
32 }
33
34 /// Start a new import job from CSV data.
35 ///
36 /// Creates the job in `pending` status, then spawns a background task
37 /// to parse the CSV and run the import pipeline.
38 #[tracing::instrument(skip_all, name = "imports::start_import")]
39 pub(super) async fn start_import(
40 State(db): State<PgPool>,
41 State(bg): State<BackgroundTx>,
42 AuthUser(user): AuthUser,
43 Json(req): Json<StartImportRequest>,
44 ) -> Result<impl IntoResponse> {
45 user.check_not_suspended()?;
46 user.check_not_sandbox()?;
47
48 // Validate project ownership
49 super::verify_project_ownership(&db, req.project_id, user.id).await?;
50
51 // Only generic_csv is currently supported
52 if req.source != ImportSource::GenericCsv {
53 return Err(AppError::validation(format!(
54 "Import source '{}' is not yet supported. Only 'generic_csv' is available.",
55 req.source,
56 )));
57 }
58
59 // Decode base64 CSV data
60 let csv_bytes = base64::engine::general_purpose::STANDARD
61 .decode(&req.csv_data)
62 .map_err(|_| AppError::validation("Invalid base64-encoded CSV data"))?;
63
64 if csv_bytes.len() > MAX_CSV_SIZE {
65 return Err(AppError::validation(format!(
66 "CSV data exceeds maximum size of {} MB",
67 MAX_CSV_SIZE / (1024 * 1024),
68 )));
69 }
70
71 // Parse CSV into payload
72 let payload = import::csv_converter::parse_csv(&csv_bytes, &req.column_mapping)?;
73 let total_rows = payload.total_rows() as i32;
74
75 let job = db::imports::create_import_job(&db, user.id, req.project_id, req.source, total_rows)
76 .await?;
77
78 let job_id = job.id;
79 let pool = db.clone();
80 let project_id = req.project_id;
81 let user_id = user.id;
82
83 // Run the import on the bounded background pool rather than a raw
84 // `tokio::spawn`: an unbounded spawn let K concurrent imports each drive a
85 // per-item write loop against the shared 25-connection pool, starving request
86 // handlers to their acquire timeout. `state.bg` caps concurrent execution and
87 // queues the rest (Run 21 perf, chronic import N+1/uncapped-spawn).
88 bg.spawn("import-pipeline", async move {
89 if let Err(e) =
90 import::pipeline::run_import(&pool, job_id, project_id, user_id, payload).await
91 {
92 tracing::error!(error = %e, job_id = %job_id, "import pipeline failed");
93 let _ = db::imports::fail_import_job(&pool, job_id, &e.to_string()).await;
94 }
95 });
96
97 Ok(Json(json!({ "job_id": job_id })))
98 }
99
100 /// Get the status and progress of an import job.
101 #[tracing::instrument(skip_all, name = "imports::get_import_status")]
102 pub(super) async fn get_import_status(
103 State(db): State<PgPool>,
104 AuthUser(user): AuthUser,
105 Path(id): Path<String>,
106 ) -> Result<impl IntoResponse> {
107 let job_id: ImportJobId = id.parse().map_err(|_| AppError::NotFound)?;
108
109 let job = db::imports::get_import_job(&db, job_id, user.id)
110 .await?
111 .ok_or(AppError::NotFound)?;
112
113 Ok(Json(json!({
114 "id": job.id,
115 "source": job.source.to_string(),
116 "status": job.status.to_string(),
117 "total_rows": job.total_rows,
118 "processed_rows": job.processed_rows,
119 "created_rows": job.created_rows,
120 "skipped_rows": job.skipped_rows,
121 "error_log": job.error_log,
122 "created_at": job.created_at,
123 "completed_at": job.completed_at,
124 })))
125 }
126
127 /// List all import jobs for the authenticated user.
128 #[tracing::instrument(skip_all, name = "imports::list_imports")]
129 pub(super) async fn list_imports(
130 State(db): State<PgPool>,
131 AuthUser(user): AuthUser,
132 ) -> Result<impl IntoResponse> {
133 let jobs = db::imports::list_import_jobs(&db, user.id).await?;
134
135 let data: Vec<serde_json::Value> = jobs
136 .into_iter()
137 .map(|j| {
138 json!({
139 "id": j.id,
140 "source": j.source.to_string(),
141 "status": j.status.to_string(),
142 "total_rows": j.total_rows,
143 "processed_rows": j.processed_rows,
144 "created_rows": j.created_rows,
145 "skipped_rows": j.skipped_rows,
146 "created_at": j.created_at,
147 "completed_at": j.completed_at,
148 })
149 })
150 .collect();
151
152 Ok(Json(json!({ "data": data })))
153 }
154