use crate::features::databases::MigrationContext;
use crate::{
features::databases::env_switch,
features::databases::shared_db::{self, Engine},
features::deploy::{env_resolve, shared_net},
store::{self, migration::Record},
};
use anyhow::{Result, anyhow, bail};
use chrono::Utc;
use shared::{Migration, MigrationStep, Service};
use std::{collections::HashMap, process::Stdio};
use tracing::{info, warn};
use ulid::Ulid;
pub const STEPS: [&str; 7] = [
"Pré-checagem",
"Parar escritas",
"Dump + restore",
"Verificação",
"Trocar conexão",
"Subir serviços",
"Desligar banco antigo",
];
pub fn new_record(req: &shared::MigrationStartReq, project_id: &str, env_var: &str) -> Record {
Record {
m: Migration {
id: format!("mig_{}", Ulid::new()),
project_id: project_id.to_string(),
source_service_id: req.source_service_id.clone(),
source_database: req.source_database.clone(),
dest_database_id: req.dest_database_id.clone(),
env_var: env_var.to_string(),
status: "Running".into(),
steps: STEPS
.iter()
.map(|n| MigrationStep {
name: n.to_string(),
state: "pending".into(),
detail: String::new(),
})
.collect(),
log: vec![],
created_at: Utc::now(),
},
env_changes: vec![],
stopped_services: vec![],
}
}
pub fn valid_source_db(name: &str) -> bool {
!name.is_empty()
&& name.len() <= 63
&& name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
}
async fn step(state: &MigrationContext, rec: &mut Record, i: usize, st: &str, detail: &str) {
rec.m.steps[i].state = st.into();
if !detail.is_empty() {
rec.m.steps[i].detail = detail.into();
}
let _ = store::migration::save(&state.db, rec).await;
}
async fn log(state: &MigrationContext, rec: &mut Record, text: &str) {
for l in text.lines().filter(|l| !l.trim().is_empty()) {
rec.m.log.push(l.to_string());
}
let n = rec.m.log.len();
if n > 400 {
rec.m.log.drain(..n - 400);
}
let _ = store::migration::save(&state.db, rec).await;
}
pub fn pipe_script(e: Engine) -> &'static str {
match e {
Engine::Postgres => {
r#"set -euo pipefail
PGPASSWORD="$ORIGEM_PASS" pg_dump --format=custom --no-owner --no-acl \
-h "$ORIGEM_HOST" -p "$ORIGEM_PORT" -U "$ORIGEM_USER" "$ORIGEM_DB" \
| PGPASSWORD="$DESTINO_PASS" pg_restore --no-owner --role="$DESTINO_USER" --exit-on-error \
-h "$DESTINO_HOST" -p "$DESTINO_PORT" -U "$DESTINO_USER" --dbname="$DESTINO_DB"
"#
}
Engine::MySql | Engine::MariaDb => {
if e == Engine::MySql {
r#"set -euo pipefail
umask 077
printf '[client]\nuser=%s\npassword=%s\nhost=%s\nport=%s\n' "$ORIGEM_USER" "$ORIGEM_PASS" "$ORIGEM_HOST" "$ORIGEM_PORT" > /tmp/o.cnf
printf '[client]\nuser=%s\npassword=%s\nhost=%s\nport=%s\n' "$DESTINO_USER" "$DESTINO_PASS" "$DESTINO_HOST" "$DESTINO_PORT" > /tmp/d.cnf
mysqldump --defaults-extra-file=/tmp/o.cnf --single-transaction --routines --triggers --events \
--no-tablespaces --set-gtid-purged=OFF "$ORIGEM_DB" \
| sed -E 's/DEFINER=`[^`]+`@`[^`]+`/DEFINER=CURRENT_USER/g' \
| mysql --defaults-extra-file=/tmp/d.cnf "$DESTINO_DB"
"#
} else {
r#"set -euo pipefail
umask 077
printf '[client]\nuser=%s\npassword=%s\nhost=%s\nport=%s\n' "$ORIGEM_USER" "$ORIGEM_PASS" "$ORIGEM_HOST" "$ORIGEM_PORT" > /tmp/o.cnf
printf '[client]\nuser=%s\npassword=%s\nhost=%s\nport=%s\n' "$DESTINO_USER" "$DESTINO_PASS" "$DESTINO_HOST" "$DESTINO_PORT" > /tmp/d.cnf
mariadb-dump --defaults-extra-file=/tmp/o.cnf --single-transaction --routines --triggers --events "$ORIGEM_DB" \
| sed -E 's/DEFINER=`[^`]+`@`[^`]+`/DEFINER=CURRENT_USER/g' \
| mariadb --defaults-extra-file=/tmp/d.cnf "$DESTINO_DB"
"#
}
}
Engine::Mongo => {
r#"set -euo pipefail
umask 077
q() { printf "password: '%s'\n" "${1//\'/\'\'}"; }
q "$ORIGEM_PASS" > /tmp/o.yaml
q "$DESTINO_PASS" > /tmp/d.yaml
mongodump --config=/tmp/o.yaml --host="$ORIGEM_HOST" --port="$ORIGEM_PORT" --username="$ORIGEM_USER" \
--authenticationDatabase=admin --db="$ORIGEM_DB" --archive \
| mongorestore --config=/tmp/d.yaml --host="$DESTINO_HOST" --port="$DESTINO_PORT" --username="$DESTINO_USER" \
--authenticationDatabase="$DESTINO_DB" --archive --nsFrom="$ORIGEM_DB.*" --nsTo="$DESTINO_DB.*"
"#
}
}
}
pub fn counts_plan(e: Engine, db: &str) -> (String, String) {
match e {
Engine::Postgres => (
format!(r#"exec psql -U "$POSTGRES_USER" -d {db} -At -v ON_ERROR_STOP=1"#),
"SELECT format('SELECT %L || ''|'' || count(*) FROM %I.%I', table_schema || '.' || table_name, table_schema, table_name) \
FROM information_schema.tables WHERE table_type = 'BASE TABLE' \
AND table_schema NOT IN ('pg_catalog', 'information_schema') ORDER BY table_schema, table_name \\gexec\n"
.into(),
),
Engine::MySql | Engine::MariaDb => (
format!(
r#"B=$(command -v mariadb || command -v mysql); export MYSQL_PWD="$MYSQL_ROOT_PASSWORD"
for t in $($B -uroot -N -e "select table_name from information_schema.tables where table_schema='{db}' and table_type='BASE TABLE' order by 1"); do
echo "$t|$($B -uroot -N -e "select count(*) from \`{db}\`.\`$t\`")"
done"#
),
String::new(),
),
Engine::Mongo => (
Engine::Mongo.admin_sh_pub().into(),
format!(
"const d = db.getSiblingDB(\"{db}\"); d.getCollectionNames().filter(c => !c.startsWith('system.')).sort()\
.forEach(c => print(c + '|' + d.getCollection(c).countDocuments({{}})));\n"
),
),
}
}
struct Side {
host: String,
port: u16,
user: String,
pass: String,
}
fn source_side(
e: Engine,
svc: &Service,
host: &str,
env: &HashMap<String, String>,
) -> Result<Side> {
let g = |k: &str| env.get(k).cloned().unwrap_or_default();
let (user, pass) = match e {
Engine::Postgres => (g("POSTGRES_USER"), g("POSTGRES_PASSWORD")),
Engine::MySql | Engine::MariaDb => {
if !g("MYSQL_ROOT_PASSWORD").is_empty() {
("root".to_string(), g("MYSQL_ROOT_PASSWORD"))
} else {
(g("MYSQL_USER"), g("MYSQL_PASSWORD"))
}
}
Engine::Mongo => (
g("MONGO_INITDB_ROOT_USERNAME"),
g("MONGO_INITDB_ROOT_PASSWORD"),
),
};
if user.is_empty() {
bail!("não achei o usuário de administração do banco antigo nas env vars");
}
Ok(Side {
host: host.to_string(),
port: svc.spec.port,
user,
pass,
})
}
fn dest_image(server: &Service) -> Result<String> {
let shared::ServiceSource::Compose(c) = &server.spec.source else {
bail!("servidor de destino não é Compose");
};
let host = crate::platform::connection::compose_host(
&c.content,
c.ingress_service.as_deref(),
server.spec.db_kind.as_deref(),
);
crate::platform::connection::compose_services(&c.content)
.into_iter()
.find(|(k, _)| Some(k) == host.as_ref())
.and_then(|(_, img)| img)
.ok_or_else(|| anyhow!("não achei a imagem do servidor de destino"))
}
pub fn image_version(image: &str) -> Option<(u32, u32)> {
let tag = image.rsplit_once(':')?.1;
let mut it = tag.split(|c: char| !c.is_ascii_digit());
let major = it.next()?.parse().ok()?;
let minor = it.next().and_then(|m| m.parse().ok()).unwrap_or(0);
Some((major, minor))
}
fn source_image(svc: &Service) -> Option<String> {
let shared::ServiceSource::Compose(c) = &svc.spec.source else {
return None;
};
let host = crate::platform::connection::compose_host(
&c.content,
c.ingress_service.as_deref(),
svc.spec.db_kind.as_deref(),
);
crate::platform::connection::compose_services(&c.content)
.into_iter()
.find(|(k, _)| Some(k) == host.as_ref())
.and_then(|(_, i)| i)
}
async fn dependents(
state: &MigrationContext,
source: &Service,
host: &str,
env_var: &str,
) -> Result<Vec<Service>> {
let mut out = vec![];
for s in store::services::list(&state.db, &source.spec.project_id).await? {
if s.id == source.id || s.spec.db_kind.is_some() {
continue;
}
if !matches!(
s.status,
shared::ServiceStatus::Running | shared::ServiceStatus::Degraded
) {
continue;
}
let env = env_resolve::resolve(&state.db, &state.secrets, &s).await?;
if env.iter().any(|(k, v)| k == env_var || v.contains(host)) {
out.push(s);
}
}
Ok(out)
}
async fn docker_run_pipe(
engine: Engine,
image: &str,
network: &str,
origem: &Side,
origem_db: &str,
destino: &Side,
destino_db: &str,
) -> Result<String> {
let mut cmd = crate::engine::cli();
cmd.args(["run", "--rm", "--network", network]);
let vars = [
("ORIGEM_HOST", origem.host.clone()),
("ORIGEM_PORT", origem.port.to_string()),
("ORIGEM_USER", origem.user.clone()),
("ORIGEM_PASS", origem.pass.clone()),
("ORIGEM_DB", origem_db.to_string()),
("DESTINO_HOST", destino.host.clone()),
("DESTINO_PORT", destino.port.to_string()),
("DESTINO_USER", destino.user.clone()),
("DESTINO_PASS", destino.pass.clone()),
("DESTINO_DB", destino_db.to_string()),
];
for (k, v) in &vars {
cmd.args(["-e", k]).env(k, v);
}
cmd.args(["--entrypoint", "bash", image, "-c", pipe_script(engine)]);
let out = cmd
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.output()
.await
.map_err(|e| anyhow!("falha ao executar docker: {e}"))?;
let text = format!(
"{}{}",
String::from_utf8_lossy(&out.stdout),
String::from_utf8_lossy(&out.stderr)
);
if !out.status.success() {
bail!("{}", text.trim());
}
Ok(text)
}
pub async fn run(state: MigrationContext, mut rec: Record) {
let mut cur = 0usize;
let res = run_inner(&state, &mut rec, &mut cur).await;
match res {
Ok(()) => {
rec.m.status = "Completed".into();
let _ = store::migration::save(&state.db, &rec).await;
info!(id = %rec.m.id, "migração concluída");
}
Err(e) => {
warn!(id = %rec.m.id, step = cur, error = %e, "migração falhou");
step(&state, &mut rec, cur, "failed", &e.to_string()).await;
for i in cur + 1..rec.m.steps.len() {
rec.m.steps[i].state = "skipped".into();
}
if let Err(u) = undo(&state, &mut rec, false).await {
log(
&state,
&mut rec,
&format!("rollback automático falhou: {u}"),
)
.await;
}
rec.m.status = "Failed".into();
let _ = store::migration::save(&state.db, &rec).await;
}
}
}
async fn run_inner(state: &MigrationContext, rec: &mut Record, cur: &mut usize) -> Result<()> {
let source = store::services::get(&state.db, &rec.m.source_service_id)
.await?
.ok_or_else(|| anyhow!("banco antigo não encontrado"))?;
let dest = store::managed_database::get(&state.db, &rec.m.dest_database_id)
.await?
.ok_or_else(|| anyhow!("database de destino não encontrado"))?;
let server = store::services::get(&state.db, &dest.server_service_id)
.await?
.ok_or_else(|| anyhow!("servidor de destino não encontrado"))?;
let engine = Engine::from_kind(source.spec.db_kind.as_deref())
.ok_or_else(|| anyhow!("motor não suportado"))?;
let network = store::projects::network_name(&state.db, &source.spec.project_id).await?;
let src_db = rec.m.source_database.clone();
*cur = 0;
step(state, rec, 0, "running", "").await;
if Engine::from_kind(server.spec.db_kind.as_deref()) != Some(engine) {
bail!(
"origem e destino precisam ser do mesmo motor (MySQL ↔ MariaDB também não é permitido)"
);
}
if let (Some(s), Some(d)) = (
source_image(&source).and_then(|i| image_version(&i)),
dest_image(&server).ok().and_then(|i| image_version(&i)),
) && d < s
{
bail!(
"o servidor de destino ({}.{}) é mais antigo que a origem ({}.{})",
d.0,
d.1,
s.0,
s.1
);
}
let src_cid = shared_net::server_container(&state.docker.inner, &source)
.await
.ok_or_else(|| anyhow!("o banco antigo não está no ar"))?;
let dst_cid = shared_net::server_container(&state.docker.inner, &server)
.await
.ok_or_else(|| anyhow!("o servidor compartilhado não está no ar"))?;
shared_db::wait_ready(engine, &dst_cid, 10).await?;
let (sh, stdin) = counts_plan(engine, &dest.name);
let dest_now = shared_db::exec_sh(engine, &dst_cid, &sh, &stdin).await?;
if !dest_now.trim().is_empty() {
bail!("o database de destino não está vazio; crie um novo ou apague o conteúdo");
}
let (sh, stdin) = counts_plan(engine, &src_db);
let env: HashMap<String, String> = env_resolve::resolve(&state.db, &state.secrets, &source)
.await?
.into_iter()
.collect();
let src_host = crate::platform::connection::compose_host(
match &source.spec.source {
shared::ServiceSource::Compose(c) => &c.content,
_ => "",
},
match &source.spec.source {
shared::ServiceSource::Compose(c) => c.ingress_service.as_deref(),
_ => None,
},
source.spec.db_kind.as_deref(),
)
.unwrap_or_else(|| shared::app_network_alias(&source.spec.name));
let _ = shared_db::exec_sh(engine, &src_cid, &sh, &stdin)
.await
.map_err(|e| anyhow!("não consegui ler o database de origem \"{src_db}\": {e}"))?;
step(
state,
rec,
0,
"ok",
"motor, versão, origem e destino conferidos; destino vazio",
)
.await;
*cur = 1;
step(state, rec, 1, "running", "").await;
let deps = dependents(state, &source, &src_host, &rec.m.env_var).await?;
for d in &deps {
rec.stopped_services.push(d.id.clone());
}
let _ = store::migration::save(&state.db, rec).await;
for d in &deps {
let r = crate::features::deploy::stop::handle(state.deploy.clone(), d.id.clone()).await;
if let shared::Response::Err { message, .. } = r {
bail!("falha ao parar {}: {message}", d.spec.name);
}
}
step(
state,
rec,
1,
"ok",
&format!("{} serviço(s) parado(s)", deps.len()),
)
.await;
let (sh, stdin) = counts_plan(engine, &src_db);
let counts_src = shared_db::exec_sh(engine, &src_cid, &sh, &stdin).await?;
*cur = 2;
step(state, rec, 2, "running", "").await;
let origem = source_side(engine, &source, &src_host, &env)?;
let destino = Side {
host: shared_net::alias_of(&server),
port: server.spec.port,
user: dest.name.clone(),
pass: dest.password.clone(),
};
let image = dest_image(&server)?;
let out = docker_run_pipe(
engine, &image, &network, &origem, &src_db, &destino, &dest.name,
)
.await?;
log(state, rec, &out).await;
step(state, rec, 2, "ok", "").await;
*cur = 3;
step(state, rec, 3, "running", "").await;
let (sh, stdin) = counts_plan(engine, &dest.name);
let counts_dst = shared_db::exec_sh(engine, &dst_cid, &sh, &stdin).await?;
let norm = |s: &str| -> Vec<String> {
let mut v: Vec<String> = s
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.map(String::from)
.collect();
v.sort();
v
};
let (a, b) = (norm(&counts_src), norm(&counts_dst));
if a != b {
log(
state,
rec,
&format!("origem:\n{}\ndestino:\n{}", a.join("\n"), b.join("\n")),
)
.await;
bail!("contagens diferentes entre origem e destino (veja o log)");
}
step(
state,
rec,
3,
"ok",
&format!("{} tabela(s)/coleção(ões) conferidas", a.len()),
)
.await;
*cur = 4;
step(state, rec, 4, "running", "").await;
let view = super::managed_view::to_view(&server, engine, &dest);
rec.env_changes = env_switch::switch(
&state.db,
&rec.m.project_id,
&rec.m.env_var,
&view.connection_url,
)
.await?;
step(
state,
rec,
4,
"ok",
&format!(
"{} aponta para o banco novo (valor antigo guardado)",
rec.m.env_var
),
)
.await;
*cur = 5;
step(state, rec, 5, "running", "").await;
for id in rec.stopped_services.clone() {
let r = crate::features::deploy::start::handle(state.deploy.clone(), id.clone()).await;
if let shared::Response::Err { message, .. } = r {
bail!("falha ao subir {id}: {message}");
}
}
step(
state,
rec,
5,
"ok",
"redeploy enfileirado; acompanhe o healthcheck de cada serviço",
)
.await;
*cur = 6;
step(state, rec, 6, "running", "").await;
let r = crate::features::deploy::stop::handle(state.deploy.clone(), source.id.clone()).await;
if let shared::Response::Err { message, .. } = r {
bail!("falha ao parar o banco antigo: {message}");
}
step(
state,
rec,
6,
"ok",
"banco antigo parado, mantido para rollback",
)
.await;
Ok(())
}
pub async fn undo(state: &MigrationContext, rec: &mut Record, source_stopped: bool) -> Result<()> {
if !rec.env_changes.is_empty() {
env_switch::restore(&state.db, &rec.m.project_id, &rec.env_changes).await?;
rec.env_changes.clear();
}
if source_stopped {
let r = crate::features::deploy::start::handle(
state.deploy.clone(),
rec.m.source_service_id.clone(),
)
.await;
if let shared::Response::Err { message, .. } = r {
bail!("falha ao subir o banco antigo: {message}");
}
}
for id in rec.stopped_services.clone() {
let _ = crate::features::deploy::start::handle(state.deploy.clone(), id).await;
}
rec.stopped_services.clear();
let _ = store::migration::save(&state.db, rec).await;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn versao_da_imagem() {
assert_eq!(image_version("postgres:18"), Some((18, 0)));
assert_eq!(image_version("mysql:8.4"), Some((8, 4)));
assert_eq!(image_version("mariadb:11.4.2"), Some((11, 4)));
assert_eq!(image_version("postgres:latest"), None);
assert_eq!(image_version("postgres"), None);
}
#[test]
fn nome_de_origem_seguro() {
assert!(valid_source_db("db_2024"));
for bad in ["", "a b", "a;b", "a'b", "a`b", "a-b"] {
assert!(!valid_source_db(bad), "{bad}");
}
}
#[test]
fn scripts_sem_senha_em_argumento() {
for e in [
Engine::Postgres,
Engine::MySql,
Engine::MariaDb,
Engine::Mongo,
] {
let s = pipe_script(e);
assert!(s.starts_with("set -euo pipefail"));
assert!(!s.contains("--password") && !s.contains("-p\"$"), "{e:?}");
}
assert!(pipe_script(Engine::MariaDb).contains("mariadb-dump"));
assert!(!pipe_script(Engine::MariaDb).contains("set-gtid-purged"));
}
}