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;
18use crate::ir::difference::Difference;
19use crate::ir::eq::{Equiv, field_difference};
20
21/// Declarative model of a Postgres `SUBSCRIPTION`.
22#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
23pub struct Subscription {
24 /// Subscription name (not schema-qualified — subscriptions are global).
25 pub name: Identifier,
26 /// libpq connection string. May contain `${VAR}` env-var references that
27 /// are resolved at apply-time preflight. Stored verbatim through parse,
28 /// canon, diff, and plan serialization.
29 pub connection: String,
30 /// Publications this subscription reads from. Sorted + deduped by canon.
31 /// Non-empty (enforced by canon).
32 pub publications: Vec<Identifier>,
33 /// Per-field lenient WITH options.
34 pub options: SubscriptionOptions,
35 /// Object owner. `None` = unmanaged (the differ ignores ownership).
36 /// `Some(role)` = managed: diff emits `ALTER SUBSCRIPTION ... OWNER TO role`.
37 pub owner: Option<Identifier>,
38 /// Optional comment.
39 pub comment: Option<String>,
40}
41
42impl Equiv for Subscription {
43 fn differences(&self, other: &Self) -> Vec<Difference> {
44 // Field-completeness guard: the compiler errors if a field is added
45 // without being handled below. Bindings are unused (read via `self`).
46 let Self {
47 name: _,
48 connection: _,
49 publications: _,
50 options: _,
51 owner: _,
52 comment: _,
53 } = self;
54 let mut out = Vec::new();
55 out.extend(field_difference("name", &self.name, &other.name));
56 out.extend(field_difference(
57 "connection",
58 &self.connection,
59 &other.connection,
60 ));
61 out.extend(field_difference(
62 "publications",
63 &format!("{:?}", self.publications),
64 &format!("{:?}", other.publications),
65 ));
66 out.extend(field_difference(
67 "options",
68 &format!("{:?}", self.options),
69 &format!("{:?}", other.options),
70 ));
71 out.extend(field_difference(
72 "owner",
73 &format!("{:?}", self.owner),
74 &format!("{:?}", other.owner),
75 ));
76 out.extend(field_difference(
77 "comment",
78 &format!("{:?}", self.comment),
79 &format!("{:?}", other.comment),
80 ));
81 out
82 }
83}
84
85/// Per-field lenient WITH options for a `Subscription`.
86///
87/// Every field is `Option<T>`. `None` = unmanaged (pgevolve neither sets
88/// nor resets); `Some(value)` = managed (diff emits an ALTER to match).
89/// Matches the v0.3.3 reloptions per-field-lenient pattern.
90///
91/// **CREATE-only fields**: `connect`, `create_slot`, and `copy_data` are
92/// PG-CREATE-only (no `ALTER SUBSCRIPTION s SET (connect = …)` exists).
93/// They flow into the IR from source CREATE statements so users can declare
94/// them, but the differ NEVER includes them in `AlterSubscriptionSetOptions`
95/// deltas, and the catalog reader ALWAYS returns `None` for them
96/// (`pg_subscription` doesn't store the CREATE-time decision).
97/// See `diff::subscriptions::options_delta`.
98#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
99pub struct SubscriptionOptions {
100 /// Whether the subscription is enabled. PG default: true.
101 pub enabled: Option<bool>,
102 /// Replication slot name on the publisher. `None` = use subscription name.
103 pub slot_name: Option<Identifier>,
104 /// Whether `CREATE SUBSCRIPTION` should dial the publisher to set up the
105 /// replication slot and verify the connection. PG default: true.
106 ///
107 /// `connect = false` creates the subscription without connecting to the
108 /// publisher; no slot is created at CREATE time. This is a CREATE-only
109 /// directive: it does not appear in `pg_subscription` and cannot be
110 /// changed via `ALTER SUBSCRIPTION`. The catalog reader always returns
111 /// `None` for this field.
112 ///
113 /// Set `connect: Some(false)` in generated catalogs that use synthetic
114 /// DSNs (e.g. `replica.example.com`) to avoid network errors during
115 /// `CREATE SUBSCRIPTION`.
116 pub connect: Option<bool>,
117 /// Whether `CREATE SUBSCRIPTION` should create the publisher-side slot.
118 /// PG default: true.
119 pub create_slot: Option<bool>,
120 /// Whether to copy existing rows during initial sync. PG default: true.
121 pub copy_data: Option<bool>,
122 /// `synchronous_commit` GUC value for the subscription's apply worker.
123 /// Free-form string (e.g., `"on"`, `"off"`, `"remote_write"`, `"local"`).
124 pub synchronous_commit: Option<String>,
125 /// Use binary copy / binary replication protocol. PG default: false.
126 pub binary: Option<bool>,
127 /// Streaming mode for large in-progress transactions.
128 pub streaming: Option<StreamingMode>,
129 /// Two-phase commit handling. PG 14+; default: false.
130 pub two_phase: Option<bool>,
131 /// Disable the subscription on apply error. PG 15+; default: false.
132 pub disable_on_error: Option<bool>,
133 /// Whether the subscription owner must supply a password. PG 16+; default: true.
134 pub password_required: Option<bool>,
135 /// Run the apply worker as the subscription owner (instead of the table owner).
136 /// PG 16+; default: false.
137 pub run_as_owner: Option<bool>,
138 /// Replication origin handling. PG 16+; default: Any.
139 pub origin: Option<OriginMode>,
140 /// Whether the subscription survives failover. PG 17+; default: false.
141 pub failover: Option<bool>,
142}
143
144/// `streaming` mode for in-progress transactions on a subscription.
145#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
146pub enum StreamingMode {
147 /// Stream nothing; spool to disk at the subscriber.
148 Off,
149 /// Stream in-progress transactions to disk on the subscriber.
150 On,
151 /// Stream in-progress transactions to parallel apply workers. PG 16+.
152 Parallel,
153}
154
155/// `origin` mode for replication-origin handling on a subscription.
156#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
157pub enum OriginMode {
158 /// Replicate all changes regardless of origin (default).
159 Any,
160 /// Replicate only changes from non-replicated sources (avoid loops).
161 None,
162}
163
164#[cfg(test)]
165mod tests {
166 use super::*;
167
168 #[test]
169 fn options_default_all_none() {
170 let o = SubscriptionOptions::default();
171 assert!(o.enabled.is_none());
172 assert!(o.slot_name.is_none());
173 assert!(o.connect.is_none());
174 assert!(o.create_slot.is_none());
175 assert!(o.copy_data.is_none());
176 assert!(o.synchronous_commit.is_none());
177 assert!(o.binary.is_none());
178 assert!(o.streaming.is_none());
179 assert!(o.two_phase.is_none());
180 assert!(o.disable_on_error.is_none());
181 assert!(o.password_required.is_none());
182 assert!(o.run_as_owner.is_none());
183 assert!(o.origin.is_none());
184 assert!(o.failover.is_none());
185 }
186
187 #[test]
188 fn options_connect_false_round_trips() {
189 // `connect: Some(false)` must be preserved through clone + equality.
190 let o = SubscriptionOptions {
191 connect: Some(false),
192 ..Default::default()
193 };
194 assert_eq!(o.connect, Some(false));
195 let o2 = o.clone();
196 assert_eq!(o, o2);
197 }
198
199 #[test]
200 fn streaming_off_does_not_equal_on() {
201 assert_ne!(StreamingMode::Off, StreamingMode::On);
202 assert_ne!(StreamingMode::On, StreamingMode::Parallel);
203 }
204
205 #[test]
206 fn origin_any_does_not_equal_none() {
207 assert_ne!(OriginMode::Any, OriginMode::None);
208 }
209}