pgevolve_core/ir/
subscription.rs

1//! Subscription IR — declarative logical-replication subscriber-side metadata.
2//!
3//! A `Subscription` is a Postgres `CREATE SUBSCRIPTION` object. It lives at
4//! the Catalog top level (not schema-qualified) because Postgres treats
5//! subscriptions as a per-database global namespace.
6//!
7//! The `connection` field stores the libpq connection string verbatim,
8//! including unresolved `${VAR}` env-var references. Resolution happens at
9//! apply-time preflight (`crates/pgevolve/src/executor/env_interp.rs`),
10//! never at parse or canon time. The source IR — and therefore plan.sql —
11//! never contains resolved secrets.
12//!
13//! Spec: `docs/superpowers/specs/2026-05-26-subscriptions-design.md`.
14
15use serde::{Deserialize, Serialize};
16
17use crate::identifier::Identifier;
18
19/// Declarative model of a Postgres `SUBSCRIPTION`.
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct Subscription {
22    /// Subscription name (not schema-qualified — subscriptions are global).
23    pub name: Identifier,
24    /// libpq connection string. May contain `${VAR}` env-var references that
25    /// are resolved at apply-time preflight. Stored verbatim through parse,
26    /// canon, diff, and plan serialization.
27    pub connection: String,
28    /// Publications this subscription reads from. Sorted + deduped by canon.
29    /// Non-empty (enforced by canon).
30    pub publications: Vec<Identifier>,
31    /// Per-field lenient WITH options.
32    pub options: SubscriptionOptions,
33    /// Object owner. `None` = unmanaged (the differ ignores ownership).
34    /// `Some(role)` = managed: diff emits `ALTER SUBSCRIPTION ... OWNER TO role`.
35    pub owner: Option<Identifier>,
36    /// Optional comment.
37    pub comment: Option<String>,
38}
39
40/// Per-field lenient WITH options for a `Subscription`.
41///
42/// Every field is `Option<T>`. `None` = unmanaged (pgevolve neither sets
43/// nor resets); `Some(value)` = managed (diff emits an ALTER to match).
44/// Matches the v0.3.3 reloptions per-field-lenient pattern.
45///
46/// **CREATE-only fields**: `create_slot` and `copy_data` are PG-CREATE-only
47/// (no `ALTER SUBSCRIPTION s SET (create_slot = …)` exists). They flow into
48/// the IR from source CREATE statements so users can declare them, but the
49/// differ NEVER includes them in `AlterSubscriptionSetOptions` deltas, and
50/// the catalog reader ALWAYS returns `None` for them (`pg_subscription`
51/// doesn't store the CREATE-time decision). See `diff::subscriptions::options_delta`.
52#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
53pub struct SubscriptionOptions {
54    /// Whether the subscription is enabled. PG default: true.
55    pub enabled: Option<bool>,
56    /// Replication slot name on the publisher. `None` = use subscription name.
57    pub slot_name: Option<Identifier>,
58    /// Whether `CREATE SUBSCRIPTION` should create the publisher-side slot.
59    /// PG default: true.
60    pub create_slot: Option<bool>,
61    /// Whether to copy existing rows during initial sync. PG default: true.
62    pub copy_data: Option<bool>,
63    /// `synchronous_commit` GUC value for the subscription's apply worker.
64    /// Free-form string (e.g., `"on"`, `"off"`, `"remote_write"`, `"local"`).
65    pub synchronous_commit: Option<String>,
66    /// Use binary copy / binary replication protocol. PG default: false.
67    pub binary: Option<bool>,
68    /// Streaming mode for large in-progress transactions.
69    pub streaming: Option<StreamingMode>,
70    /// Two-phase commit handling. PG 14+; default: false.
71    pub two_phase: Option<bool>,
72    /// Disable the subscription on apply error. PG 15+; default: false.
73    pub disable_on_error: Option<bool>,
74    /// Whether the subscription owner must supply a password. PG 16+; default: true.
75    pub password_required: Option<bool>,
76    /// Run the apply worker as the subscription owner (instead of the table owner).
77    /// PG 16+; default: false.
78    pub run_as_owner: Option<bool>,
79    /// Replication origin handling. PG 16+; default: Any.
80    pub origin: Option<OriginMode>,
81    /// Whether the subscription survives failover. PG 17+; default: false.
82    pub failover: Option<bool>,
83}
84
85/// `streaming` mode for in-progress transactions on a subscription.
86#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
87pub enum StreamingMode {
88    /// Stream nothing; spool to disk at the subscriber.
89    Off,
90    /// Stream in-progress transactions to disk on the subscriber.
91    On,
92    /// Stream in-progress transactions to parallel apply workers. PG 16+.
93    Parallel,
94}
95
96/// `origin` mode for replication-origin handling on a subscription.
97#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
98pub enum OriginMode {
99    /// Replicate all changes regardless of origin (default).
100    Any,
101    /// Replicate only changes from non-replicated sources (avoid loops).
102    None,
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108
109    #[test]
110    fn options_default_all_none() {
111        let o = SubscriptionOptions::default();
112        assert!(o.enabled.is_none());
113        assert!(o.slot_name.is_none());
114        assert!(o.create_slot.is_none());
115        assert!(o.copy_data.is_none());
116        assert!(o.synchronous_commit.is_none());
117        assert!(o.binary.is_none());
118        assert!(o.streaming.is_none());
119        assert!(o.two_phase.is_none());
120        assert!(o.disable_on_error.is_none());
121        assert!(o.password_required.is_none());
122        assert!(o.run_as_owner.is_none());
123        assert!(o.origin.is_none());
124        assert!(o.failover.is_none());
125    }
126
127    #[test]
128    fn streaming_off_does_not_equal_on() {
129        assert_ne!(StreamingMode::Off, StreamingMode::On);
130        assert_ne!(StreamingMode::On, StreamingMode::Parallel);
131    }
132
133    #[test]
134    fn origin_any_does_not_equal_none() {
135        assert_ne!(OriginMode::Any, OriginMode::None);
136    }
137}
pgevolve_core/ir/subscription.rs

pgevolve_core/ir/
subscription.rs
