#[workflow(name = "file_processing", description = "Process incoming files")]
pub mod file_processing {
    use super::*;

    #[task]
    pub async fn validate_file(ctx: &mut Context<serde_json::Value>) -> Result<(), TaskError> {
        let filename = ctx.get("filename").and_then(|v| v.as_str()).unwrap_or("unknown").to_string();
        ctx.insert("validated", serde_json::json!(true))?;
        info!("File '{}' validated", filename);
        Ok(())
    }

    #[task(dependencies = ["validate_file"])]
    pub async fn process_file(ctx: &mut Context<serde_json::Value>) -> Result<(), TaskError> {
        let filename = ctx.get("filename").and_then(|v| v.as_str()).unwrap_or("unknown").to_string();
        ctx.insert("processed", serde_json::json!(true))?;
        info!("File '{}' processed", filename);
        Ok(())
    }
}

with cloaca.WorkflowBuilder("file_processing") as builder:
    builder.description("Process incoming files")

    @cloaca.task()
    def validate_file(context):
        filename = context.get("filename", "unknown")
        context.set("validated", True)
        print(f"Validated: {filename}")
        return context

    @cloaca.task(dependencies=["validate_file"])
    def process_file(context):
        filename = context.get("filename", "unknown")
        print(f"Processing: {filename}")
        context.set("processed", True)
        return context

use cloacina::trigger;
use cloacina::{Context, TriggerError, TriggerResult};
use std::sync::atomic::{AtomicUsize, Ordering};

// The poll fn has no `self`, so polling state lives in a module-level static.
static FILE_COUNTER: AtomicUsize = AtomicUsize::new(0);

// `#[trigger]` generates the `Trigger` impl plus a zero-arg constructor and
// submits it to the runtime inventory at compile time, so `Runtime::new()`
// (inside `DefaultRunner`) auto-registers the *polling function* — you never
// hand-construct the trigger. `on = "..."` declares which workflow it fires.
//
// Auto-registration only makes the poll loop aware of the trigger. Binding it
// to `file_processing` as a persisted schedule still takes one explicit
// `scheduler.register_trigger(trigger, "file_processing")` call (shown below) —
// pass the same name you put in `on`. Packaged workflows skip even that: the
// daemon's reconciler reads `on` and creates the schedule row for you.
#[trigger(
    name = "file_watcher",
    on = "file_processing",
    poll_interval = "2s",
    allow_concurrent = false
)]
async fn file_watcher() -> Result<TriggerResult, TriggerError> {
    // Pretend a new file lands on disk every fifth poll. Real code would
    // `std::fs::read_dir` or call an object-storage API.
    let count = FILE_COUNTER.fetch_add(1, Ordering::SeqCst);
    if count % 5 != 4 {
        return Ok(TriggerResult::Skip);
    }

    let filename = format!("data_{}.csv", chrono::Utc::now().timestamp());
    let mut ctx = Context::new();
    ctx.insert("filename", serde_json::json!(filename))
        .map_err(|e| TriggerError::PollError { message: e.to_string() })?;
    ctx.insert("watch_path", serde_json::json!("/data/inbox"))
        .map_err(|e| TriggerError::PollError { message: e.to_string() })?;
    Ok(TriggerResult::Fire(Some(ctx)))
}

import cloaca
import random

@cloaca.trigger(
    name="file_watcher",
    on="file_processing",       # the workflow this trigger fires
    poll_interval="5s",
    allow_concurrent=False
)
def file_watcher():
    # Check for new files (simulated)
    if random.randint(1, 10) == 5:
        ctx = cloaca.Context({"filename": "data_123.csv"})
        return cloaca.TriggerResult.fire(ctx)
    return cloaca.TriggerResult.skip()

use cloacina::runner::{DefaultRunner, DefaultRunnerConfig};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Enable trigger scheduling so the runner spins up its poll loop.
    let config = DefaultRunnerConfig::builder()
        .enable_trigger_scheduling(true)
        .trigger_base_poll_interval(Duration::from_secs(1))
        .build()
        .unwrap();

    let runner = DefaultRunner::with_config("sqlite://triggers.db?mode=rwc", config).await?;

    // `#[trigger]` already auto-registered `file_watcher` into the runtime
    // inventory, so look it up by name rather than constructing it.
    let trigger = runner
        .runtime()
        .get_trigger("file_watcher")
        .ok_or("trigger 'file_watcher' not in runtime inventory")?;

    // Persist the trigger -> workflow schedule row so the unified scheduler
    // dispatches `file_processing` whenever the trigger fires.
    let scheduler = runner
        .unified_scheduler()
        .await
        .ok_or("unified scheduler not enabled — check enable_trigger_scheduling()")?;
    scheduler
        .register_trigger(trigger.as_ref(), "file_processing")
        .await?;

    info!("Trigger registered. Running for 30 seconds...");
    tokio::time::sleep(Duration::from_secs(30)).await;

    runner.shutdown().await?;
    Ok(())
}

import cloaca

# The @cloaca.trigger decorator already registered file_watcher at import
# time, so constructing the runner starts the poll loop for it.
runner = cloaca.DefaultRunner("sqlite://triggers.db")

# List the registered trigger schedules
for schedule in runner.list_trigger_schedules():
    print(f"{schedule['trigger_name']} -> {schedule['workflow_name']}")

# Enable or disable a trigger at runtime
runner.set_trigger_enabled("file_watcher", False)
runner.set_trigger_enabled("file_watcher", True)

# Inspect execution history
for execution in runner.get_trigger_execution_history("file_watcher"):
    print(f"Started: {execution['started_at']}")

runner.shutdown()

gdoc_arrow_left_alt 06 — Multi-Tenancy 08 — Task Deferral gdoc_arrow_right_alt