rustcdc 0.4.0

//! Table-name glob-pattern router for CDC events.
//!
//! See [`TableRouter`] for the main entry point.

use crate::core::{Error, Event, Result};
use crate::sink::SinkAdapter;

// ─── Glob helpers ─────────────────────────────────────────────────────────────

/// Returns `true` if `s` matches the glob `pattern`.
///
/// Supported wildcards:
/// - `*`  — zero or more of any character *within a single segment*.
/// - `?`  — exactly one character.
///
/// To match across the schema/table boundary, use `*.*`.  A bare `*` pattern
/// matches any event regardless of whether it carries a schema qualifier
/// (`"*"` is a true catch-all and is the recommended default route pattern).
fn glob_segment_matches(pattern: &str, s: &str) -> bool {
    // Fast paths.
    if pattern == "*" {
        return true;
    }
    if !pattern.contains(['*', '?']) {
        return pattern == s;
    }
    glob_match(pattern.as_bytes(), s.as_bytes())
}

/// Recursive glob match: `*` = zero-or-more of any byte, `?` = any single byte.
fn glob_match(pat: &[u8], s: &[u8]) -> bool {
    match (pat.split_first(), s.split_first()) {
        (None, None) => true,
        (None, Some(_)) => false,
        (Some((&b'*', rest_pat)), _) => {
            // Try skip (consume nothing from s) or consume one byte from s.
            glob_match(rest_pat, s) || (!s.is_empty() && glob_match(pat, &s[1..]))
        }
        (Some((&b'?', rest_pat)), Some((_, rest_s))) => glob_match(rest_pat, rest_s),
        (Some((p, rest_pat)), Some((c, rest_s))) => p == c && glob_match(rest_pat, rest_s),
        _ => false,
    }
}

/// Match `pattern` against `table_key`, where `table_key` is either `"schema.table"`
/// or a bare `"table"` string (no dot).
///
/// Matching semantics:
///
/// | Pattern          | Matches                                          |
/// |------------------|--------------------------------------------------|
/// | `"*"`            | anything (bare or qualified)                     |
/// | `"*.*"`          | any qualified `schema.table`                     |
/// | `"schema.*"`     | any table in the given schema                    |
/// | `"*.table"`      | `table` in any schema                            |
/// | `"schema.table"` | exact qualified match                            |
/// | `"table"`        | exact bare-table match (no schema)               |
/// | `"pre*"`         | tables starting with `pre` (bare or right-side)  |
fn table_matches(pattern: &str, table_key: &str) -> bool {
    // A bare "*" is a true catch-all.
    if pattern == "*" {
        return true;
    }

    let pat_dot = pattern.find('.');
    let tbl_dot = table_key.find('.');

    match (pat_dot, tbl_dot) {
        // Both qualified: "schema.table" vs "schema.table"
        (Some(pi), Some(ti)) => {
            let (ps, pt) = pattern.split_at(pi);
            let (ts, tt) = table_key.split_at(ti);
            glob_segment_matches(ps, ts) && glob_segment_matches(&pt[1..], &tt[1..])
        }
        // Pattern qualified, table bare: never matches (schema required but missing).
        (Some(_), None) => false,
        // Pattern bare, table qualified: match pattern against the table-name part only.
        (None, Some(ti)) => {
            let (_, tt) = table_key.split_at(ti);
            glob_segment_matches(pattern, &tt[1..])
        }
        // Both bare.
        (None, None) => glob_segment_matches(pattern, table_key),
    }
}

// ─── TableRoute ───────────────────────────────────────────────────────────────

/// A single entry in a [`TableRouter`]: a glob pattern paired with a sink.
#[derive(Debug)]
pub struct TableRoute<S> {
    /// Glob pattern matched against the event's `schema.table` qualified name.
    pub pattern: String,
    /// Sink that receives events matching [`pattern`](Self::pattern).
    pub sink: S,
}

impl<S> TableRoute<S> {
    /// Create a new route.
    pub fn new(pattern: impl Into<String>, sink: S) -> Self {
        Self {
            pattern: pattern.into(),
            sink,
        }
    }
}

// ─── TableRouterBuilder ───────────────────────────────────────────────────────

/// Ergonomic builder for [`TableRouter`].
///
/// Obtain an instance via [`TableRouter::builder`].
pub struct TableRouterBuilder<S> {
    name: String,
    routes: Vec<TableRoute<S>>,
    default: Option<S>,
    drop_unrouted: bool,
}

impl<S: SinkAdapter> TableRouterBuilder<S> {
    fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            routes: Vec::new(),
            default: None,
            drop_unrouted: false,
        }
    }

    /// Append a route: events whose `schema.table` matches `pattern` are sent to `sink`.
    ///
    /// Routes are evaluated in the order they are added; the first matching pattern wins.
    pub fn route(mut self, pattern: impl Into<String>, sink: S) -> Self {
        self.routes.push(TableRoute::new(pattern, sink));
        self
    }

    /// Set a fallback sink for events that match no explicit pattern.
    ///
    /// When no default is provided, unmatched events are silently dropped (or return an
    /// error depending on [`drop_unrouted`](Self::drop_unrouted)).
    pub fn default(mut self, sink: S) -> Self {
        self.default = Some(sink);
        self
    }

    /// When `true` (the default), events that match no route and have no default sink are
    /// silently dropped.  When `false`, such events return a [`Error::StateError`].
    ///
    /// Silently dropping unmatched events is the right default for fan-out pipelines where
    /// you only care about a subset of tables.  Set this to `false` when you want strict
    /// auditability guarantees (every event must land somewhere).
    pub fn drop_unrouted(mut self, drop: bool) -> Self {
        self.drop_unrouted = drop;
        self
    }

    /// Build the [`TableRouter`].
    pub fn build(self) -> TableRouter<S> {
        TableRouter {
            name: self.name,
            routes: self.routes,
            default: self.default,
            drop_unrouted: self.drop_unrouted,
            closed: false,
        }
    }
}

// ─── TableRouter ──────────────────────────────────────────────────────────────

/// Routes CDC events to named sinks based on table glob patterns.
///
/// Each incoming event is compared against the registered routes (in insertion order)
/// using glob pattern matching.  The first matching sink receives the event.  If no route
/// matches and a default sink is configured, the event goes to the default sink.
/// Otherwise the event is silently dropped (or an error is returned — see
/// [`TableRouterBuilder::drop_unrouted`]).
///
/// `TableRouter` itself implements [`SinkAdapter`], so it can be nested or composed
/// freely with other pipeline components.
///
/// # Generic parameter
///
/// `TableRouter<S>` is generic over a single `S: SinkAdapter`.  All sinks in the
/// router must be the same concrete type.  When you need heterogeneous sink types,
/// wrap each behind a common enum that implements `SinkAdapter`.
///
/// # Example
///
/// ```rust,no_run
/// use rustcdc::pipeline::{TableRouter, TableRoute};
/// use rustcdc::sink::MemorySinkAdapter;
///
/// let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("demo")
///     .route("public.orders", MemorySinkAdapter::new("orders"))
///     .route("public.products", MemorySinkAdapter::new("products"))
///     .default(MemorySinkAdapter::new("fallback"))
///     .build();
/// ```
#[derive(Debug)]
pub struct TableRouter<S> {
    name: String,
    routes: Vec<TableRoute<S>>,
    default: Option<S>,
    drop_unrouted: bool,
    closed: bool,
}

impl<S: SinkAdapter> TableRouter<S> {
    /// Return an ergonomic builder.
    pub fn builder(name: impl Into<String>) -> TableRouterBuilder<S> {
        TableRouterBuilder::new(name)
    }

    /// Create a `TableRouter` from an explicit list of routes and an optional default sink.
    ///
    /// Prefer [`builder`](Self::builder) for ergonomic construction.
    pub fn new(name: impl Into<String>, routes: Vec<TableRoute<S>>, default: Option<S>) -> Self {
        Self {
            name: name.into(),
            routes,
            default,
            drop_unrouted: false,
            closed: false,
        }
    }

    /// Borrow all configured routes.
    pub fn routes(&self) -> &[TableRoute<S>] {
        &self.routes
    }

    /// Borrow the default (fallback) sink, if any.
    pub fn default_sink(&self) -> Option<&S> {
        self.default.as_ref()
    }

    /// Return the sink that would receive `event`, if any.
    ///
    /// This is a read-only probe — use [`SinkAdapter::send`] to actually deliver the event.
    pub fn route_for(&self, event: &Event) -> Option<&S> {
        let key = event.qualified_table_name();
        for route in &self.routes {
            if table_matches(&route.pattern, &key) {
                return Some(&route.sink);
            }
        }
        self.default.as_ref()
    }

    /// Flush all sinks (routes + default).
    pub async fn flush_all(&mut self) -> Result<()> {
        let mut errors: Vec<String> = Vec::new();
        for route in &mut self.routes {
            if let Err(e) = route.sink.flush().await {
                errors.push(format!("route '{}': {e}", route.pattern));
            }
        }
        if let Some(ref mut d) = self.default {
            if let Err(e) = d.flush().await {
                errors.push(format!("default: {e}"));
            }
        }
        if errors.is_empty() {
            Ok(())
        } else {
            Err(Error::StateError(errors.join("; ")))
        }
    }

    /// Close all sinks (routes + default).
    pub async fn close_all(&mut self) -> Result<()> {
        let mut errors: Vec<String> = Vec::new();
        for route in &mut self.routes {
            if let Err(e) = route.sink.close().await {
                errors.push(format!("route '{}': {e}", route.pattern));
            }
        }
        if let Some(ref mut d) = self.default {
            if let Err(e) = d.close().await {
                errors.push(format!("default: {e}"));
            }
        }
        self.closed = true;
        if errors.is_empty() {
            Ok(())
        } else {
            Err(Error::StateError(errors.join("; ")))
        }
    }
}

impl<S: SinkAdapter> SinkAdapter for TableRouter<S> {
    async fn send(&mut self, event: &Event) -> Result<()> {
        if self.closed {
            return Err(Error::StateError("TableRouter is closed".into()));
        }
        let key = event.qualified_table_name();
        for route in &mut self.routes {
            if table_matches(&route.pattern, &key) {
                return route.sink.send(event).await;
            }
        }
        if let Some(ref mut default_sink) = self.default {
            return default_sink.send(event).await;
        }
        // No match and no default.
        if self.drop_unrouted {
            Ok(())
        } else {
            Err(Error::StateError(format!(
                "TableRouter '{}': no route matched '{key}' and no default sink is configured",
                self.name,
            )))
        }
    }

    async fn flush(&mut self) -> Result<()> {
        self.flush_all().await
    }

    async fn close(&mut self) -> Result<()> {
        self.close_all().await
    }

    fn name(&self) -> &str {
        &self.name
    }
}

// ─── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::{Operation, SourceMetadata, EVENT_ENVELOPE_VERSION};
    use crate::sink::MemorySinkAdapter;

    fn make_event(schema: Option<&str>, table: &str) -> Event {
        Event {
            before: None,
            after: Some(serde_json::json!({"id": 1})),
            op: Operation::Insert,
            source: SourceMetadata {
                source_name: "test".into(),
                offset: "0".into(),
                timestamp: 0,
            },
            ts: 0,
            schema: schema.map(Into::into),
            table: table.into(),
            primary_key: None,
            snapshot: None,
            transaction: None,
            envelope_version: EVENT_ENVELOPE_VERSION,
            before_is_key_only: false,
        }
    }

    // ─── glob helpers ──────────────────────────────────────────────────────────

    #[test]
    fn glob_star_matches_anything() {
        assert!(glob_segment_matches("*", "foo"));
        assert!(glob_segment_matches("*", ""));
    }

    #[test]
    fn glob_exact_matches() {
        assert!(glob_segment_matches("orders", "orders"));
        assert!(!glob_segment_matches("orders", "order"));
    }

    #[test]
    fn glob_suffix_wildcard() {
        assert!(glob_segment_matches("order*", "orders"));
        assert!(glob_segment_matches("order*", "order_items"));
        assert!(!glob_segment_matches("order*", "my_orders"));
    }

    #[test]
    fn glob_prefix_wildcard() {
        assert!(glob_segment_matches("*_audit", "user_audit"));
        assert!(!glob_segment_matches("*_audit", "audit_log"));
    }

    #[test]
    fn glob_question_mark() {
        assert!(glob_segment_matches("t_?", "t_1"));
        assert!(glob_segment_matches("t_?", "t_a"));
        assert!(!glob_segment_matches("t_?", "t_12"));
    }

    // ─── table_matches ─────────────────────────────────────────────────────────

    #[test]
    fn catch_all_matches_qualified() {
        assert!(table_matches("*", "public.orders"));
    }

    #[test]
    fn catch_all_matches_bare() {
        assert!(table_matches("*", "orders"));
    }

    #[test]
    fn schema_wildcard_matches_correct_schema() {
        assert!(table_matches("public.*", "public.orders"));
        assert!(!table_matches("public.*", "private.orders"));
    }

    #[test]
    fn table_wildcard_matches_correct_table() {
        assert!(table_matches("*.orders", "public.orders"));
        assert!(!table_matches("*.orders", "public.products"));
    }

    #[test]
    fn exact_qualified_match() {
        assert!(table_matches("public.orders", "public.orders"));
        assert!(!table_matches("public.orders", "public.products"));
    }

    #[test]
    fn bare_pattern_matches_table_part_of_qualified() {
        // bare "orders" matches "public.orders"
        assert!(table_matches("orders", "public.orders"));
    }

    #[test]
    fn qualified_pattern_does_not_match_bare_table() {
        assert!(!table_matches("public.orders", "orders"));
    }

    // ─── TableRouter ───────────────────────────────────────────────────────────

    #[tokio::test]
    async fn routes_event_to_matching_sink() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("public.orders", MemorySinkAdapter::new("orders"))
            .route("public.products", MemorySinkAdapter::new("products"))
            .build();

        let event = make_event(Some("public"), "orders");
        router.send(&event).await.unwrap();

        let orders = router.routes()[0].sink.events();
        let products = router.routes()[1].sink.events();
        assert_eq!(orders.len(), 1);
        assert_eq!(products.len(), 0);
    }

    #[tokio::test]
    async fn unmatched_event_goes_to_default_sink() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("public.orders", MemorySinkAdapter::new("orders"))
            .default(MemorySinkAdapter::new("fallback"))
            .build();

        let event = make_event(Some("public"), "customers");
        router.send(&event).await.unwrap();

        assert_eq!(router.routes()[0].sink.events().len(), 0);
        assert_eq!(
            router
                .default_sink()
                .unwrap()
                .exported_events()
                .unwrap()
                .len(),
            1
        );
    }

    #[tokio::test]
    async fn drop_unrouted_silently_discards_event() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("public.orders", MemorySinkAdapter::new("orders"))
            .drop_unrouted(true)
            .build();

        // no default, drop_unrouted = true
        let event = make_event(Some("public"), "unrelated");
        router.send(&event).await.unwrap(); // must not error
        assert_eq!(router.routes()[0].sink.events().len(), 0);
    }

    #[tokio::test]
    async fn no_route_and_no_default_returns_error_when_strict() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("strict")
            .route("public.orders", MemorySinkAdapter::new("orders"))
            .drop_unrouted(false) // explicit strict mode
            .build();

        let event = make_event(Some("public"), "unrelated");
        let result = router.send(&event).await;
        assert!(
            result.is_err(),
            "should error on unmatched event in strict mode"
        );
    }

    #[tokio::test]
    async fn first_matching_route_wins() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("*", MemorySinkAdapter::new("catch-all"))
            .route("public.orders", MemorySinkAdapter::new("orders"))
            .build();

        let event = make_event(Some("public"), "orders");
        router.send(&event).await.unwrap();

        // catch-all is first, so it wins
        assert_eq!(router.routes()[0].sink.events().len(), 1);
        assert_eq!(router.routes()[1].sink.events().len(), 0);
    }

    #[tokio::test]
    async fn flush_all_propagates_errors() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("*", MemorySinkAdapter::new("a"))
            .build();
        // MemorySinkAdapter flush always succeeds — this just verifies no panic.
        router.flush_all().await.unwrap();
    }

    #[tokio::test]
    async fn sink_adapter_send_after_close_returns_error() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("*", MemorySinkAdapter::new("a"))
            .build();
        router.close().await.unwrap();
        let event = make_event(Some("public"), "orders");
        assert!(router.send(&event).await.is_err());
    }

    #[tokio::test]
    async fn route_for_returns_correct_sink() {
        let router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("public.orders", MemorySinkAdapter::new("orders"))
            .default(MemorySinkAdapter::new("fallback"))
            .build();

        let orders_event = make_event(Some("public"), "orders");
        let other_event = make_event(Some("public"), "customers");

        let orders_sink = router.route_for(&orders_event).unwrap();
        assert_eq!(orders_sink.name(), "orders");

        let fallback = router.route_for(&other_event).unwrap();
        assert_eq!(fallback.name(), "fallback");
    }

    #[tokio::test]
    async fn glob_prefix_route_matches_multiple_tables() {
        let mut router: TableRouter<MemorySinkAdapter> = TableRouter::builder("test")
            .route("public.order*", MemorySinkAdapter::new("orders"))
            .build();

        router
            .send(&make_event(Some("public"), "orders"))
            .await
            .unwrap();
        router
            .send(&make_event(Some("public"), "order_items"))
            .await
            .unwrap();
        assert_eq!(router.routes()[0].sink.events().len(), 2);
    }
}