use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use crate::context::DjogiContext;
use crate::error::DjogiError;
use super::guard::WorkspaceGuard;
use super::ledger::{
self, ExecutionMode, LedgerRow, LedgerStatus, SHA256_HEX_LEN, compute_checksum,
};
use super::naming::{down_filename, up_filename};
use super::projection::BucketKey;
use super::replay_plan::committed_replay_plan_filename;
use super::reset::{
ResetSqlSide, compute_committed_down_sql_checksum, compute_committed_sql_checksum,
};
use super::schema::SNAPSHOT_FORMAT_VERSION;
use super::target::bucket_dir;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttuneMode {
DiffOnly,
Record {
reason: String,
},
Squash {
from: String,
publish: bool,
app: Option<String>,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AttuneEntry {
pub bucket: BucketKey,
pub version: String,
pub kind: AttuneEntryKind,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AttuneEntryKind {
Unrecorded,
Orphaned,
Recorded,
Squashed,
}
impl AttuneEntryKind {
pub fn as_str(self) -> &'static str {
match self {
AttuneEntryKind::Unrecorded => "unrecorded",
AttuneEntryKind::Orphaned => "orphaned",
AttuneEntryKind::Recorded => "recorded",
AttuneEntryKind::Squashed => "squashed",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AttuneReport {
pub entries: Vec<AttuneEntry>,
pub mutated: bool,
pub squashed_to: Option<String>,
pub published: bool,
pub resolved_target: Option<String>,
pub parent_pointer_updated: bool,
pub diagnostics: Vec<AttuneDiagnostic>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttuneDiagnostic {
LedgerTableMissing { database: String },
DryRunMutationsSkipped { mode: &'static str },
DryRunRecordSkipped { resolved_target: Option<String> },
DryRunSquashRecordSkipped { resolved_target: Option<String> },
}
impl AttuneDiagnostic {
pub fn code(&self) -> &'static str {
match self {
AttuneDiagnostic::LedgerTableMissing { .. } => "ATTUNE-001",
AttuneDiagnostic::DryRunMutationsSkipped { .. } => "ATTUNE-002",
AttuneDiagnostic::DryRunRecordSkipped { .. }
| AttuneDiagnostic::DryRunSquashRecordSkipped { .. } => "ATTUNE-003",
}
}
}
impl std::fmt::Display for AttuneDiagnostic {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AttuneDiagnostic::LedgerTableMissing { database } => write!(
f,
"[{}] ledger table `djogi_schema_migrations` does not exist in database \
`{database}`; ledger not bootstrapped — run `migrations apply` or \
`attune --record --apply` first (Codex umbrella round-3 U-8: plain \
`attune --record` without `--apply` is dry-run after U-5 and will not \
bootstrap; the apply flag is load-bearing)",
self.code(),
),
AttuneDiagnostic::DryRunMutationsSkipped { mode } => write!(
f,
"[{}] {mode} mode requested without `--apply`; diff reported, \
no database or disk mutation happened — re-run with `--apply` to commit",
self.code(),
),
AttuneDiagnostic::DryRunRecordSkipped { resolved_target } => match resolved_target {
Some(sha) => write!(
f,
"[{}] `--record` was requested without `--apply`; parent submodule \
pointer would be updated to `{sha}` but no parent index mutation \
happened — re-run with `--apply` to commit",
self.code(),
),
None => write!(
f,
"[{}] `--record` was requested without `--apply`; no parent submodule \
pointer would be updated because no target was resolved — pass \
`--target <ref>` to populate, then re-run with `--apply` to commit",
self.code(),
),
},
AttuneDiagnostic::DryRunSquashRecordSkipped { resolved_target } => {
match resolved_target {
Some(sha) => write!(
f,
"[{}] would update parent submodule pointer to `{sha}` but `--apply` \
was not provided; no parent index mutation happened — re-run with \
`--apply` to commit",
self.code(),
),
None => write!(
f,
"[{}] no parent submodule pointer would be updated because no target \
was resolved; pass `--target <ref>` to populate, then re-run with \
`--apply` to commit",
self.code(),
),
}
}
}
}
}
#[derive(Debug)]
pub enum AttuneError {
Refused(AttuneRefusal),
FilesystemScanFailed { source: std::io::Error },
LedgerQueryFailed { source: DjogiError },
SqlReadFailed {
path: PathBuf,
source: std::io::Error,
},
SqlWriteFailed {
path: PathBuf,
source: std::io::Error,
},
SqlDeleteFailed {
path: PathBuf,
source: std::io::Error,
},
GitPublishFailed {
stderr: String,
status_code: Option<i32>,
},
GitTargetResolveFailed {
target: String,
stderr: String,
status_code: Option<i32>,
},
GitFetchFailed {
stderr: String,
status_code: Option<i32>,
},
GitUpdateSubmodulePointerFailed {
new_sha: String,
stderr: String,
status_code: Option<i32>,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttuneRefusal {
SquashNotLocalhost { database_url: String },
SquashNotDevProfile { profile: String },
SquashDevModeOff,
SquashEnvIsProduction { env_value: String },
SquashFromVersionNotFound { version: String },
SquashFromVersionAmbiguous {
version: String,
buckets: Vec<String>,
},
}
impl std::fmt::Display for AttuneError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AttuneError::Refused(r) => write!(f, "attune refused: {r}"),
AttuneError::FilesystemScanFailed { source } => {
write!(f, "attune filesystem scan failed: {source}")
}
AttuneError::LedgerQueryFailed { source } => {
write!(f, "attune ledger query failed: {source}")
}
AttuneError::SqlReadFailed { path, source } => write!(
f,
"attune could not read SQL file at {}: {source}",
path.display()
),
AttuneError::SqlWriteFailed { path, source } => write!(
f,
"attune could not write squashed SQL at {}: {source}",
path.display()
),
AttuneError::SqlDeleteFailed { path, source } => write!(
f,
"attune could not delete subsumed SQL at {}: {source}",
path.display()
),
AttuneError::GitPublishFailed {
stderr,
status_code,
} => match status_code {
Some(c) => write!(
f,
"attune --publish: git push exited with status {c}: {stderr}"
),
None => write!(
f,
"attune --publish: git push terminated by signal: {stderr}"
),
},
AttuneError::GitTargetResolveFailed {
target,
stderr,
status_code,
} => match status_code {
Some(c) => write!(
f,
"attune target `{target}`: git rev-parse failed locally and after \
a remote fetch retry (status {c}): {stderr}; either the target \
does not exist on any configured remote, or `migrations/` has no \
remote at all"
),
None => write!(
f,
"attune target `{target}`: git rev-parse terminated by signal: {stderr}"
),
},
AttuneError::GitFetchFailed {
stderr,
status_code,
} => match status_code {
Some(c) => write!(
f,
"attune target resolution: git fetch failed (status {c}): {stderr}"
),
None => write!(
f,
"attune target resolution: git fetch terminated by signal: {stderr}"
),
},
AttuneError::GitUpdateSubmodulePointerFailed {
new_sha,
stderr,
status_code,
} => match status_code {
Some(c) => write!(
f,
"attune --record: failed to update parent repo's submodule \
pointer to `{new_sha}` (status {c}): {stderr}"
),
None => write!(
f,
"attune --record: parent submodule pointer update terminated \
by signal: {stderr}"
),
},
}
}
}
impl std::fmt::Display for AttuneRefusal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
AttuneRefusal::SquashNotLocalhost { database_url } => write!(
f,
"attune --squash refuses to run when DATABASE_URL is not localhost \
(got `{database_url}`); squash is a destructive history rewrite and \
must not be invoked against shared / production databases"
),
AttuneRefusal::SquashNotDevProfile { profile } => write!(
f,
"attune --squash refuses to run with profile=`{profile}`; squash \
is dev-only — set `profile = \"development\"` (or remove the \
production override) before retrying"
),
AttuneRefusal::SquashDevModeOff => f.write_str(
"attune --squash refuses to run when `[database].dev_mode = false` \
in Djogi.toml; squash is a destructive history rewrite and the \
`dev_mode` flag is the explicit opt-in. Set `dev_mode = true` in \
the `[database]` block before retrying",
),
AttuneRefusal::SquashEnvIsProduction { env_value } => write!(
f,
"attune --squash refuses to run when DJOGI_ENV=`{env_value}` \
(case-insensitive `production`); the deployment-environment \
signal overrides any local Djogi.toml profile override. Unset \
DJOGI_ENV (or set it to a non-production value) before retrying"
),
AttuneRefusal::SquashFromVersionNotFound { version } => write!(
f,
"attune --squash --from `{version}` did not match any version on disk \
in the connected database; list `migrations/<database>/<app>/` to find \
a valid starting version"
),
AttuneRefusal::SquashFromVersionAmbiguous { version, buckets } => write!(
f,
"attune --squash --from `{version}` exists in multiple buckets ({}); \
squash refuses to silently pick one — each bucket's history is \
independent. Disambiguate by passing `--app <app_label>` to scope the \
squash to a single bucket",
buckets.join(", ")
),
}
}
}
impl std::error::Error for AttuneError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
AttuneError::FilesystemScanFailed { source } => Some(source),
AttuneError::LedgerQueryFailed { source } => Some(source),
AttuneError::SqlReadFailed { source, .. } => Some(source),
AttuneError::SqlWriteFailed { source, .. } => Some(source),
AttuneError::SqlDeleteFailed { source, .. } => Some(source),
AttuneError::GitTargetResolveFailed { .. }
| AttuneError::GitFetchFailed { .. }
| AttuneError::GitUpdateSubmodulePointerFailed { .. } => None,
_ => None,
}
}
}
pub struct AttuneRequest<'a> {
pub workspace_root: &'a Path,
pub database_url: &'a str,
pub profile: &'a str,
pub dev_mode: bool,
pub target: Option<&'a str>,
pub apply: bool,
pub record: bool,
pub mode: AttuneMode,
pub _guard: &'a WorkspaceGuard,
}
fn djogi_env_is_production() -> Option<String> {
let raw = std::env::var("DJOGI_ENV").ok()?;
if super::policy::ascii_eq_ignore_case(raw.as_bytes(), b"production") {
Some(raw)
} else {
None
}
}
fn push_record_skipped(
diagnostics: &mut Vec<AttuneDiagnostic>,
req: &AttuneRequest<'_>,
resolved_target: &Option<String>,
) {
if resolved_target.is_none() {
return;
}
if req.record {
diagnostics.push(AttuneDiagnostic::DryRunRecordSkipped {
resolved_target: resolved_target.clone(),
});
} else if matches!(req.mode, AttuneMode::Squash { .. }) {
diagnostics.push(AttuneDiagnostic::DryRunSquashRecordSkipped {
resolved_target: resolved_target.clone(),
});
}
}
pub async fn attune(
ctx: &mut DjogiContext,
req: AttuneRequest<'_>,
) -> Result<AttuneReport, AttuneError> {
if let AttuneMode::Squash { .. } = &req.mode {
if !super::policy::is_localhost_connection(req.database_url) {
return Err(AttuneError::Refused(AttuneRefusal::SquashNotLocalhost {
database_url: req.database_url.to_string(),
}));
}
if req.profile == "production" {
return Err(AttuneError::Refused(AttuneRefusal::SquashNotDevProfile {
profile: req.profile.to_string(),
}));
}
if !req.dev_mode {
return Err(AttuneError::Refused(AttuneRefusal::SquashDevModeOff));
}
if let Some(env_value) = djogi_env_is_production() {
return Err(AttuneError::Refused(AttuneRefusal::SquashEnvIsProduction {
env_value,
}));
}
}
let active_database = active_database_name(ctx).await?;
let ledger_exists = match (&req.mode, req.apply) {
(AttuneMode::DiffOnly, _)
| (AttuneMode::Record { .. }, false)
| (AttuneMode::Squash { .. }, false) => ledger_table_exists(ctx).await?,
(AttuneMode::Record { .. }, true) | (AttuneMode::Squash { .. }, true) => {
ledger::bootstrap(ctx)
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
true
}
};
let disk = scan_disk_for_database(req.workspace_root, &active_database)?;
let ledger_versions = if ledger_exists {
scan_ledger(ctx, &active_database).await?
} else {
BTreeMap::new()
};
let mut entries: Vec<AttuneEntry> = Vec::new();
for (bucket, versions) in &disk {
let ledger_for_bucket = ledger_versions.get(bucket).cloned().unwrap_or_default();
for version in versions.keys() {
if !ledger_for_bucket.contains_key(version) {
entries.push(AttuneEntry {
bucket: bucket.clone(),
version: version.clone(),
kind: AttuneEntryKind::Unrecorded,
});
}
}
}
for (bucket, versions) in &ledger_versions {
let disk_for_bucket = disk.get(bucket).cloned().unwrap_or_default();
for version in versions.keys() {
if !disk_for_bucket.contains_key(version) {
entries.push(AttuneEntry {
bucket: bucket.clone(),
version: version.clone(),
kind: AttuneEntryKind::Orphaned,
});
}
}
}
let mut diagnostics: Vec<AttuneDiagnostic> = Vec::new();
if !ledger_exists {
diagnostics.push(AttuneDiagnostic::LedgerTableMissing {
database: active_database.clone(),
});
}
let resolved_target = match req.target {
Some(t) if !t.is_empty() => Some(resolve_git_target(req.workspace_root, t)?),
_ => None,
};
let mut report = AttuneReport {
entries: Vec::new(),
mutated: false,
squashed_to: None,
published: false,
resolved_target: resolved_target.clone(),
parent_pointer_updated: false,
diagnostics,
};
let apply = req.apply;
let effective_record = req.record || matches!(req.mode, AttuneMode::Squash { .. });
match &req.mode {
AttuneMode::DiffOnly => {
entries.sort_by(|a, b| sort_key(a).cmp(&sort_key(b)));
report.entries = entries;
if !apply {
push_record_skipped(&mut report.diagnostics, &req, &resolved_target);
}
if req.record
&& apply
&& let Some(sha) = resolved_target.as_ref()
{
update_parent_submodule_pointer(req.workspace_root, sha)?;
report.parent_pointer_updated = true;
}
Ok(report)
}
AttuneMode::Record { reason } => {
let mut out: Vec<AttuneEntry> = Vec::new();
for entry in entries {
match entry.kind {
AttuneEntryKind::Unrecorded => {
let path = disk
.get(&entry.bucket)
.and_then(|m| m.get(&entry.version))
.cloned();
if let Some(path) = path {
if apply {
insert_recorded_row(
ctx,
&entry.bucket,
&entry.version,
&path,
reason,
)
.await?;
report.mutated = true;
out.push(AttuneEntry {
kind: AttuneEntryKind::Recorded,
..entry
});
continue;
}
out.push(AttuneEntry {
kind: AttuneEntryKind::Unrecorded,
..entry
});
continue;
}
out.push(entry);
}
_ => out.push(entry),
}
}
out.sort_by(|a, b| sort_key(a).cmp(&sort_key(b)));
report.entries = out;
if !apply {
report
.diagnostics
.push(AttuneDiagnostic::DryRunMutationsSkipped { mode: "Record" });
push_record_skipped(&mut report.diagnostics, &req, &resolved_target);
} else if req.record {
if let Some(sha) = &resolved_target {
update_parent_submodule_pointer(req.workspace_root, sha)?;
report.parent_pointer_updated = true;
}
}
Ok(report)
}
AttuneMode::Squash { from, publish, app } => {
if !apply {
entries.sort_by(|a, b| sort_key(a).cmp(&sort_key(b)));
report.entries = entries;
report
.diagnostics
.push(AttuneDiagnostic::DryRunMutationsSkipped { mode: "Squash" });
if effective_record {
push_record_skipped(&mut report.diagnostics, &req, &resolved_target);
}
return Ok(report);
}
run_squash(
ctx,
req.workspace_root,
from,
*publish,
app.as_deref(),
&disk,
&mut report,
entries,
)
.await?;
if effective_record && let Some(sha) = &resolved_target {
update_parent_submodule_pointer(req.workspace_root, sha)?;
report.parent_pointer_updated = true;
}
Ok(report)
}
}
}
fn scan_disk_filtered(
workspace_root: &Path,
database_filter: Option<&str>,
) -> Result<BTreeMap<BucketKey, BTreeMap<String, PathBuf>>, AttuneError> {
super::target::scan_filesystem_with_files(workspace_root, database_filter)
.map_err(|source| AttuneError::FilesystemScanFailed { source })
}
#[cfg(test)]
fn scan_disk(
workspace_root: &Path,
) -> Result<BTreeMap<BucketKey, BTreeMap<String, PathBuf>>, AttuneError> {
scan_disk_filtered(workspace_root, None)
}
fn scan_disk_for_database(
workspace_root: &Path,
database: &str,
) -> Result<BTreeMap<BucketKey, BTreeMap<String, PathBuf>>, AttuneError> {
scan_disk_filtered(workspace_root, Some(database))
}
async fn scan_ledger(
ctx: &mut DjogiContext,
database: &str,
) -> Result<BTreeMap<BucketKey, BTreeMap<String, LedgerStatus>>, AttuneError> {
let rows = ctx
.query_all(
"SELECT version, app_label, status \
FROM djogi_schema_migrations \
ORDER BY app_label, version",
&[],
)
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
let mut out: BTreeMap<BucketKey, BTreeMap<String, LedgerStatus>> = BTreeMap::new();
for row in &rows {
let version: String = row.try_get(0).map_err(|e| AttuneError::LedgerQueryFailed {
source: DjogiError::from(e),
})?;
let app_label: String = row.try_get(1).map_err(|e| AttuneError::LedgerQueryFailed {
source: DjogiError::from(e),
})?;
let status_s: String = row.try_get(2).map_err(|e| AttuneError::LedgerQueryFailed {
source: DjogiError::from(e),
})?;
let status = LedgerStatus::from_db_str(&status_s).unwrap_or(LedgerStatus::Failed);
if !matches!(
status,
LedgerStatus::Applied | LedgerStatus::Faked | LedgerStatus::Baseline
) {
continue;
}
let bucket = BucketKey {
database: database.to_string(),
app: app_label,
};
out.entry(bucket).or_default().insert(version, status);
}
Ok(out)
}
async fn active_database_name(ctx: &mut DjogiContext) -> Result<String, AttuneError> {
let row = ctx
.query_one("SELECT current_database()::text", &[])
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
let name: String = row.try_get(0).map_err(|e| AttuneError::LedgerQueryFailed {
source: DjogiError::from(e),
})?;
Ok(name)
}
async fn ledger_table_exists(ctx: &mut DjogiContext) -> Result<bool, AttuneError> {
let row = ctx
.query_one(
"SELECT EXISTS(SELECT 1 FROM pg_class \
WHERE relname = 'djogi_schema_migrations' AND relkind = 'r')",
&[],
)
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
let exists: bool = row.try_get(0).map_err(|e| AttuneError::LedgerQueryFailed {
source: DjogiError::from(e),
})?;
Ok(exists)
}
async fn insert_recorded_row(
ctx: &mut DjogiContext,
bucket: &BucketKey,
version: &str,
up_path: &Path,
reason: &str,
) -> Result<(), AttuneError> {
let up_sql = std::fs::read_to_string(up_path).map_err(|e| AttuneError::SqlReadFailed {
path: up_path.to_path_buf(),
source: e,
})?;
let checksum_up = compute_committed_sql_checksum(&up_sql, ResetSqlSide::Up);
let down_path = up_path
.parent()
.map(|p| p.join(super::naming::down_filename(version)));
let checksum_down = down_path
.as_ref()
.and_then(|p| std::fs::read_to_string(p).ok())
.and_then(|down_sql| compute_committed_down_sql_checksum(&down_sql));
let _ = SHA256_HEX_LEN;
let timestamp = time::OffsetDateTime::now_utc()
.format(&time::format_description::well_known::Rfc3339)
.unwrap_or_else(|_| "<unknown>".to_string());
let note = format!("attune --record at {timestamp}: {reason}");
let row = LedgerRow {
version: version.to_string(),
description: format!("<attune --record> {version}"),
checksum_up,
checksum_down,
execution_mode: ExecutionMode::Transactional,
status: LedgerStatus::Applied,
execution_time_ms: 0,
out_of_order_flag: false,
applied_steps_count: 0,
total_steps: None,
partial_apply_note: Some(note),
run_id: 0, snapshot_version: SNAPSHOT_FORMAT_VERSION.to_string(),
app_label: bucket.app.clone(),
leaf_identity: None,
};
let ledger_id = ledger::insert_pending(ctx, &row)
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
ctx.execute(
"UPDATE djogi_schema_migrations SET status = 'applied' WHERE id = $1",
&[&ledger_id],
)
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
let _ = down_filename(version); let _ = up_filename(version);
Ok(())
}
#[allow(clippy::too_many_arguments)]
async fn run_squash(
ctx: &mut DjogiContext,
workspace_root: &Path,
from: &str,
publish: bool,
app_filter: Option<&str>,
disk: &BTreeMap<BucketKey, BTreeMap<String, PathBuf>>,
report: &mut AttuneReport,
diff_entries: Vec<AttuneEntry>,
) -> Result<(), AttuneError> {
let mut entries = diff_entries;
let bucket = locate_squash_target(disk, from, app_filter)?.clone();
let versions = disk.get(&bucket).expect("target_bucket came from disk map");
let to_squash: Vec<(&String, &PathBuf)> = versions
.iter()
.filter(|(v, _)| v.as_str() >= from)
.collect();
if to_squash.len() <= 1 {
entries.sort_by(|a, b| sort_key(a).cmp(&sort_key(b)));
report.entries = entries;
return Ok(());
}
let dir = bucket_dir(workspace_root, &bucket);
let (combined_up, combined_down_segments) = build_combined_sql(&dir, &to_squash)?;
let mut combined_down = String::new();
for seg in combined_down_segments.iter().rev() {
combined_down.push_str(seg);
combined_down.push('\n');
}
let wrote_down = !combined_down.is_empty();
let new_up_path = dir.join(up_filename(from));
std::fs::write(&new_up_path, combined_up.as_bytes()).map_err(|e| {
AttuneError::SqlWriteFailed {
path: new_up_path.clone(),
source: e,
}
})?;
if wrote_down {
let new_down_path = dir.join(down_filename(from));
std::fs::write(&new_down_path, combined_down.as_bytes()).map_err(|e| {
AttuneError::SqlWriteFailed {
path: new_down_path.clone(),
source: e,
}
})?;
}
delete_replay_plan_sidecar_if_exists(&dir, from)?;
delete_subsumed(ctx, &dir, &to_squash, from, &bucket, &mut entries).await?;
refresh_retained_row(
ctx,
from,
&bucket,
&combined_up,
if wrote_down {
Some(combined_down.as_str())
} else {
None
},
)
.await?;
report.mutated = true;
report.squashed_to = Some(from.to_string());
if publish && report.mutated {
run_git_commit_and_publish(workspace_root, from)?;
report.published = true;
}
entries.sort_by(|a, b| sort_key(a).cmp(&sort_key(b)));
report.entries = entries;
Ok(())
}
fn locate_squash_target<'a>(
disk: &'a BTreeMap<BucketKey, BTreeMap<String, PathBuf>>,
from: &str,
app_filter: Option<&str>,
) -> Result<&'a BucketKey, AttuneError> {
let mut matching: Vec<&BucketKey> = Vec::new();
for (bucket, versions) in disk {
if let Some(label) = app_filter
&& bucket.app.as_str() != label
{
continue;
}
if versions.contains_key(from) {
matching.push(bucket);
}
}
match matching.len() {
0 => Err(AttuneError::Refused(
AttuneRefusal::SquashFromVersionNotFound {
version: from.to_string(),
},
)),
1 => Ok(matching[0]),
_ => {
let mut buckets_rendered: Vec<String> = matching
.iter()
.map(|b| {
let app_render = if b.app.is_empty() { "_global_" } else { &b.app };
format!("{}/{}", b.database, app_render)
})
.collect();
buckets_rendered.sort();
Err(AttuneError::Refused(
AttuneRefusal::SquashFromVersionAmbiguous {
version: from.to_string(),
buckets: buckets_rendered,
},
))
}
}
}
fn delete_replay_plan_sidecar_if_exists(dir: &Path, version: &str) -> Result<(), AttuneError> {
let path = dir.join(committed_replay_plan_filename(version));
match std::fs::remove_file(&path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(AttuneError::SqlDeleteFailed { path, source: e }),
}
}
fn build_combined_sql(
dir: &Path,
to_squash: &[(&String, &PathBuf)],
) -> Result<(String, Vec<String>), AttuneError> {
let mut combined_up = String::new();
let mut combined_down_segments: Vec<String> = Vec::new();
for (version, path) in to_squash {
let up_sql = std::fs::read_to_string(path).map_err(|e| AttuneError::SqlReadFailed {
path: (*path).clone(),
source: e,
})?;
combined_up.push_str(&format!("-- begin {version}\n"));
combined_up.push_str(&up_sql);
if !up_sql.ends_with('\n') {
combined_up.push('\n');
}
combined_up.push_str(&format!("-- end {version}\n\n"));
let down_path = dir.join(down_filename(version));
if let Ok(down_sql) = std::fs::read_to_string(&down_path) {
combined_down_segments.push(format!(
"-- begin {version} (reverse)\n{down_sql}\n-- end {version}\n",
));
}
}
Ok((combined_up, combined_down_segments))
}
async fn delete_subsumed(
ctx: &mut DjogiContext,
dir: &Path,
to_squash: &[(&String, &PathBuf)],
from: &str,
bucket: &BucketKey,
entries: &mut Vec<AttuneEntry>,
) -> Result<(), AttuneError> {
for (version, path) in to_squash {
if version.as_str() == from {
continue;
}
std::fs::remove_file(path).map_err(|e| AttuneError::SqlDeleteFailed {
path: (*path).clone(),
source: e,
})?;
let down = dir.join(down_filename(version));
if down.exists() {
std::fs::remove_file(&down).map_err(|e| AttuneError::SqlDeleteFailed {
path: down.clone(),
source: e,
})?;
}
delete_replay_plan_sidecar_if_exists(dir, version)?;
ctx.execute(
"DELETE FROM djogi_schema_migrations WHERE version = $1 AND app_label = $2",
&[version, &bucket.app],
)
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
entries.push(AttuneEntry {
bucket: bucket.clone(),
version: (*version).clone(),
kind: AttuneEntryKind::Squashed,
});
}
Ok(())
}
async fn refresh_retained_row(
ctx: &mut DjogiContext,
from: &str,
bucket: &BucketKey,
combined_up: &str,
combined_down: Option<&str>,
) -> Result<(), AttuneError> {
let new_checksum_up = compute_checksum([combined_up]);
let new_checksum_down = combined_down.and_then(compute_committed_down_sql_checksum);
let prior_description: Option<String> = ctx
.query_one(
"SELECT description FROM djogi_schema_migrations \
WHERE version = $1 AND app_label = $2",
&[&from.to_string(), &bucket.app],
)
.await
.ok()
.and_then(|row| row.try_get::<_, String>(0).ok());
let new_description = match prior_description {
Some(prev) => format!("squashed migration starting at {from} (was: {prev})"),
None => format!("squashed migration starting at {from}"),
};
ctx.execute(
"UPDATE djogi_schema_migrations \
SET checksum_up = $1, checksum_down = $2, description = $3 \
WHERE version = $4 AND app_label = $5",
&[
&new_checksum_up,
&new_checksum_down,
&new_description,
&from.to_string(),
&bucket.app,
],
)
.await
.map_err(|e| AttuneError::LedgerQueryFailed { source: e })?;
Ok(())
}
fn run_git_commit_and_publish(workspace_root: &Path, from: &str) -> Result<(), AttuneError> {
let migrations_root = super::target::migrations_root(workspace_root);
let commit_msg = format!("djogi attune --squash from {from}");
let head_subject_out = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("log")
.arg("-1")
.arg("--pretty=%s")
.output()
.map_err(|e| AttuneError::GitPublishFailed {
stderr: format!("failed to spawn `git log`: {e}"),
status_code: None,
})?;
let head_subject = String::from_utf8_lossy(head_subject_out.stderr_or_stdout())
.trim()
.to_string();
let commit_already_landed = head_subject_out.status.success() && head_subject == commit_msg;
if !commit_already_landed {
let add_out = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("add")
.arg("-A")
.output()
.map_err(|e| AttuneError::GitPublishFailed {
stderr: format!("failed to spawn `git add`: {e}"),
status_code: None,
})?;
if !add_out.status.success() {
let stderr = String::from_utf8_lossy(&add_out.stderr).into_owned();
return Err(AttuneError::GitPublishFailed {
stderr: format!("`git add -A` failed: {stderr}"),
status_code: add_out.status.code(),
});
}
let commit_out = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("commit")
.arg("-m")
.arg(&commit_msg)
.output()
.map_err(|e| AttuneError::GitPublishFailed {
stderr: format!("failed to spawn `git commit`: {e}"),
status_code: None,
})?;
if !commit_out.status.success() {
let stderr = String::from_utf8_lossy(&commit_out.stderr).into_owned();
return Err(AttuneError::GitPublishFailed {
stderr: format!("`git commit` failed: {stderr}"),
status_code: commit_out.status.code(),
});
}
}
let remote_check = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("config")
.arg("--get")
.arg("remote.origin.url")
.output()
.map_err(|e| AttuneError::GitPublishFailed {
stderr: format!("failed to probe remote.origin.url: {e}"),
status_code: None,
})?;
if !remote_check.status.success() {
return Err(AttuneError::GitPublishFailed {
stderr: "remote `origin` is not configured on the migrations submodule; \
set a remote with `git -C migrations remote add origin <url>` \
before retrying `--publish`"
.to_string(),
status_code: remote_check.status.code(),
});
}
let push_out = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("push")
.arg("origin")
.arg("HEAD")
.output()
.map_err(|e| AttuneError::GitPublishFailed {
stderr: format!("failed to spawn `git push`: {e}"),
status_code: None,
})?;
if !push_out.status.success() {
let stderr = String::from_utf8_lossy(&push_out.stderr).into_owned();
return Err(AttuneError::GitPublishFailed {
stderr: format!(
"`git push origin HEAD` failed: {stderr}\n\
Hint: the squash commit IS on the local migrations branch; \
retry with `attune --squash --publish` once the push issue \
is resolved (auth, network, etc.)."
),
status_code: push_out.status.code(),
});
}
Ok(())
}
trait OutputReadStdoutOrStderr {
fn stderr_or_stdout(&self) -> &[u8];
}
impl OutputReadStdoutOrStderr for std::process::Output {
fn stderr_or_stdout(&self) -> &[u8] {
if self.status.success() {
&self.stdout
} else {
&self.stderr
}
}
}
fn resolve_git_target(workspace_root: &Path, target: &str) -> Result<String, AttuneError> {
let migrations_root = super::target::migrations_root(workspace_root);
let local_arg = format!("{target}^{{commit}}");
let local = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("rev-parse")
.arg("--verify")
.arg(&local_arg)
.output()
.map_err(|e| AttuneError::GitTargetResolveFailed {
target: target.to_string(),
stderr: format!("failed to spawn `git rev-parse`: {e}"),
status_code: None,
})?;
if local.status.success() {
return Ok(String::from_utf8_lossy(&local.stdout).trim().to_string());
}
let fetch = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("fetch")
.arg("--all")
.arg("--tags")
.output()
.map_err(|e| AttuneError::GitFetchFailed {
stderr: format!("failed to spawn `git fetch`: {e}"),
status_code: None,
})?;
if !fetch.status.success() {
return Err(AttuneError::GitFetchFailed {
stderr: String::from_utf8_lossy(&fetch.stderr).into_owned(),
status_code: fetch.status.code(),
});
}
let retry = std::process::Command::new("git")
.arg("-C")
.arg(&migrations_root)
.arg("rev-parse")
.arg("--verify")
.arg(&local_arg)
.output()
.map_err(|e| AttuneError::GitTargetResolveFailed {
target: target.to_string(),
stderr: format!("failed to spawn `git rev-parse` (retry): {e}"),
status_code: None,
})?;
if retry.status.success() {
return Ok(String::from_utf8_lossy(&retry.stdout).trim().to_string());
}
Err(AttuneError::GitTargetResolveFailed {
target: target.to_string(),
stderr: String::from_utf8_lossy(&retry.stderr).into_owned(),
status_code: retry.status.code(),
})
}
fn update_parent_submodule_pointer(
workspace_root: &Path,
new_sha: &str,
) -> Result<(), AttuneError> {
let cacheinfo = format!("160000,{new_sha},migrations");
let out = std::process::Command::new("git")
.arg("-C")
.arg(workspace_root)
.arg("update-index")
.arg("--cacheinfo")
.arg(&cacheinfo)
.output()
.map_err(|e| AttuneError::GitUpdateSubmodulePointerFailed {
new_sha: new_sha.to_string(),
stderr: format!("failed to spawn `git update-index`: {e}"),
status_code: None,
})?;
if !out.status.success() {
return Err(AttuneError::GitUpdateSubmodulePointerFailed {
new_sha: new_sha.to_string(),
stderr: String::from_utf8_lossy(&out.stderr).into_owned(),
status_code: out.status.code(),
});
}
Ok(())
}
fn sort_key(e: &AttuneEntry) -> (String, String, String, &'static str) {
(
e.bucket.database.clone(),
e.bucket.app.clone(),
e.version.clone(),
e.kind.as_str(),
)
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
use std::sync::atomic::{AtomicUsize, Ordering};
fn temp_root(tag: &str) -> PathBuf {
static COUNTER: AtomicUsize = AtomicUsize::new(0);
let n = COUNTER.fetch_add(1, Ordering::SeqCst);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_nanos();
let p = std::env::temp_dir().join(format!("djogi-attune-{tag}-{nanos}-{n}"));
fs::create_dir_all(&p).unwrap();
p
}
#[test]
fn scan_disk_picks_up_only_up_files() {
let root = temp_root("scan_disk_up_only");
let dir = root.join("migrations/main/billing");
fs::create_dir_all(&dir).unwrap();
fs::write(
dir.join("V20260425010203__init.sdjql"),
"CREATE TABLE foo();",
)
.unwrap();
fs::write(
dir.join("V20260425010203__init.down.sdjql"),
"DROP TABLE foo;",
)
.unwrap();
fs::write(dir.join("README.md"), "noop").unwrap();
let scanned = scan_disk(&root).expect("scan ok");
let bucket = BucketKey {
database: "main".to_string(),
app: "billing".to_string(),
};
let versions = scanned.get(&bucket).expect("billing bucket");
assert_eq!(versions.len(), 1, "down + readme must be ignored");
assert!(versions.contains_key("V20260425010203__init"));
let _ = fs::remove_dir_all(&root);
}
#[test]
fn squash_sidecar_cleanup_removes_committed_replay_plan() {
let root = temp_root("squash_sidecar_cleanup");
let version = "V20260425010203__init";
let dir = root.join("migrations/main/billing");
fs::create_dir_all(&dir).unwrap();
let sidecar = dir.join(committed_replay_plan_filename(version));
fs::write(&sidecar, "{}").unwrap();
delete_replay_plan_sidecar_if_exists(&dir, version).expect("delete sidecar");
assert!(
!sidecar.exists(),
"squash must not leave stale replay manifests beside rewritten SQL"
);
delete_replay_plan_sidecar_if_exists(&dir, version).expect("missing sidecar is ok");
let _ = fs::remove_dir_all(&root);
}
#[test]
fn scan_disk_groups_by_bucket() {
let root = temp_root("scan_disk_buckets");
fs::create_dir_all(root.join("migrations/main/billing")).unwrap();
fs::create_dir_all(root.join("migrations/main/_global_")).unwrap();
fs::write(
root.join("migrations/main/billing/V20260101000001__init.sdjql"),
"",
)
.unwrap();
fs::write(
root.join("migrations/main/_global_/V20260101000002__init.sdjql"),
"",
)
.unwrap();
let scanned = scan_disk(&root).expect("scan ok");
assert_eq!(scanned.len(), 2);
let global = BucketKey {
database: "main".to_string(),
app: "".to_string(),
};
let billing = BucketKey {
database: "main".to_string(),
app: "billing".to_string(),
};
assert!(scanned.contains_key(&global), "global bucket present");
assert!(scanned.contains_key(&billing), "billing bucket present");
let _ = fs::remove_dir_all(&root);
}
#[test]
fn sort_key_is_stable() {
let a = AttuneEntry {
bucket: BucketKey {
database: "main".to_string(),
app: "billing".to_string(),
},
version: "V20260101000001__init".to_string(),
kind: AttuneEntryKind::Unrecorded,
};
let b = AttuneEntry {
bucket: BucketKey {
database: "main".to_string(),
app: "billing".to_string(),
},
version: "V20260201000001__add".to_string(),
kind: AttuneEntryKind::Unrecorded,
};
assert!(sort_key(&a) < sort_key(&b));
}
#[test]
fn entry_kind_as_str_is_lowercase() {
assert_eq!(AttuneEntryKind::Unrecorded.as_str(), "unrecorded");
assert_eq!(AttuneEntryKind::Orphaned.as_str(), "orphaned");
assert_eq!(AttuneEntryKind::Recorded.as_str(), "recorded");
assert_eq!(AttuneEntryKind::Squashed.as_str(), "squashed");
}
#[test]
fn refusal_displays_actionable_message() {
let r = AttuneRefusal::SquashNotLocalhost {
database_url: "postgres://prod.example.com/main".to_string(),
};
let s = format!("{r}");
assert!(s.contains("postgres://prod.example.com/main"));
assert!(s.contains("not localhost"));
let r2 = AttuneRefusal::SquashNotDevProfile {
profile: "production".to_string(),
};
let s2 = format!("{r2}");
assert!(s2.contains("production"));
assert!(s2.contains("dev-only"));
}
#[test]
fn u2_squash_dev_mode_off_message_names_field() {
let r = AttuneRefusal::SquashDevModeOff;
let s = format!("{r}");
assert!(s.contains("dev_mode"), "must name the dev_mode field: {s}");
assert!(
s.contains("Djogi.toml") || s.contains("[database]"),
"must hint at the config location: {s}"
);
}
#[test]
fn u2_squash_env_is_production_message_names_env_value() {
let r = AttuneRefusal::SquashEnvIsProduction {
env_value: "production".to_string(),
};
let s = format!("{r}");
assert!(s.contains("DJOGI_ENV"), "must name the env-var: {s}");
assert!(s.contains("production"), "must echo the value: {s}");
}
#[test]
fn u1_dry_run_diagnostic_codes_are_stable() {
assert_eq!(
AttuneDiagnostic::DryRunMutationsSkipped { mode: "Record" }.code(),
"ATTUNE-002"
);
assert_eq!(
AttuneDiagnostic::DryRunRecordSkipped {
resolved_target: None,
}
.code(),
"ATTUNE-003"
);
assert_eq!(
AttuneDiagnostic::DryRunSquashRecordSkipped {
resolved_target: None,
}
.code(),
"ATTUNE-003"
);
}
#[test]
fn u1_dry_run_diagnostic_messages_are_actionable() {
let d = AttuneDiagnostic::DryRunMutationsSkipped { mode: "Squash" };
let s = d.to_string();
assert!(s.contains("Squash"), "must name mode: {s}");
assert!(s.contains("--apply"), "must hint at remediation: {s}");
let d_with = AttuneDiagnostic::DryRunRecordSkipped {
resolved_target: Some("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef".to_string()),
};
let s_with = d_with.to_string();
assert!(
s_with.contains("deadbeef"),
"must echo resolved SHA: {s_with}"
);
assert!(s_with.contains("--apply"));
assert!(
s_with.contains("parent submodule pointer would be updated to"),
"must describe skipped parent-pointer update: {s_with}"
);
assert!(
s_with.contains("`--record` was requested without `--apply`"),
"explicit record case must name the causal flag: {s_with}"
);
let d_none = AttuneDiagnostic::DryRunRecordSkipped {
resolved_target: None,
};
let s_none = d_none.to_string();
assert!(
s_none.contains("no parent submodule pointer would be updated"),
"must describe the unresolved-target outcome accurately: {s_none}"
);
assert!(s_none.contains("--target <ref>"));
assert!(
s_none.contains("`--record` was requested without `--apply`"),
"explicit record case must still name the causal flag when unresolved: {s_none}"
);
}
#[test]
fn dry_run_squash_record_skipped_none_wording_is_neutral() {
let s = AttuneDiagnostic::DryRunSquashRecordSkipped {
resolved_target: None,
}
.to_string();
assert!(
s.contains("no parent submodule pointer would be updated"),
"must describe the unresolved-target outcome accurately: {s}"
);
assert!(s.contains("--target <ref>"));
assert!(s.contains("--apply"));
assert!(
!s.contains("`--record`"),
"squash-implied wording must NOT name --record: {s}"
);
}
#[test]
fn u9_record_source_drives_diagnostic_wording() {
let explicit = AttuneDiagnostic::DryRunRecordSkipped {
resolved_target: Some("abcd".to_string()),
}
.to_string();
let squash = AttuneDiagnostic::DryRunSquashRecordSkipped {
resolved_target: Some("abcd".to_string()),
}
.to_string();
assert!(
explicit.contains(
"`--record` was requested without `--apply`; parent submodule pointer \
would be updated to `abcd`"
),
"direct --record wording must name the explicit flag: {explicit}"
);
assert!(
squash.contains(
"would update parent submodule pointer to `abcd` but `--apply` was not \
provided; no parent index mutation happened"
),
"squash-implied wording must match neutral prose: {squash}"
);
assert!(
explicit.contains("--record"),
"direct --record wording must mention the explicit flag: {explicit}"
);
assert!(
!squash.contains("--record requested"),
"squash-implied wording must NOT say --record requested: {squash}"
);
}
#[test]
fn attune_diagnostic_dry_run_record_skipped_api_stability() {
let diagnostic = AttuneDiagnostic::DryRunRecordSkipped {
resolved_target: Some("abc123".to_string()),
};
let resolved_target = match diagnostic.clone() {
AttuneDiagnostic::DryRunRecordSkipped { resolved_target } => resolved_target,
AttuneDiagnostic::LedgerTableMissing { .. }
| AttuneDiagnostic::DryRunMutationsSkipped { .. }
| AttuneDiagnostic::DryRunSquashRecordSkipped { .. } => {
panic!("expected DryRunRecordSkipped")
}
};
assert_eq!(resolved_target.as_deref(), Some("abc123"));
assert_eq!(
diagnostic.to_string(),
"[ATTUNE-003] `--record` was requested without `--apply`; parent submodule \
pointer would be updated to `abc123` but no parent index mutation happened \
— re-run with `--apply` to commit"
);
}
#[test]
fn u1_git_target_resolve_failed_message_names_target() {
let e = AttuneError::GitTargetResolveFailed {
target: "feature/branch".to_string(),
stderr: "fatal: ambiguous argument".to_string(),
status_code: Some(128),
};
let s = e.to_string();
assert!(s.contains("feature/branch"));
assert!(s.contains("128"));
assert!(s.contains("rev-parse"));
}
#[test]
fn u1_git_update_submodule_pointer_failed_message_names_sha() {
let e = AttuneError::GitUpdateSubmodulePointerFailed {
new_sha: "deadbeef".to_string(),
stderr: "fatal: refusing to overwrite".to_string(),
status_code: Some(128),
};
let s = e.to_string();
assert!(s.contains("deadbeef"));
assert!(s.contains("--record"));
assert!(s.contains("submodule pointer"));
}
#[test]
fn u2_djogi_env_is_production_predicate_matches_only_exact_word() {
let prior = std::env::var("DJOGI_ENV").ok();
unsafe { std::env::remove_var("DJOGI_ENV") };
assert_eq!(djogi_env_is_production(), None);
unsafe { std::env::set_var("DJOGI_ENV", "production") };
assert_eq!(djogi_env_is_production(), Some("production".to_string()));
unsafe { std::env::set_var("DJOGI_ENV", "Production") };
assert_eq!(djogi_env_is_production(), Some("Production".to_string()));
unsafe { std::env::set_var("DJOGI_ENV", "PRODUCTION") };
assert_eq!(djogi_env_is_production(), Some("PRODUCTION".to_string()));
unsafe { std::env::set_var("DJOGI_ENV", "prod") };
assert_eq!(djogi_env_is_production(), None);
unsafe { std::env::set_var("DJOGI_ENV", "productionx") };
assert_eq!(djogi_env_is_production(), None);
unsafe { std::env::set_var("DJOGI_ENV", "development") };
assert_eq!(djogi_env_is_production(), None);
unsafe { std::env::set_var("DJOGI_ENV", "staging") };
assert_eq!(djogi_env_is_production(), None);
unsafe { std::env::set_var("DJOGI_ENV", "") };
assert_eq!(djogi_env_is_production(), None);
match prior {
Some(v) => unsafe { std::env::set_var("DJOGI_ENV", v) },
None => unsafe { std::env::remove_var("DJOGI_ENV") },
}
}
}