datum-cdc 0.10.3

#![forbid(unsafe_code)]
//! Compile-checked snippets backing docs/guides/datum-cdc.md.
//!
//! These are construction/blueprint-level only: they build `CdcSource`
//! blueprints and never open a PostgreSQL connection, so they compile and run
//! in CI without a database. The connection is established at materialization,
//! not at `build()`. The pg-gated checkpoint/resume examples in the guide are
//! fenced blocks copied from `tests/pg_integration.rs` and labelled as such.

use datum_cdc::{CdcSource, ChangeEvent, MemoryCheckpointStore, PostgresCdcConfig, SlotLifecycle};

#[test]
fn cdc_source_blueprint() {
    // #region cdc-source
    use datum::Source;
    use datum_cdc::{CdcHandle, CdcResult};

    fn build_orders_source() -> CdcResult<Source<ChangeEvent, CdcHandle>> {
        CdcSource::postgres()
            // The MVP replication build is plaintext only (sslmode=disable/prefer);
            // see the TLS note in the guide.
            .connect(PostgresCdcConfig::from_url(
                "postgresql://datum_cdc@127.0.0.1:5432/datum_cdc?sslmode=disable",
            )?)
            .slot("datum_orders")
            .publication("orders_pub")
            .slot_lifecycle(SlotLifecycle::Existing)
            .buffer_capacity(8192)
            .build()
    }

    // Building the blueprint starts nothing — no PostgreSQL connection, no
    // replication task. Materialization (`run`/`run_with`) is what connects.
    let source = build_orders_source().expect("CDC blueprint builds");
    let _ = source;
    // #endregion cdc-source
}

#[test]
fn cdc_source_with_context_and_checkpoint_store() {
    // #region cdc-context
    use datum::SourceWithContext;
    use datum_cdc::{CdcHandle, CdcOffset};

    // `build_with_context()` keeps the CdcOffset attached to each event so it
    // survives context-preserving operators — this is the recommended
    // at-least-once shape. A durable checkpoint store lets the source resume
    // after a process restart.
    let source: SourceWithContext<ChangeEvent, CdcOffset, CdcHandle> = CdcSource::postgres()
        .connect(
            PostgresCdcConfig::from_url("postgresql://datum_cdc@127.0.0.1:5432/datum_cdc")
                .expect("valid url"),
        )
        .slot("datum_orders")
        .publication("orders_pub")
        .slot_lifecycle(SlotLifecycle::CreateIfMissing)
        .checkpoint_store(MemoryCheckpointStore::new())
        .build_with_context()
        .expect("CDC context blueprint builds");
    let _ = source;
    // #endregion cdc-context
}