use crate::platform::config::RustployConfig;
use crate::{
engine,
engine::{DockerClient, containers, images, networks},
features::ingress::{IngressController, TlsManager},
platform::event_bus::EventBus,
platform::secrets::SecretsManager,
store::Db,
};
use anyhow::{Result, anyhow};
use bollard::models::HealthStatusEnum;
use chrono::Utc;
use shared::{
DeployState, Deployment, Event, HealthcheckKind, Service, ServiceSource, ServiceStatus,
};
use std::{path::PathBuf, sync::Arc, time::Duration};
use tokio::time::sleep;
use tracing::{debug, error, info, warn};
mod compose_flow;
mod drain;
mod healthcheck;
mod promote;
mod resolve;
mod rollback;
mod source;
mod staging;
mod state_machine;
#[cfg(test)]
use {
rollback::{rollback_cause, truncate_chars},
source::missing_dockerfile_msg,
state_machine::failure_log_lines,
};
#[cfg(test)]
mod tests;
pub struct DeployExecutor {
pub db: Arc<Db>,
pub docker: Arc<DockerClient>,
pub ingress: Arc<IngressController>,
pub bus: Arc<EventBus>,
pub secrets: Arc<SecretsManager>,
pub tls: Arc<TlsManager>,
pub db_path: PathBuf,
pub drain_secs: u64,
pub registry_internal_token: Option<Arc<str>>,
}
fn is_embedded_registry_image(image: &str, port: u16, domain: Option<&str>) -> bool {
if image.starts_with(&format!("127.0.0.1:{port}/")) {
return true;
}
if image.starts_with(&format!("localhost:{port}/")) {
return true;
}
if let Some(d) = domain
&& image.starts_with(&format!("{d}/"))
{
return true;
}
false
}
impl DeployExecutor {
pub(super) fn clone_dir(&self, deployment_id: &str) -> PathBuf {
self.db_path.join("builds").join(deployment_id)
}
pub(super) fn short<'a>(&self, id: &'a str) -> &'a str {
let s = id.find('_').map(|i| &id[i + 1..]).unwrap_or(id);
&s[..8.min(s.len())]
}
pub(super) async fn ensure_live_names_free(&self, svc: &Service) -> Result<()> {
for i in 0..svc.spec.replicas.max(1) {
let live = containers::replica_live_name(&svc.id, &svc.spec.name, i);
let Some((cid, owner)) = containers::name_owner(&self.docker.inner, &live).await?
else {
continue;
};
if owner
.as_deref()
.is_some_and(|o| containers::label_owned_by(o, &svc.id))
{
continue;
}
let dono = match owner.as_deref() {
Some(owner_id) => self.describe_service(owner_id).await,
None => "um container que não foi criado pelo rustploy".to_string(),
};
warn!(
service_id = %svc.id,
container_name = %live,
container_id = %cid,
owner = ?owner,
"deploy recusado: nome de container ocupado por outro serviço"
);
return Err(anyhow!(
"o nome de container `{live}` já é usado por {dono}; o deploy foi \
recusado para não derrubá-lo. Nome de container é único no servidor \
inteiro: renomeie um dos dois serviços ou remova o container órfão"
));
}
Ok(())
}
pub(super) async fn describe_service(&self, service_id: &str) -> String {
let candidates = [service_id.to_string(), format!("svc_{service_id}")];
for id in &candidates {
if let Ok(Some(owner)) = crate::store::services::get(&self.db, id).await {
let projeto =
match crate::store::projects::get(&self.db, &owner.spec.project_id).await {
Ok(Some(p)) => p.name,
_ => owner.spec.project_id.clone(),
};
return format!("o serviço `{}` do projeto `{projeto}`", owner.spec.name);
}
}
format!("o serviço `{service_id}` (que não existe mais — container órfão)")
}
pub(super) async fn ensure_firewall(
&self,
deployment_id: &str,
service_id: &str,
host_port: u16,
) {
let line = match crate::features::ingress::firewall::ensure_allowed(host_port).await {
Ok(backend) if backend == "none" => {
format!("--> Porta externa {host_port} exposta (nenhum firewall ativo no host)")
}
Ok(backend) => {
format!("--> Porta externa {host_port} liberada no firewall ({backend})")
}
Err(e) => format!(
"--> Aviso: não foi possível liberar a porta {host_port} no firewall: {e}. \
Se a conexão externa falhar, libere-a manualmente."
),
};
self.log_step(deployment_id, service_id, &line).await;
}
pub(super) async fn log_step(&self, deployment_id: &str, service_id: &str, line: &str) {
let ts = chrono::Utc::now();
let _ = crate::store::build_logs::append(&self.db, deployment_id, line, ts).await;
self.bus.publish(Event::BuildLog {
deployment_id: deployment_id.to_string(),
service_id: service_id.to_string(),
line: line.to_string(),
timestamp: ts,
});
}
pub(super) async fn registry_credentials_for(
&self,
image: &str,
) -> Option<bollard::auth::DockerCredentials> {
let token = self.registry_internal_token.as_ref()?;
let port = RustployConfig::global().registry.port;
let mut domain = RustployConfig::global().registry.domain.clone();
if let Ok(Some(d)) = crate::store::daemon_settings::get(
&self.db,
crate::store::daemon_settings::KEY_REGISTRY_DOMAIN,
)
.await
&& !d.trim().is_empty()
{
domain = Some(d);
}
if is_embedded_registry_image(image, port, domain.as_deref()) {
Some(bollard::auth::DockerCredentials {
username: Some("rp-internal".to_string()),
password: Some(token.to_string()),
..Default::default()
})
} else {
None
}
}
pub(super) fn image_for(&self, dep: &Deployment, svc: &Service) -> String {
match &svc.spec.source {
ServiceSource::Registry { image } => image.clone(),
ServiceSource::Git(_) => format!("rp_{}:{}", svc.spec.safe_name(), self.short(&dep.id)),
ServiceSource::Archive(_) => {
format!("rp_{}:{}", svc.spec.safe_name(), self.short(&dep.id))
}
ServiceSource::Compose(c) => format!("compose:{}", c.content),
}
}
pub(super) fn archive_dir(&self, service_id: &str, archive_id: &str) -> PathBuf {
self.db_path
.join("uploads")
.join("services")
.join(service_id)
.join(archive_id)
}
pub(super) async fn network_name(&self, project_id: &str) -> Result<String> {
crate::store::projects::network_name(&self.db, project_id).await
}
pub(super) async fn ensure_network(&self, project_id: &str) -> Result<String> {
let name = self.network_name(project_id).await?;
networks::ensure_project_network(&self.docker.inner, &name).await
}
pub(super) async fn resolve_env(&self, svc: &Service) -> Result<Vec<(String, String)>> {
super::env_resolve::resolve(&self.db, &self.secrets, svc).await
}
pub(super) async fn load_deployment(&self, id: &str) -> Result<Deployment> {
crate::store::deployments::get(&self.db, id)
.await?
.ok_or_else(|| anyhow!("deployment not found: {id}"))
}
pub(super) async fn load_service(&self, id: &str) -> Result<Option<Service>> {
crate::store::services::get(&self.db, id).await
}
pub(super) async fn transition(
&self,
deployment_id: &str,
from: &DeployState,
to: DeployState,
message: Option<String>,
) -> Result<()> {
info!(
deployment_id = %deployment_id,
from = from.label(),
to = to.label(),
message = ?message,
"executor: gravando transição no banco"
);
let dep = crate::store::deployments::transition(
&self.db,
deployment_id,
from,
to.clone(),
message.clone(),
)
.await?;
self.bus.publish(Event::DeployStateChanged {
deployment_id: deployment_id.to_string(),
service_id: dep.service_id.clone(),
state: to.clone(),
timestamp: Utc::now(),
message,
});
info!(
deployment_id = %deployment_id,
state = to.label(),
"executor: evento DeployStateChanged publicado"
);
Ok(())
}
}