use cratefield_core::Harness;
#[cfg(feature = "postgres")]
#[cfg(feature = "postgres")]
const FLEET_PARALLELISM: usize = 8;
pub fn apply(
harness: &Harness,
dialect: &str,
url: &str,
plan: bool,
fleet: bool,
tenant: Option<&str>,
strict: bool,
) -> Result<(), String> {
if dialect != "postgres" {
return Err(format!(
"dialect {dialect:?} is not supported for apply (only postgres, issue #18)"
));
}
if plan && fleet {
return Err(
"--plan and --fleet contradict each other: --plan applies nothing, --fleet \
applies to every registered tenant"
.to_owned(),
);
}
if tenant.is_some() && !plan {
return Err(
"--tenant only means something with --plan today: fleet apply is the runtime \
reconciler's job, and one tenant cannot be applied without its neighbours' \
registry being read anyway"
.to_owned(),
);
}
if strict && !fleet {
return Err(
"--strict says what to do about a degraded tenant, and only --fleet has \
tenants: a plain apply either applies its migrations or fails"
.to_owned(),
);
}
#[cfg(not(feature = "postgres"))]
{
let _ = (harness, url, plan, fleet, tenant, strict);
Err(
"this fz binary was built without cratefield-cli's `postgres` feature — \
rebuild the fz bin with `--features cratefield-cli/postgres` to apply \
migrations to Postgres (sqlite/D1 migrations run through \
`wrangler d1 migrations apply`)"
.to_owned(),
)
}
#[cfg(feature = "postgres")]
{
if plan {
apply_plan(harness, url, tenant)
} else if fleet {
apply_fleet(harness, url, strict)
} else {
apply_postgres(harness, url)
}
}
}
#[cfg(feature = "postgres")]
fn runtime() -> Result<tokio::runtime::Runtime, String> {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|err| format!("cannot start the async runtime: {err}"))
}
#[cfg(feature = "postgres")]
fn apply_postgres(harness: &Harness, url: &str) -> Result<(), String> {
runtime()?.block_on(async {
let db = cratefield_adapter_postgres::Postgres::connect(url)
.await
.map_err(|err| format!("cannot connect (check --url): {err}"))?;
db.apply_harness_migrations(harness)
.await
.map_err(|err| format!("migrations failed: {err}"))
})
}
#[cfg(feature = "postgres")]
fn apply_fleet(harness: &Harness, url: &str, strict: bool) -> Result<(), String> {
runtime()?.block_on(async {
let db = cratefield_adapter_postgres::Postgres::connect(url)
.await
.map_err(|err| {
format!("cannot connect to the control database (check --url): {err}")
})?;
let reports = db
.reconcile_fleet(harness, FLEET_PARALLELISM)
.await
.map_err(|err| format!("reconciliation failed: {err}"))?;
if reports.is_empty() {
println!(
"no tenants registered in the control database — nothing to reconcile \
(a venture database is applied without --fleet)"
);
return Ok(());
}
let degraded: Vec<&cratefield_adapter_postgres::TenantReport> = reports
.iter()
.filter(|report| report.status == cratefield_adapter_postgres::TenantStatus::Degraded)
.collect();
for report in &reports {
println!(
"tenant {}: {} applied, {} skipped, {}{}",
report.tenant,
report.applied,
report.skipped,
report.status,
report
.error
.as_deref()
.map(|err| format!(" — {err}"))
.unwrap_or_default()
);
}
if degraded.is_empty() {
Ok(())
} else if strict {
Err(format!(
"{} tenant(s) degraded ({}): boot would abort in strict mode",
degraded.len(),
degraded
.iter()
.map(|report| report.tenant.as_str())
.collect::<Vec<_>>()
.join(", ")
))
} else {
println!(
"{} tenant(s) degraded; every other tenant serves (re-run with --strict \
to make this a failure)",
degraded.len()
);
Ok(())
}
})
}
#[cfg(feature = "postgres")]
fn apply_plan(harness: &Harness, url: &str, tenant: Option<&str>) -> Result<(), String> {
runtime()?.block_on(async {
let control = cratefield_adapter_postgres::Postgres::connect(url)
.await
.map_err(|err| {
format!("cannot connect to the control database (check --url): {err}")
})?;
control
.bootstrap_registry()
.await
.map_err(|err| format!("cannot bootstrap the registry: {err}"))?;
let records = control
.tenants()
.await
.map_err(|err| format!("cannot read the tenant registry: {err}"))?;
let records: Vec<_> = match tenant {
Some(name) => {
let found: Vec<_> = records
.iter()
.filter(|record| record.tenant == name)
.collect();
if found.is_empty() {
return Err(format!("tenant {name:?} is not in the registry"));
}
found
}
None => records.iter().collect(),
};
if records.is_empty() {
println!("no tenants registered");
return Ok(());
}
for record in records {
match control.reconcile_plan(record, harness).await {
Ok(plan) => {
if plan.modules.is_empty() {
println!("tenant {}: nothing to apply", plan.tenant);
} else {
println!("tenant {}:", plan.tenant);
for module in &plan.modules {
println!(" {}: {}", module.module, module.migrations.join(", "));
}
}
}
Err(err) => println!("tenant {}: unreachable — {err}", record.tenant),
}
}
Ok(())
})
}