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**: `connect`, `create_slot`, and `copy_data` are
47/// PG-CREATE-only (no `ALTER SUBSCRIPTION s SET (connect = …)` exists).
48/// They flow into the IR from source CREATE statements so users can declare
49/// them, but the differ NEVER includes them in `AlterSubscriptionSetOptions`
50/// deltas, and the catalog reader ALWAYS returns `None` for them
51/// (`pg_subscription` doesn't store the CREATE-time decision).
52/// See `diff::subscriptions::options_delta`.
53#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
54pub struct SubscriptionOptions {
55    /// Whether the subscription is enabled. PG default: true.
56    pub enabled: Option<bool>,
57    /// Replication slot name on the publisher. `None` = use subscription name.
58    pub slot_name: Option<Identifier>,
59    /// Whether `CREATE SUBSCRIPTION` should dial the publisher to set up the
60    /// replication slot and verify the connection. PG default: true.
61    ///
62    /// `connect = false` creates the subscription without connecting to the
63    /// publisher; no slot is created at CREATE time. This is a CREATE-only
64    /// directive: it does not appear in `pg_subscription` and cannot be
65    /// changed via `ALTER SUBSCRIPTION`. The catalog reader always returns
66    /// `None` for this field.
67    ///
68    /// Set `connect: Some(false)` in generated catalogs that use synthetic
69    /// DSNs (e.g. `replica.example.com`) to avoid network errors during
70    /// `CREATE SUBSCRIPTION`.
71    pub connect: Option<bool>,
72    /// Whether `CREATE SUBSCRIPTION` should create the publisher-side slot.
73    /// PG default: true.
74    pub create_slot: Option<bool>,
75    /// Whether to copy existing rows during initial sync. PG default: true.
76    pub copy_data: Option<bool>,
77    /// `synchronous_commit` GUC value for the subscription's apply worker.
78    /// Free-form string (e.g., `"on"`, `"off"`, `"remote_write"`, `"local"`).
79    pub synchronous_commit: Option<String>,
80    /// Use binary copy / binary replication protocol. PG default: false.
81    pub binary: Option<bool>,
82    /// Streaming mode for large in-progress transactions.
83    pub streaming: Option<StreamingMode>,
84    /// Two-phase commit handling. PG 14+; default: false.
85    pub two_phase: Option<bool>,
86    /// Disable the subscription on apply error. PG 15+; default: false.
87    pub disable_on_error: Option<bool>,
88    /// Whether the subscription owner must supply a password. PG 16+; default: true.
89    pub password_required: Option<bool>,
90    /// Run the apply worker as the subscription owner (instead of the table owner).
91    /// PG 16+; default: false.
92    pub run_as_owner: Option<bool>,
93    /// Replication origin handling. PG 16+; default: Any.
94    pub origin: Option<OriginMode>,
95    /// Whether the subscription survives failover. PG 17+; default: false.
96    pub failover: Option<bool>,
97}
98
99/// `streaming` mode for in-progress transactions on a subscription.
100#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
101pub enum StreamingMode {
102    /// Stream nothing; spool to disk at the subscriber.
103    Off,
104    /// Stream in-progress transactions to disk on the subscriber.
105    On,
106    /// Stream in-progress transactions to parallel apply workers. PG 16+.
107    Parallel,
108}
109
110/// `origin` mode for replication-origin handling on a subscription.
111#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
112pub enum OriginMode {
113    /// Replicate all changes regardless of origin (default).
114    Any,
115    /// Replicate only changes from non-replicated sources (avoid loops).
116    None,
117}
118
119#[cfg(test)]
120mod tests {
121    use super::*;
122
123    #[test]
124    fn options_default_all_none() {
125        let o = SubscriptionOptions::default();
126        assert!(o.enabled.is_none());
127        assert!(o.slot_name.is_none());
128        assert!(o.connect.is_none());
129        assert!(o.create_slot.is_none());
130        assert!(o.copy_data.is_none());
131        assert!(o.synchronous_commit.is_none());
132        assert!(o.binary.is_none());
133        assert!(o.streaming.is_none());
134        assert!(o.two_phase.is_none());
135        assert!(o.disable_on_error.is_none());
136        assert!(o.password_required.is_none());
137        assert!(o.run_as_owner.is_none());
138        assert!(o.origin.is_none());
139        assert!(o.failover.is_none());
140    }
141
142    #[test]
143    fn options_connect_false_round_trips() {
144        // `connect: Some(false)` must be preserved through clone + equality.
145        let o = SubscriptionOptions {
146            connect: Some(false),
147            ..Default::default()
148        };
149        assert_eq!(o.connect, Some(false));
150        let o2 = o.clone();
151        assert_eq!(o, o2);
152    }
153
154    #[test]
155    fn streaming_off_does_not_equal_on() {
156        assert_ne!(StreamingMode::Off, StreamingMode::On);
157        assert_ne!(StreamingMode::On, StreamingMode::Parallel);
158    }
159
160    #[test]
161    fn origin_any_does_not_equal_none() {
162        assert_ne!(OriginMode::Any, OriginMode::None);
163    }
164}
pgevolve_core/ir/subscription.rs

pgevolve_core/ir/
subscription.rs
