1use crate::{ConxianError, ConxianResult, Persistence, PersistentState};
2use std::fs;
3use std::path::PathBuf;
4
5pub struct FilePersistence {
6 path: PathBuf,
7}
8
9impl FilePersistence {
10 pub fn new(path: &str) -> Self {
11 Self {
12 path: PathBuf::from(path),
13 }
14 }
15}
16
17impl Persistence for FilePersistence {
18 fn save(&self, state: &PersistentState) -> ConxianResult<()> {
19 let json = serde_json::to_string(state)
20 .map_err(|e| ConxianError::Io(format!("Serialization failed: {}", e)))?;
21
22 let tmp_path = self.path.with_extension("tmp");
23 fs::write(&tmp_path, json)
24 .map_err(|e| ConxianError::Io(format!("Write to temporary file failed: {}", e)))?;
25
26 fs::rename(&tmp_path, &self.path)
27 .map_err(|e| ConxianError::Io(format!("Atomic rename failed: {}", e)))?;
28
29 Ok(())
30 }
31
32 fn load(&self) -> ConxianResult<PersistentState> {
33 if !self.path.exists() {
34 return Ok(PersistentState::default());
35 }
36
37 let content = fs::read_to_string(&self.path)
38 .map_err(|e| ConxianError::Io(format!("Read failed: {}", e)))?;
39
40 let state: PersistentState = serde_json::from_str(&content)
41 .map_err(|e| ConxianError::Io(format!("Deserialization failed: {}", e)))?;
42
43 Ok(state)
44 }
45}
46
47use aes_gcm::{
48 aead::{Aead, KeyInit},
49 Aes256Gcm, Nonce,
50};
51use rusqlite::{params, Connection};
52use std::sync::Mutex;
53
54pub struct EncryptedOfflineQueue {
55 conn: Mutex<Connection>,
56 encryption_key: [u8; 32],
57}
58
59impl EncryptedOfflineQueue {
60 pub fn new(path: &str, key: [u8; 32]) -> ConxianResult<Self> {
61 let conn = Connection::open(path)
62 .map_err(|e| ConxianError::Io(format!("Failed to open SQLite: {}", e)))?;
63
64 conn.execute(
65 "CREATE TABLE IF NOT EXISTS offline_receipts (
66 id TEXT PRIMARY KEY,
67 encrypted_payload BLOB NOT NULL,
68 nonce BLOB NOT NULL,
69 status TEXT NOT NULL,
70 timestamp INTEGER NOT NULL
71 )",
72 [],
73 )
74 .map_err(|e| ConxianError::Io(format!("Failed to create table: {}", e)))?;
75
76 conn.execute(
77 "CREATE TABLE IF NOT EXISTS webhook_replay_keys (
78 replay_key TEXT PRIMARY KEY,
79 expires_at INTEGER NOT NULL,
80 created_at INTEGER NOT NULL
81 )",
82 [],
83 )
84 .map_err(|e| ConxianError::Io(format!("Failed to create replay table: {}", e)))?;
85
86 conn.execute(
87 "CREATE INDEX IF NOT EXISTS idx_webhook_replay_keys_expires_at
88 ON webhook_replay_keys (expires_at)",
89 [],
90 )
91 .map_err(|e| ConxianError::Io(format!("Failed to create replay index: {}", e)))?;
92
93 Ok(Self {
94 conn: Mutex::new(conn),
95 encryption_key: key,
96 })
97 }
98
99 fn encrypt(&self, data: &[u8]) -> ConxianResult<(Vec<u8>, [u8; 12])> {
100 let cipher = Aes256Gcm::new_from_slice(&self.encryption_key)
101 .map_err(|e| ConxianError::Security(format!("Cipher init failed: {}", e)))?;
102
103 let mut nonce_bytes = [0u8; 12];
104 let now = std::time::SystemTime::now()
105 .duration_since(std::time::UNIX_EPOCH)
106 .unwrap()
107 .as_nanos();
108 let now_bytes = now.to_le_bytes();
109 nonce_bytes.copy_from_slice(&now_bytes[..12]);
110 let nonce = Nonce::from_slice(&nonce_bytes);
111
112 let ciphertext = cipher
113 .encrypt(nonce, data)
114 .map_err(|e| ConxianError::Security(format!("Encryption failed: {}", e)))?;
115
116 Ok((ciphertext, nonce_bytes))
117 }
118
119 fn decrypt(&self, ciphertext: &[u8], nonce_bytes: &[u8]) -> ConxianResult<Vec<u8>> {
120 let cipher = Aes256Gcm::new_from_slice(&self.encryption_key)
121 .map_err(|e| ConxianError::Security(format!("Cipher init failed: {}", e)))?;
122 let nonce = Nonce::from_slice(nonce_bytes);
123
124 cipher
125 .decrypt(nonce, ciphertext)
126 .map_err(|e| ConxianError::Security(format!("Decryption failed: {}", e)))
127 }
128}
129
130impl crate::OfflineQueue for EncryptedOfflineQueue {
131 fn enqueue(&self, receipt: &crate::OfflineReceipt) -> ConxianResult<()> {
132 let json = serde_json::to_vec(receipt)
133 .map_err(|e| ConxianError::Io(format!("Serialization failed: {}", e)))?;
134
135 let (encrypted, nonce) = self.encrypt(&json)?;
136 let status = format!("{:?}", receipt.status);
137 let timestamp = i64::try_from(receipt.timestamp)
138 .map_err(|_| ConxianError::Io("Timestamp overflow".to_string()))?;
139
140 let conn = self.conn.lock().expect("lock poisoned");
141 conn.execute(
142 "INSERT INTO offline_receipts (id, encrypted_payload, nonce, status, timestamp) VALUES (?1, ?2, ?3, ?4, ?5)",
143 params![receipt.receipt_id, encrypted, nonce.to_vec(), status, timestamp],
144 ).map_err(|e| ConxianError::Io(format!("Enqueue failed: {}", e)))?;
145
146 Ok(())
147 }
148
149 fn dequeue_pending(&self) -> ConxianResult<Vec<crate::OfflineReceipt>> {
150 let conn = self.conn.lock().expect("lock poisoned");
151 let mut stmt = conn
152 .prepare("SELECT encrypted_payload, nonce FROM offline_receipts WHERE UPPER(status) = 'PENDING' OR UPPER(status) = 'GOSSIPED'")
153 .map_err(|e| ConxianError::Io(format!("Prepare failed: {}", e)))?;
154
155 let rows = stmt
156 .query_map([], |row| {
157 Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, Vec<u8>>(1)?))
158 })
159 .map_err(|e| ConxianError::Io(format!("Query failed: {}", e)))?;
160
161 let mut receipts = Vec::new();
162 for row in rows {
163 let (encrypted, nonce): (Vec<u8>, Vec<u8>) =
164 row.map_err(|e| ConxianError::Io(format!("Row error: {}", e)))?;
165 let decrypted = self.decrypt(&encrypted, &nonce)?;
166 let receipt: crate::OfflineReceipt = serde_json::from_slice(&decrypted)
167 .map_err(|e| ConxianError::Io(format!("Deserialization failed: {}", e)))?;
168 receipts.push(receipt);
169 }
170
171 Ok(receipts)
172 }
173
174 fn mark_broadcasted(&self, receipt_id: &str) -> ConxianResult<()> {
175 let conn = self.conn.lock().expect("lock poisoned");
176 conn.execute(
177 "UPDATE offline_receipts SET status = 'BROADCASTED' WHERE id = ?1",
178 params![receipt_id],
179 )
180 .map_err(|e| ConxianError::Io(format!("Update failed: {}", e)))?;
181 Ok(())
182 }
183
184 fn claim_replay_key(&self, replay_key: &str, ttl_seconds: u64) -> ConxianResult<bool> {
185 if replay_key.trim().is_empty() {
186 return Err(ConxianError::Persistence(
187 "Replay key cannot be empty".to_string(),
188 ));
189 }
190
191 if ttl_seconds == 0 {
192 return Err(ConxianError::Persistence(
193 "Replay key TTL must be greater than zero".to_string(),
194 ));
195 }
196
197 let now_seconds = std::time::SystemTime::now()
198 .duration_since(std::time::UNIX_EPOCH)
199 .map_err(|e| ConxianError::Persistence(format!("Clock error: {}", e)))?
200 .as_secs();
201
202 let expires_at = now_seconds
203 .checked_add(ttl_seconds)
204 .ok_or_else(|| ConxianError::Persistence("Replay key TTL overflow".to_string()))?;
205
206 let now_i64 = i64::try_from(now_seconds)
207 .map_err(|_| ConxianError::Persistence("Timestamp overflow".to_string()))?;
208 let expires_at_i64 = i64::try_from(expires_at)
209 .map_err(|_| ConxianError::Persistence("Expiry timestamp overflow".to_string()))?;
210
211 let conn = self.conn.lock().expect("lock poisoned");
212
213 conn.execute(
214 "DELETE FROM webhook_replay_keys WHERE expires_at < ?1",
215 params![now_i64],
216 )
217 .map_err(|e| ConxianError::Persistence(format!("Replay cleanup failed: {}", e)))?;
218
219 let affected = conn
220 .execute(
221 "INSERT INTO webhook_replay_keys (replay_key, expires_at, created_at)
222 VALUES (?1, ?2, ?3)
223 ON CONFLICT(replay_key) DO UPDATE SET
224 expires_at = excluded.expires_at,
225 created_at = excluded.created_at
226 WHERE webhook_replay_keys.expires_at < excluded.created_at",
227 params![replay_key, expires_at_i64, now_i64],
228 )
229 .map_err(|e| ConxianError::Persistence(format!("Replay claim failed: {}", e)))?;
230
231 Ok(affected > 0)
232 }
233}
234
235#[cfg(test)]
236mod tests {
237 use super::EncryptedOfflineQueue;
238 use crate::OfflineQueue;
239 use std::time::{Duration, SystemTime, UNIX_EPOCH};
240
241 fn tmp_db_path(prefix: &str) -> String {
242 let suffix = SystemTime::now()
243 .duration_since(UNIX_EPOCH)
244 .expect("system clock moved backwards")
245 .as_nanos();
246 format!("{}_{}.db", prefix, suffix)
247 }
248
249 #[test]
250 fn claim_replay_key_rejects_duplicate_until_expiry() {
251 let db_path = tmp_db_path("replay_claim");
252 let queue = EncryptedOfflineQueue::new(&db_path, [7u8; 32]).unwrap();
253
254 assert!(queue.claim_replay_key("ramp:sig:hash", 1).unwrap());
255 assert!(!queue.claim_replay_key("ramp:sig:hash", 1).unwrap());
256
257 std::thread::sleep(Duration::from_secs(2));
258 assert!(queue.claim_replay_key("ramp:sig:hash", 1).unwrap());
259
260 let _ = std::fs::remove_file(db_path);
261 }
262
263 #[test]
264 fn claim_replay_key_allows_distinct_keys() {
265 let db_path = tmp_db_path("replay_distinct");
266 let queue = EncryptedOfflineQueue::new(&db_path, [3u8; 32]).unwrap();
267
268 assert!(queue.claim_replay_key("ramp:sig1:hash", 60).unwrap());
269 assert!(queue.claim_replay_key("ramp:sig2:hash", 60).unwrap());
270
271 let _ = std::fs::remove_file(db_path);
272 }
273}