1use crate::config::{load_from_pool, resolve, FullConfig};
5use crate::db::pool::Pool;
6use crate::db::{parse_canonical, CanonicalType, Dialect};
7use crate::error::AppError;
8use crate::extractors::tenant::TenantId;
9use crate::handlers::config::{reload_model, replace_config};
10use crate::handlers::entity::{get_or_create_tenant_pool, resolve_tenant_context};
11use crate::migration::{
12 apply_migrations, apply_rls_to_tables, compute_migration_plan, execute_migration_plan,
13 revert_migrations, MigrationPlan,
14};
15use crate::state::AppState;
16use crate::store::{
17 count_package_kind, delete_package_and_config, get_migration_plan, get_package,
18 list_package_ids, list_packages, mark_migration_plan_applied, save_migration_plan,
19 upsert_package,
20};
21use crate::tenant::TenantStrategy;
22use axum::extract::{Multipart, Path, State};
23use axum::Json;
24use serde::Deserialize;
25use serde_json::{json, Value};
26use std::collections::HashSet;
27use std::io::Cursor;
28use uuid::Uuid;
29use zip::ZipArchive;
30
31pub(crate) fn reject_asset_columns_without_storage(
35 config: &FullConfig,
36 storage: &Option<std::sync::Arc<dyn crate::storage::StorageProvider>>,
37) -> Result<(), AppError> {
38 if storage.is_some() {
39 return Ok(());
40 }
41 let asset_cols: Vec<String> = config
42 .columns
43 .iter()
44 .filter(|c| {
45 matches!(
46 parse_canonical(&c.type_),
47 CanonicalType::Asset | CanonicalType::AssetArray
48 )
49 })
50 .map(|c| c.name.clone())
51 .collect();
52 if !asset_cols.is_empty() {
53 return Err(AppError::BadRequest(format!(
54 "Package defines asset column(s) [{}] but no storage provider is configured. \
55 Set STORAGE_PROVIDER (s3 | azure | gcs | rustfs) before installing packages \
56 that use asset or asset[] columns.",
57 asset_cols.join(", ")
58 )));
59 }
60 Ok(())
61}
62
63#[derive(serde::Serialize)]
65struct TenantMigrationOutcome {
66 target: String,
68 strategy: String,
70 status: String,
72 warnings: Vec<String>,
73 #[serde(default, skip_serializing_if = "Vec::is_empty")]
75 skipped: Vec<String>,
76 #[serde(skip_serializing_if = "Option::is_none")]
77 error: Option<String>,
78}
79
80const CONFIG_KINDS: &[&str] = &[
82 "schemas",
83 "enums",
84 "tables",
85 "columns",
86 "indexes",
87 "relationships",
88 "api_entities",
89 "kv_stores",
90 "reports",
91];
92
93fn dependencies(kind: &str) -> &'static [&'static str] {
96 match kind {
97 "schemas" => &[],
98 "enums" => &["schemas"],
99 "tables" => &["schemas"],
100 "columns" => &["tables"],
101 "indexes" => &["schemas", "tables"],
102 "relationships" => &["schemas", "tables", "columns"],
103 "api_entities" => &["tables"],
104 "kv_stores" => &[],
105 "reports" => &["tables", "columns"],
108 _ => &[],
109 }
110}
111
112fn config_apply_order() -> Vec<&'static str> {
115 let mut order = Vec::with_capacity(CONFIG_KINDS.len());
116 let mut done: HashSet<&'static str> = HashSet::new();
117 while order.len() < CONFIG_KINDS.len() {
118 let mut made_progress = false;
119 for &kind in CONFIG_KINDS {
120 if done.contains(kind) {
121 continue;
122 }
123 let deps = dependencies(kind);
124 if deps.iter().all(|d| done.contains(d)) {
125 order.push(kind);
126 done.insert(kind);
127 made_progress = true;
128 }
129 }
130 if !made_progress {
131 break;
132 }
133 }
134 order
135}
136
137const DEFAULT_SCHEMA_ID: &str = "default";
139
140fn assemble_config(bodies: &[(&'static str, Vec<Value>)]) -> Result<FullConfig, AppError> {
144 fn de<T: serde::de::DeserializeOwned>(
145 bodies: &[(&'static str, Vec<Value>)],
146 kind: &str,
147 ) -> Result<Vec<T>, AppError> {
148 let arr = bodies
149 .iter()
150 .find(|(k, _)| *k == kind)
151 .map(|(_, v)| v.clone())
152 .unwrap_or_default();
153 serde_json::from_value(Value::Array(arr))
154 .map_err(|e| AppError::BadRequest(format!("invalid {}: {}", kind, e)))
155 }
156 Ok(FullConfig {
157 schemas: de(bodies, "schemas")?,
158 enums: de(bodies, "enums")?,
159 tables: de(bodies, "tables")?,
160 columns: de(bodies, "columns")?,
161 indexes: de(bodies, "indexes")?,
162 relationships: de(bodies, "relationships")?,
163 api_entities: de(bodies, "api_entities")?,
164 kv_stores: de(bodies, "kv_stores")?,
165 reports: de(bodies, "reports")?,
166 })
167}
168
169fn inject_schema_id(body: &mut [Value], schema_id: &str) {
170 for rec in body.iter_mut() {
171 if let Some(obj) = rec.as_object_mut() {
172 if !obj.contains_key("schema_id") {
173 obj.insert("schema_id".into(), Value::String(schema_id.to_string()));
174 }
175 }
176 }
177}
178
179fn inject_relationship_schema_ids(body: &mut [Value], schema_id: &str) {
180 for rec in body.iter_mut() {
181 if let Some(obj) = rec.as_object_mut() {
182 if !obj.contains_key("from_schema_id") {
184 obj.insert(
185 "from_schema_id".into(),
186 Value::String(schema_id.to_string()),
187 );
188 }
189 let is_cross_package = obj
192 .get("to_package_id")
193 .and_then(Value::as_str)
194 .map(|s| !s.is_empty())
195 .unwrap_or(false);
196 if !is_cross_package && !obj.contains_key("to_schema_id") {
197 obj.insert("to_schema_id".into(), Value::String(schema_id.to_string()));
198 }
199 }
200 }
201}
202
203fn read_zip_entry_to_string<R: std::io::Read + std::io::Seek>(
204 archive: &mut ZipArchive<R>,
205 name: &str,
206) -> Result<String, AppError> {
207 let mut f = archive
208 .by_name(name)
209 .map_err(|e| AppError::BadRequest(e.to_string()))?;
210 let mut s = String::new();
211 std::io::Read::read_to_string(&mut f, &mut s)
212 .map_err(|e| AppError::BadRequest(e.to_string()))?;
213 Ok(s)
214}
215
216fn read_kind_from_zip<R: std::io::Read + std::io::Seek>(
220 archive: &mut ZipArchive<R>,
221 kind: &str,
222) -> Result<Vec<Value>, AppError> {
223 let flat = format!("{}.json", kind);
224 if let Ok(content) = read_zip_entry_to_string(archive, &flat) {
225 return serde_json::from_str(&content)
226 .map_err(|e| AppError::BadRequest(format!("invalid {}: {}", flat, e)));
227 }
228
229 let prefix = format!("{}/", kind);
230 let mut names: Vec<String> = archive
231 .file_names()
232 .filter(|n| n.starts_with(&prefix) && n.ends_with(".json"))
233 .map(String::from)
234 .collect();
235 names.sort();
236
237 let mut merged: Vec<Value> = Vec::new();
238 for name in names {
239 let content = read_zip_entry_to_string(archive, &name)?;
240 let mut items: Vec<Value> = serde_json::from_str(&content)
241 .map_err(|e| AppError::BadRequest(format!("invalid {}: {}", name, e)))?;
242 merged.append(&mut items);
243 }
244 Ok(merged)
245}
246
247#[allow(clippy::too_many_arguments)]
252async fn apply_ddl_to_pool(
253 migration_pool: &Pool,
254 config_pool: &Pool,
255 config: &FullConfig,
256 plan: Option<&MigrationPlan>,
257 package_id: &str,
258 target: &str,
259 strategy: &str,
260 from_version: Option<&str>,
261 to_version: &str,
262 rls_tenant_column: Option<&str>,
263 dialect: &dyn Dialect,
264 cross_package_configs: &std::collections::HashMap<String, FullConfig>,
265) -> TenantMigrationOutcome {
266 match plan {
267 Some(p) => {
269 let migration_id = Uuid::new_v4().to_string();
270 match execute_migration_plan(
271 migration_pool,
272 config_pool,
273 p,
274 &migration_id,
275 package_id,
276 target,
277 from_version,
278 to_version,
279 dialect,
280 )
281 .await
282 {
283 Ok(result) => {
284 if let Some(col) = rls_tenant_column {
288 if let Err(e) =
289 apply_rls_to_tables(migration_pool, config, None, col, dialect).await
290 {
291 tracing::warn!(target, strategy, error = %e, "RLS reconciliation failed after upgrade");
292 return TenantMigrationOutcome {
293 target: target.to_string(),
294 strategy: strategy.to_string(),
295 status: "failed".to_string(),
296 warnings: result.warnings,
297 skipped: result.skips,
298 error: Some(format!("RLS reconciliation failed: {}", e)),
299 };
300 }
301 }
302 TenantMigrationOutcome {
303 target: target.to_string(),
304 strategy: strategy.to_string(),
305 status: if result.warned > 0 {
306 "applied_with_warnings".to_string()
307 } else {
308 "applied".to_string()
309 },
310 warnings: result.warnings,
311 skipped: result.skips,
312 error: None,
313 }
314 }
315 Err(e) => {
316 tracing::warn!(target, strategy, error = %e, "DDL broadcast failed (upgrade)");
317 TenantMigrationOutcome {
318 target: target.to_string(),
319 strategy: strategy.to_string(),
320 status: "failed".to_string(),
321 warnings: vec![],
322 skipped: vec![],
323 error: Some(e.to_string()),
324 }
325 }
326 }
327 }
328 None => {
330 match apply_migrations(
331 migration_pool,
332 config,
333 None,
334 rls_tenant_column,
335 dialect,
336 cross_package_configs,
337 )
338 .await
339 {
340 Ok(()) => TenantMigrationOutcome {
341 target: target.to_string(),
342 strategy: strategy.to_string(),
343 status: "applied".to_string(),
344 warnings: vec![],
345 skipped: vec![],
346 error: None,
347 },
348 Err(e) => {
349 tracing::warn!(target, strategy, error = %e, "DDL broadcast failed (fresh install)");
350 TenantMigrationOutcome {
351 target: target.to_string(),
352 strategy: strategy.to_string(),
353 status: "failed".to_string(),
354 warnings: vec![],
355 skipped: vec![],
356 error: Some(e.to_string()),
357 }
358 }
359 }
360 }
361 }
362}
363
364async fn broadcast_ddl(
374 state: &AppState,
375 config_pool: &Pool,
376 config: &FullConfig,
377 old_config: Option<&FullConfig>,
378 package_id: &str,
379 from_version: Option<&str>,
380 to_version: &str,
381) -> Vec<TenantMigrationOutcome> {
382 let mut outcomes = Vec::new();
383
384 let cross_package_configs: std::collections::HashMap<String, FullConfig> = {
386 match list_package_ids(config_pool).await {
387 Ok(ids) => {
388 let mut map = std::collections::HashMap::new();
389 for pid in ids {
390 if pid == package_id {
391 continue;
392 }
393 match load_from_pool(config_pool, &pid).await {
394 Ok(cfg) => {
395 map.insert(pid, cfg);
396 }
397 Err(e) => {
398 tracing::warn!(pkg = %pid, error = %e, "could not load cross-package config for FK resolution")
399 }
400 }
401 }
402 map
403 }
404 Err(e) => {
405 tracing::warn!(error = %e, "could not list packages for cross-package FK resolution");
406 std::collections::HashMap::new()
407 }
408 }
409 };
410
411 let plan: Option<MigrationPlan> = match old_config {
417 Some(old) => {
418 match compute_migration_plan(
419 old,
420 config,
421 None,
422 None,
423 state.dialect.as_ref(),
424 &cross_package_configs,
425 ) {
426 Ok(p) => Some(p),
427 Err(e) => {
428 tracing::error!(error = %e, "could not compute migration plan for broadcast");
429 return vec![TenantMigrationOutcome {
430 target: "all".to_string(),
431 strategy: "n/a".to_string(),
432 status: "failed".to_string(),
433 warnings: vec![],
434 skipped: vec![],
435 error: Some(format!("migration plan error: {}", e)),
436 }];
437 }
438 }
439 }
440 None => None,
441 };
442
443 if state.tenant_registry.has_shared_rls_tenants() {
445 let outcome = apply_ddl_to_pool(
446 &state.pool,
447 config_pool,
448 config,
449 plan.as_ref(),
450 package_id,
451 "central_rls_db",
452 "rls",
453 from_version,
454 to_version,
455 Some(crate::migration::RLS_TENANT_COLUMN),
456 state.dialect.as_ref(),
457 &cross_package_configs,
458 )
459 .await;
460 outcomes.push(outcome);
461 }
462
463 let mut seen_rls_urls: HashSet<String> = HashSet::new();
466 for (tid, db_url) in state.tenant_registry.rls_dedicated_db_targets() {
467 if !seen_rls_urls.insert(db_url.clone()) {
468 continue; }
470 let pool = match get_or_create_tenant_pool(state, &tid, &db_url).await {
471 Ok(p) => p,
472 Err(e) => {
473 tracing::warn!(target = %tid, error = %e, "could not connect to dedicated RLS tenant DB");
474 outcomes.push(TenantMigrationOutcome {
475 target: tid.clone(),
476 strategy: "rls".to_string(),
477 status: "failed".to_string(),
478 warnings: vec![],
479 skipped: vec![],
480 error: Some(format!("connection failed: {}", e)),
481 });
482 continue;
483 }
484 };
485 let outcome = apply_ddl_to_pool(
486 &pool,
487 config_pool,
488 config,
489 plan.as_ref(),
490 package_id,
491 &tid,
492 "rls",
493 from_version,
494 to_version,
495 Some(crate::migration::RLS_TENANT_COLUMN),
496 state.dialect.as_ref(),
497 &cross_package_configs,
498 )
499 .await;
500 outcomes.push(outcome);
501 }
502
503 for (tid, db_url) in state.tenant_registry.database_tenant_targets() {
505 let pool = match get_or_create_tenant_pool(state, &tid, &db_url).await {
506 Ok(p) => p,
507 Err(e) => {
508 tracing::warn!(target = %tid, error = %e, "could not connect to Database tenant DB");
509 outcomes.push(TenantMigrationOutcome {
510 target: tid.clone(),
511 strategy: "database".to_string(),
512 status: "failed".to_string(),
513 warnings: vec![],
514 skipped: vec![],
515 error: Some(format!("connection failed: {}", e)),
516 });
517 continue;
518 }
519 };
520 let outcome = apply_ddl_to_pool(
521 &pool,
522 config_pool,
523 config,
524 plan.as_ref(),
525 package_id,
526 &tid,
527 "database",
528 from_version,
529 to_version,
530 None,
531 state.dialect.as_ref(),
532 &cross_package_configs,
533 )
534 .await;
535 outcomes.push(outcome);
536 }
537
538 outcomes
539}
540
541pub async fn install_package(
545 TenantId(tenant_id_opt): TenantId,
546 State(state): State<AppState>,
547 mut multipart: Multipart,
548) -> Result<impl axum::response::IntoResponse, AppError> {
549 let tenant_id = tenant_id_opt
550 .as_deref()
551 .filter(|s| !s.is_empty())
552 .ok_or_else(|| AppError::BadRequest("X-Tenant-ID header is required".into()))?;
553 state
554 .tenant_registry
555 .get(tenant_id)
556 .ok_or_else(|| AppError::NotFound(format!("tenant not found: {}", tenant_id)))?;
557
558 let mut zip_bytes: Option<Vec<u8>> = None;
559 while let Ok(Some(field)) = multipart.next_field().await {
560 let name = field.name().unwrap_or("").to_string();
561 if name == "file" || name == "package" {
562 let data = field
563 .bytes()
564 .await
565 .map_err(|e| AppError::BadRequest(e.to_string()))?;
566 zip_bytes = Some(data.to_vec());
567 break;
568 }
569 }
570 let zip_bytes = zip_bytes.ok_or_else(|| {
571 AppError::BadRequest("missing 'file' or 'package' field in multipart body".into())
572 })?;
573
574 let mut archive = ZipArchive::new(Cursor::new(zip_bytes))
575 .map_err(|e| AppError::BadRequest(format!("invalid zip: {}", e)))?;
576
577 let manifest_name = archive
578 .file_names()
579 .find(|n| *n == "manifest.json" || n.ends_with("/manifest.json"))
580 .map(String::from)
581 .ok_or_else(|| AppError::BadRequest("zip must contain manifest.json at root".into()))?;
582
583 let manifest_value: Value = {
584 let mut file = archive
585 .by_name(&manifest_name)
586 .map_err(|e| AppError::BadRequest(e.to_string()))?;
587 let mut buf = String::new();
588 std::io::Read::read_to_string(&mut file, &mut buf)
589 .map_err(|e| AppError::BadRequest(e.to_string()))?;
590 serde_json::from_str(&buf)
591 .map_err(|e| AppError::BadRequest(format!("invalid manifest.json: {}", e)))?
592 };
593
594 let manifest_obj = manifest_value
595 .as_object()
596 .ok_or_else(|| AppError::BadRequest("manifest.json must be an object".into()))?;
597 let id = manifest_obj
598 .get("id")
599 .and_then(Value::as_str)
600 .ok_or_else(|| AppError::BadRequest("manifest must have 'id' (string)".into()))?;
601 let _name = manifest_obj
602 .get("name")
603 .and_then(Value::as_str)
604 .ok_or_else(|| AppError::BadRequest("manifest must have 'name' (string)".into()))?;
605 let _version = manifest_obj
606 .get("version")
607 .and_then(Value::as_str)
608 .ok_or_else(|| AppError::BadRequest("manifest must have 'version' (string)".into()))?;
609 let schema_name = manifest_obj
610 .get("schema")
611 .and_then(Value::as_str)
612 .ok_or_else(|| {
613 AppError::BadRequest(
614 "manifest must have 'schema' (string) - the schema name for all configs".into(),
615 )
616 })?;
617
618 let ctx = resolve_tenant_context(&state, Some(tenant_id), None, Some(id)).await?;
619 let config_pool = ctx.config_pool();
620 let package_cache_key = ctx.package_cache_key().to_string();
622
623 if let Some(deps) = manifest_obj.get("dependencies").and_then(Value::as_array) {
625 let installed_ids: std::collections::HashSet<String> =
626 list_package_ids(config_pool).await?.into_iter().collect();
627 let missing: Vec<&str> = deps
628 .iter()
629 .filter_map(Value::as_str)
630 .filter(|dep| !installed_ids.contains(*dep))
631 .collect();
632 if !missing.is_empty() {
633 return Err(AppError::BadRequest(format!(
634 "package '{}' depends on [{}] which are not installed; install them first",
635 id,
636 missing.join(", ")
637 )));
638 }
639 }
640
641 let incoming_version = manifest_obj
642 .get("version")
643 .and_then(Value::as_str)
644 .unwrap_or("");
645
646 let is_upgrade = if let Some(existing) = get_package(config_pool, id).await? {
648 if existing.semantic_version.as_deref() == Some(incoming_version) {
649 return Err(AppError::Conflict(format!(
650 "package '{}' version '{}' is already installed",
651 id, incoming_version
652 )));
653 }
654 true
655 } else {
656 false
657 };
658
659 let old_config = if is_upgrade {
660 Some(
661 load_from_pool(config_pool, id)
662 .await
663 .map_err(AppError::Config)?,
664 )
665 } else {
666 None
667 };
668
669 let schemas_body: Vec<Value> = vec![serde_json::json!({
670 "id": DEFAULT_SCHEMA_ID,
671 "name": schema_name
672 })];
673
674 let apply_order = config_apply_order();
678 let mut bodies: Vec<(&'static str, Vec<Value>)> = Vec::with_capacity(apply_order.len());
679 for kind in &apply_order {
680 let body: Vec<Value> = if *kind == "schemas" {
681 schemas_body.clone()
682 } else {
683 let mut body = read_kind_from_zip(&mut archive, kind)?;
684 match *kind {
685 "enums" | "tables" | "indexes" => inject_schema_id(&mut body, DEFAULT_SCHEMA_ID),
686 "relationships" => inject_relationship_schema_ids(&mut body, DEFAULT_SCHEMA_ID),
687 _ => {}
688 }
689 body
690 };
691 bodies.push((*kind, body));
692 }
693
694 let config = assemble_config(&bodies)?;
697 let new_model = resolve(&config)
698 .map_err(AppError::Config)?
699 .with_package_id(id);
700
701 reject_asset_columns_without_storage(&config, &state.storage)?;
703
704 let tenant_outcomes = broadcast_ddl(
707 &state,
708 config_pool,
709 &config,
710 old_config.as_ref(),
711 id,
712 old_config
713 .as_ref()
714 .and_then(|_| manifest_value.get("version").and_then(Value::as_str)),
715 incoming_version,
716 )
717 .await;
718
719 if !is_upgrade {
723 let failures: Vec<String> = tenant_outcomes
724 .iter()
725 .filter(|o| o.status == "failed")
726 .map(|o| format!("{}: {}", o.target, o.error.clone().unwrap_or_default()))
727 .collect();
728 if !failures.is_empty() {
729 return Err(AppError::BadRequest(format!(
730 "package '{}' installation failed during schema creation; no configuration was \
731 saved. Errors: {}",
732 id,
733 failures.join("; ")
734 )));
735 }
736 }
737
738 let mut applied = Vec::with_capacity(bodies.len());
740 for (kind, body) in bodies {
741 replace_config(config_pool, kind, body, false, id, None).await?;
742 applied.push(kind.to_string());
743 }
744 upsert_package(config_pool, id, &manifest_value).await?;
745
746 let migration_warnings: Vec<String> = tenant_outcomes
747 .iter()
748 .flat_map(|o| o.warnings.iter().cloned())
749 .collect();
750
751 {
753 let mut model_guard = state
754 .model
755 .write()
756 .map_err(|_| AppError::BadRequest("state lock".into()))?;
757 *model_guard = new_model.clone();
758 let mut pkg_guard = state
759 .package_models
760 .write()
761 .map_err(|_| AppError::BadRequest("state lock".into()))?;
762 pkg_guard.insert(id.to_string(), new_model.clone());
764 for (tid, _) in state.tenant_registry.database_tenant_targets() {
766 pkg_guard.insert(format!("{}:{}", id, tid), new_model.clone());
767 }
768 pkg_guard.insert(package_cache_key, new_model);
770 }
771 crate::handlers::entity::invalidate_cross_package_index(&state);
773
774 #[derive(serde::Serialize)]
775 struct PackageInstallResponse {
776 package: Value,
777 applied: Vec<String>,
778 warnings: Vec<String>,
779 tenant_migrations: Vec<TenantMigrationOutcome>,
781 }
782 Ok((
783 axum::http::StatusCode::OK,
784 Json(crate::response::SuccessOne {
785 data: PackageInstallResponse {
786 package: manifest_value,
787 applied,
788 warnings: migration_warnings,
789 tenant_migrations: tenant_outcomes,
790 },
791 meta: None,
792 }),
793 ))
794}
795
796#[derive(Deserialize)]
797pub struct UninstallPath {
798 pub package_id: String,
799}
800
801pub async fn uninstall_package(
803 TenantId(tenant_id_opt): TenantId,
804 State(state): State<AppState>,
805 Path(UninstallPath { package_id }): Path<UninstallPath>,
806) -> Result<impl axum::response::IntoResponse, AppError> {
807 let tenant_id = tenant_id_opt
808 .as_deref()
809 .filter(|s| !s.is_empty())
810 .ok_or_else(|| AppError::BadRequest("X-Tenant-ID header is required".into()))?;
811
812 let ctx = resolve_tenant_context(&state, Some(tenant_id), None, Some(&package_id)).await?;
813 let config_pool = ctx.config_pool();
814 let migration_pool = ctx.migration_pool();
815 let schema_override = ctx.schema_override();
816 let package_cache_key = ctx.package_cache_key().to_string();
817
818 let installed = list_package_ids(config_pool).await?;
819 if !installed.contains(&package_id) {
820 return Err(AppError::NotFound(format!(
821 "package not found: {}",
822 package_id
823 )));
824 }
825
826 let all_packages = list_packages(config_pool).await?;
828 let dependents: Vec<String> = all_packages
829 .iter()
830 .filter(|row| row.id != package_id)
831 .filter(|row| {
832 row.payload
833 .get("dependencies")
834 .and_then(Value::as_array)
835 .map(|deps| deps.iter().any(|d| d.as_str() == Some(package_id.as_str())))
836 .unwrap_or(false)
837 })
838 .map(|row| row.id.clone())
839 .collect();
840 if !dependents.is_empty() {
841 return Err(AppError::Conflict(format!(
842 "cannot uninstall '{}': packages [{}] depend on it; uninstall them first",
843 package_id,
844 dependents.join(", ")
845 )));
846 }
847
848 let config = load_from_pool(config_pool, &package_id)
849 .await
850 .map_err(AppError::Config)?;
851 revert_migrations(migration_pool, &config, schema_override).await?;
852 delete_package_and_config(config_pool, &package_id).await?;
853
854 {
855 state
856 .package_models
857 .write()
858 .map_err(|_| AppError::BadRequest("state lock".into()))?
859 .remove(&package_cache_key);
860 }
861 crate::handlers::entity::invalidate_cross_package_index(&state);
863
864 if std::ptr::eq(&state.pool as *const _, config_pool as *const _) {
866 let _ = reload_model(&state).await;
867 }
868
869 #[derive(serde::Serialize)]
870 struct UninstallResponse {
871 package_id: String,
872 }
873 Ok((
874 axum::http::StatusCode::OK,
875 Json(crate::response::SuccessOne {
876 data: UninstallResponse { package_id },
877 meta: None,
878 }),
879 ))
880}
881
882async fn package_detail_data(
884 pool: &Pool,
885 package_id: &str,
886) -> Result<Value, crate::error::AppError> {
887 use crate::handlers::config::get_config;
888
889 let (schemas, enums, tables, columns, indexes, relationships, api_entities, kv_stores) = tokio::try_join!(
890 get_config(pool, "schemas", package_id),
891 get_config(pool, "enums", package_id),
892 get_config(pool, "tables", package_id),
893 get_config(pool, "columns", package_id),
894 get_config(pool, "indexes", package_id),
895 get_config(pool, "relationships", package_id),
896 get_config(pool, "api_entities", package_id),
897 get_config(pool, "kv_stores", package_id),
898 )?;
899
900 Ok(json!({
901 "stats": {
902 "schemas": schemas.len(),
903 "enums": enums.len(),
904 "tables": tables.len(),
905 "columns": columns.len(),
906 "indexes": indexes.len(),
907 "relationships": relationships.len(),
908 "apiEntities": api_entities.len(),
909 "kvStores": kv_stores.len(),
910 },
911 "schemas": schemas,
912 "enums": enums,
913 "tables": tables,
914 "columns": columns,
915 "indexes": indexes,
916 "relationships": relationships,
917 "apiEntities": api_entities,
918 "kvStores": kv_stores,
919 }))
920}
921
922pub async fn list_packages_handler(
924 State(state): State<AppState>,
925) -> Result<impl axum::response::IntoResponse, crate::error::AppError> {
926 let packages = list_packages(&state.pool).await?;
927
928 let mut items: Vec<Value> = Vec::with_capacity(packages.len());
929 for pkg in packages {
930 let (schemas, enums, tables, columns, indexes, relationships, api_entities, kv_stores) = tokio::try_join!(
931 count_package_kind(&state.pool, "schemas", &pkg.id),
932 count_package_kind(&state.pool, "enums", &pkg.id),
933 count_package_kind(&state.pool, "tables", &pkg.id),
934 count_package_kind(&state.pool, "columns", &pkg.id),
935 count_package_kind(&state.pool, "indexes", &pkg.id),
936 count_package_kind(&state.pool, "relationships", &pkg.id),
937 count_package_kind(&state.pool, "api_entities", &pkg.id),
938 count_package_kind(&state.pool, "kv_stores", &pkg.id),
939 )?;
940
941 let name = pkg
942 .payload
943 .get("name")
944 .and_then(Value::as_str)
945 .map(String::from);
946 let version = pkg
947 .payload
948 .get("version")
949 .and_then(Value::as_str)
950 .map(String::from);
951 let schema = pkg
952 .payload
953 .get("schema")
954 .and_then(Value::as_str)
955 .map(String::from);
956 let dependencies: Vec<&str> = pkg
957 .payload
958 .get("dependencies")
959 .and_then(Value::as_array)
960 .map(|arr| arr.iter().filter_map(Value::as_str).collect())
961 .unwrap_or_default();
962
963 items.push(json!({
964 "id": pkg.id,
965 "name": name,
966 "version": version,
967 "schema": schema,
968 "installedVersion": pkg.version,
969 "updatedAt": pkg.updated_at,
970 "dependencies": dependencies,
971 "stats": {
972 "schemas": schemas,
973 "enums": enums,
974 "tables": tables,
975 "columns": columns,
976 "indexes": indexes,
977 "relationships": relationships,
978 "apiEntities": api_entities,
979 "kvStores": kv_stores,
980 },
981 }));
982 }
983
984 let count = items.len() as u64;
985 Ok((
986 axum::http::StatusCode::OK,
987 Json(crate::response::SuccessMany {
988 data: items,
989 meta: crate::response::MetaCount { count },
990 }),
991 ))
992}
993
994#[derive(Deserialize)]
995pub struct PackageIdPath {
996 pub package_id: String,
997}
998
999pub async fn get_package_handler(
1001 State(state): State<AppState>,
1002 Path(PackageIdPath { package_id }): Path<PackageIdPath>,
1003) -> Result<impl axum::response::IntoResponse, crate::error::AppError> {
1004 let pkg = get_package(&state.pool, &package_id)
1005 .await?
1006 .ok_or_else(|| {
1007 crate::error::AppError::NotFound(format!("package not found: {}", package_id))
1008 })?;
1009
1010 let name = pkg
1011 .payload
1012 .get("name")
1013 .and_then(Value::as_str)
1014 .map(String::from);
1015 let version = pkg
1016 .payload
1017 .get("version")
1018 .and_then(Value::as_str)
1019 .map(String::from);
1020 let schema = pkg
1021 .payload
1022 .get("schema")
1023 .and_then(Value::as_str)
1024 .map(String::from);
1025
1026 let mut detail = package_detail_data(&state.pool, &package_id).await?;
1027 let obj = detail.as_object_mut().unwrap();
1028 obj.insert("id".into(), json!(pkg.id));
1029 obj.insert("name".into(), json!(name));
1030 obj.insert("version".into(), json!(version));
1031 obj.insert("schema".into(), json!(schema));
1032 obj.insert("installedVersion".into(), json!(pkg.version));
1033 obj.insert("updatedAt".into(), json!(pkg.updated_at));
1034 obj.insert("manifest".into(), pkg.payload);
1035
1036 Ok((
1037 axum::http::StatusCode::OK,
1038 Json(crate::response::SuccessOne {
1039 data: detail,
1040 meta: None,
1041 }),
1042 ))
1043}
1044
1045pub async fn preview_migration_handler(
1052 TenantId(tenant_id_opt): TenantId,
1053 State(state): State<AppState>,
1054 mut multipart: Multipart,
1055) -> Result<impl axum::response::IntoResponse, AppError> {
1056 let tenant_id = tenant_id_opt
1057 .as_deref()
1058 .filter(|s| !s.is_empty())
1059 .ok_or_else(|| AppError::BadRequest("X-Tenant-ID header is required".into()))?;
1060
1061 let mut zip_bytes_raw: Option<Vec<u8>> = None;
1062 while let Ok(Some(field)) = multipart.next_field().await {
1063 let name = field.name().unwrap_or("").to_string();
1064 if name == "file" || name == "package" {
1065 let data = field
1066 .bytes()
1067 .await
1068 .map_err(|e| AppError::BadRequest(e.to_string()))?;
1069 zip_bytes_raw = Some(data.to_vec());
1070 break;
1071 }
1072 }
1073 let zip_bytes = zip_bytes_raw
1074 .ok_or_else(|| AppError::BadRequest("missing 'file' or 'package' field".into()))?;
1075
1076 let mut archive = ZipArchive::new(Cursor::new(zip_bytes.clone()))
1077 .map_err(|e| AppError::BadRequest(format!("invalid zip: {}", e)))?;
1078
1079 let manifest_name = archive
1080 .file_names()
1081 .find(|n| *n == "manifest.json" || n.ends_with("/manifest.json"))
1082 .map(String::from)
1083 .ok_or_else(|| AppError::BadRequest("zip must contain manifest.json".into()))?;
1084
1085 let manifest_value: Value = {
1086 let mut file = archive
1087 .by_name(&manifest_name)
1088 .map_err(|e| AppError::BadRequest(e.to_string()))?;
1089 let mut buf = String::new();
1090 std::io::Read::read_to_string(&mut file, &mut buf)
1091 .map_err(|e| AppError::BadRequest(e.to_string()))?;
1092 serde_json::from_str(&buf)
1093 .map_err(|e| AppError::BadRequest(format!("invalid manifest.json: {}", e)))?
1094 };
1095 let manifest_obj = manifest_value
1096 .as_object()
1097 .ok_or_else(|| AppError::BadRequest("manifest.json must be an object".into()))?;
1098
1099 let id = manifest_obj
1100 .get("id")
1101 .and_then(Value::as_str)
1102 .ok_or_else(|| AppError::BadRequest("manifest must have 'id'".into()))?;
1103 let incoming_version = manifest_obj
1104 .get("version")
1105 .and_then(Value::as_str)
1106 .unwrap_or("");
1107 let schema_name = manifest_obj
1108 .get("schema")
1109 .and_then(Value::as_str)
1110 .ok_or_else(|| AppError::BadRequest("manifest must have 'schema'".into()))?;
1111
1112 let existing = get_package(&state.pool, id).await?.ok_or_else(|| {
1113 AppError::NotFound(format!(
1114 "package '{}' is not installed — preview is only for upgrades",
1115 id
1116 ))
1117 })?;
1118
1119 if existing.semantic_version.as_deref() == Some(incoming_version) {
1120 return Err(AppError::Conflict(format!(
1121 "package '{}' version '{}' is already installed",
1122 id, incoming_version
1123 )));
1124 }
1125
1126 let from_version = existing.semantic_version.clone();
1127 let ctx = resolve_tenant_context(&state, Some(tenant_id), None, Some(id)).await?;
1128 let config_pool = ctx.config_pool();
1129
1130 let old_config = load_from_pool(config_pool, id)
1131 .await
1132 .map_err(AppError::Config)?;
1133
1134 let schemas_body = vec![serde_json::json!({ "id": DEFAULT_SCHEMA_ID, "name": schema_name })];
1136 let config_kinds = [
1137 "schemas",
1138 "enums",
1139 "tables",
1140 "columns",
1141 "indexes",
1142 "relationships",
1143 "api_entities",
1144 "kv_stores",
1145 ];
1146 let mut all_values: std::collections::HashMap<String, Vec<Value>> =
1147 std::collections::HashMap::new();
1148 for kind in &config_kinds {
1149 let body: Vec<Value> = if *kind == "schemas" {
1150 serde_json::from_value(Value::Array(schemas_body.clone())).unwrap_or_default()
1151 } else {
1152 let mut body = read_kind_from_zip(&mut archive, kind).unwrap_or_default();
1153 match *kind {
1154 "enums" | "tables" | "indexes" => inject_schema_id(&mut body, DEFAULT_SCHEMA_ID),
1155 "relationships" => inject_relationship_schema_ids(&mut body, DEFAULT_SCHEMA_ID),
1156 _ => {}
1157 }
1158 body
1159 };
1160 all_values.insert(kind.to_string(), body);
1161 }
1162
1163 let new_config = build_full_config_from_values(&all_values)?;
1165
1166 let plan = compute_migration_plan(
1167 &old_config,
1168 &new_config,
1169 ctx.schema_override(),
1170 ctx.rls_tenant_column(),
1171 state.dialect.as_ref(),
1172 &std::collections::HashMap::new(),
1173 )
1174 .map_err(|e| AppError::BadRequest(format!("migration plan error: {}", e)))?;
1175
1176 let summary = plan.summary();
1177 let plan_json = serde_json::to_value(&plan).map_err(|e| AppError::BadRequest(e.to_string()))?;
1178 let migration_id = Uuid::new_v4().to_string();
1179
1180 save_migration_plan(
1181 config_pool,
1182 &migration_id,
1183 id,
1184 tenant_id,
1185 from_version.as_deref(),
1186 incoming_version,
1187 &plan_json,
1188 &zip_bytes,
1189 )
1190 .await?;
1191
1192 Ok((
1193 axum::http::StatusCode::OK,
1194 Json(crate::response::SuccessOne {
1195 data: json!({
1196 "migration_id": migration_id,
1197 "package_id": id,
1198 "from_version": from_version,
1199 "to_version": incoming_version,
1200 "expires_in_hours": 24,
1201 "summary": {
1202 "total": summary.total,
1203 "safe": summary.safe,
1204 "best_effort": summary.best_effort,
1205 "warn_only": summary.warn_only,
1206 },
1207 "steps": plan.steps,
1208 }),
1209 meta: None,
1210 }),
1211 ))
1212}
1213
1214#[derive(Deserialize)]
1215pub struct MigrationIdPath {
1216 pub migration_id: String,
1217}
1218
1219pub async fn apply_migration_handler(
1224 TenantId(tenant_id_opt): TenantId,
1225 State(state): State<AppState>,
1226 Path(MigrationIdPath { migration_id }): Path<MigrationIdPath>,
1227) -> Result<impl axum::response::IntoResponse, AppError> {
1228 let tenant_id = tenant_id_opt
1229 .as_deref()
1230 .filter(|s| !s.is_empty())
1231 .ok_or_else(|| AppError::BadRequest("X-Tenant-ID header is required".into()))?;
1232
1233 let row = get_migration_plan(&state.pool, &migration_id)
1234 .await?
1235 .ok_or_else(|| {
1236 AppError::NotFound(format!("migration plan '{}' not found", migration_id))
1237 })?;
1238
1239 if row.status == "applied" {
1240 return Err(AppError::Conflict(format!(
1241 "migration plan '{}' has already been applied",
1242 migration_id
1243 )));
1244 }
1245 if row.status != "pending" {
1246 return Err(AppError::BadRequest(format!(
1247 "migration plan '{}' has status '{}' and cannot be applied",
1248 migration_id, row.status
1249 )));
1250 }
1251
1252 let now = chrono::Utc::now();
1253 if now > row.expires_at {
1254 return Err(AppError::BadRequest(format!(
1255 "migration plan '{}' expired at {} — re-run preview to generate a new plan",
1256 migration_id, row.expires_at
1257 )));
1258 }
1259
1260 if row.tenant_id != tenant_id {
1261 return Err(AppError::BadRequest(format!(
1262 "migration plan '{}' was created for tenant '{}', not '{}'",
1263 migration_id, row.tenant_id, tenant_id
1264 )));
1265 }
1266
1267 let plan: MigrationPlan = serde_json::from_value(row.plan_json.clone())
1268 .map_err(|e| AppError::BadRequest(format!("corrupted migration plan: {}", e)))?;
1269
1270 let ctx = resolve_tenant_context(&state, Some(tenant_id), None, Some(&row.package_id)).await?;
1271 let config_pool = ctx.config_pool();
1272 let migration_pool = ctx.migration_pool();
1273 let package_cache_key = ctx.package_cache_key().to_string();
1274
1275 let mut archive = ZipArchive::new(Cursor::new(row.zip_bytes.clone()))
1277 .map_err(|e| AppError::BadRequest(format!("stored zip corrupted: {}", e)))?;
1278
1279 let manifest_name = archive
1280 .file_names()
1281 .find(|n| *n == "manifest.json" || n.ends_with("/manifest.json"))
1282 .map(String::from)
1283 .ok_or_else(|| AppError::BadRequest("stored zip missing manifest.json".into()))?;
1284
1285 let manifest_value: Value = {
1286 let mut file = archive
1287 .by_name(&manifest_name)
1288 .map_err(|e| AppError::BadRequest(e.to_string()))?;
1289 let mut buf = String::new();
1290 std::io::Read::read_to_string(&mut file, &mut buf)
1291 .map_err(|e| AppError::BadRequest(e.to_string()))?;
1292 serde_json::from_str(&buf)
1293 .map_err(|e| AppError::BadRequest(format!("invalid manifest: {}", e)))?
1294 };
1295 let schema_name = manifest_value
1296 .get("schema")
1297 .and_then(Value::as_str)
1298 .ok_or_else(|| AppError::BadRequest("manifest missing 'schema'".into()))?;
1299
1300 let schemas_body = vec![serde_json::json!({ "id": DEFAULT_SCHEMA_ID, "name": schema_name })];
1301 let apply_order = config_apply_order();
1302 for kind in &apply_order {
1303 let body: Vec<Value> = if *kind == "schemas" {
1304 serde_json::from_value(Value::Array(schemas_body.clone()))
1305 .map_err(|e| AppError::BadRequest(format!("schemas body: {}", e)))?
1306 } else {
1307 let mut body = read_kind_from_zip(&mut archive, kind)?;
1308 match *kind {
1309 "enums" | "tables" | "indexes" => inject_schema_id(&mut body, DEFAULT_SCHEMA_ID),
1310 "relationships" => inject_relationship_schema_ids(&mut body, DEFAULT_SCHEMA_ID),
1311 _ => {}
1312 }
1313 body
1314 };
1315 replace_config(config_pool, kind, body, false, &row.package_id, None).await?;
1316 }
1317 upsert_package(config_pool, &row.package_id, &manifest_value).await?;
1318
1319 let claimed = mark_migration_plan_applied(config_pool, &migration_id).await?;
1321 if !claimed {
1322 return Err(AppError::Conflict(format!(
1323 "migration plan '{}' was applied by a concurrent request",
1324 migration_id
1325 )));
1326 }
1327
1328 let result = execute_migration_plan(
1330 migration_pool,
1331 config_pool,
1332 &plan,
1333 &migration_id,
1334 &row.package_id,
1335 tenant_id,
1336 row.from_version.as_deref(),
1337 &row.to_version,
1338 state.dialect.as_ref(),
1339 )
1340 .await?;
1341
1342 let new_config = load_from_pool(config_pool, &row.package_id)
1344 .await
1345 .map_err(AppError::Config)?;
1346
1347 if let Some(col) = ctx.rls_tenant_column() {
1350 apply_rls_to_tables(
1351 migration_pool,
1352 &new_config,
1353 None,
1354 col,
1355 state.dialect.as_ref(),
1356 )
1357 .await?;
1358 }
1359
1360 let new_model = resolve(&new_config)
1361 .map_err(AppError::Config)?
1362 .with_package_id(&row.package_id);
1363 {
1364 let mut guard = state
1365 .model
1366 .write()
1367 .map_err(|_| AppError::BadRequest("state lock".into()))?;
1368 *guard = new_model.clone();
1369 state
1370 .package_models
1371 .write()
1372 .map_err(|_| AppError::BadRequest("state lock".into()))?
1373 .insert(package_cache_key, new_model);
1374 }
1375 crate::handlers::entity::invalidate_cross_package_index(&state);
1377
1378 Ok((
1379 axum::http::StatusCode::OK,
1380 Json(crate::response::SuccessOne {
1381 data: json!({
1382 "migration_id": migration_id,
1383 "package_id": row.package_id,
1384 "from_version": row.from_version,
1385 "to_version": row.to_version,
1386 "steps_applied": result.applied,
1387 "steps_warned": result.warned,
1388 "steps_skipped": result.skipped,
1389 "warnings": result.warnings,
1390 "skipped": result.skips,
1391 }),
1392 meta: None,
1393 }),
1394 ))
1395}
1396
1397pub async fn bootstrap_tenant_handler(
1410 TenantId(tenant_id_opt): TenantId,
1411 State(state): State<AppState>,
1412 Path(PackageIdPath { package_id }): Path<PackageIdPath>,
1413) -> Result<impl axum::response::IntoResponse, AppError> {
1414 let tenant_id = tenant_id_opt
1415 .as_deref()
1416 .filter(|s| !s.is_empty())
1417 .ok_or_else(|| AppError::BadRequest("X-Tenant-ID header is required".into()))?;
1418
1419 let entry = state
1420 .tenant_registry
1421 .get(tenant_id)
1422 .ok_or_else(|| AppError::NotFound(format!("tenant not found: {}", tenant_id)))?;
1423
1424 if !matches!(entry.strategy, TenantStrategy::Database) {
1427 return Err(AppError::BadRequest(
1428 "bootstrap only applies to Database-strategy tenants; RLS tenants share the central DB which is migrated by install_package".into(),
1429 ));
1430 }
1431
1432 let database_url = entry.database_url.as_deref().ok_or_else(|| {
1433 AppError::BadRequest(format!("tenant {}: missing database_url", tenant_id))
1434 })?;
1435
1436 let _ = get_package(&state.pool, &package_id)
1438 .await?
1439 .ok_or_else(|| AppError::NotFound(format!("package '{}' is not installed", package_id)))?;
1440
1441 let config = load_from_pool(&state.pool, &package_id)
1442 .await
1443 .map_err(AppError::Config)?;
1444
1445 let pool = get_or_create_tenant_pool(&state, tenant_id, database_url).await?;
1446
1447 apply_migrations(
1449 &pool,
1450 &config,
1451 None,
1452 None,
1453 state.dialect.as_ref(),
1454 &std::collections::HashMap::new(),
1455 )
1456 .await?;
1457
1458 let model = crate::config::resolve(&config)
1460 .map_err(AppError::Config)?
1461 .with_package_id(&package_id);
1462 {
1463 state
1464 .package_models
1465 .write()
1466 .map_err(|_| AppError::BadRequest("state lock".into()))?
1467 .insert(format!("{}:{}", package_id, tenant_id), model);
1468 }
1469
1470 Ok((
1471 axum::http::StatusCode::OK,
1472 Json(crate::response::SuccessOne {
1473 data: serde_json::json!({
1474 "tenant_id": tenant_id,
1475 "package_id": package_id,
1476 "status": "bootstrapped",
1477 }),
1478 meta: None,
1479 }),
1480 ))
1481}
1482
1483fn build_full_config_from_values(
1487 values: &std::collections::HashMap<String, Vec<Value>>,
1488) -> Result<crate::config::FullConfig, AppError> {
1489 fn parse_kind<T: serde::de::DeserializeOwned>(
1490 values: &std::collections::HashMap<String, Vec<Value>>,
1491 key: &str,
1492 ) -> Result<Vec<T>, AppError> {
1493 let arr = values.get(key).cloned().unwrap_or_default();
1494 arr.into_iter()
1495 .map(|v| {
1496 serde_json::from_value(v)
1497 .map_err(|e| AppError::BadRequest(format!("{} parse error: {}", key, e)))
1498 })
1499 .collect()
1500 }
1501
1502 Ok(crate::config::FullConfig {
1503 schemas: parse_kind(values, "schemas")?,
1504 enums: parse_kind(values, "enums")?,
1505 tables: parse_kind(values, "tables")?,
1506 columns: parse_kind(values, "columns")?,
1507 indexes: parse_kind(values, "indexes")?,
1508 relationships: parse_kind(values, "relationships")?,
1509 api_entities: parse_kind(values, "api_entities")?,
1510 kv_stores: parse_kind(values, "kv_stores")?,
1511 reports: parse_kind(values, "reports")?,
1512 })
1513}