eventuary_postgres/
reader.rs

1use std::collections::{HashMap, VecDeque};
2use std::num::NonZeroU32;
3use std::sync::Arc;
4use std::time::Duration;
5
6use chrono::{DateTime, Utc};
7use sqlx::{PgPool, Row};
8use tokio::sync::Mutex;
9use tokio::sync::Notify;
10use tokio::sync::mpsc;
11
12use eventuary_core::io::cursor::{CursorOrder, JsonCursorCodec};
13use eventuary_core::io::filter::{EventFilter, NamespacePattern, TopicPattern};
14use eventuary_core::io::reader::{
15    CoordinatedAcker, CoordinatedCursor, CoordinatedReader, CoordinatedReaderConfig,
16    CoordinatedStream, CoordinatedSubscription, PartitionAcker, PartitionedCoordAdapter,
17    PartitionedCursor,
18};
19use eventuary_core::io::stream::SpawnedStream;
20use eventuary_core::io::{Acker, Cursor, Filter, Message, Reader};
21use eventuary_core::partition::{HasPartition, Partition, PartitionGroup, PartitionSelection};
22use eventuary_core::{
23    Error, PartitionableSubscription, Result, SerializedEvent, SerializedPayload, StartFrom,
24    StartableSubscription, StopAt,
25};
26
27use crate::coordinator::PgPartitionCoordinator;
28use crate::event_log::{PgEventLogSchema, PgEventLogSchemaConfig};
29use crate::relation::PgRelationName;
30
31#[derive(
32    Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, serde::Serialize, serde::Deserialize,
33)]
34pub struct PgCursor {
35    pub sequence: i64,
36    pub partition: Partition,
37}
38
39impl PgCursor {
40    pub fn new(sequence: i64, partition: Partition) -> Self {
41        Self {
42            sequence,
43            partition,
44        }
45    }
46
47    pub fn sequence(&self) -> i64 {
48        self.sequence
49    }
50
51    pub fn partition(&self) -> Partition {
52        self.partition
53    }
54}
55
56impl Cursor for PgCursor {
57    fn order_key(&self) -> CursorOrder {
58        CursorOrder::from_i64(self.sequence)
59    }
60}
61
62impl HasPartition for PgCursor {
63    fn partition(&self) -> Partition {
64        self.partition
65    }
66}
67
68impl PgCursor {
69    pub fn codec() -> Result<JsonCursorCodec<Self>> {
70        JsonCursorCodec::new("eventuary.postgres.pg_cursor.v1")
71    }
72}
73
74#[derive(Debug, Clone)]
75pub struct PgSubscription {
76    pub start: StartFrom<PgCursor>,
77    pub stop_at: StopAt<PgCursor>,
78    pub filter: EventFilter,
79    pub batch_size: Option<usize>,
80    pub limit: Option<usize>,
81    pub partitions: PartitionSelection,
82}
83
84impl Default for PgSubscription {
85    fn default() -> Self {
86        Self {
87            start: StartFrom::Latest,
88            stop_at: StopAt::Never,
89            filter: EventFilter::default(),
90            batch_size: None,
91            limit: None,
92            partitions: PartitionSelection::All,
93        }
94    }
95}
96
97impl StartableSubscription<PgCursor> for PgSubscription {
98    fn with_start(mut self, start: StartFrom<PgCursor>) -> Self {
99        self.start = start;
100        self
101    }
102}
103
104impl PartitionableSubscription<PgCursor> for PgSubscription {
105    /// Restrict this subscription to a validated group of partitions sharing
106    /// the same `partition_count`. The reader emits a single SQL query per
107    /// poll using `partition_id = ANY($::bigint[])` instead of one query per
108    /// partition. Single-partition uses fall through the default trait impl
109    /// which wraps in a singleton group; `partition_id = ANY(ARRAY[$1])`
110    /// plans identically to `partition_id = $1` on modern Postgres.
111    fn with_partitions(mut self, group: PartitionGroup) -> Self {
112        self.partitions = PartitionSelection::Many(group);
113        self
114    }
115}
116
117#[derive(Debug, Clone)]
118pub struct PgReaderConfig {
119    pub events_relation: PgRelationName,
120    pub poll_interval: Duration,
121    pub default_batch_size: usize,
122}
123
124impl Default for PgReaderConfig {
125    fn default() -> Self {
126        Self {
127            events_relation: PgRelationName::new("events").expect("default events relation"),
128            poll_interval: Duration::from_millis(100),
129            default_batch_size: 100,
130        }
131    }
132}
133
134/// Source-side acker. Holds shared cursor state so an unacked message is
135/// re-emitted on the next stream poll instead of being dropped.
136#[derive(Clone)]
137pub struct PgCursorAcker {
138    state: Arc<Mutex<CursorState>>,
139    notify: Arc<Notify>,
140    sequence: i64,
141}
142
143struct CursorState {
144    last_acked: i64,
145    pending_nack: bool,
146}
147
148impl PgCursorAcker {
149    #[doc(hidden)]
150    pub fn dummy(sequence: i64) -> Self {
151        Self {
152            state: Arc::new(Mutex::new(CursorState {
153                last_acked: 0,
154                pending_nack: false,
155            })),
156            notify: Arc::new(Notify::new()),
157            sequence,
158        }
159    }
160}
161
162impl Acker for PgCursorAcker {
163    async fn ack(&self) -> Result<()> {
164        let mut state = self.state.lock().await;
165        if self.sequence > state.last_acked {
166            state.last_acked = self.sequence;
167        }
168        state.pending_nack = false;
169        self.notify.notify_waiters();
170        Ok(())
171    }
172
173    async fn nack(&self) -> Result<()> {
174        let mut state = self.state.lock().await;
175        state.pending_nack = true;
176        self.notify.notify_waiters();
177        Ok(())
178    }
179}
180
181#[derive(Clone)]
182pub struct PgReader {
183    pool: PgPool,
184    config: PgReaderConfig,
185}
186
187impl PgReader {
188    pub fn new(pool: PgPool, config: PgReaderConfig) -> Self {
189        Self { pool, config }
190    }
191
192    pub async fn connect(pool: PgPool, config: PgReaderConfig) -> Result<Self> {
193        Self::prepare_schema(&pool, &config).await?;
194        Ok(Self::new(pool, config))
195    }
196
197    pub async fn prepare_schema(pool: &PgPool, config: &PgReaderConfig) -> Result<()> {
198        PgEventLogSchema::prepare(
199            pool,
200            &PgEventLogSchemaConfig {
201                events_relation: config.events_relation.clone(),
202            },
203        )
204        .await
205    }
206
207    pub fn schema_sql(config: &PgReaderConfig) -> String {
208        PgEventLogSchema::schema_sql(&PgEventLogSchemaConfig {
209            events_relation: config.events_relation.clone(),
210        })
211    }
212}
213
214impl Reader for PgReader {
215    type Subscription = PgSubscription;
216    type Acker = PgCursorAcker;
217    type Cursor = PgCursor;
218    type Stream = SpawnedStream<PgCursorAcker, PgCursor>;
219
220    async fn read(&self, subscription: Self::Subscription) -> Result<Self::Stream> {
221        let pool = self.pool.clone();
222        let config = self.config.clone();
223        let (tx, rx) = mpsc::channel(64);
224        let events_relation = config.events_relation.render();
225        let poll_interval = config.poll_interval;
226        let batch_size = subscription
227            .batch_size
228            .unwrap_or(config.default_batch_size)
229            .clamp(1, 1000);
230        let filter = subscription.filter.clone();
231        let limit = subscription.limit;
232        let partitions = subscription.partitions.clone();
233
234        let (mut after_seq, lower_bound_ts) =
235            match resolve_initial_position(&pool, &events_relation, &subscription).await {
236                Ok(pos) => pos,
237                Err(e) => {
238                    let _ = tx.send(Err(e)).await;
239                    return Ok(SpawnedStream::from_receiver(rx));
240                }
241            };
242
243        let stop_seq = match resolve_stop_position(&pool, &events_relation, &subscription).await {
244            Ok(pos) => pos,
245            Err(e) => {
246                let _ = tx.send(Err(e)).await;
247                return Ok(SpawnedStream::from_receiver(rx));
248            }
249        };
250
251        let state = Arc::new(Mutex::new(CursorState {
252            last_acked: after_seq,
253            pending_nack: false,
254        }));
255        let notify = Arc::new(Notify::new());
256
257        let handle = tokio::spawn(async move {
258            let mut delivered = 0usize;
259            let mut buffer: VecDeque<(SerializedEvent, i64, Partition)> = VecDeque::new();
260            loop {
261                if buffer.is_empty() {
262                    let fetched = match fetch_batch(
263                        &pool,
264                        FetchBatchParams {
265                            events_relation: &events_relation,
266                            after_seq,
267                            stop_seq,
268                            take: batch_size,
269                            lower_bound_ts,
270                            filter: &filter,
271                            partitions: &partitions,
272                        },
273                    )
274                    .await
275                    {
276                        Ok(b) => b,
277                        Err(e) => {
278                            let _ = tx.send(Err(e)).await;
279                            return;
280                        }
281                    };
282                    if fetched.is_empty() {
283                        if stop_seq.is_some() {
284                            return;
285                        }
286                        tokio::time::sleep(poll_interval).await;
287                        continue;
288                    }
289                    buffer.extend(fetched);
290                }
291
292                while let Some((serialized, sequence, partition)) = buffer.front() {
293                    let sequence = *sequence;
294                    let partition = *partition;
295                    let event = match serialized.to_event() {
296                        Ok(e) => e,
297                        Err(e) => {
298                            let _ = tx
299                                .send(Err(Error::Serialization(format!(
300                                    "decode event at sequence {sequence}: {e}"
301                                ))))
302                                .await;
303                            return;
304                        }
305                    };
306                    if !filter.matches(&event) {
307                        buffer.pop_front();
308                        after_seq = sequence;
309                        continue;
310                    }
311                    if let Some(l) = limit
312                        && delivered >= l
313                    {
314                        return;
315                    }
316                    let acker = PgCursorAcker {
317                        state: Arc::clone(&state),
318                        notify: Arc::clone(&notify),
319                        sequence,
320                    };
321                    let cursor = PgCursor {
322                        sequence,
323                        partition,
324                    };
325                    if tx
326                        .send(Ok(Message::new(event, acker, cursor)))
327                        .await
328                        .is_err()
329                    {
330                        return;
331                    }
332                    delivered += 1;
333
334                    loop {
335                        {
336                            let guard = state.lock().await;
337                            if guard.last_acked >= sequence {
338                                after_seq = sequence;
339                                buffer.pop_front();
340                                break;
341                            }
342                            if guard.pending_nack {
343                                break;
344                            }
345                            if tx.is_closed() {
346                                return;
347                            }
348                        }
349                        notify.notified().await;
350                    }
351                }
352            }
353        });
354
355        Ok(SpawnedStream::new(rx, handle))
356    }
357}
358
359pub type PgPartitionedCursor = PartitionedCursor<PgCursor>;
360pub type PgCoordinatedReaderConfig = CoordinatedReaderConfig;
361pub type PgCoordinatedSubscription = CoordinatedSubscription<PgSubscription, PgCursor>;
362pub type PgCoordinatedReader = CoordinatedReader<PgReader, PgPartitionCoordinator>;
363/// Standalone `PartitionLease`-fenced acker over the raw `PgCursor`. This
364/// alias matches the simple shape used by code paths that wire a coordinator
365/// outside of `CoordinatedReader::read`. The stream-emitted acker after the
366/// shared-fetch rewrite is [`PgCoordinatedStreamAcker`].
367pub type PgCoordinatedAcker = CoordinatedAcker<PgCursorAcker, PgCursor, PgPartitionCoordinator>;
368/// Acker carried on every message emitted by [`PgCoordinatedReader`].
369pub type PgCoordinatedStreamAcker = CoordinatedAcker<
370    PartitionAcker<PgCursorAcker, PgCursor>,
371    PartitionedCursor<PgCursor>,
372    PartitionedCoordAdapter<PgPartitionCoordinator, PgCursor>,
373>;
374pub type PgCoordinatedCursor = CoordinatedCursor<PartitionedCursor<PgCursor>>;
375pub type PgCoordinatedStream = CoordinatedStream<
376    PartitionAcker<PgCursorAcker, PgCursor>,
377    PartitionedCursor<PgCursor>,
378    PartitionedCoordAdapter<PgPartitionCoordinator, PgCursor>,
379>;
380
381async fn resolve_initial_position(
382    pool: &PgPool,
383    events_relation: &str,
384    subscription: &PgSubscription,
385) -> Result<(i64, Option<DateTime<Utc>>)> {
386    match subscription.start.clone() {
387        StartFrom::After(cursor) => Ok((cursor.sequence, None)),
388        StartFrom::Earliest => Ok((0, None)),
389        StartFrom::Latest => {
390            let sql = match subscription.filter.organization.as_ref() {
391                Some(_) => format!(
392                    "SELECT COALESCE(MAX(sequence), 0) AS s FROM {events_relation} WHERE organization = $1",
393                ),
394                None => format!("SELECT COALESCE(MAX(sequence), 0) AS s FROM {events_relation}"),
395            };
396            let mut q = sqlx::query(&sql);
397            if let Some(org) = subscription.filter.organization.as_ref() {
398                q = q.bind(org.as_str());
399            }
400            let row = q
401                .fetch_one(pool)
402                .await
403                .map_err(|e| Error::Store(e.to_string()))?;
404            Ok((row.get::<i64, _>("s"), None))
405        }
406        StartFrom::Timestamp(ts) => {
407            let sql = match subscription.filter.organization.as_ref() {
408                Some(_) => format!(
409                    "SELECT COALESCE(MIN(sequence), 1) - 1 AS s FROM {events_relation} \
410                     WHERE organization = $1 AND timestamp >= $2::timestamptz",
411                ),
412                None => format!(
413                    "SELECT COALESCE(MIN(sequence), 1) - 1 AS s FROM {events_relation} \
414                     WHERE timestamp >= $1::timestamptz",
415                ),
416            };
417            let mut q = sqlx::query(&sql);
418            if let Some(org) = subscription.filter.organization.as_ref() {
419                q = q.bind(org.as_str());
420            }
421            q = q.bind(ts.to_rfc3339());
422            let row = q
423                .fetch_one(pool)
424                .await
425                .map_err(|e| Error::Store(e.to_string()))?;
426            Ok((row.get::<i64, _>("s").max(0), Some(ts)))
427        }
428    }
429}
430
431async fn resolve_stop_position(
432    pool: &PgPool,
433    events_relation: &str,
434    subscription: &PgSubscription,
435) -> Result<Option<i64>> {
436    match subscription.stop_at {
437        StopAt::Never => Ok(None),
438        StopAt::Cursor(cursor) => Ok(Some(cursor.sequence)),
439        StopAt::CurrentEnd => {
440            let sql = match subscription.filter.organization.as_ref() {
441                Some(_) => format!(
442                    "SELECT COALESCE(MAX(sequence), 0) AS s FROM {events_relation} WHERE organization = $1",
443                ),
444                None => format!("SELECT COALESCE(MAX(sequence), 0) AS s FROM {events_relation}"),
445            };
446            let mut query = sqlx::query(&sql);
447            if let Some(org) = subscription.filter.organization.as_ref() {
448                query = query.bind(org.as_str());
449            }
450            let row = query
451                .fetch_one(pool)
452                .await
453                .map_err(|e| Error::Store(e.to_string()))?;
454            Ok(Some(row.get::<i64, _>("s")))
455        }
456    }
457}
458
459struct FetchBatchParams<'a> {
460    events_relation: &'a str,
461    after_seq: i64,
462    stop_seq: Option<i64>,
463    take: usize,
464    lower_bound_ts: Option<DateTime<Utc>>,
465    filter: &'a EventFilter,
466    partitions: &'a PartitionSelection,
467}
468
469async fn fetch_batch(
470    pool: &PgPool,
471    p: FetchBatchParams<'_>,
472) -> Result<Vec<(SerializedEvent, i64, Partition)>> {
473    let FetchBatchParams {
474        events_relation,
475        after_seq,
476        stop_seq,
477        take,
478        lower_bound_ts,
479        filter,
480        partitions,
481    } = p;
482    let mut sql = format!(
483        "SELECT sequence, id::text AS id_text, organization, namespace, topic, event_key, \
484         payload::text AS payload_text, content_type, metadata::text AS metadata_text, \
485         timestamp::text AS timestamp_text, version, parent_id::text AS parent_id_text, \
486         correlation_id, causation_id, partition_id, partition_count \
487         FROM {events_relation} WHERE sequence > $1",
488    );
489    let mut bind_index = 2usize;
490
491    if stop_seq.is_some() {
492        sql.push_str(&format!(" AND sequence <= ${bind_index}"));
493        bind_index += 1;
494    }
495
496    if filter.organization.is_some() {
497        sql.push_str(&format!(" AND organization = ${bind_index}"));
498        bind_index += 1;
499    }
500
501    let exact_topic: Option<String> = filter.topic.as_ref().map(|p| match p {
502        TopicPattern::Exact(t) => t.as_str().to_owned(),
503    });
504    if exact_topic.is_some() {
505        sql.push_str(&format!(" AND topic = ${bind_index}"));
506        bind_index += 1;
507    }
508    let ns_filter = filter.namespace.as_ref().and_then(|p| match p {
509        NamespacePattern::Prefix(ns) if !ns.is_root() => Some(ns.as_str().to_owned()),
510        _ => None,
511    });
512    if ns_filter.is_some() {
513        sql.push_str(&format!(
514            " AND (namespace = ${bind_index} OR namespace LIKE ${bind_index} || '/%')"
515        ));
516        bind_index += 1;
517    }
518    if lower_bound_ts.is_some() {
519        sql.push_str(&format!(" AND timestamp >= ${bind_index}::timestamptz"));
520        bind_index += 1;
521    }
522    match partitions {
523        PartitionSelection::All => {}
524        PartitionSelection::One(_) => {
525            sql.push_str(&format!(
526                " AND partition_count = ${bind_index} AND partition_id = ${}",
527                bind_index + 1
528            ));
529            bind_index += 2;
530        }
531        PartitionSelection::Many(_) => {
532            sql.push_str(&format!(
533                " AND partition_count = ${bind_index} AND partition_id = ANY(${}::bigint[])",
534                bind_index + 1
535            ));
536            bind_index += 2;
537        }
538    }
539    sql.push_str(&format!(" ORDER BY sequence ASC LIMIT ${bind_index}"));
540
541    let mut q = sqlx::query(&sql).bind(after_seq);
542
543    if let Some(stop_seq) = stop_seq {
544        q = q.bind(stop_seq);
545    }
546
547    if let Some(org) = &filter.organization {
548        q = q.bind(org.as_str());
549    }
550    if let Some(topic) = exact_topic {
551        q = q.bind(topic);
552    }
553    if let Some(prefix) = ns_filter {
554        q = q.bind(prefix);
555    }
556    if let Some(ts) = lower_bound_ts {
557        q = q.bind(ts.to_rfc3339());
558    }
559    match partitions {
560        PartitionSelection::All => {}
561        PartitionSelection::One(partition) => {
562            q = q.bind(partition.count() as i64);
563            q = q.bind(partition.id() as i64);
564        }
565        PartitionSelection::Many(group) => {
566            q = q.bind(group.count() as i64);
567            let ids: Vec<i64> = group.partitions().iter().map(|p| p.id() as i64).collect();
568            q = q.bind(ids);
569        }
570    }
571    q = q.bind(take as i64);
572
573    let rows = q
574        .fetch_all(pool)
575        .await
576        .map_err(|e| Error::Store(e.to_string()))?;
577
578    rows.into_iter()
579        .map(|row| {
580            let sequence: i64 = row.get("sequence");
581            let id_text: String = row.get("id_text");
582            let id = uuid::Uuid::parse_str(&id_text)
583                .map_err(|e| Error::Serialization(format!("decode id: {e}")))?;
584            let parent_id = row
585                .get::<Option<String>, _>("parent_id_text")
586                .as_deref()
587                .map(uuid::Uuid::parse_str)
588                .transpose()
589                .map_err(|e| Error::Serialization(format!("decode parent_id: {e}")))?;
590            let payload_str: String = row.get("payload_text");
591            let payload: SerializedPayload = serde_json::from_str(&payload_str)
592                .map_err(|e| Error::Serialization(format!("decode payload: {e}")))?;
593            let metadata_str: String = row.get("metadata_text");
594            let metadata: HashMap<String, String> = serde_json::from_str(&metadata_str)
595                .map_err(|e| Error::Serialization(format!("decode metadata: {e}")))?;
596            let timestamp_str: String = row.get("timestamp_text");
597            let timestamp = parse_pg_timestamp(&timestamp_str).map_err(|e| {
598                Error::Serialization(format!("decode timestamp at sequence {sequence}: {e}"))
599            })?;
600            let serialized = SerializedEvent {
601                id,
602                organization: row.get("organization"),
603                namespace: row.get("namespace"),
604                topic: row.get("topic"),
605                payload,
606                metadata,
607                timestamp,
608                version: row.get::<i64, _>("version") as u64,
609                key: row.get("event_key"),
610                parent_id,
611                correlation_id: row.get("correlation_id"),
612                causation_id: row.get("causation_id"),
613            };
614            let partition_id: Option<i64> = row.get("partition_id");
615            let partition_count: Option<i64> = row.get("partition_count");
616            let partition = decode_partition(partition_id, partition_count)?;
617            Ok((serialized, sequence, partition))
618        })
619        .collect()
620}
621
622fn decode_partition(partition_id: Option<i64>, partition_count: Option<i64>) -> Result<Partition> {
623    match (partition_id, partition_count) {
624        (Some(id), Some(count)) => {
625            let id = u32::try_from(id)
626                .map_err(|_| Error::Store(format!("partition_id {id} exceeds u32::MAX")))?;
627            let count = u32::try_from(count)
628                .map_err(|_| Error::Store(format!("partition_count {count} exceeds u32::MAX")))?;
629            let count = NonZeroU32::new(count)
630                .ok_or_else(|| Error::Store("partition_count must be positive".to_owned()))?;
631            Partition::new(id, count).map_err(|e| Error::Store(format!("invalid partition: {e}")))
632        }
633        (None, None) => Ok(Partition::new(0, NonZeroU32::new(1).unwrap())
634            .expect("synthetic single partition is valid")),
635        _ => Err(Error::Store(
636            "event has incomplete partition columns — run partition backfill first".to_owned(),
637        )),
638    }
639}
640
641fn parse_pg_timestamp(s: &str) -> std::result::Result<DateTime<Utc>, chrono::ParseError> {
642    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
643        return Ok(dt.with_timezone(&Utc));
644    }
645    DateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f%#z").map(|dt| dt.with_timezone(&Utc))
646}
647
648#[cfg(test)]
649mod tests {
650    use super::*;
651    use eventuary_core::io::cursor::{CursorCodec, CursorOrder};
652    use eventuary_core::io::{Cursor, CursorId};
653
654    fn test_partition() -> Partition {
655        Partition::new(0, NonZeroU32::new(1).unwrap()).unwrap()
656    }
657
658    #[test]
659    fn pg_subscription_with_partition_wraps_in_singleton_partition_group() {
660        use eventuary_core::PartitionableSubscription;
661        let count = NonZeroU32::new(8).unwrap();
662        let partition = Partition::new(3, count).unwrap();
663        let sub = PgSubscription::default().with_partition(partition);
664        match sub.partitions {
665            PartitionSelection::Many(g) => {
666                assert_eq!(g.len(), 1);
667                assert_eq!(g.partitions()[0].id(), 3);
668                assert_eq!(g.count(), 8);
669            }
670            _ => panic!("expected PartitionSelection::Many(singleton)"),
671        }
672    }
673
674    #[test]
675    fn pg_subscription_with_partitions_sets_partition_selection_many() {
676        let count = NonZeroU32::new(8).unwrap();
677        let group = PartitionGroup::new(vec![
678            Partition::new(1, count).unwrap(),
679            Partition::new(4, count).unwrap(),
680            Partition::new(7, count).unwrap(),
681        ])
682        .unwrap();
683        let sub = PgSubscription::default().with_partitions(group);
684        match sub.partitions {
685            PartitionSelection::Many(g) => {
686                assert_eq!(g.len(), 3);
687                assert_eq!(g.count(), 8);
688                let ids: Vec<u32> = g.partitions().iter().map(|p| p.id()).collect();
689                assert_eq!(ids, vec![1, 4, 7]);
690            }
691            _ => panic!("expected PartitionSelection::Many"),
692        }
693    }
694
695    #[test]
696    fn pg_cursor_id_is_global() {
697        assert_eq!(PgCursor::new(42, test_partition()).id(), CursorId::global());
698    }
699
700    #[test]
701    fn pg_cursor_order_key_from_sequence() {
702        assert_eq!(
703            PgCursor::new(42, test_partition()).order_key(),
704            CursorOrder::from_i64(42)
705        );
706        assert!(
707            PgCursor::new(9, test_partition()).order_key()
708                < PgCursor::new(10, test_partition()).order_key()
709        );
710    }
711
712    #[test]
713    fn pg_cursor_codec_roundtrips() {
714        let codec = PgCursor::codec().unwrap();
715        let cursor = PgCursor::new(42, test_partition());
716        let encoded = codec.encode(&cursor).unwrap();
717        assert_eq!(encoded.kind().as_str(), "eventuary.postgres.pg_cursor.v1");
718        assert_eq!(encoded.order(), &CursorOrder::from_i64(42));
719        assert_eq!(codec.decode(&encoded).unwrap(), cursor);
720    }
721
722    #[test]
723    fn pg_cursor_codec_preserves_typed_ord() {
724        let codec = PgCursor::codec().unwrap();
725        let lo = codec.encode(&PgCursor::new(9, test_partition())).unwrap();
726        let hi = codec.encode(&PgCursor::new(10, test_partition())).unwrap();
727        assert!(lo < hi);
728    }
729}
eventuary_postgres/reader.rs

eventuary_postgres/
reader.rs
