ironflow_ops_postgres/
maintenance.rs

1//! Maintenance operations: `VACUUM`, `ANALYZE`, `REINDEX`.
2
3use async_trait::async_trait;
4use ironflow_core::error::OperationError;
5use ironflow_core::operation::{Operation, OperationContext, TypedOperation};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use sqlx::{Executor, PgPool};
9
10use crate::helpers::{pg_error, quote_identifier, to_value};
11
12/// Output of maintenance operations.
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct MaintenanceOutput {
15    /// The operation that was performed.
16    pub operation: String,
17    /// The target (table or index name).
18    pub target: String,
19}
20
21/// Run `VACUUM` on a table.
22///
23/// # Examples
24///
25/// ```no_run
26/// use ironflow_ops_postgres::maintenance::Vacuum;
27/// use ironflow_core::operation::Operation;
28///
29/// # fn example(pool: sqlx::PgPool) {
30/// let op = Vacuum::new(pool, "public", "users");
31/// assert_eq!(op.kind(), "postgres");
32/// # }
33/// ```
34pub struct Vacuum {
35    pool: PgPool,
36    schema: String,
37    table: String,
38}
39
40impl Vacuum {
41    /// Create a new vacuum operation.
42    pub fn new(pool: PgPool, schema: impl Into<String>, table: impl Into<String>) -> Self {
43        Self {
44            pool,
45            schema: schema.into(),
46            table: table.into(),
47        }
48    }
49
50    /// Execute and return a typed result.
51    ///
52    /// # Errors
53    ///
54    /// Returns [`OperationError::External`] on connection or permission errors.
55    pub async fn run(&self, _ctx: &OperationContext) -> Result<MaintenanceOutput, OperationError> {
56        let schema = quote_identifier(&self.schema)?;
57        let table = quote_identifier(&self.table)?;
58        let sql = format!("VACUUM {schema}.{table}");
59        self.pool.execute(sql.as_str()).await.map_err(pg_error)?;
60        Ok(MaintenanceOutput {
61            operation: "VACUUM".to_string(),
62            target: format!("{}.{}", self.schema, self.table),
63        })
64    }
65}
66
67#[async_trait]
68impl Operation for Vacuum {
69    fn kind(&self) -> &str {
70        "postgres"
71    }
72    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
73        to_value(&self.run(ctx).await?)
74    }
75    fn input(&self) -> Option<Value> {
76        Some(serde_json::json!({ "schema": self.schema, "table": self.table }))
77    }
78}
79
80impl TypedOperation for Vacuum {
81    type Output = MaintenanceOutput;
82}
83
84/// Run `ANALYZE` on a table.
85///
86/// # Examples
87///
88/// ```no_run
89/// use ironflow_ops_postgres::maintenance::Analyze;
90/// use ironflow_core::operation::Operation;
91///
92/// # fn example(pool: sqlx::PgPool) {
93/// let op = Analyze::new(pool, "public", "users");
94/// assert_eq!(op.kind(), "postgres");
95/// # }
96/// ```
97pub struct Analyze {
98    pool: PgPool,
99    schema: String,
100    table: String,
101}
102
103impl Analyze {
104    /// Create a new analyze operation.
105    pub fn new(pool: PgPool, schema: impl Into<String>, table: impl Into<String>) -> Self {
106        Self {
107            pool,
108            schema: schema.into(),
109            table: table.into(),
110        }
111    }
112
113    /// Execute and return a typed result.
114    ///
115    /// # Errors
116    ///
117    /// Returns [`OperationError::External`] on connection or permission errors.
118    pub async fn run(&self, _ctx: &OperationContext) -> Result<MaintenanceOutput, OperationError> {
119        let schema = quote_identifier(&self.schema)?;
120        let table = quote_identifier(&self.table)?;
121        let sql = format!("ANALYZE {schema}.{table}");
122        self.pool.execute(sql.as_str()).await.map_err(pg_error)?;
123        Ok(MaintenanceOutput {
124            operation: "ANALYZE".to_string(),
125            target: format!("{}.{}", self.schema, self.table),
126        })
127    }
128}
129
130#[async_trait]
131impl Operation for Analyze {
132    fn kind(&self) -> &str {
133        "postgres"
134    }
135    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
136        to_value(&self.run(ctx).await?)
137    }
138    fn input(&self) -> Option<Value> {
139        Some(serde_json::json!({ "schema": self.schema, "table": self.table }))
140    }
141}
142
143impl TypedOperation for Analyze {
144    type Output = MaintenanceOutput;
145}
146
147/// Run `REINDEX TABLE` on a table.
148///
149/// # Examples
150///
151/// ```no_run
152/// use ironflow_ops_postgres::maintenance::Reindex;
153/// use ironflow_core::operation::Operation;
154///
155/// # fn example(pool: sqlx::PgPool) {
156/// let op = Reindex::new(pool, "public", "users");
157/// assert_eq!(op.kind(), "postgres");
158/// # }
159/// ```
160pub struct Reindex {
161    pool: PgPool,
162    schema: String,
163    table: String,
164}
165
166impl Reindex {
167    /// Create a new reindex operation.
168    pub fn new(pool: PgPool, schema: impl Into<String>, table: impl Into<String>) -> Self {
169        Self {
170            pool,
171            schema: schema.into(),
172            table: table.into(),
173        }
174    }
175
176    /// Execute and return a typed result.
177    ///
178    /// # Errors
179    ///
180    /// Returns [`OperationError::External`] on connection or permission errors.
181    pub async fn run(&self, _ctx: &OperationContext) -> Result<MaintenanceOutput, OperationError> {
182        let schema = quote_identifier(&self.schema)?;
183        let table = quote_identifier(&self.table)?;
184        let sql = format!("REINDEX TABLE {schema}.{table}");
185        self.pool.execute(sql.as_str()).await.map_err(pg_error)?;
186        Ok(MaintenanceOutput {
187            operation: "REINDEX".to_string(),
188            target: format!("{}.{}", self.schema, self.table),
189        })
190    }
191}
192
193#[async_trait]
194impl Operation for Reindex {
195    fn kind(&self) -> &str {
196        "postgres"
197    }
198    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
199        to_value(&self.run(ctx).await?)
200    }
201    fn input(&self) -> Option<Value> {
202        Some(serde_json::json!({ "schema": self.schema, "table": self.table }))
203    }
204}
205
206impl TypedOperation for Reindex {
207    type Output = MaintenanceOutput;
208}
209
210#[cfg(test)]
211mod tests {
212    use super::*;
213
214    #[tokio::test]
215    async fn vacuum_kind() {
216        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
217        let op = Vacuum::new(pool, "public", "users");
218        assert_eq!(op.kind(), "postgres");
219    }
220
221    #[tokio::test]
222    async fn analyze_kind() {
223        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
224        let op = Analyze::new(pool, "public", "users");
225        assert_eq!(op.kind(), "postgres");
226    }
227
228    #[tokio::test]
229    async fn reindex_kind() {
230        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
231        let op = Reindex::new(pool, "public", "users");
232        assert_eq!(op.kind(), "postgres");
233    }
234
235    #[tokio::test]
236    async fn vacuum_input_no_secrets() {
237        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
238        let op = Vacuum::new(pool, "public", "users");
239        let input = op.input().unwrap();
240        let text = input.to_string();
241        assert!(!text.contains("postgres://"), "leaked URL: {text}");
242    }
243}
ironflow_ops_postgres/maintenance.rs

ironflow_ops_postgres/
maintenance.rs
