use super::Db;
use super::infra::model::{InfraState, ServiceInfra};
use anyhow::Result;
use chrono::{DateTime, Utc};
use shared::{Service, ServiceSource, ServiceSpec, ServiceStatus};
use std::collections::HashMap;
use tracing::{debug, info, warn};
use ulid::Ulid;
type Runtime = (String, Option<String>, String, DateTime<Utc>);
type RuntimeRow = (String, String, Option<String>, String, DateTime<Utc>);
async fn runtimes(db: &Db) -> Result<HashMap<String, Runtime>> {
let rows: Vec<RuntimeRow> = sqlx::query_as(
"SELECT service_id, status, live_container_id, desired_state, updated_at \
FROM service_runtime",
)
.fetch_all(db.pool())
.await?;
Ok(rows
.into_iter()
.map(|(id, st, c, d, u)| (id, (st, c, d, u)))
.collect())
}
fn parse_status(s: &str) -> ServiceStatus {
match s {
"Stopped" => ServiceStatus::Stopped,
"Stopping" => ServiceStatus::Stopping,
"Deploying" => ServiceStatus::Deploying,
"Queued" => ServiceStatus::Queued,
"Running" => ServiceStatus::Running,
"Degraded" => ServiceStatus::Degraded,
s if s.starts_with("Error:") => {
ServiceStatus::Error(s.trim_start_matches("Error:").trim().to_string())
}
_ => ServiceStatus::Stopped,
}
}
fn assemble(s: &ServiceInfra, rt: Option<&Runtime>) -> Service {
Service {
id: s.id.clone(),
spec: s.spec.clone(),
compose_project: s.compose_project.clone(),
status: rt
.map(|r| parse_status(&r.0))
.unwrap_or(ServiceStatus::Stopped),
live_container_id: rt.and_then(|r| r.1.clone()),
created_at: s.created_at,
updated_at: rt.map(|r| r.3).unwrap_or(s.created_at),
}
}
async fn all_services(db: &Db) -> Result<Vec<(Service, Option<Runtime>)>> {
let rts = runtimes(db).await?;
let st = db.infra().snapshot();
let mut out: Vec<(Service, Option<Runtime>)> = st
.projects
.iter()
.flat_map(|p| p.services.iter())
.map(|s| {
let rt = rts.get(&s.id).cloned();
(assemble(s, rt.as_ref()), rt)
})
.collect();
out.sort_by(|a, b| (a.0.created_at, &a.0.id).cmp(&(b.0.created_at, &b.0.id)));
Ok(out)
}
async fn filtered(db: &Db, keep: impl Fn(&Runtime) -> bool) -> Result<Vec<Service>> {
Ok(all_services(db)
.await?
.into_iter()
.filter(|(_, rt)| rt.as_ref().is_some_and(&keep))
.map(|(s, _)| s)
.collect())
}
fn new_compose_project(st: &InfraState, id: &str, name: &str) -> String {
let short = shared::new_compose_project_name(id, name);
let taken = st
.projects
.iter()
.flat_map(|p| p.services.iter())
.any(|s| s.compose_project.as_deref() == Some(short.as_str()));
if taken {
shared::new_compose_project_name_long(id, name)
} else {
short
}
}
fn check_shared_rules(spec: &ServiceSpec) -> Result<()> {
crate::platform::connection::validate_shared_rules(
&spec.name,
&spec.source,
spec.product.as_deref(),
spec.shared.is_some(),
)
.map_err(|e| anyhow::anyhow!(e))
}
pub async fn create(db: &Db, spec: ServiceSpec) -> Result<Service> {
check_shared_rules(&spec)?;
let id = format!("svc_{}", Ulid::new());
info!(id = %id, name = %spec.name, project_id = %spec.project_id, "db::services::create");
let now = Utc::now();
let new_safe = spec.safe_name();
let rec = db.infra().update(|st| {
let compose_project = match &spec.source {
ServiceSource::Compose(_) => Some(new_compose_project(st, &id, &spec.name)),
_ => None,
};
let Some(project) = st.projects.iter_mut().find(|p| p.id == spec.project_id) else {
anyhow::bail!("projeto {} não encontrado", spec.project_id);
};
if project
.services
.iter()
.any(|s| shared::normalize_name(&s.spec.name) == new_safe)
{
anyhow::bail!(
"já existe um serviço com o nome \"{}\" neste projeto",
spec.name
);
}
let rec = ServiceInfra {
id: id.clone(),
spec: spec.clone(),
compose_project,
created_at: now,
webhook_token: None,
secrets: vec![],
};
project.services.push(rec.clone());
Ok(rec)
})?;
sqlx::query(
"INSERT OR IGNORE INTO service_runtime (service_id, status, live_container_id, desired_state, updated_at)
VALUES (?, 'Stopped', NULL, 'Stopped', ?)",
)
.bind(&id)
.bind(now)
.execute(db.pool())
.await?;
let svc = assemble(&rec, None);
info!(service_id = %svc.id, name = %svc.spec.name, "db::services::create: salvo");
Ok(svc)
}
pub async fn list(db: &Db, project_id: &str) -> Result<Vec<Service>> {
Ok(all_services(db)
.await?
.into_iter()
.map(|(s, _)| s)
.filter(|s| s.spec.project_id == project_id)
.collect())
}
pub async fn get(db: &Db, id: &str) -> Result<Option<Service>> {
let st = db.infra().snapshot();
let Some(rec) = st
.projects
.iter()
.flat_map(|p| p.services.iter())
.find(|s| s.id == id)
else {
return Ok(None);
};
let rt = runtimes(db).await?.remove(id);
Ok(Some(assemble(rec, rt.as_ref())))
}
pub async fn update_spec(db: &Db, id: &str, spec: ServiceSpec) -> Result<Option<Service>> {
check_shared_rules(&spec)?;
let new_safe = spec.safe_name();
let found = db.infra().update(|st| {
let compose_name = new_compose_project(st, id, &spec.name);
let Some(project) = st
.projects
.iter_mut()
.find(|p| p.services.iter().any(|s| s.id == id))
else {
return Ok(false);
};
if project
.services
.iter()
.any(|s| s.id != id && shared::normalize_name(&s.spec.name) == new_safe)
{
anyhow::bail!(
"já existe um serviço com o nome \"{}\" neste projeto",
spec.name
);
}
let owner = project.id.clone();
let svc = project.services.iter_mut().find(|s| s.id == id).unwrap();
if matches!(spec.source, ServiceSource::Compose(_)) && svc.compose_project.is_none() {
svc.compose_project = Some(compose_name);
}
svc.spec = spec.clone();
svc.spec.project_id = owner;
Ok(true)
})?;
if !found {
return Ok(None);
}
get(db, id).await
}
pub async fn clear_pre_deploy_job(db: &Db, job_id: &str) -> Result<u64> {
db.infra().update(|st| {
let mut cleared = 0u64;
for svc in st.projects.iter_mut().flat_map(|p| p.services.iter_mut()) {
let legacy = svc.spec.pre_deploy_job_id.as_deref() == Some(job_id);
let queued = svc.spec.pre_deploy_job_ids.iter().any(|j| j == job_id);
if !legacy && !queued {
continue;
}
if legacy {
svc.spec.pre_deploy_job_id = None;
}
svc.spec.pre_deploy_job_ids.retain(|j| j != job_id);
cleared += 1;
}
Ok(cleared)
})
}
pub async fn update_status(
db: &Db,
id: &str,
status: &ServiceStatus,
container_id: Option<&str>,
) -> Result<()> {
info!(
service_id = %id,
status = %status,
container_id = ?container_id.map(
|container_id| format!("...{}", &container_id[..container_id.len().min(10)])
),
"db::services::update_status"
);
let exists = db
.infra()
.snapshot()
.projects
.iter()
.flat_map(|p| p.services.iter())
.any(|s| s.id == id);
if exists {
sqlx::query(
"INSERT INTO service_runtime (service_id, status, live_container_id, desired_state, updated_at)
VALUES (?, ?, ?, 'Stopped', ?)
ON CONFLICT(service_id) DO UPDATE SET
status = excluded.status,
-- Queued/Deploying sem container: o deploy novo ainda está sendo construído e o container
-- ANTIGO segue no ar. Apagar o id aqui cegava logs, métricas, watchdog e o parar durante
-- todo o build, e fazia o reconcile achar que o Live antigo não rodava (vira Stopped).
live_container_id = CASE
WHEN excluded.live_container_id IS NULL
AND excluded.status IN ('Queued', 'Deploying')
THEN service_runtime.live_container_id
ELSE excluded.live_container_id
END,
updated_at = excluded.updated_at",
)
.bind(id)
.bind(status.to_string())
.bind(container_id)
.bind(Utc::now())
.execute(db.pool())
.await?;
}
debug!(service_id = %id, status = %status, "db::services::update_status: atualizado");
Ok(())
}
pub async fn delete(db: &Db, id: &str) -> Result<bool> {
super::shared_access::delete_for_server(db, id).await?;
super::managed_database::delete_for_server(db, id).await?;
let removed = db.infra().update(|st| {
let mut removed = false;
for p in &mut st.projects {
let before = p.services.len();
p.services.retain(|s| s.id != id);
removed |= p.services.len() != before;
}
Ok(removed)
})?;
sqlx::query("DELETE FROM service_runtime WHERE service_id = ?")
.bind(id)
.execute(db.pool())
.await?;
Ok(removed)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DesiredState {
Running,
Stopped,
}
impl DesiredState {
fn as_str(self) -> &'static str {
match self {
Self::Running => "Running",
Self::Stopped => "Stopped",
}
}
}
pub async fn set_desired_state(db: &Db, id: &str, state: DesiredState) -> Result<()> {
if get(db, id).await?.is_none() {
return Ok(());
}
sqlx::query(
"INSERT INTO service_runtime (service_id, status, live_container_id, desired_state, updated_at)
VALUES (?, 'Stopped', NULL, ?, ?)
ON CONFLICT(service_id) DO UPDATE SET desired_state = excluded.desired_state",
)
.bind(id)
.bind(state.as_str())
.bind(Utc::now())
.execute(db.pool())
.await?;
Ok(())
}
#[allow(dead_code)] pub async fn desired_state(db: &Db, id: &str) -> Result<DesiredState> {
let v: Option<String> =
sqlx::query_scalar("SELECT desired_state FROM service_runtime WHERE service_id = ?")
.bind(id)
.fetch_optional(db.pool())
.await?;
Ok(match v.as_deref() {
Some("Running") => DesiredState::Running,
_ => DesiredState::Stopped,
})
}
pub async fn get_desired_running(db: &Db) -> Result<Vec<Service>> {
filtered(db, |r| r.2 == "Running").await
}
pub async fn get_running(db: &Db) -> Result<Vec<Service>> {
filtered(db, |r| r.0 == "Running").await
}
pub async fn count_by_project(db: &Db, project_id: &str) -> Result<i64> {
Ok(db
.infra()
.snapshot()
.projects
.iter()
.find(|p| p.id == project_id)
.map(|p| p.services.len() as i64)
.unwrap_or(0))
}
pub async fn get_watchable(db: &Db) -> Result<Vec<Service>> {
filtered(db, |r| r.0 == "Running" || r.0 == "Degraded").await
}
pub async fn list_all(db: &Db) -> Result<Vec<Service>> {
Ok(all_services(db)
.await?
.into_iter()
.map(|(s, _)| s)
.collect())
}
#[cfg(test)]
fn test_compose_spec(name: String, project_id: String) -> ServiceSpec {
ServiceSpec {
name,
project_id,
source: ServiceSource::Compose(shared::ComposeSource {
content: String::new(),
ingress_service: None,
}),
port: 80,
host_port: None,
domain: None,
tls_enabled: false,
env_vars: vec![],
env_comments: vec![],
volumes: vec![],
healthcheck: shared::Healthcheck::default(),
replicas: 1,
resources: shared::ResourceLimits::default(),
run_command: None,
run_args: vec![],
product: None,
domains: vec![],
pre_deploy_job_id: None,
pre_deploy_job_ids: vec![],
depends_on: vec![],
registry_auto_deploy: false,
autostart: true,
shared: None,
}
}
#[cfg(test)]
mod tests {
use super::*;
use shared::ServiceSource;
fn spec(name: &str, project: &str) -> ServiceSpec {
ServiceSpec {
name: name.into(),
project_id: project.into(),
source: ServiceSource::Registry {
image: "nginx:latest".into(),
},
port: 80,
host_port: None,
domain: None,
tls_enabled: false,
env_vars: vec![],
env_comments: vec![],
volumes: vec![],
healthcheck: shared::Healthcheck::default(),
replicas: 1,
resources: shared::ResourceLimits::default(),
run_command: None,
run_args: vec![],
product: None,
domains: vec![],
pre_deploy_job_id: None,
pre_deploy_job_ids: vec![],
depends_on: vec![],
registry_auto_deploy: false,
autostart: true,
shared: None,
}
}
async fn mem_db() -> Db {
let db = Db::memory_for_test().await;
db.seed_project_for_test("proj_a");
db.seed_project_for_test("proj_b");
db
}
#[tokio::test]
async fn rejeita_nome_duplicado_no_mesmo_projeto() {
let db = mem_db().await;
create(&db, spec("api", "proj_a")).await.unwrap();
assert!(create(&db, spec("api", "proj_a")).await.is_err());
create(&db, spec("my-api", "proj_a")).await.unwrap();
assert!(create(&db, spec("my_api", "proj_a")).await.is_err());
}
#[tokio::test]
async fn spec_vai_ao_yaml_e_runtime_fica_no_sql() {
let db = mem_db().await;
let svc = create(&db, spec("api", "proj_a")).await.unwrap();
update_status(
&db,
&svc.id,
&ServiceStatus::Running,
Some("cont_secreto_1"),
)
.await
.unwrap();
let yaml = String::from_utf8(
db.infra()
.tree()
.read("projects/proj_a/project.yml")
.unwrap()
.unwrap(),
)
.unwrap();
assert!(yaml.contains("nginx:latest"), "o spec deve estar no YAML");
assert!(
!yaml.contains("Running") && !yaml.contains("cont_secreto_1"),
"status e container são runtime e não podem ir ao YAML:\n{yaml}"
);
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM service")
.fetch_one(db.pool())
.await
.unwrap();
assert_eq!(n, 0);
}
#[tokio::test]
async fn criar_em_projeto_inexistente_falha_sem_deixar_rastro() {
let db = mem_db().await;
let e = create(&db, spec("api", "nao_existe")).await.unwrap_err();
assert!(e.to_string().contains("não encontrado"), "{e}");
assert!(list_all(&db).await.unwrap().is_empty());
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM service_runtime")
.fetch_one(db.pool())
.await
.unwrap();
assert_eq!(n, 0);
}
#[tokio::test]
async fn webhook_token_acompanha_o_servico_no_yaml_cifrado() {
use crate::store::webhook_tokens as wh;
let db = std::sync::Arc::new(mem_db().await);
let sm = crate::platform::secrets::SecretsManager::new(db.clone()).unwrap();
let svc = create(&db, spec("api", "proj_a")).await.unwrap();
assert!(wh::get(&db, &sm, &svc.id).await.unwrap().is_none());
wh::upsert(&db, &sm, &svc.id, "tok1").await.unwrap();
wh::upsert(&db, &sm, &svc.id, "tok2-visivel").await.unwrap();
assert_eq!(
wh::get(&db, &sm, &svc.id).await.unwrap().as_deref(),
Some("tok2-visivel")
);
let yaml = String::from_utf8(
db.infra()
.tree()
.read("projects/proj_a/project.yml")
.unwrap()
.unwrap(),
)
.unwrap();
assert!(yaml.contains("ENC[age:") && !yaml.contains("tok2-visivel"));
assert!(wh::upsert(&db, &sm, "fantasma", "x").await.is_err());
delete(&db, &svc.id).await.unwrap();
assert!(wh::get(&db, &sm, &svc.id).await.unwrap().is_none());
}
#[tokio::test]
async fn status_e_container_vivem_em_service_runtime() {
let db = mem_db().await;
let svc = create(&db, spec("api", "proj_a")).await.unwrap();
assert_eq!(svc.status, ServiceStatus::Stopped);
update_status(&db, &svc.id, &ServiceStatus::Running, Some("cont_1"))
.await
.unwrap();
let got = get(&db, &svc.id).await.unwrap().unwrap();
assert_eq!(got.status, ServiceStatus::Running);
assert_eq!(got.live_container_id.as_deref(), Some("cont_1"));
let (st,): (String,) =
sqlx::query_as("SELECT status FROM service_runtime WHERE service_id = ?")
.bind(&svc.id)
.fetch_one(db.pool())
.await
.unwrap();
assert_eq!(st, "Running");
assert_eq!(get_running(&db).await.unwrap().len(), 1);
assert_eq!(get_watchable(&db).await.unwrap().len(), 1);
update_status(&db, &svc.id, &ServiceStatus::Stopped, None)
.await
.unwrap();
assert!(get_running(&db).await.unwrap().is_empty());
}
#[tokio::test]
async fn update_status_de_servico_inexistente_nao_cria_linha_de_runtime() {
let db = mem_db().await;
update_status(&db, "svc_fantasma", &ServiceStatus::Running, None)
.await
.unwrap();
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM service_runtime")
.fetch_one(db.pool())
.await
.unwrap();
assert_eq!(n, 0);
}
#[tokio::test]
async fn desired_state_comeca_parado_e_so_o_boot_ve_os_desejados_no_ar() {
let db = mem_db().await;
let a = create(&db, spec("a", "proj_a")).await.unwrap();
let b = create(&db, spec("b", "proj_a")).await.unwrap();
assert_eq!(
desired_state(&db, &a.id).await.unwrap(),
DesiredState::Stopped
);
assert!(get_desired_running(&db).await.unwrap().is_empty());
set_desired_state(&db, &a.id, DesiredState::Running)
.await
.unwrap();
let ids: Vec<String> = get_desired_running(&db)
.await
.unwrap()
.into_iter()
.map(|s| s.id)
.collect();
assert_eq!(ids, vec![a.id.clone()]);
update_status(&db, &a.id, &ServiceStatus::Stopped, None)
.await
.unwrap();
assert_eq!(
desired_state(&db, &a.id).await.unwrap(),
DesiredState::Running
);
set_desired_state(&db, &a.id, DesiredState::Stopped)
.await
.unwrap();
assert!(get_desired_running(&db).await.unwrap().is_empty());
let _ = b;
}
#[tokio::test]
async fn queued_e_deploying_preservam_o_container_ao_vivo_e_os_demais_estados_limpam() {
let db = mem_db().await;
let svc = create(&db, spec("api", "proj_a")).await.unwrap();
let cid = |db: &Db, id: String| {
let db = db.clone();
async move { get(&db, &id).await.unwrap().unwrap().live_container_id }
};
update_status(&db, &svc.id, &ServiceStatus::Running, Some("c_velho"))
.await
.unwrap();
update_status(&db, &svc.id, &ServiceStatus::Queued, None)
.await
.unwrap();
assert_eq!(cid(&db, svc.id.clone()).await.as_deref(), Some("c_velho"));
update_status(&db, &svc.id, &ServiceStatus::Deploying, None)
.await
.unwrap();
assert_eq!(cid(&db, svc.id.clone()).await.as_deref(), Some("c_velho"));
update_status(&db, &svc.id, &ServiceStatus::Running, Some("c_novo"))
.await
.unwrap();
assert_eq!(cid(&db, svc.id.clone()).await.as_deref(), Some("c_novo"));
update_status(&db, &svc.id, &ServiceStatus::Stopped, None)
.await
.unwrap();
assert_eq!(cid(&db, svc.id.clone()).await, None);
}
#[tokio::test]
async fn delete_apaga_tambem_o_runtime() {
let db = mem_db().await;
let svc = create(&db, spec("api", "proj_a")).await.unwrap();
assert!(delete(&db, &svc.id).await.unwrap());
let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM service_runtime")
.fetch_one(db.pool())
.await
.unwrap();
assert_eq!(n, 0);
}
#[tokio::test]
async fn migracao_preenche_runtime_de_banco_antigo_e_preserva_a_intencao() {
let db = mem_db().await;
let up = create(&db, spec("up", "proj_a")).await.unwrap();
let down = create(&db, spec("down", "proj_a")).await.unwrap();
for (svc, status) in [(&up, "Running"), (&down, "Stopped")] {
sqlx::query(
"INSERT INTO service (id, name, project_id, spec, status, created_at, updated_at)
VALUES (?, ?, 'proj_a', '{}', ?, ?, ?)",
)
.bind(&svc.id)
.bind(&svc.spec.name)
.bind(status)
.bind(Utc::now())
.bind(Utc::now())
.execute(db.pool())
.await
.unwrap();
}
sqlx::query("DELETE FROM service_runtime")
.execute(db.pool())
.await
.unwrap();
super::super::connect_existing_for_test(&db).await;
assert_eq!(
get(&db, &up.id).await.unwrap().unwrap().status,
ServiceStatus::Running
);
assert_eq!(
desired_state(&db, &up.id).await.unwrap(),
DesiredState::Running
);
assert_eq!(
desired_state(&db, &down.id).await.unwrap(),
DesiredState::Stopped
);
set_desired_state(&db, &up.id, DesiredState::Stopped)
.await
.unwrap();
super::super::connect_existing_for_test(&db).await;
assert_eq!(
desired_state(&db, &up.id).await.unwrap(),
DesiredState::Stopped
);
}
#[tokio::test]
async fn permite_mesmo_nome_em_projetos_diferentes() {
let db = mem_db().await;
create(&db, spec("api", "proj_a")).await.unwrap();
assert!(create(&db, spec("api", "proj_b")).await.is_ok());
}
#[tokio::test]
async fn rename_nao_pode_colidir_mas_permite_o_proprio_nome() {
let db = mem_db().await;
create(&db, spec("api", "proj_a")).await.unwrap();
let web = create(&db, spec("web", "proj_a")).await.unwrap();
assert!(
update_spec(&db, &web.id, spec("api", "proj_a"))
.await
.is_err()
);
assert!(
update_spec(&db, &web.id, spec("web2", "proj_a"))
.await
.unwrap()
.is_some()
);
assert!(
update_spec(&db, &web.id, spec("web2", "proj_a"))
.await
.unwrap()
.is_some()
);
}
#[tokio::test]
async fn clear_pre_deploy_job_limpa_so_servicos_que_referenciam_o_job_legado() {
let db = mem_db().await;
let mut with_job = spec("api", "proj_a");
with_job.pre_deploy_job_id = Some("job_alvo".into());
let api = create(&db, with_job).await.unwrap();
let mut with_other_job = spec("web", "proj_a");
with_other_job.pre_deploy_job_id = Some("job_outro".into());
let web = create(&db, with_other_job).await.unwrap();
let cleared = clear_pre_deploy_job(&db, "job_alvo").await.unwrap();
assert_eq!(cleared, 1);
assert_eq!(
get(&db, &api.id)
.await
.unwrap()
.unwrap()
.spec
.pre_deploy_job_id,
None
);
assert_eq!(
get(&db, &web.id)
.await
.unwrap()
.unwrap()
.spec
.pre_deploy_job_id,
Some("job_outro".into())
);
}
#[tokio::test]
async fn clear_pre_deploy_job_remove_so_o_item_da_fila_preservando_ordem() {
let db = mem_db().await;
let mut with_queue = spec("api", "proj_a");
with_queue.pre_deploy_job_ids =
vec!["job_migration".into(), "job_alvo".into(), "job_test".into()];
let api = create(&db, with_queue).await.unwrap();
let cleared = clear_pre_deploy_job(&db, "job_alvo").await.unwrap();
assert_eq!(cleared, 1);
assert_eq!(
get(&db, &api.id)
.await
.unwrap()
.unwrap()
.spec
.pre_deploy_job_ids,
vec!["job_migration".to_string(), "job_test".to_string()]
);
}
#[tokio::test]
async fn clear_pre_deploy_job_no_op_quando_ninguem_referencia() {
let db = mem_db().await;
let mut with_queue = spec("api", "proj_a");
with_queue.pre_deploy_job_ids = vec!["job_migration".into()];
create(&db, with_queue).await.unwrap();
let cleared = clear_pre_deploy_job(&db, "job_nunca_referenciado")
.await
.unwrap();
assert_eq!(cleared, 0);
}
}
fn compose_service_keys(content: &str) -> std::collections::BTreeSet<String> {
serde_yaml::from_str::<serde_yaml::Value>(content)
.ok()
.and_then(|v| v.get("services").and_then(|s| s.as_mapping().cloned()))
.map(|m| {
m.keys()
.filter_map(|k| k.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default()
}
pub async fn backfill_compose_projects(
db: &Db,
live: &std::collections::BTreeMap<String, std::collections::BTreeSet<String>>,
) -> Result<()> {
let services: Vec<Service> = list_all(db).await?;
let mut decided: Vec<(String, String)> = vec![];
let compose: Vec<&Service> = services
.iter()
.filter(|s| matches!(s.spec.source, ServiceSource::Compose(_)))
.collect();
let mut taken: std::collections::HashSet<String> = compose
.iter()
.filter_map(|s| s.compose_project.clone())
.collect();
let legacy_of = |s: &Service| shared::compose_project_name(&s.id, &s.spec.name);
let legacy_names: std::collections::HashSet<String> =
compose.iter().map(|s| legacy_of(s)).collect();
let (exact, rest): (Vec<&&Service>, Vec<&&Service>) = compose
.iter()
.filter(|s| s.compose_project.is_none())
.partition(|s| live.contains_key(&legacy_of(s)));
for svc in exact {
let legacy = legacy_of(svc);
decided.push((svc.id.clone(), legacy.clone()));
taken.insert(legacy);
}
for svc in rest {
let legacy = legacy_of(svc);
let ServiceSource::Compose(c) = &svc.spec.source else {
continue;
};
let keys = compose_service_keys(&c.content);
let id_prefix = {
let legacy_id = svc.id.strip_prefix("svc_").unwrap_or(&svc.id);
format!(
"rp_{}_",
legacy_id.get(..8).unwrap_or(legacy_id).to_lowercase()
)
};
let candidates: Vec<&String> = live
.iter()
.filter(|(p, _)| p.starts_with(&id_prefix))
.filter(|(p, _)| !taken.contains(*p) && !legacy_names.contains(*p))
.filter(|(_, stack_keys)| !stack_keys.is_empty() && stack_keys.is_subset(&keys))
.map(|(p, _)| p)
.collect();
let chosen = if candidates.len() == 1 {
info!(
service_id = %svc.id, name = %svc.spec.name, stack = %candidates[0],
"db::services::backfill: stack viva com nome antigo (serviço renomeado sem redeploy)"
);
candidates[0].clone()
} else {
if candidates.len() > 1 {
warn!(
service_id = %svc.id, name = %svc.spec.name, ?candidates,
"db::services::backfill: mais de uma stack candidata; usando o nome legado"
);
}
legacy
};
decided.push((svc.id.clone(), chosen.clone()));
taken.insert(chosen);
}
if !decided.is_empty() {
db.infra().update(|st| {
for svc in st.projects.iter_mut().flat_map(|p| p.services.iter_mut()) {
if svc.compose_project.is_none()
&& let Some((_, name)) = decided.iter().find(|(id, _)| *id == svc.id)
{
svc.compose_project = Some(name.clone());
}
}
Ok(())
})?;
}
let mut seen = std::collections::HashSet::new();
for s in db
.infra()
.snapshot()
.projects
.iter()
.flat_map(|p| p.services.iter())
{
if let Some(c) = &s.compose_project
&& !seen.insert(c.clone())
{
warn!(stack = %c, "db::services::backfill: duas stacks Compose com o mesmo nome; separar exige decidir de quem são os volumes");
}
}
Ok(())
}
#[cfg(test)]
mod nome_gravado_tests {
use super::*;
use crate::store::projects;
async fn temp_db() -> Db {
let dir = std::env::temp_dir().join(format!("rustploy-test-{}", Ulid::new()));
crate::store::connect(&dir).await.unwrap()
}
fn esquece_stack(db: &Db, id: Option<&str>) {
db.infra()
.update(|st| {
for s in st.projects.iter_mut().flat_map(|p| p.services.iter_mut()) {
if id.is_none_or(|i| i == s.id) {
s.compose_project = None;
}
}
Ok(())
})
.unwrap();
}
fn troca_id(db: &Db, de: &str, para: &str) {
db.infra()
.update(|st| {
for s in st.projects.iter_mut().flat_map(|p| p.services.iter_mut()) {
if s.id == de {
s.id = para.to_string();
}
}
Ok(())
})
.unwrap();
}
fn live(
v: &[(&str, &[&str])],
) -> std::collections::BTreeMap<String, std::collections::BTreeSet<String>> {
v.iter()
.map(|(p, keys)| (p.to_string(), keys.iter().map(|k| k.to_string()).collect()))
.collect()
}
#[tokio::test]
async fn projeto_novo_grava_rede_com_id_inteiro_e_dois_juntos_nao_colidem() {
let db = temp_db().await;
let a = projects::create(&db, "a".into(), None).await.unwrap();
let b = projects::create(&db, "b".into(), None).await.unwrap();
let na = projects::network_name(&db, &a.id).await.unwrap();
let nb = projects::network_name(&db, &b.id).await.unwrap();
assert_eq!(na, shared::new_project_network_name(&a.id));
assert!(na.len() > "rp_net_".len() + 20, "ID inteiro: {na}");
assert_ne!(na, nb);
}
#[tokio::test]
async fn migracao_preenche_rede_com_o_nome_legado() {
let db = temp_db().await;
sqlx::query(
"INSERT INTO project (id, name, env_vars, created_at) VALUES ('prj_01ABCDEFGHJK', 'velho', '[]', ?)",
)
.bind(Utc::now())
.execute(db.pool())
.await
.unwrap();
crate::store::connect_existing_for_test(&db).await;
assert_eq!(
projects::network_name(&db, "prj_01ABCDEFGHJK")
.await
.unwrap(),
"rp_net_01ABCDEF"
);
}
#[tokio::test]
async fn rename_de_servico_compose_nao_muda_a_stack() {
let db = temp_db().await;
let p = projects::create(&db, "p".into(), None).await.unwrap();
let spec = test_compose_spec("postgres".into(), p.id.clone());
let svc = create(&db, spec).await.unwrap();
let stack = svc.compose_project.clone().expect("Compose grava a stack");
assert_eq!(stack, shared::new_compose_project_name(&svc.id, "postgres"));
let mut spec = svc.spec.clone();
spec.name = "banco".into();
let renomeado = update_spec(&db, &svc.id, spec).await.unwrap().unwrap();
assert_eq!(renomeado.spec.name, "banco");
assert_eq!(renomeado.compose_project_name(), stack);
}
#[tokio::test]
async fn servico_que_nao_e_compose_nao_grava_stack() {
let db = temp_db().await;
let p = projects::create(&db, "p".into(), None).await.unwrap();
let spec = test_compose_spec("x".into(), p.id.clone());
let mut spec = spec;
spec.source = ServiceSource::Registry {
image: "nginx".into(),
};
let svc = create(&db, spec).await.unwrap();
assert!(svc.compose_project.is_none());
}
#[tokio::test]
async fn backfill_usa_a_stack_viva_de_servico_renomeado_sem_redeploy() {
let db = temp_db().await;
let p = projects::create(&db, "p".into(), None).await.unwrap();
let mut spec = test_compose_spec("postgres".into(), p.id.clone());
spec.source = ServiceSource::Compose(shared::ComposeSource {
content: "services:\n app:\n image: x\n".into(),
ingress_service: None,
});
let svc = create(&db, spec).await.unwrap();
esquece_stack(&db, Some(&svc.id));
let mut spec = svc.spec.clone();
spec.name = "banco".into();
update_spec(&db, &svc.id, spec).await.unwrap();
esquece_stack(&db, Some(&svc.id));
let antiga = shared::compose_project_name(&svc.id, "postgres");
backfill_compose_projects(&db, &live(&[("outra_coisa", &["x"]), (&antiga, &["app"])]))
.await
.unwrap();
let depois = get(&db, &svc.id).await.unwrap().unwrap();
assert_eq!(depois.compose_project.as_deref(), Some(antiga.as_str()));
}
#[tokio::test]
async fn backfill_sem_stack_no_docker_usa_a_formula_legada_com_o_nome_atual() {
let db = temp_db().await;
let p = projects::create(&db, "p".into(), None).await.unwrap();
let svc = create(&db, test_compose_spec("db".into(), p.id.clone()))
.await
.unwrap();
esquece_stack(&db, Some(&svc.id));
backfill_compose_projects(&db, &live(&[])).await.unwrap();
let depois = get(&db, &svc.id).await.unwrap().unwrap();
assert_eq!(
depois.compose_project,
Some(shared::compose_project_name(&svc.id, "db"))
);
}
#[tokio::test]
async fn backfill_nao_rouba_a_stack_de_outro_servico() {
let db = temp_db().await;
let p = projects::create(&db, "p".into(), None).await.unwrap();
let a = create(&db, test_compose_spec("a".into(), p.id.clone()))
.await
.unwrap();
let b = create(&db, test_compose_spec("b".into(), p.id.clone()))
.await
.unwrap();
esquece_stack(&db, None);
let stack_a = shared::compose_project_name(&a.id, "a");
backfill_compose_projects(&db, &live(&[(&stack_a, &["app"])]))
.await
.unwrap();
let a2 = get(&db, &a.id).await.unwrap().unwrap();
let b2 = get(&db, &b.id).await.unwrap().unwrap();
assert_ne!(a2.compose_project, b2.compose_project);
assert_eq!(
b2.compose_project,
Some(shared::compose_project_name(&b.id, "b"))
);
}
#[tokio::test]
async fn backfill_de_import_no_mesmo_segundo_liga_cada_stack_ao_seu_servico() {
let db = temp_db().await;
let p = projects::create(&db, "p".into(), None).await.unwrap();
let mk = |name: &str, key: &str| {
let mut spec = test_compose_spec(name.into(), p.id.clone());
spec.source = ServiceSource::Compose(shared::ComposeSource {
content: format!("services:\n {key}:\n image: x\n"),
ingress_service: None,
});
spec
};
let db_svc = create(&db, mk("db", "rp_db")).await.unwrap();
let broker = create(&db, mk("broker", "rp_broker")).await.unwrap();
for (svc, sufixo) in [
(&db_svc, "AAAAAAAAAAAAAAAAAAAAAAAA"),
(&broker, "BBBBBBBBBBBBBBBBBBBBBBBB"),
] {
troca_id(&db, &svc.id, &format!("svc_01M3Q7T1{sufixo}"));
}
let broker_id = "svc_01M3Q7T1BBBBBBBBBBBBBBBBBBBBBBBB";
let db_id = "svc_01M3Q7T1AAAAAAAAAAAAAAAAAAAAAAAA";
let mut spec = get(&db, broker_id).await.unwrap().unwrap().spec;
spec.name = "mq".into();
update_spec(&db, broker_id, spec).await.unwrap();
esquece_stack(&db, None);
backfill_compose_projects(&db, &live(&[("rp_01m3q7t1_broker", &["rp_broker"])]))
.await
.unwrap();
let mq = get(&db, broker_id).await.unwrap().unwrap();
let db2 = get(&db, db_id).await.unwrap().unwrap();
assert_eq!(mq.compose_project.as_deref(), Some("rp_01m3q7t1_broker"));
assert_eq!(db2.compose_project.as_deref(), Some("rp_01m3q7t1_db"));
}
}