Skip to main content

pgevolve_core/plan/
deserialize.rs

1//! Readers for the three on-disk plan files. See spec §7.
2//!
3//! `plan.sql` is parsed by a small line-based scanner that recognizes the
4//! `-- @pgevolve ...` directive lines and groups intervening lines into the
5//! preceding step's SQL body. `intent.toml` and `manifest.toml` are
6//! deserialized through `serde` / `toml` directly; manifest's embedded
7//! `target_snapshot_yaml` round-trips through `serde_yaml`.
8
9use std::path::Path;
10
11use serde::Deserialize;
12use time::OffsetDateTime;
13use time::format_description::well_known::Rfc3339;
14
15use crate::identifier::{Identifier, QualifiedName};
16use crate::ir::catalog::Catalog;
17use crate::plan::grouping::TransactionGroup;
18use crate::plan::io_error::PlanIoError;
19use crate::plan::plan::{
20    DestructiveIntent, LintWaiver, Plan, PlanId, PlanMetadata, RecordedFinding, StepOverride,
21    parse_kind_name,
22};
23use crate::plan::raw_step::{RawStep, StepKind, TransactionConstraint};
24
25// ---------------------------------------------------------------------------
26// plan.sql
27// ---------------------------------------------------------------------------
28
29/// Loosely-typed view of a parsed `plan.sql`. Final `Plan` assembly happens in
30/// [`read_plan_dir`], which cross-references this with `intent.toml` and
31/// `manifest.toml`.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct PartialPlan {
34    /// `plan id=` value — the 16-char short hash.
35    pub id_short: String,
36    /// `version=` value.
37    pub pgevolve_version: String,
38    /// `ruleset=` value.
39    pub planner_ruleset_version: u32,
40    /// `source_rev=` value, if present.
41    pub source_rev: Option<String>,
42    /// `target=` value (opaque target identity).
43    pub target_identity: String,
44    /// `intents_required=` value.
45    pub intents_required: u32,
46    /// Parsed `created=` timestamp.
47    pub created_at: OffsetDateTime,
48    /// Recovered groups (each step's `destructive_reason` is `None` here; the
49    /// reason lives in `intent.toml` and is grafted on in [`read_plan_dir`]).
50    pub groups: Vec<TransactionGroup>,
51}
52
53/// Parse `plan.sql` from a string.
54#[allow(clippy::too_many_lines)]
55pub fn read_plan_sql(s: &str) -> Result<PartialPlan, PlanIoError> {
56    let mut id_short: Option<String> = None;
57    let mut version: Option<String> = None;
58    let mut ruleset: Option<u32> = None;
59    let mut created: Option<OffsetDateTime> = None;
60    let mut source_rev: Option<String> = None;
61    let mut target_identity: Option<String> = None;
62    let mut intents_required: Option<u32> = None;
63
64    let mut groups: Vec<TransactionGroup> = Vec::new();
65    let mut current_group: Option<TransactionGroup> = None;
66    let mut current_step: Option<(RawStep, Vec<String>)> = None;
67    // Tracks the active dollar-quote tag while we're inside a `$tag$...$tag$`
68    // literal in a step's SQL body. Function/procedure bodies are emitted
69    // wrapped in `$pgevolve$...$pgevolve$` and may contain literal
70    // `-- @pgevolve dep: ...` directives, `BEGIN;`, `COMMIT;`, etc. — all of
71    // which would otherwise be mistaken for plan-level directives by the
72    // line scanner. Track open/close and treat in-string lines as body text.
73    let mut active_dollar_tag: Option<String> = None;
74
75    for raw_line in s.lines() {
76        let line = raw_line;
77        let trimmed = line.trim_end();
78
79        // If we're inside a dollar-quote, accumulate body text and only check
80        // for the matching close tag; do not parse directives.
81        if let Some(tag) = active_dollar_tag.as_ref() {
82            if let Some((_, ref mut body)) = current_step {
83                body.push(line.to_string());
84            }
85            // Check if this line closes the dollar-quote.
86            let close = format!("${tag}$");
87            if line.contains(&close) {
88                active_dollar_tag = None;
89            }
90            continue;
91        }
92
93        if trimmed.is_empty() {
94            // Blank lines are layout-only between directives; they end the
95            // current step's SQL body if one is accumulating.
96            continue;
97        }
98
99        // Pre-scan for dollar-quote OPEN on this line. The emitter writes
100        // `AS $pgevolve$...` followed by the body and matching close. We need
101        // to enter dollar-quote state if the line opens one and the close
102        // isn't on the same line.
103        if let Some(tag) = detect_dollar_quote_open(line) {
104            if let Some((_, ref mut body)) = current_step {
105                body.push(line.to_string());
106            }
107            active_dollar_tag = Some(tag);
108            continue;
109        }
110
111        if let Some(rest) = trimmed.strip_prefix("-- @pgevolve ") {
112            // Finalize any in-flight step before starting a new directive.
113            flush_step(&mut current_step, &mut current_group);
114
115            let kv = parse_kv(rest)?;
116            let head = kv
117                .first()
118                .ok_or_else(|| PlanIoError::MalformedDirective(rest.into()))?;
119            match head.0.as_str() {
120                "plan" => {
121                    for (k, v) in &kv {
122                        match k.as_str() {
123                            "plan" | "id" => id_short = Some(v.clone()),
124                            "version" => version = Some(v.clone()),
125                            "ruleset" => ruleset = Some(parse_u32(v)?),
126                            "created" => {
127                                created =
128                                    Some(OffsetDateTime::parse(v, &Rfc3339).map_err(|e| {
129                                        PlanIoError::MalformedDirective(format!("created={v}: {e}"))
130                                    })?);
131                            }
132                            _ => {}
133                        }
134                    }
135                }
136                "source_rev" => source_rev = Some(head.1.clone()),
137                "target" => target_identity = Some(head.1.clone()),
138                "intents_required" => intents_required = Some(parse_u32(&head.1)?),
139                "group" => {
140                    if let Some(g) = current_group.take() {
141                        groups.push(g);
142                    }
143                    let mut id = 0u32;
144                    let mut transactional = true;
145                    for (k, v) in &kv {
146                        match k.as_str() {
147                            "id" => id = parse_u32(v)?,
148                            "transactional" => transactional = parse_bool(v)?,
149                            // "group" sentinel head and any unknown key fall through.
150                            _ => {}
151                        }
152                    }
153                    current_group = Some(TransactionGroup {
154                        id,
155                        transactional,
156                        steps: Vec::new(),
157                    });
158                }
159                "step" => {
160                    let mut step_no = 0u32;
161                    let mut kind: Option<StepKind> = None;
162                    let mut destructive = false;
163                    let mut intent_id: Option<u32> = None;
164                    let mut targets: Vec<QualifiedName> = Vec::new();
165                    for (k, v) in &kv {
166                        match k.as_str() {
167                            "step" => step_no = parse_u32(v)?,
168                            "kind" => {
169                                kind = Some(parse_kind_name(v).ok_or_else(|| {
170                                    PlanIoError::MalformedDirective(format!("kind={v}"))
171                                })?);
172                            }
173                            "destructive" => destructive = parse_bool(v)?,
174                            "intent_id" => intent_id = Some(parse_u32(v)?),
175                            "targets" => targets = parse_targets(v)?,
176                            _ => {}
177                        }
178                    }
179                    let kind = kind.ok_or_else(|| {
180                        PlanIoError::MalformedDirective("step missing kind".into())
181                    })?;
182                    let g = current_group.as_ref().ok_or_else(|| {
183                        PlanIoError::MalformedDirective("step outside group".into())
184                    })?;
185                    let transactional = if g.transactional {
186                        TransactionConstraint::InTransaction
187                    } else {
188                        TransactionConstraint::OutsideTransaction
189                    };
190                    let step = RawStep {
191                        step_no,
192                        kind,
193                        destructive,
194                        destructive_reason: None,
195                        intent_id,
196                        targets,
197                        sql: String::new(),
198                        transactional,
199                    };
200                    current_step = Some((step, Vec::new()));
201                }
202                other => {
203                    return Err(PlanIoError::MalformedDirective(format!(
204                        "unknown directive: {other}"
205                    )));
206                }
207            }
208            continue;
209        }
210
211        // Non-directive lines are either BEGIN/COMMIT or step SQL body lines.
212        if trimmed == "BEGIN;" {
213            // Group framing; not part of any step's SQL.
214            continue;
215        }
216        if trimmed == "COMMIT;" {
217            flush_step(&mut current_step, &mut current_group);
218            if let Some(g) = current_group.take() {
219                groups.push(g);
220            }
221            continue;
222        }
223        if let Some((_, ref mut body)) = current_step {
224            body.push(line.to_string());
225        }
226        // Lines outside any step (e.g., blank padding) are silently dropped.
227    }
228
229    // Flush any trailing step / group at EOF.
230    flush_step(&mut current_step, &mut current_group);
231    if let Some(g) = current_group.take() {
232        groups.push(g);
233    }
234
235    Ok(PartialPlan {
236        id_short: id_short
237            .ok_or_else(|| PlanIoError::MalformedDirective("missing plan id".into()))?,
238        pgevolve_version: version
239            .ok_or_else(|| PlanIoError::MalformedDirective("missing version".into()))?,
240        planner_ruleset_version: ruleset
241            .ok_or_else(|| PlanIoError::MalformedDirective("missing ruleset".into()))?,
242        source_rev,
243        target_identity: target_identity
244            .ok_or_else(|| PlanIoError::MalformedDirective("missing target".into()))?,
245        intents_required: intents_required
246            .ok_or_else(|| PlanIoError::MalformedDirective("missing intents_required".into()))?,
247        created_at: created
248            .ok_or_else(|| PlanIoError::MalformedDirective("missing created".into()))?,
249        groups,
250    })
251}
252
253/// If `line` opens a `$tag$` dollar-quoted literal that isn't closed on the
254/// same line, return the tag (excluding the surrounding `$` markers). Returns
255/// `None` for lines with no open quote, or lines whose dollar-quotes are fully
256/// closed on the same line.
257///
258/// Scans left-to-right looking for `$<tag>$` markers (tag = empty or
259/// `[A-Za-z_][A-Za-z0-9_]*` per PG's dollar-quote grammar). Each occurrence
260/// toggles "inside-quote" state; if we end the line still inside, return the
261/// active tag.
262fn detect_dollar_quote_open(line: &str) -> Option<String> {
263    let bytes = line.as_bytes();
264    let mut i = 0;
265    let mut active: Option<String> = None;
266    while i < bytes.len() {
267        if bytes[i] == b'$' {
268            // Scan tag chars.
269            let tag_start = i + 1;
270            let mut j = tag_start;
271            while j < bytes.len() {
272                let b = bytes[j];
273                let valid = if j == tag_start {
274                    b.is_ascii_alphabetic() || b == b'_'
275                } else {
276                    b.is_ascii_alphanumeric() || b == b'_'
277                };
278                if !valid {
279                    break;
280                }
281                j += 1;
282            }
283            if j < bytes.len() && bytes[j] == b'$' {
284                let tag = std::str::from_utf8(&bytes[tag_start..j])
285                    .unwrap_or("")
286                    .to_string();
287                if let Some(open) = active.as_ref() {
288                    if open == &tag {
289                        active = None;
290                    }
291                } else {
292                    active = Some(tag);
293                }
294                i = j + 1;
295                continue;
296            }
297        }
298        i += 1;
299    }
300    active
301}
302
303fn flush_step(
304    current_step: &mut Option<(RawStep, Vec<String>)>,
305    current_group: &mut Option<TransactionGroup>,
306) {
307    if let Some((mut step, body)) = current_step.take() {
308        step.sql = body.join("\n").trim_end_matches('\n').to_string();
309        if let Some(g) = current_group.as_mut() {
310            g.steps.push(step);
311        }
312    }
313}
314
315fn parse_kv(s: &str) -> Result<Vec<(String, String)>, PlanIoError> {
316    // Tokens are whitespace-separated; each token is `key=value`. The first
317    // token may be a bare `head` keyword (e.g., `plan`, `group`, `step`) — we
318    // treat that as `(head, "")` so callers can look it up uniformly.
319    let mut out = Vec::new();
320    let mut tokens = s.split_whitespace();
321    if let Some(first) = tokens.next() {
322        if let Some((k, v)) = first.split_once('=') {
323            out.push((k.to_string(), v.to_string()));
324        } else {
325            out.push((first.to_string(), String::new()));
326        }
327    }
328    for t in tokens {
329        let (k, v) = t
330            .split_once('=')
331            .ok_or_else(|| PlanIoError::MalformedDirective(format!("bare token: {t}")))?;
332        out.push((k.to_string(), v.to_string()));
333    }
334    Ok(out)
335}
336
337fn parse_u32(s: &str) -> Result<u32, PlanIoError> {
338    s.parse()
339        .map_err(|_| PlanIoError::MalformedDirective(format!("not a u32: {s}")))
340}
341
342fn parse_bool(s: &str) -> Result<bool, PlanIoError> {
343    match s {
344        "true" => Ok(true),
345        "false" => Ok(false),
346        _ => Err(PlanIoError::MalformedDirective(format!("not a bool: {s}"))),
347    }
348}
349
350fn parse_targets(s: &str) -> Result<Vec<QualifiedName>, PlanIoError> {
351    if s.is_empty() {
352        return Ok(Vec::new());
353    }
354    let mut out = Vec::new();
355    for part in s.split(',') {
356        out.push(parse_qname(part)?);
357    }
358    Ok(out)
359}
360
361fn parse_qname(s: &str) -> Result<QualifiedName, PlanIoError> {
362    let (schema, name) = s
363        .split_once('.')
364        .ok_or_else(|| PlanIoError::MalformedDirective(format!("expected schema.name: {s}")))?;
365    let schema = Identifier::from_unquoted(schema)
366        .map_err(|e| PlanIoError::MalformedDirective(format!("schema {schema}: {e}")))?;
367    let name = Identifier::from_unquoted(name)
368        .map_err(|e| PlanIoError::MalformedDirective(format!("name {name}: {e}")))?;
369    Ok(QualifiedName::new(schema, name))
370}
371
372// ---------------------------------------------------------------------------
373// intent.toml
374// ---------------------------------------------------------------------------
375
376/// Parsed view of `intent.toml`.
377#[derive(Debug, Clone, PartialEq, Eq)]
378pub struct ParsedIntent {
379    /// The short plan id field used for cross-check.
380    pub plan_id: String,
381    /// The intent rows (one per destructive step).
382    pub intents: Vec<DestructiveIntent>,
383    /// Lint waivers from `[[lint_waiver]]` rows.
384    pub lint_waivers: Vec<LintWaiver>,
385    /// Step overrides from `[[step_override]]` rows.
386    pub step_overrides: Vec<StepOverride>,
387}
388
389#[derive(Deserialize)]
390struct IntentDocDe {
391    plan_id: String,
392    #[serde(default, rename = "intent")]
393    intents: Vec<IntentRowDe>,
394    #[serde(default, rename = "lint_waiver")]
395    lint_waivers: Vec<LintWaiver>,
396    #[serde(default, rename = "step_override")]
397    step_overrides: Vec<StepOverride>,
398}
399
400#[derive(Deserialize)]
401struct IntentRowDe {
402    id: u32,
403    step: u32,
404    kind: String,
405    target: String,
406    reason: String,
407    /// `approved = true/false` in `intent.toml`. Defaults to `false` (the
408    /// writer always emits `approved = false`; the user flips it manually).
409    /// Retained on `DestructiveIntent` so preflight can enforce approval.
410    #[serde(default)]
411    approved: bool,
412}
413
414/// Parse `intent.toml` from a string.
415pub fn read_intent_toml(s: &str) -> Result<ParsedIntent, PlanIoError> {
416    let doc: IntentDocDe = toml::from_str(s)?;
417    Ok(ParsedIntent {
418        plan_id: doc.plan_id,
419        intents: doc
420            .intents
421            .into_iter()
422            .map(|r| DestructiveIntent {
423                id: r.id,
424                step: r.step,
425                kind: r.kind,
426                target: r.target,
427                reason: r.reason,
428                approved: r.approved,
429            })
430            .collect(),
431        lint_waivers: doc.lint_waivers,
432        step_overrides: doc.step_overrides,
433    })
434}
435
436// ---------------------------------------------------------------------------
437// manifest.toml
438// ---------------------------------------------------------------------------
439
440/// Parsed view of `manifest.toml`.
441#[derive(Debug, Clone, PartialEq)]
442pub struct ParsedManifest {
443    /// Short plan id for cross-check.
444    pub plan_id: String,
445    /// Full 64-char hex plan hash.
446    pub plan_hash: String,
447    /// pgevolve version at plan time.
448    pub pgevolve_version: String,
449    /// Planner ruleset version at plan time.
450    pub planner_ruleset_version: u32,
451    /// Optional source-tree revision.
452    pub source_rev: Option<String>,
453    /// Stable target-database identity.
454    pub target_identity: String,
455    /// Plan creation timestamp.
456    pub created_at: OffsetDateTime,
457    /// Recovered target catalog snapshot.
458    pub target_snapshot: Catalog,
459    /// `LintAtPlan` findings captured at plan time. Empty on older plans that
460    /// predate this field (`#[serde(default)]` on the de side).
461    pub lint_at_plan_findings: Vec<RecordedFinding>,
462}
463
464#[derive(Deserialize)]
465struct ManifestDocDe {
466    plan_id: String,
467    plan_hash: String,
468    pgevolve_version: String,
469    planner_ruleset_version: u32,
470    source_rev: Option<String>,
471    target_identity: String,
472    created_at: String,
473    target_snapshot_json: String,
474    #[serde(default)]
475    lint_at_plan_findings: Vec<RecordedFinding>,
476}
477
478/// Parse `manifest.toml` from a string.
479pub fn read_manifest_toml(s: &str) -> Result<ParsedManifest, PlanIoError> {
480    let doc: ManifestDocDe = toml::from_str(s)?;
481    let created_at = OffsetDateTime::parse(&doc.created_at, &Rfc3339).map_err(|e| {
482        PlanIoError::MalformedDirective(format!("manifest created_at={}: {e}", doc.created_at))
483    })?;
484    let target_snapshot: Catalog = serde_json::from_str(&doc.target_snapshot_json)?;
485    Ok(ParsedManifest {
486        plan_id: doc.plan_id,
487        plan_hash: doc.plan_hash,
488        pgevolve_version: doc.pgevolve_version,
489        planner_ruleset_version: doc.planner_ruleset_version,
490        source_rev: doc.source_rev,
491        target_identity: doc.target_identity,
492        created_at,
493        target_snapshot,
494        lint_at_plan_findings: doc.lint_at_plan_findings,
495    })
496}
497
498// ---------------------------------------------------------------------------
499// Plan::read_from_dir
500// ---------------------------------------------------------------------------
501
502/// Read a plan directory (three files) back into a `Plan`.
503///
504/// Cross-checks the short `plan_id` value across the three files and rejects
505/// inconsistent inputs. Destructive-step `destructive_reason` is grafted from
506/// `intent.toml`'s `reason` field (the SQL writer does not carry it).
507pub fn read_plan_dir(dir: &Path) -> Result<Plan, PlanIoError> {
508    let sql_path = dir.join("plan.sql");
509    let intent_path = dir.join("intent.toml");
510    let manifest_path = dir.join("manifest.toml");
511
512    let sql = std::fs::read_to_string(&sql_path).map_err(|e| PlanIoError::io(&sql_path, e))?;
513    let intent_str =
514        std::fs::read_to_string(&intent_path).map_err(|e| PlanIoError::io(&intent_path, e))?;
515    let manifest_str =
516        std::fs::read_to_string(&manifest_path).map_err(|e| PlanIoError::io(&manifest_path, e))?;
517
518    let partial = read_plan_sql(&sql)?;
519    let intent = read_intent_toml(&intent_str)?;
520    let manifest = read_manifest_toml(&manifest_str)?;
521
522    if partial.id_short != intent.plan_id || partial.id_short != manifest.plan_id {
523        return Err(PlanIoError::PlanIdMismatch {
524            sql: partial.id_short,
525            intent: intent.plan_id,
526            manifest: manifest.plan_id,
527        });
528    }
529
530    let id = PlanId::from_full_hex(&manifest.plan_hash)?;
531
532    // Graft destructive_reason from intent rows onto the matching steps.
533    let mut groups = partial.groups;
534    for g in &mut groups {
535        for step in &mut g.steps {
536            if let Some(intent_id) = step.intent_id
537                && let Some(row) = intent.intents.iter().find(|i| i.id == intent_id)
538            {
539                step.destructive_reason = Some(row.reason.clone());
540            }
541        }
542    }
543
544    let metadata = PlanMetadata {
545        pgevolve_version: manifest.pgevolve_version,
546        planner_ruleset_version: manifest.planner_ruleset_version,
547        source_rev: manifest.source_rev,
548        target_identity: manifest.target_identity,
549        target_snapshot: manifest.target_snapshot,
550        created_at: manifest.created_at,
551        lint_at_plan_findings: manifest.lint_at_plan_findings,
552    };
553
554    Ok(Plan {
555        id,
556        groups,
557        intents: intent.intents,
558        lint_waivers: intent.lint_waivers,
559        step_overrides: intent.step_overrides,
560        metadata,
561        advisory_findings: Vec::new(),
562    })
563}
564
565#[cfg(test)]
566mod tests {
567    use super::*;
568    use crate::plan::serialize::{write_plan_dir, write_plan_sql};
569    use crate::plan::{
570        grouping::TransactionGroup, plan::Plan, raw_step::RawStep, raw_step::StepKind,
571        raw_step::TransactionConstraint,
572    };
573
574    fn id_id(s: &str) -> Identifier {
575        Identifier::from_unquoted(s).unwrap()
576    }
577
578    fn qn(schema: &str, name: &str) -> QualifiedName {
579        QualifiedName::new(id_id(schema), id_id(name))
580    }
581
582    fn step(
583        kind: StepKind,
584        sql: &str,
585        destructive: bool,
586        targets: Vec<QualifiedName>,
587        c: TransactionConstraint,
588    ) -> RawStep {
589        RawStep {
590            step_no: 0,
591            kind,
592            destructive,
593            destructive_reason: destructive.then(|| "test reason".to_string()),
594            intent_id: None,
595            targets,
596            sql: sql.to_string(),
597            transactional: c,
598        }
599    }
600
601    fn simple_plan() -> Plan {
602        use crate::ir::schema::Schema;
603        let groups = vec![
604            TransactionGroup {
605                id: 1,
606                transactional: true,
607                steps: vec![
608                    step(
609                        StepKind::CreateSchema,
610                        "CREATE SCHEMA app;",
611                        false,
612                        vec![qn("app", "app")],
613                        TransactionConstraint::InTransaction,
614                    ),
615                    step(
616                        StepKind::DropTable,
617                        "DROP TABLE app.legacy;",
618                        true,
619                        vec![qn("app", "legacy")],
620                        TransactionConstraint::InTransaction,
621                    ),
622                ],
623            },
624            TransactionGroup {
625                id: 2,
626                transactional: false,
627                steps: vec![step(
628                    StepKind::CreateIndexConcurrent,
629                    "CREATE INDEX CONCURRENTLY users_idx ON app.users USING btree (id);",
630                    false,
631                    vec![qn("app", "users_idx"), qn("app", "users")],
632                    TransactionConstraint::OutsideTransaction,
633                )],
634            },
635        ];
636        let mut snapshot = Catalog::empty();
637        snapshot.schemas.push(Schema::new(id_id("app")));
638        Plan::from_grouped(
639            groups,
640            &Catalog::empty(),
641            &snapshot,
642            "tid-xyz".into(),
643            Some("git:abcdef0".into()),
644            "0.1.0",
645            1,
646        )
647        .unwrap()
648    }
649
650    #[test]
651    fn read_plan_sql_round_trips_simple_plan() {
652        let plan = simple_plan();
653        let mut buf = Vec::new();
654        write_plan_sql(&plan, &mut buf).unwrap();
655        let s = String::from_utf8(buf).unwrap();
656        let partial = read_plan_sql(&s).unwrap();
657
658        assert_eq!(partial.id_short, plan.id.short());
659        assert_eq!(partial.pgevolve_version, "0.1.0");
660        assert_eq!(partial.planner_ruleset_version, 1);
661        assert_eq!(partial.source_rev.as_deref(), Some("git:abcdef0"));
662        assert_eq!(partial.target_identity, "tid-xyz");
663        assert_eq!(partial.intents_required, 1);
664        assert_eq!(partial.groups.len(), 2);
665        assert!(partial.groups[0].transactional);
666        assert!(!partial.groups[1].transactional);
667        assert_eq!(partial.groups[0].steps.len(), 2);
668        assert_eq!(partial.groups[1].steps.len(), 1);
669        // Step bodies survive verbatim.
670        assert_eq!(partial.groups[0].steps[0].sql, "CREATE SCHEMA app;");
671        assert_eq!(
672            partial.groups[1].steps[0].sql,
673            "CREATE INDEX CONCURRENTLY users_idx ON app.users USING btree (id);"
674        );
675        // Step numbers preserved.
676        assert_eq!(partial.groups[0].steps[0].step_no, 1);
677        assert_eq!(partial.groups[0].steps[1].step_no, 2);
678        assert_eq!(partial.groups[1].steps[0].step_no, 3);
679        // Destructive step recovered with intent_id.
680        assert!(partial.groups[0].steps[1].destructive);
681        assert_eq!(partial.groups[0].steps[1].intent_id, Some(1));
682        // Targets list recovered.
683        assert_eq!(
684            partial.groups[1].steps[0].targets,
685            vec![qn("app", "users_idx"), qn("app", "users")],
686        );
687    }
688
689    #[test]
690    fn read_plan_sql_rejects_missing_plan_header() {
691        let s = "-- @pgevolve target=t\n";
692        assert!(matches!(
693            read_plan_sql(s),
694            Err(PlanIoError::MalformedDirective(_))
695        ));
696    }
697
698    #[test]
699    fn read_plan_sql_rejects_unknown_directive() {
700        let s = "-- @pgevolve plan id=abc version=0.1.0 ruleset=1 created=2026-05-09T18:42:11Z\n\
701                 -- @pgevolve nope=true\n";
702        assert!(matches!(
703            read_plan_sql(s),
704            Err(PlanIoError::MalformedDirective(_))
705        ));
706    }
707
708    #[test]
709    fn read_intent_toml_round_trips() {
710        let plan = simple_plan();
711        let mut buf = Vec::new();
712        crate::plan::serialize::write_intent_toml(&plan, &mut buf).unwrap();
713        let s = String::from_utf8(buf).unwrap();
714        let parsed = read_intent_toml(&s).unwrap();
715        assert_eq!(parsed.plan_id, plan.id.short());
716        assert_eq!(parsed.intents.len(), 1);
717        assert_eq!(parsed.intents[0].id, 1);
718        assert_eq!(parsed.intents[0].kind, "drop_table");
719        assert_eq!(parsed.intents[0].reason, "test reason");
720    }
721
722    #[test]
723    fn read_manifest_toml_round_trips_catalog() {
724        let plan = simple_plan();
725        let mut buf = Vec::new();
726        crate::plan::serialize::write_manifest_toml(&plan, &mut buf).unwrap();
727        let s = String::from_utf8(buf).unwrap();
728        let parsed = read_manifest_toml(&s).unwrap();
729        assert_eq!(parsed.plan_id, plan.id.short());
730        assert_eq!(parsed.plan_hash, plan.id.to_hex());
731        assert_eq!(parsed.target_snapshot, plan.metadata.target_snapshot);
732        assert_eq!(parsed.planner_ruleset_version, 1);
733        assert_eq!(parsed.target_identity, "tid-xyz");
734    }
735
736    #[test]
737    fn read_plan_dir_round_trips_whole_plan() {
738        let plan = simple_plan();
739        let dir = tempfile::tempdir().unwrap();
740        write_plan_dir(&plan, dir.path()).unwrap();
741        let recovered = read_plan_dir(dir.path()).unwrap();
742
743        // The full Plan should compare equal modulo timestamp truncation; RFC3339
744        // round-trip preserves UTC offsets and nanosecond precision via `time` v0.3,
745        // so equality should hold.
746        assert_eq!(recovered.id, plan.id);
747        assert_eq!(recovered.intents, plan.intents);
748        assert_eq!(
749            recovered.metadata.target_snapshot,
750            plan.metadata.target_snapshot
751        );
752        assert_eq!(
753            recovered.metadata.pgevolve_version,
754            plan.metadata.pgevolve_version
755        );
756        assert_eq!(
757            recovered.metadata.target_identity,
758            plan.metadata.target_identity
759        );
760        assert_eq!(recovered.groups.len(), plan.groups.len());
761        for (a, b) in recovered.groups.iter().zip(plan.groups.iter()) {
762            assert_eq!(a.id, b.id);
763            assert_eq!(a.transactional, b.transactional);
764            assert_eq!(a.steps.len(), b.steps.len());
765            for (sa, sb) in a.steps.iter().zip(b.steps.iter()) {
766                assert_eq!(sa.step_no, sb.step_no);
767                assert_eq!(sa.kind, sb.kind);
768                assert_eq!(sa.destructive, sb.destructive);
769                assert_eq!(sa.intent_id, sb.intent_id);
770                assert_eq!(sa.targets, sb.targets);
771                assert_eq!(sa.sql, sb.sql);
772                assert_eq!(sa.transactional, sb.transactional);
773                // destructive_reason is grafted from intent.toml.
774                if sb.destructive {
775                    assert_eq!(sa.destructive_reason, sb.destructive_reason);
776                }
777            }
778        }
779    }
780
781    #[test]
782    fn read_plan_dir_rejects_mismatched_plan_id() {
783        let plan = simple_plan();
784        let dir = tempfile::tempdir().unwrap();
785        write_plan_dir(&plan, dir.path()).unwrap();
786        // Tamper with intent.toml's plan_id.
787        let intent_path = dir.path().join("intent.toml");
788        let s = std::fs::read_to_string(&intent_path).unwrap();
789        let tampered = s.replacen(&plan.id.short(), "deadbeef00000000", 1);
790        std::fs::write(&intent_path, tampered).unwrap();
791
792        let err = read_plan_dir(dir.path()).unwrap_err();
793        assert!(matches!(err, PlanIoError::PlanIdMismatch { .. }));
794    }
795
796    #[test]
797    fn read_plan_sql_handles_multi_line_step_body() {
798        let plan_text = "\
799-- @pgevolve plan id=abc1234567890123 version=0.1.0 ruleset=1 created=2026-05-09T18:42:11Z
800-- @pgevolve target=tid
801-- @pgevolve intents_required=0
802
803-- @pgevolve group id=1 transactional=true
804BEGIN;
805-- @pgevolve step=1 kind=create_table destructive=false targets=app.t
806CREATE TABLE app.t (
807    id bigint NOT NULL,
808    name text
809);
810COMMIT;
811";
812        let partial = read_plan_sql(plan_text).unwrap();
813        assert_eq!(partial.groups.len(), 1);
814        let body = &partial.groups[0].steps[0].sql;
815        assert!(body.starts_with("CREATE TABLE app.t ("));
816        assert!(body.contains("id bigint NOT NULL,"));
817        assert!(body.ends_with(");"));
818    }
819
820    #[test]
821    fn read_plan_dir_round_trips_view_and_mv_step_kinds() {
822        use crate::ir::catalog::Catalog;
823        use crate::ir::schema::Schema;
824        use crate::plan::grouping::TransactionGroup;
825        use crate::plan::plan::Plan;
826        use crate::plan::raw_step::{RawStep, StepKind, TransactionConstraint};
827        use crate::plan::serialize::write_plan_dir;
828
829        let id_id = |s: &str| Identifier::from_unquoted(s).unwrap();
830        let qn = |schema: &str, name: &str| QualifiedName::new(id_id(schema), id_id(name));
831
832        let view_step = |kind: StepKind, sql: &str, destructive: bool| -> RawStep {
833            RawStep {
834                step_no: 0,
835                kind,
836                destructive,
837                destructive_reason: destructive.then(|| "test".to_string()),
838                intent_id: None,
839                targets: vec![qn("app", "my_view")],
840                sql: sql.to_string(),
841                transactional: TransactionConstraint::InTransaction,
842            }
843        };
844
845        let groups = vec![TransactionGroup {
846            id: 1,
847            transactional: true,
848            steps: vec![
849                view_step(
850                    StepKind::CreateView,
851                    "CREATE VIEW app.my_view AS\nSELECT 1;",
852                    false,
853                ),
854                view_step(StepKind::DropView, "DROP VIEW app.my_view;", true),
855                view_step(
856                    StepKind::CreateMaterializedView,
857                    "CREATE MATERIALIZED VIEW app.my_view AS\nSELECT 1\nWITH NO DATA;",
858                    false,
859                ),
860                view_step(
861                    StepKind::DropMaterializedView,
862                    "DROP MATERIALIZED VIEW app.my_view;",
863                    false,
864                ),
865                view_step(
866                    StepKind::RefreshMaterializedView,
867                    "REFRESH MATERIALIZED VIEW app.my_view;",
868                    false,
869                ),
870                view_step(
871                    StepKind::AlterViewSetReloption,
872                    "ALTER VIEW app.my_view SET (security_barrier = true);",
873                    false,
874                ),
875                view_step(
876                    StepKind::CommentOnView,
877                    "COMMENT ON VIEW app.my_view IS 'a view';",
878                    false,
879                ),
880            ],
881        }];
882
883        let mut snapshot = Catalog::empty();
884        snapshot.schemas.push(Schema::new(id_id("app")));
885        let plan = Plan::from_grouped(
886            groups,
887            &Catalog::empty(),
888            &snapshot,
889            "test-views-target".into(),
890            None,
891            "0.2.0",
892            2,
893        )
894        .unwrap();
895
896        let dir = tempfile::tempdir().unwrap();
897        write_plan_dir(&plan, dir.path()).unwrap();
898        let recovered = read_plan_dir(dir.path()).unwrap();
899
900        assert_eq!(recovered.groups.len(), 1);
901        let steps = &recovered.groups[0].steps;
902        assert_eq!(steps.len(), 7);
903        assert_eq!(steps[0].kind, StepKind::CreateView);
904        assert_eq!(steps[1].kind, StepKind::DropView);
905        assert_eq!(steps[2].kind, StepKind::CreateMaterializedView);
906        assert_eq!(steps[3].kind, StepKind::DropMaterializedView);
907        assert_eq!(steps[4].kind, StepKind::RefreshMaterializedView);
908        assert_eq!(steps[5].kind, StepKind::AlterViewSetReloption);
909        assert_eq!(steps[6].kind, StepKind::CommentOnView);
910
911        // Destructive step (DropView) gets an intent_id.
912        assert_eq!(steps[1].intent_id, Some(1));
913        assert!(steps[1].destructive);
914
915        // Non-destructive steps have no intent_id.
916        assert!(steps[0].intent_id.is_none());
917        assert!(steps[3].intent_id.is_none()); // DropMaterializedView is NOT destructive
918
919        // SQL bodies survive the round-trip.
920        assert_eq!(steps[0].sql, "CREATE VIEW app.my_view AS\nSELECT 1;");
921        assert_eq!(
922            steps[2].sql,
923            "CREATE MATERIALIZED VIEW app.my_view AS\nSELECT 1\nWITH NO DATA;"
924        );
925        assert_eq!(steps[4].sql, "REFRESH MATERIALIZED VIEW app.my_view;");
926    }
927}