ig_client/storage/
utils.rs

1use crate::error::AppError;
2use crate::presentation::transaction::StoreTransaction;
3use crate::storage::config::DatabaseConfig;
4use serde::Serialize;
5use serde::de::DeserializeOwned;
6use serde_json;
7use sqlx::{Executor, PgPool};
8use tracing::info;
9
10/// Initializes the `ig_options` table used by [`store_transactions`].
11///
12/// Ships the DDL in code (like the other storage tables) so persistence works
13/// against a fresh database. The `raw_hash` column is the deduplication key:
14/// it stores the MD5 digest of the raw transaction JSON and carries a `UNIQUE`
15/// constraint so re-ingesting an identical payload is a no-op. Call this once
16/// before [`store_transactions`].
17///
18/// # Arguments
19/// * `pool` - PostgreSQL connection pool
20///
21/// # Errors
22/// Returns [`AppError`] if the `CREATE TABLE` statement fails.
23#[must_use = "table initialization can fail and must be handled"]
24pub async fn initialize_ig_options_table(pool: &PgPool) -> Result<(), AppError> {
25    info!("Initializing ig_options table...");
26
27    sqlx::query(
28        r#"
29        CREATE TABLE IF NOT EXISTS ig_options (
30            id BIGSERIAL PRIMARY KEY,
31            reference TEXT NOT NULL,
32            deal_date TIMESTAMPTZ NOT NULL,
33            underlying TEXT,
34            strike DOUBLE PRECISION,
35            option_type TEXT,
36            expiry DATE,
37            transaction_type TEXT NOT NULL,
38            pnl_eur DOUBLE PRECISION NOT NULL,
39            is_fee BOOLEAN NOT NULL,
40            raw TEXT NOT NULL,
41            raw_hash TEXT NOT NULL UNIQUE,
42            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
43        )
44        "#,
45    )
46    .execute(pool)
47    .await?;
48
49    sqlx::query("CREATE INDEX IF NOT EXISTS idx_ig_options_reference ON ig_options(reference)")
50        .execute(pool)
51        .await?;
52
53    info!("ig_options table initialized successfully");
54    Ok(())
55}
56
57/// Stores a list of transactions in the database.
58///
59/// Requires the `ig_options` table to exist; call
60/// [`initialize_ig_options_table`] first. Deduplication is keyed on
61/// `raw_hash`, which is computed database-side as `md5(raw)` from the bound
62/// raw JSON (a stable, dependency-free digest); `ON CONFLICT (raw_hash) DO
63/// NOTHING` makes re-ingesting an identical payload a no-op.
64///
65/// # Arguments
66/// * `pool` - PostgreSQL connection pool
67/// * `txs` - List of transactions to store
68///
69/// # Returns
70/// * `Result<usize, AppError>` - Number of transactions inserted or an error
71#[must_use = "storage writes can fail and the inserted count must be handled"]
72pub async fn store_transactions(
73    pool: &sqlx::PgPool,
74    txs: &[StoreTransaction],
75) -> Result<usize, AppError> {
76    let mut tx = pool.begin().await?;
77    let mut inserted = 0;
78
79    for t in txs {
80        // `raw_hash` is derived from the raw JSON via Postgres' built-in
81        // `md5()` on the bound `$10` value (no string interpolation), giving a
82        // deterministic dedup key without an extra dependency.
83        let result = tx
84            .execute(
85                sqlx::query(
86                    r#"
87                    INSERT INTO ig_options (
88                        reference, deal_date, underlying, strike,
89                        option_type, expiry, transaction_type, pnl_eur, is_fee, raw,
90                        raw_hash
91                    )
92                    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10, md5($10))
93                    ON CONFLICT (raw_hash) DO NOTHING
94                    "#,
95                )
96                .bind(&t.reference)
97                .bind(t.deal_date)
98                .bind(&t.underlying)
99                .bind(t.strike)
100                .bind(&t.option_type)
101                .bind(t.expiry)
102                .bind(&t.transaction_type)
103                .bind(t.pnl_eur)
104                .bind(t.is_fee)
105                .bind(&t.raw_json),
106            )
107            .await?;
108
109        inserted += result.rows_affected() as usize;
110    }
111
112    tx.commit().await?;
113    Ok(inserted)
114}
115
116/// Serializes a value to a JSON string
117pub fn serialize_to_json<T: Serialize>(value: &T) -> Result<String, serde_json::Error> {
118    serde_json::to_string(value)
119}
120
121/// Deserializes a JSON string into a value
122pub fn deserialize_from_json<T: DeserializeOwned>(s: &str) -> Result<T, serde_json::Error> {
123    serde_json::from_str(s)
124}
125
126/// Creates a PostgreSQL connection pool from database configuration
127///
128/// # Arguments
129/// * `config` - Database configuration containing URL and max connections
130///
131/// # Returns
132/// * `Result<PgPool, AppError>` - Connection pool or an error
133pub async fn create_connection_pool(config: &DatabaseConfig) -> Result<PgPool, AppError> {
134    info!(
135        "Creating PostgreSQL connection pool with max {} connections",
136        config.max_connections
137    );
138
139    let pool = sqlx::postgres::PgPoolOptions::new()
140        .max_connections(config.max_connections)
141        .connect(&config.url)
142        .await
143        .map_err(AppError::Db)?;
144
145    info!("PostgreSQL connection pool created successfully");
146    Ok(pool)
147}
148
149/// Creates a database configuration from environment variables
150///
151/// # Returns
152/// * `Result<DatabaseConfig, AppError>` - Database configuration or an error
153pub fn create_database_config_from_env() -> Result<DatabaseConfig, AppError> {
154    dotenv::dotenv().ok();
155    let url = std::env::var("DATABASE_URL").map_err(|_| {
156        AppError::InvalidInput("DATABASE_URL environment variable is required".to_string())
157    })?;
158
159    let max_connections = std::env::var("DATABASE_MAX_CONNECTIONS")
160        .unwrap_or_else(|_| "10".to_string())
161        .parse::<u32>()
162        .map_err(|e| {
163            AppError::InvalidInput(format!("Invalid DATABASE_MAX_CONNECTIONS value: {e}"))
164        })?;
165
166    Ok(DatabaseConfig {
167        url,
168        max_connections,
169    })
170}
171
172#[cfg(test)]
173mod tests {
174    use super::*;
175    use serde::{Deserialize, Serialize};
176
177    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
178    struct TestStruct {
179        name: String,
180        value: i32,
181        optional: Option<f64>,
182    }
183
184    #[test]
185    fn test_serialize_to_json_simple_struct() {
186        let test_data = TestStruct {
187            name: "test".to_string(),
188            value: 42,
189            optional: Some(3.15),
190        };
191
192        let result = serialize_to_json(&test_data);
193        assert!(result.is_ok());
194
195        let json = result.expect("should serialize");
196        assert!(json.contains("\"name\":\"test\""));
197        assert!(json.contains("\"value\":42"));
198        assert!(json.contains("\"optional\":3.15"));
199    }
200
201    #[test]
202    fn test_serialize_to_json_with_none() {
203        let test_data = TestStruct {
204            name: "none_test".to_string(),
205            value: 0,
206            optional: None,
207        };
208
209        let result = serialize_to_json(&test_data);
210        assert!(result.is_ok());
211
212        let json = result.expect("should serialize");
213        assert!(json.contains("\"optional\":null"));
214    }
215
216    #[test]
217    fn test_deserialize_from_json_valid() {
218        let json = r#"{"name":"deserialized","value":100,"optional":2.5}"#;
219
220        let result: Result<TestStruct, _> = deserialize_from_json(json);
221        assert!(result.is_ok());
222
223        let data = result.expect("should deserialize");
224        assert_eq!(data.name, "deserialized");
225        assert_eq!(data.value, 100);
226        assert_eq!(data.optional, Some(2.5));
227    }
228
229    #[test]
230    fn test_deserialize_from_json_with_null() {
231        let json = r#"{"name":"null_test","value":50,"optional":null}"#;
232
233        let result: Result<TestStruct, _> = deserialize_from_json(json);
234        assert!(result.is_ok());
235
236        let data = result.expect("should deserialize");
237        assert_eq!(data.name, "null_test");
238        assert_eq!(data.value, 50);
239        assert!(data.optional.is_none());
240    }
241
242    #[test]
243    fn test_deserialize_from_json_invalid() {
244        let json = r#"{"invalid": "json for TestStruct"}"#;
245
246        let result: Result<TestStruct, _> = deserialize_from_json(json);
247        assert!(result.is_err());
248    }
249
250    #[test]
251    fn test_deserialize_from_json_malformed() {
252        let json = r#"{"name": "incomplete"#;
253
254        let result: Result<TestStruct, _> = deserialize_from_json(json);
255        assert!(result.is_err());
256    }
257
258    #[test]
259    fn test_serialize_deserialize_roundtrip() {
260        let original = TestStruct {
261            name: "roundtrip".to_string(),
262            value: 999,
263            optional: Some(1.234),
264        };
265
266        let json = serialize_to_json(&original).expect("should serialize");
267        let deserialized: TestStruct = deserialize_from_json(&json).expect("should deserialize");
268
269        assert_eq!(original, deserialized);
270    }
271
272    #[test]
273    fn test_serialize_vec() {
274        let vec = vec![
275            TestStruct {
276                name: "first".to_string(),
277                value: 1,
278                optional: None,
279            },
280            TestStruct {
281                name: "second".to_string(),
282                value: 2,
283                optional: Some(2.0),
284            },
285        ];
286
287        let result = serialize_to_json(&vec);
288        assert!(result.is_ok());
289
290        let json = result.expect("should serialize");
291        assert!(json.starts_with('['));
292        assert!(json.ends_with(']'));
293        assert!(json.contains("\"first\""));
294        assert!(json.contains("\"second\""));
295    }
296
297    #[test]
298    fn test_deserialize_vec() {
299        let json =
300            r#"[{"name":"a","value":1,"optional":null},{"name":"b","value":2,"optional":3.0}]"#;
301
302        let result: Result<Vec<TestStruct>, _> = deserialize_from_json(json);
303        assert!(result.is_ok());
304
305        let vec = result.expect("should deserialize");
306        assert_eq!(vec.len(), 2);
307        assert_eq!(vec[0].name, "a");
308        assert_eq!(vec[1].name, "b");
309    }
310
311    #[test]
312    fn test_serialize_empty_string() {
313        let test_data = TestStruct {
314            name: String::new(),
315            value: 0,
316            optional: None,
317        };
318
319        let result = serialize_to_json(&test_data);
320        assert!(result.is_ok());
321
322        let json = result.expect("should serialize");
323        assert!(json.contains("\"name\":\"\""));
324    }
325
326    #[test]
327    fn test_serialize_special_characters() {
328        let test_data = TestStruct {
329            name: "test\"with\\special\nchars".to_string(),
330            value: 0,
331            optional: None,
332        };
333
334        let result = serialize_to_json(&test_data);
335        assert!(result.is_ok());
336
337        let json = result.expect("should serialize");
338        // JSON should escape special characters
339        assert!(json.contains("\\\""));
340        assert!(json.contains("\\\\"));
341        assert!(json.contains("\\n"));
342    }
343
344    #[test]
345    fn test_database_config_creation() {
346        let config = DatabaseConfig {
347            url: "postgres://localhost/test".to_string(),
348            max_connections: 5,
349        };
350        assert_eq!(config.url, "postgres://localhost/test");
351        assert_eq!(config.max_connections, 5);
352    }
353}
ig_client/storage/utils.rs

ig_client/storage/
utils.rs
