1use std::fs;
2use std::path::PathBuf;
3
4use anyhow::{Context, Result};
5use serde::{Deserialize, Serialize};
6use sha2::{Digest, Sha256};
7
8use crate::approvals::ApprovalRecord;
9use crate::config::{
10 get_connection_policy, get_environment_kind, is_protected_environment, EnvironmentKind,
11 MongoConfig,
12};
13use crate::connections::ConnectionPolicy;
14use crate::core::sync::SyncConfig;
15use crate::storage;
16
17const PLAN_VERSION: u8 = 1;
18
19#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
20#[serde(rename_all = "snake_case")]
21pub enum PlanStatus {
22 Planned,
23 Approved,
24 Running,
25 Completed,
26 Failed,
27}
28
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct SyncPlanRecord {
31 pub version: u8,
32 pub id: String,
33 pub created_at: String,
34 pub updated_at: String,
35 pub status: PlanStatus,
36 pub hash: String,
37 pub policy_hash: String,
38 pub config: SyncConfig,
39 pub source_kind: EnvironmentKind,
40 pub target_kind: EnvironmentKind,
41 pub source_protected: bool,
42 pub target_protected: bool,
43 pub source_policy: ConnectionPolicy,
44 pub target_policy: ConnectionPolicy,
45 pub destructive: bool,
46 pub requires_full_backup: bool,
47 pub requires_human_approval: bool,
48 pub warnings: Vec<String>,
49}
50
51#[derive(Serialize)]
52struct PlanHashPayload<'a> {
53 version: u8,
54 config: &'a SyncConfig,
55 source_kind: EnvironmentKind,
56 target_kind: EnvironmentKind,
57 source_protected: bool,
58 target_protected: bool,
59 source_policy: &'a ConnectionPolicy,
60 target_policy: &'a ConnectionPolicy,
61 destructive: bool,
62 requires_full_backup: bool,
63 requires_human_approval: bool,
64 warnings: &'a [String],
65}
66
67#[derive(Serialize)]
68struct PolicyHashPayload<'a> {
69 source_policy: &'a ConnectionPolicy,
70 target_policy: &'a ConnectionPolicy,
71 source_kind: EnvironmentKind,
72 target_kind: EnvironmentKind,
73 source_protected: bool,
74 target_protected: bool,
75}
76
77pub fn create_sync_plan(mut config: SyncConfig) -> Result<SyncPlanRecord> {
78 config.options.update_collection_settings();
79 crate::utils::mongodb::validate_db_name(&config.source_db)?;
80 crate::utils::mongodb::validate_db_name(&config.target_db)?;
81 let _source_config = MongoConfig::from_env(config.source_env.clone()).with_context(|| {
82 format!(
83 "Source connection '{}' is not configured",
84 config.source_env
85 )
86 })?;
87 let _target_config = MongoConfig::from_env(config.target_env.clone()).with_context(|| {
88 format!(
89 "Target connection '{}' is not configured",
90 config.target_env
91 )
92 })?;
93
94 let source_kind = get_environment_kind(&config.source_env);
95 let target_kind = get_environment_kind(&config.target_env);
96 let source_protected = is_protected_environment(&config.source_env);
97 let target_protected = is_protected_environment(&config.target_env);
98 let source_policy = get_connection_policy(&config.source_env);
99 let target_policy = get_connection_policy(&config.target_env);
100 let destructive = config.options.is_destructive();
101 let requires_full_backup = destructive && target_policy.destructive_requires_backup;
102 let requires_human_approval = target_policy.human_approval_required;
103
104 if !source_policy.allow_as_source {
105 anyhow::bail!(
106 "Connection '{}' is not allowed as a sync source",
107 config.source_env
108 );
109 }
110 if !target_policy.allow_as_target {
111 anyhow::bail!(
112 "Connection '{}' is not allowed as a sync target",
113 config.target_env
114 );
115 }
116 if requires_full_backup && !config.options.create_backup {
117 anyhow::bail!(
118 "Refusing destructive sync to protected/production target '{}:{}' without a full backup. Set --backup true.",
119 config.target_env,
120 config.target_db
121 );
122 }
123
124 let mut warnings = Vec::new();
125 if config.source_env == config.target_env {
126 warnings.push("Source and target connections are the same".to_string());
127 }
128 if destructive {
129 warnings.push("Target collections will be dropped or cleared before import".to_string());
130 }
131 if target_protected || target_kind.is_prod() {
132 warnings.push("Target is protected/production".to_string());
133 }
134 if requires_human_approval {
135 warnings.push("Human OS approval is required before execution".to_string());
136 }
137
138 let now = chrono::Utc::now().to_rfc3339();
139 let mut plan = SyncPlanRecord {
140 version: PLAN_VERSION,
141 id: generate_plan_id(),
142 created_at: now.clone(),
143 updated_at: now,
144 status: PlanStatus::Planned,
145 hash: String::new(),
146 policy_hash: String::new(),
147 config,
148 source_kind,
149 target_kind,
150 source_protected,
151 target_protected,
152 source_policy,
153 target_policy,
154 destructive,
155 requires_full_backup,
156 requires_human_approval,
157 warnings,
158 };
159 plan.hash = compute_plan_hash(&plan)?;
160 plan.policy_hash = compute_policy_hash(&plan)?;
161 Ok(plan)
162}
163
164pub fn save_plan(plan: &SyncPlanRecord) -> Result<()> {
165 storage::atomic_write_json(&plan_path(&plan.id), plan)
166}
167
168pub fn create_and_save_sync_plan(config: SyncConfig) -> Result<SyncPlanRecord> {
169 let plan = create_sync_plan(config)?;
170 save_plan(&plan)?;
171 Ok(plan)
172}
173
174pub fn load_plan(id: &str) -> Result<SyncPlanRecord> {
175 let path = plan_path(id);
176 let contents = fs::read_to_string(&path)
177 .with_context(|| format!("Failed to read plan {} from {}", id, path.display()))?;
178 let plan: SyncPlanRecord =
179 serde_json::from_str(&contents).with_context(|| format!("Failed to parse plan {}", id))?;
180 let expected_hash = compute_plan_hash(&plan)?;
181 if plan.hash != expected_hash {
182 anyhow::bail!(
183 "Plan '{}' hash mismatch. Expected {}, found {}. Refusing to use possibly edited plan.",
184 id,
185 expected_hash,
186 plan.hash
187 );
188 }
189 Ok(plan)
190}
191
192pub fn list_plans() -> Result<Vec<SyncPlanRecord>> {
193 let dir = storage::plans_dir();
194 if !dir.exists() {
195 return Ok(Vec::new());
196 }
197
198 let mut plans = Vec::new();
199 for entry in fs::read_dir(&dir).with_context(|| format!("Failed to read {}", dir.display()))? {
200 let entry = entry?;
201 let path = entry.path().join("plan.json");
202 if path.exists() {
203 let contents = fs::read_to_string(&path)?;
204 let plan: SyncPlanRecord = serde_json::from_str(&contents)?;
205 plans.push(plan);
206 }
207 }
208 plans.sort_by(|a, b| b.created_at.cmp(&a.created_at));
209 Ok(plans)
210}
211
212pub fn update_plan_status(id: &str, status: PlanStatus) -> Result<SyncPlanRecord> {
213 let mut plan = load_plan(id)?;
214 plan.status = status;
215 plan.updated_at = chrono::Utc::now().to_rfc3339();
216 save_plan(&plan)?;
217 Ok(plan)
218}
219
220pub fn save_approval(plan_id: &str, approval: &ApprovalRecord) -> Result<()> {
221 storage::atomic_write_json(&approval_path(plan_id), approval)
222}
223
224pub fn load_approval(plan_id: &str) -> Result<Option<ApprovalRecord>> {
225 let path = approval_path(plan_id);
226 if !path.exists() {
227 return Ok(None);
228 }
229 let contents = fs::read_to_string(&path)
230 .with_context(|| format!("Failed to read approval from {}", path.display()))?;
231 let approval = serde_json::from_str(&contents)
232 .with_context(|| format!("Failed to parse approval for plan {plan_id}"))?;
233 Ok(Some(approval))
234}
235
236pub fn render_plan_text(plan: &SyncPlanRecord) -> String {
237 let mut text = String::new();
238 text.push_str(&format!("Plan: {}\n", plan.id));
239 text.push_str(&format!("Status: {:?}\n", plan.status));
240 text.push_str(&format!("Hash: {}\n", plan.hash));
241 text.push_str(&format!(
242 "Source: {}:{} ({})\n",
243 plan.config.source_env, plan.config.source_db, plan.source_kind
244 ));
245 text.push_str(&format!(
246 "Target: {}:{} ({})\n",
247 plan.config.target_env, plan.config.target_db, plan.target_kind
248 ));
249 text.push_str(&format!("Target protected: {}\n", plan.target_protected));
250 text.push_str(&format!("Backup: {}\n", plan.config.options.create_backup));
251 text.push_str(&format!("Drop: {}\n", plan.config.options.drop_collections));
252 text.push_str(&format!(
253 "Clear: {}\n",
254 plan.config.options.clear_collections
255 ));
256 text.push_str(&format!(
257 "Requires human approval: {}\n",
258 plan.requires_human_approval
259 ));
260 text.push_str(&format!(
261 "Requires full backup: {}\n",
262 plan.requires_full_backup
263 ));
264 if !plan.warnings.is_empty() {
265 text.push_str("Warnings:\n");
266 for warning in &plan.warnings {
267 text.push_str(&format!(" - {warning}\n"));
268 }
269 }
270 text
271}
272
273pub fn render_plan_markdown(plan: &SyncPlanRecord) -> String {
274 let mut md = String::new();
275 md.push_str(&format!("# Arcula sync plan `{}`\n\n", plan.id));
276 md.push_str(&format!("- **Status:** `{:?}`\n", plan.status));
277 md.push_str(&format!("- **Plan hash:** `{}`\n", plan.hash));
278 md.push_str(&format!(
279 "- **Source:** `{}` / `{}` / `{}`\n",
280 plan.config.source_env, plan.config.source_db, plan.source_kind
281 ));
282 md.push_str(&format!(
283 "- **Target:** `{}` / `{}` / `{}`\n",
284 plan.config.target_env, plan.config.target_db, plan.target_kind
285 ));
286 md.push_str(&format!(
287 "- **Target protected:** `{}`\n",
288 plan.target_protected
289 ));
290 md.push_str(&format!(
291 "- **Backup:** `{}`\n",
292 plan.config.options.create_backup
293 ));
294 md.push_str(&format!(
295 "- **Drop:** `{}`\n",
296 plan.config.options.drop_collections
297 ));
298 md.push_str(&format!(
299 "- **Clear:** `{}`\n",
300 plan.config.options.clear_collections
301 ));
302 md.push_str(&format!(
303 "- **Requires human approval:** `{}`\n",
304 plan.requires_human_approval
305 ));
306 md.push_str(&format!(
307 "- **Requires full backup:** `{}`\n",
308 plan.requires_full_backup
309 ));
310 if !plan.warnings.is_empty() {
311 md.push_str("\n## Warnings\n\n");
312 for warning in &plan.warnings {
313 md.push_str(&format!("- {warning}\n"));
314 }
315 }
316 md
317}
318
319pub fn plan_dir(id: &str) -> PathBuf {
320 storage::plans_dir().join(id)
321}
322
323fn plan_path(id: &str) -> PathBuf {
324 plan_dir(id).join("plan.json")
325}
326
327fn approval_path(plan_id: &str) -> PathBuf {
328 plan_dir(plan_id).join("approval.json")
329}
330
331fn compute_plan_hash(plan: &SyncPlanRecord) -> Result<String> {
332 let payload = PlanHashPayload {
333 version: plan.version,
334 config: &plan.config,
335 source_kind: plan.source_kind,
336 target_kind: plan.target_kind,
337 source_protected: plan.source_protected,
338 target_protected: plan.target_protected,
339 source_policy: &plan.source_policy,
340 target_policy: &plan.target_policy,
341 destructive: plan.destructive,
342 requires_full_backup: plan.requires_full_backup,
343 requires_human_approval: plan.requires_human_approval,
344 warnings: &plan.warnings,
345 };
346 hash_json(&payload)
347}
348
349fn compute_policy_hash(plan: &SyncPlanRecord) -> Result<String> {
350 let payload = PolicyHashPayload {
351 source_policy: &plan.source_policy,
352 target_policy: &plan.target_policy,
353 source_kind: plan.source_kind,
354 target_kind: plan.target_kind,
355 source_protected: plan.source_protected,
356 target_protected: plan.target_protected,
357 };
358 hash_json(&payload)
359}
360
361fn hash_json<T: Serialize>(value: &T) -> Result<String> {
362 let bytes = serde_json::to_vec(value)?;
363 let mut hasher = Sha256::new();
364 hasher.update(bytes);
365 Ok(to_hex(&hasher.finalize()))
366}
367
368fn generate_plan_id() -> String {
369 let timestamp = chrono::Utc::now().format("%Y%m%d%H%M%S");
370 let suffix: u32 = rand::random();
371 format!("sync_{timestamp}_{suffix:08x}")
372}
373
374fn to_hex(bytes: &[u8]) -> String {
375 const HEX: &[u8; 16] = b"0123456789abcdef";
376 let mut output = String::with_capacity(bytes.len() * 2);
377 for byte in bytes {
378 output.push(HEX[(byte >> 4) as usize] as char);
379 output.push(HEX[(byte & 0x0f) as usize] as char);
380 }
381 output
382}