microclaw_engine/internal/storage/db/
outbox.rs

1use super::*;
2
3/// A final reply queued for redelivery after a failed channel send.
4#[derive(Debug, Clone)]
5pub struct OutboxMessageRecord {
6    pub id: i64,
7    pub delivery_id: String,
8    pub chat_id: i64,
9    pub channel: String,
10    pub payload_text: String,
11    pub full_payload_text: String,
12    pub chunk_index: i64,
13    pub total_chunks: i64,
14    pub idempotency_key: String,
15    pub status: String,
16    pub attempts: i64,
17    pub next_attempt_at: Option<String>,
18    pub last_error: Option<String>,
19}
20
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct OutboundDeliveryHealth {
23    pub total_deliveries: i64,
24    pub delivered_deliveries: i64,
25    pub pending_chunks: i64,
26    pub sending_chunks: i64,
27    pub retry_chunks: i64,
28    pub failed_chunks: i64,
29    pub oldest_unfinished_at: Option<String>,
30}
31
32pub(crate) fn refresh_delivery_status(
33    tx: &Transaction<'_>,
34    chunk_id: i64,
35    now: &str,
36) -> Result<(), MicroClawError> {
37    let delivery_id: String = tx.query_row(
38        "SELECT delivery_id FROM outbound_delivery_chunks WHERE id=?1",
39        params![chunk_id],
40        |row| row.get(0),
41    )?;
42    let (total, delivered, failed): (i64, i64, i64) = tx.query_row(
43        "SELECT COUNT(*),
44                SUM(CASE WHEN status='delivered' THEN 1 ELSE 0 END),
45                SUM(CASE WHEN status='failed' THEN 1 ELSE 0 END)
46         FROM outbound_delivery_chunks WHERE delivery_id=?1",
47        params![delivery_id],
48        |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
49    )?;
50    let status = if failed > 0 {
51        "failed"
52    } else if total > 0 && delivered == total {
53        "delivered"
54    } else if delivered > 0 {
55        "partial"
56    } else {
57        "pending"
58    };
59    tx.execute(
60        "UPDATE outbound_deliveries SET status=?2, updated_at=?3 WHERE delivery_id=?1",
61        params![delivery_id, status, now],
62    )?;
63    Ok(())
64}
65
66impl Database {
67    /// Queue a final reply whose direct channel delivery failed. The outbox
68    /// flush loop retries it with backoff until delivered or terminally
69    /// failed.
70    pub fn enqueue_outbox_message(
71        &self,
72        chat_id: i64,
73        channel: &str,
74        payload_text: &str,
75    ) -> Result<i64, MicroClawError> {
76        let delivery_id = format!("delivery-{}", uuid::Uuid::new_v4());
77        let ids = self.create_outbound_delivery(
78            &delivery_id,
79            chat_id,
80            channel,
81            payload_text,
82            &[payload_text.to_string()],
83        )?;
84        Ok(ids[0])
85    }
86
87    /// Persist a complete outbound message and its independently retryable
88    /// chunks before the first network call.
89    pub fn create_outbound_delivery(
90        &self,
91        delivery_id: &str,
92        chat_id: i64,
93        channel: &str,
94        full_payload_text: &str,
95        chunks: &[String],
96    ) -> Result<Vec<i64>, MicroClawError> {
97        let mut conn = self.lock_conn();
98        let tx = conn.transaction()?;
99        let now = chrono::Utc::now().to_rfc3339();
100        tx.execute(
101            "INSERT INTO outbound_deliveries(
102                delivery_id, chat_id, channel, full_payload_text, status,
103                created_at, updated_at
104             ) VALUES(?1, ?2, ?3, ?4, 'pending', ?5, ?5)",
105            params![delivery_id, chat_id, channel, full_payload_text, now],
106        )?;
107        let total = chunks.len().max(1) as i64;
108        let mut ids = Vec::with_capacity(chunks.len());
109        for (index, chunk) in chunks.iter().enumerate() {
110            let idempotency_key = format!("{delivery_id}:{}", index + 1);
111            tx.execute(
112                "INSERT INTO outbound_delivery_chunks(
113                    delivery_id, chunk_index, total_chunks, payload_text,
114                    idempotency_key, status, attempts, next_attempt_at,
115                    created_at, updated_at
116                 ) VALUES(?1, ?2, ?3, ?4, ?5, 'pending', 0, ?6, ?6, ?6)",
117                params![
118                    delivery_id,
119                    index as i64,
120                    total,
121                    chunk,
122                    idempotency_key,
123                    now
124                ],
125            )?;
126            ids.push(tx.last_insert_rowid());
127        }
128        tx.commit()?;
129        Ok(ids)
130    }
131
132    pub fn list_due_outbox_messages(
133        &self,
134        now_iso: &str,
135        limit: usize,
136    ) -> Result<Vec<OutboxMessageRecord>, MicroClawError> {
137        let conn = self.lock_conn();
138        let mut stmt = conn.prepare(
139            "SELECT c.id, c.delivery_id, d.chat_id, d.channel,
140                    c.payload_text, d.full_payload_text, c.chunk_index,
141                    c.total_chunks, c.idempotency_key, c.status, c.attempts,
142                    c.next_attempt_at, c.last_error
143             FROM outbound_delivery_chunks c
144             JOIN outbound_deliveries d ON d.delivery_id = c.delivery_id
145             WHERE c.status IN ('pending', 'retry')
146               AND (c.next_attempt_at IS NULL OR unixepoch(c.next_attempt_at) <= unixepoch(?1))
147             ORDER BY c.id ASC
148             LIMIT ?2",
149        )?;
150        let rows = stmt.query_map(params![now_iso, limit.max(1) as i64], |row| {
151            Ok(OutboxMessageRecord {
152                id: row.get(0)?,
153                delivery_id: row.get(1)?,
154                chat_id: row.get(2)?,
155                channel: row.get(3)?,
156                payload_text: row.get(4)?,
157                full_payload_text: row.get(5)?,
158                chunk_index: row.get(6)?,
159                total_chunks: row.get(7)?,
160                idempotency_key: row.get(8)?,
161                status: row.get(9)?,
162                attempts: row.get(10)?,
163                next_attempt_at: row.get(11)?,
164                last_error: row.get(12)?,
165            })
166        })?;
167        rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
168    }
169
170    pub fn mark_outbox_delivered(&self, id: i64) -> Result<(), MicroClawError> {
171        let mut conn = self.lock_conn();
172        let tx = conn.transaction()?;
173        let now = chrono::Utc::now().to_rfc3339();
174        tx.execute(
175            "UPDATE outbound_delivery_chunks
176             SET status='delivered', updated_at=?2 WHERE id=?1",
177            params![id, now],
178        )?;
179        refresh_delivery_status(&tx, id, &now)?;
180        tx.commit()?;
181        Ok(())
182    }
183
184    pub fn mark_outbox_sending(&self, id: i64) -> Result<bool, MicroClawError> {
185        let conn = self.lock_conn();
186        let now = chrono::Utc::now().to_rfc3339();
187        let changed = conn.execute(
188            "UPDATE outbound_delivery_chunks
189             SET status='sending', updated_at=?2
190             WHERE id=?1 AND status IN ('pending', 'retry')",
191            params![id, now],
192        )?;
193        Ok(changed == 1)
194    }
195
196    pub fn mark_outbox_retry(
197        &self,
198        id: i64,
199        attempts: i64,
200        next_attempt_at: Option<&str>,
201        last_error: &str,
202        terminal_fail: bool,
203    ) -> Result<(), MicroClawError> {
204        let mut conn = self.lock_conn();
205        let tx = conn.transaction()?;
206        let now = chrono::Utc::now().to_rfc3339();
207        let status = if terminal_fail { "failed" } else { "retry" };
208        tx.execute(
209            "UPDATE outbound_delivery_chunks
210             SET status=?2, attempts=?3, next_attempt_at=?4, last_error=?5, updated_at=?6
211             WHERE id=?1",
212            params![id, status, attempts, next_attempt_at, last_error, now],
213        )?;
214        refresh_delivery_status(&tx, id, &now)?;
215        tx.commit()?;
216        Ok(())
217    }
218
219    /// Reset chunks left in `sending` by an unclean shutdown. Stable channel
220    /// idempotency keys make replay safe on supporting transports (Weixin).
221    pub fn recover_sending_outbox_messages(&self) -> Result<usize, MicroClawError> {
222        let conn = self.lock_conn();
223        let now = chrono::Utc::now().to_rfc3339();
224        let changed = conn.execute(
225            "UPDATE outbound_delivery_chunks
226             SET status='retry', next_attempt_at=?1,
227                 last_error='recovered after interrupted delivery', updated_at=?1
228             WHERE status='sending'",
229            params![now],
230        )?;
231        Ok(changed)
232    }
233
234    pub fn outbound_delivery_is_complete(&self, delivery_id: &str) -> Result<bool, MicroClawError> {
235        let conn = self.lock_conn();
236        let remaining: i64 = conn.query_row(
237            "SELECT COUNT(*) FROM outbound_delivery_chunks
238             WHERE delivery_id=?1 AND status != 'delivered'",
239            params![delivery_id],
240            |row| row.get(0),
241        )?;
242        Ok(remaining == 0)
243    }
244
245    /// Store the full logical message once, after every external chunk is
246    /// delivered. Returns true only for the caller that performed the insert.
247    pub fn finalize_outbound_delivery(
248        &self,
249        delivery_id: &str,
250        sender_name: &str,
251    ) -> Result<bool, MicroClawError> {
252        let mut conn = self.lock_conn();
253        let tx = conn.transaction()?;
254        let delivery = tx.query_row(
255            "SELECT chat_id, full_payload_text, stored_at
256             FROM outbound_deliveries WHERE delivery_id=?1",
257            params![delivery_id],
258            |row| {
259                Ok((
260                    row.get::<_, i64>(0)?,
261                    row.get::<_, String>(1)?,
262                    row.get::<_, Option<String>>(2)?,
263                ))
264            },
265        )?;
266        if delivery.2.is_some() {
267            return Ok(false);
268        }
269        let remaining: i64 = tx.query_row(
270            "SELECT COUNT(*) FROM outbound_delivery_chunks
271             WHERE delivery_id=?1 AND status != 'delivered'",
272            params![delivery_id],
273            |row| row.get(0),
274        )?;
275        if remaining != 0 {
276            return Ok(false);
277        }
278        let now = chrono::Utc::now().to_rfc3339();
279        tx.execute(
280            "INSERT INTO messages(id, chat_id, sender_name, content, is_from_bot, timestamp)
281             VALUES(?1, ?2, ?3, ?4, 1, ?5)",
282            params![
283                uuid::Uuid::new_v4().to_string(),
284                delivery.0,
285                sender_name,
286                delivery.1,
287                now
288            ],
289        )?;
290        tx.execute(
291            "UPDATE outbound_deliveries
292             SET status='delivered', stored_at=?2, updated_at=?2
293             WHERE delivery_id=?1 AND stored_at IS NULL",
294            params![delivery_id, now],
295        )?;
296        tx.commit()?;
297        Ok(true)
298    }
299
300    /// Pending + retry outbox depth (governance/observability).
301    pub fn count_outbox_pending(&self) -> Result<i64, MicroClawError> {
302        let conn = self.lock_conn();
303        let count: i64 = conn.query_row(
304            "SELECT COUNT(*) FROM outbound_delivery_chunks
305             WHERE status IN ('pending', 'retry', 'sending')",
306            [],
307            |row| row.get(0),
308        )?;
309        Ok(count)
310    }
311
312    /// Read-only delivery health snapshot for diagnostics and monitoring.
313    pub fn outbound_delivery_health(&self) -> Result<OutboundDeliveryHealth, MicroClawError> {
314        let conn = self.lock_conn();
315        let (total_deliveries, delivered_deliveries): (i64, i64) = conn.query_row(
316            "SELECT COUNT(*),
317                    COALESCE(SUM(CASE WHEN status='delivered' THEN 1 ELSE 0 END), 0)
318             FROM outbound_deliveries",
319            [],
320            |row| Ok((row.get(0)?, row.get(1)?)),
321        )?;
322        let (pending_chunks, sending_chunks, retry_chunks, failed_chunks, oldest_unfinished_at) =
323            conn.query_row(
324                "SELECT
325                    COALESCE(SUM(CASE WHEN status='pending' THEN 1 ELSE 0 END), 0),
326                    COALESCE(SUM(CASE WHEN status='sending' THEN 1 ELSE 0 END), 0),
327                    COALESCE(SUM(CASE WHEN status='retry' THEN 1 ELSE 0 END), 0),
328                    COALESCE(SUM(CASE WHEN status='failed' THEN 1 ELSE 0 END), 0),
329                    MIN(CASE WHEN status != 'delivered' THEN created_at END)
330                 FROM outbound_delivery_chunks",
331                [],
332                |row| {
333                    Ok((
334                        row.get(0)?,
335                        row.get(1)?,
336                        row.get(2)?,
337                        row.get(3)?,
338                        row.get(4)?,
339                    ))
340                },
341            )?;
342        Ok(OutboundDeliveryHealth {
343            total_deliveries,
344            delivered_deliveries,
345            pending_chunks,
346            sending_chunks,
347            retry_chunks,
348            failed_chunks,
349            oldest_unfinished_at,
350        })
351    }
352}
353
354#[cfg(test)]
355mod tests {
356    use crate::internal::storage::db::test_support::*;
357
358    #[test]
359    fn test_outbox_lifecycle() {
360        let (db, dir) = test_db();
361        assert_eq!(db.count_outbox_pending().unwrap(), 0);
362
363        let id = db
364            .enqueue_outbox_message(42, "telegram", "the answer is 42")
365            .unwrap();
366        assert_eq!(db.count_outbox_pending().unwrap(), 1);
367
368        // Due immediately (next_attempt_at = enqueue time).
369        let due = db
370            .list_due_outbox_messages("2999-01-01T00:00:00Z", 10)
371            .unwrap();
372        assert_eq!(due.len(), 1);
373        assert_eq!(due[0].id, id);
374        assert_eq!(due[0].chat_id, 42);
375        assert_eq!(due[0].payload_text, "the answer is 42");
376        assert_eq!(due[0].status, "pending");
377
378        // Retry with a future next attempt → not due before that instant.
379        db.mark_outbox_retry(id, 1, Some("2999-06-01T00:00:00Z"), "network down", false)
380            .unwrap();
381        assert!(db
382            .list_due_outbox_messages("2999-01-01T00:00:00Z", 10)
383            .unwrap()
384            .is_empty());
385        let due = db
386            .list_due_outbox_messages("2999-07-01T00:00:00Z", 10)
387            .unwrap();
388        assert_eq!(due.len(), 1);
389        assert_eq!(due[0].attempts, 1);
390        assert_eq!(due[0].last_error.as_deref(), Some("network down"));
391
392        // Delivered → gone from the queue and the pending count.
393        db.mark_outbox_delivered(id).unwrap();
394        assert!(db
395            .list_due_outbox_messages("2999-07-01T00:00:00Z", 10)
396            .unwrap()
397            .is_empty());
398        assert_eq!(db.count_outbox_pending().unwrap(), 0);
399
400        // Terminal failure also leaves the queue.
401        let id2 = db.enqueue_outbox_message(43, "slack", "bye").unwrap();
402        db.mark_outbox_retry(id2, 8, None, "gave up", true).unwrap();
403        assert!(db
404            .list_due_outbox_messages("2999-07-01T00:00:00Z", 10)
405            .unwrap()
406            .is_empty());
407        assert_eq!(db.count_outbox_pending().unwrap(), 0);
408        cleanup(&dir);
409    }
410
411    #[test]
412    fn test_chunked_delivery_resumes_and_stores_logical_message_once() {
413        let (db, dir) = test_db();
414        let ids = db
415            .create_outbound_delivery(
416                "delivery-test",
417                77,
418                "weixin",
419                "first second third",
420                &["first".into(), "second".into(), "third".into()],
421            )
422            .unwrap();
423        assert_eq!(ids.len(), 3);
424        let health = db.outbound_delivery_health().unwrap();
425        assert_eq!(health.total_deliveries, 1);
426        assert_eq!(health.pending_chunks, 3);
427
428        assert!(db.mark_outbox_sending(ids[0]).unwrap());
429        db.mark_outbox_delivered(ids[0]).unwrap();
430        assert!(db.mark_outbox_sending(ids[1]).unwrap());
431        assert_eq!(db.recover_sending_outbox_messages().unwrap(), 1);
432
433        let due = db
434            .list_due_outbox_messages("2999-01-01T00:00:00Z", 10)
435            .unwrap();
436        assert_eq!(due.len(), 2);
437        assert_eq!(due[0].chunk_index, 1);
438        assert_eq!(due[0].idempotency_key, "delivery-test:2");
439        assert_eq!(due[1].chunk_index, 2);
440
441        for row in due {
442            assert!(db.mark_outbox_sending(row.id).unwrap());
443            db.mark_outbox_delivered(row.id).unwrap();
444        }
445        assert!(db.outbound_delivery_is_complete("delivery-test").unwrap());
446        assert!(db
447            .finalize_outbound_delivery("delivery-test", "bot")
448            .unwrap());
449        assert!(!db
450            .finalize_outbound_delivery("delivery-test", "bot")
451            .unwrap());
452
453        let messages = db.get_all_messages(77).unwrap();
454        assert_eq!(messages.len(), 1);
455        assert_eq!(messages[0].content, "first second third");
456        let health = db.outbound_delivery_health().unwrap();
457        assert_eq!(health.delivered_deliveries, 1);
458        assert_eq!(
459            health.pending_chunks + health.sending_chunks + health.retry_chunks,
460            0
461        );
462        assert_eq!(health.failed_chunks, 0);
463        cleanup(&dir);
464    }
465}
microclaw_engine/internal/storage/db/outbox.rs

microclaw_engine/internal/storage/db/
outbox.rs
