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

//! Helper functions for at-least-once delivery in the pipeline

use crate::delivery::DeduplicationKey;
use crate::models::{stream_event::Event, Position};

/// Extract position from event
pub fn extract_position_from_event(event: &Event) -> Option<Position> {
    match event {
        Event::Cdc(cdc_event) => cdc_event.position.clone(),
        Event::Insert { position, .. } => position.clone(),
        Event::Update { position, .. } => position.clone(),
        Event::Delete { position, .. } => position.clone(),
        Event::Checkpoint(position) => Some(position.clone()),
        _ => None,
    }
}

/// Create deduplication key from event
pub fn create_dedup_key_from_event(event: &Event) -> DeduplicationKey {
    match event {
        Event::Cdc(cdc_event) => {
            let lsn = match &cdc_event.position {
                Some(Position::PostgreSQL { lsn }) => lsn.clone(),
                _ => String::new(),
            };

            let mut key = DeduplicationKey::new(
                lsn,
                None, // XID not available in current model
                cdc_event.table.clone(),
            );

            // Add primary key if available (assuming 'id' field)
            if let Some(pk_value) = cdc_event.data.get("id") {
                key = key.with_primary_key(pk_value.to_string());
            }

            key
        }
        Event::Insert {
            table,
            new_data,
            position,
            ..
        } => {
            let lsn = match position {
                Some(Position::PostgreSQL { lsn }) => lsn.clone(),
                _ => String::new(),
            };

            let mut key = DeduplicationKey::new(lsn, None, table.clone());

            // Add primary key if available
            if let Some(id) = new_data.get("id") {
                key = key.with_primary_key(id.to_string());
            }

            key
        }
        Event::Update {
            table,
            new_data,
            position,
            ..
        } => {
            let lsn = match position {
                Some(Position::PostgreSQL { lsn }) => lsn.clone(),
                _ => String::new(),
            };

            let mut key = DeduplicationKey::new(lsn, None, table.clone());

            // Add primary key if available
            if let Some(id) = new_data.get("id") {
                key = key.with_primary_key(id.to_string());
            }

            key
        }
        Event::Delete {
            table,
            old_data,
            position,
            ..
        } => {
            let lsn = match position {
                Some(Position::PostgreSQL { lsn }) => lsn.clone(),
                _ => String::new(),
            };

            let mut key = DeduplicationKey::new(lsn, None, table.clone());

            // Add primary key if available
            if let Some(id) = old_data.get("id") {
                key = key.with_primary_key(id.to_string());
            }

            key
        }
        _ => DeduplicationKey::new(String::new(), None, String::new()),
    }
}