meilibridge 0.1.6

High-performance PostgreSQL to Meilisearch connector
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

//! At-least-once delivery implementation for MeiliBridge
//!
//! This module provides transaction-based checkpointing and event deduplication
//! to guarantee at-least-once delivery with minimal duplicates.

pub mod checkpoint;
pub mod deduplication;
pub mod transaction;

pub use checkpoint::TransactionalCheckpoint;
pub use deduplication::{DeduplicationKey, EventDeduplicator};
pub use transaction::{TransactionCoordinator, TwoPhaseCommit};

use crate::error::Result;
use crate::models::Position;
use std::sync::Arc;
use tokio::sync::RwLock;

/// Configuration for at-least-once delivery with deduplication
#[derive(Debug, Clone)]
pub struct AtLeastOnceConfig {
    /// Enable at-least-once delivery with deduplication
    pub enabled: bool,

    /// Deduplication window size (number of events to track)
    pub deduplication_window: usize,

    /// Transaction timeout in seconds
    pub transaction_timeout_secs: u64,

    /// Enable two-phase commit protocol
    pub two_phase_commit: bool,

    /// Checkpoint before write (atomic with event processing)
    pub checkpoint_before_write: bool,
}

impl Default for AtLeastOnceConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            deduplication_window: 10000,
            transaction_timeout_secs: 30,
            two_phase_commit: true,
            checkpoint_before_write: true,
        }
    }
}

/// Manages at-least-once delivery with deduplication
pub struct AtLeastOnceManager {
    config: AtLeastOnceConfig,
    deduplicator: Arc<RwLock<EventDeduplicator>>,
    pub transaction_coordinator: Arc<TransactionCoordinator>,
}

impl AtLeastOnceManager {
    pub fn new(config: AtLeastOnceConfig) -> Self {
        let deduplicator = Arc::new(RwLock::new(EventDeduplicator::new(
            config.deduplication_window,
        )));

        let transaction_coordinator =
            Arc::new(TransactionCoordinator::new(config.transaction_timeout_secs));

        Self {
            config,
            deduplicator,
            transaction_coordinator,
        }
    }

    /// Check if an event has already been processed
    pub async fn is_duplicate(&self, key: &DeduplicationKey) -> Result<bool> {
        if !self.config.enabled {
            return Ok(false);
        }

        let dedup = self.deduplicator.read().await;
        Ok(dedup.contains(key))
    }

    /// Mark an event as processed
    pub async fn mark_processed(&self, key: DeduplicationKey) -> Result<()> {
        if !self.config.enabled {
            return Ok(());
        }

        let mut dedup = self.deduplicator.write().await;
        dedup.add(key);
        Ok(())
    }

    /// Begin a new transaction
    pub async fn begin_transaction(&self, task_id: &str) -> Result<String> {
        if !self.config.two_phase_commit {
            return Ok(String::new());
        }

        self.transaction_coordinator.begin(task_id).await
    }

    /// Prepare phase of two-phase commit
    pub async fn prepare(&self, transaction_id: &str, position: Position) -> Result<bool> {
        if !self.config.two_phase_commit {
            return Ok(true);
        }

        self.transaction_coordinator
            .prepare(transaction_id, position)
            .await
    }

    /// Commit phase of two-phase commit
    pub async fn commit(&self, transaction_id: &str) -> Result<()> {
        if !self.config.two_phase_commit {
            return Ok(());
        }

        self.transaction_coordinator.commit(transaction_id).await
    }

    /// Rollback a transaction
    pub async fn rollback(&self, transaction_id: &str) -> Result<()> {
        if !self.config.two_phase_commit {
            return Ok(());
        }

        self.transaction_coordinator.rollback(transaction_id).await
    }
}