Enum WatchLevel 

pub enum WatchLevel {
    Collection(Vec<String>),
    Database,
    Deployment,
}

Defines the scope of collections to watch for changes.

This enum controls whether Rigatoni watches specific collections, an entire database, or all databases in a deployment.

Examples

use rigatoni_core::watch_level::WatchLevel;

// Watch specific collections
let collections = WatchLevel::Collection(vec!["users".to_string(), "orders".to_string()]);

// Watch entire database (default)
let database = WatchLevel::Database;

// Watch entire deployment (cluster-wide)
let deployment = WatchLevel::Deployment;

// Default is Database level
assert_eq!(WatchLevel::default(), WatchLevel::Database);

Variants

Collection(Vec<String>)

Watch specific collections only.

This is the most granular option, allowing you to specify exactly which collections to monitor. Each collection gets its own change stream, enabling parallel processing.

Advantages:

• Maximum parallelism (one worker per collection)
• Can apply different batching/retry settings per collection
• Lower latency for specific collections

Disadvantages:

• Must update configuration when adding new collections
• More MongoDB connections

Example

use rigatoni_core::watch_level::WatchLevel;

let level = WatchLevel::Collection(vec![
    "users".to_string(),
    "orders".to_string(),
    "products".to_string(),
]);

Database

Watch all collections in the database.

Automatically picks up new collections as they are created. Uses MongoDB’s db.watch() API to create a single change stream for the entire database.

Advantages:

• Automatic discovery of new collections
• Single change stream (simpler architecture)
• No configuration updates needed

Disadvantages:

• Single stream may become bottleneck for high-volume databases
• Cannot apply per-collection settings

Requirements:

• MongoDB replica set

Recommended for: Databases with < 50 collections

Example

use rigatoni_core::pipeline::PipelineConfig;

let config = PipelineConfig::builder()
    .mongodb_uri("mongodb://localhost:27017/?replicaSet=rs0")
    .database("mydb")
    .watch_database()  // Use Database watch level
    .build();

Deployment

Watch all databases in the deployment (cluster-wide).

Uses MongoDB’s client.watch() API to monitor changes across all databases in the deployment. This is the most comprehensive option but also the most resource-intensive.

Advantages:

• Complete visibility into all changes
• Single stream for entire cluster
• Useful for audit logging and compliance

Disadvantages:

• High event volume
• Requires cluster-wide permissions
• Not suitable for multi-tenant environments (unless intended)

Requirements:

• MongoDB 4.0+
• Cluster-wide read permissions

Recommended for: Monitoring, audit logging, compliance use cases

Example

use rigatoni_core::pipeline::PipelineConfig;

let config = PipelineConfig::builder()
    .mongodb_uri("mongodb://localhost:27017/?replicaSet=rs0")
    .database("mydb")  // Still needed for state storage keys
    .watch_deployment()  // Watch all databases
    .build();

Implementations

impl WatchLevel

pub fn is_collection(&self) -> bool

Returns true if this is collection-level watching.

Example

use rigatoni_core::watch_level::WatchLevel;

let level = WatchLevel::Collection(vec!["users".to_string()]);
assert!(level.is_collection());

let level = WatchLevel::Database;
assert!(!level.is_collection());

pub fn is_database(&self) -> bool

Returns true if this is database-level watching.

Example

use rigatoni_core::watch_level::WatchLevel;

let level = WatchLevel::Database;
assert!(level.is_database());

pub fn is_deployment(&self) -> bool

Returns true if this is deployment-level (cluster-wide) watching.

Example

use rigatoni_core::watch_level::WatchLevel;

let level = WatchLevel::Deployment;
assert!(level.is_deployment());

pub fn collections(&self) -> Option<&Vec<String>>

Returns the collections if this is collection-level watching.

Returns None for database or deployment level watching.

Example

use rigatoni_core::watch_level::WatchLevel;

let level = WatchLevel::Collection(vec!["users".to_string()]);
assert_eq!(level.collections(), Some(&vec!["users".to_string()]));

let level = WatchLevel::Database;
assert_eq!(level.collections(), None);

pub fn description(&self) -> String

Returns a human-readable description of the watch level.

Example

use rigatoni_core::watch_level::WatchLevel;

let level = WatchLevel::Collection(vec!["users".to_string(), "orders".to_string()]);
assert_eq!(level.description(), "2 collections");

let level = WatchLevel::Database;
assert_eq!(level.description(), "database");

let level = WatchLevel::Deployment;
assert_eq!(level.description(), "deployment");

pub fn resume_token_key( &self, database: &str, collection: Option<&str>, ) -> String

Returns the resume token key prefix for this watch level.

Different watch levels use different resume token keys to avoid conflicts when switching between levels.

Arguments

• database - The database name (used for collection and database levels)
• collection - Optional collection name (used for collection level)

Example

use rigatoni_core::watch_level::WatchLevel;

let level = WatchLevel::Collection(vec!["users".to_string()]);
assert_eq!(
    level.resume_token_key("mydb", Some("users")),
    "resume_token:mydb:users"
);

let level = WatchLevel::Database;
assert_eq!(
    level.resume_token_key("mydb", None),
    "resume_token:database:mydb"
);

let level = WatchLevel::Deployment;
assert_eq!(
    level.resume_token_key("mydb", None),
    "resume_token:deployment"
);

Trait Implementations

impl Clone for WatchLevel

fn clone(&self) -> WatchLevel

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for WatchLevel

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for WatchLevel

fn default() -> Self

Returns the default watch level: WatchLevel::Database.

Database-level watching is the recommended default as it provides automatic collection discovery without the complexity of deployment-level watching.

impl Display for WatchLevel

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl PartialEq for WatchLevel

fn eq(&self, other: &WatchLevel) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl Eq for WatchLevel

impl StructuralPartialEq for WatchLevel