wabot_testing/

async_jobs.rs

//! Run command and cron handlers in a test. Port of
//! `wabot-ts/src/testing/asyncHarness.ts`.
//!
//! ## Why not just call the handler
//!
//! TS's harness calls `handler.handle(command)` directly. That is the
//! fast thing, and it skips everything the job runner does *around* a
//! handler: the started → succeeded/failed transitions, the retry
//! decision, and — since Phase 6b — restoring the audit actor and
//! correlation id of whoever enqueued the work.
//!
//! Those are exactly the parts a test wants to pin, so this harness
//! runs the **real `JobRunner`** against an in-memory repository.
//! There are no polling workers and no database, so it stays a unit
//! test in cost; what it gives up is only the scheduler's timing.

use std::sync::Arc;

use parking_lot::Mutex;
use wabot_addon_async_in_memory::InMemoryJobRepository;
use wabot_core::injection::Container;
use wabot_feature_async::{
    job, register_async_runtime, register_job_repository, AsyncError, CommandData,
    CommandHandlerEntry, CommandRegistry, CronHandlerEntry, Job, JobRepository, JobRunner,
};

/// Executes handlers the way production does, minus the polling.
///
/// ```ignore
/// let harness = AsyncHarness::builder()
///     .command(SendEmailHandler::__handler_entry(&container))
///     .container(container)
///     .build();
///
/// harness.execute(&SendEmail { to: "ada@example.com".into() })
///     .await
///     .assert_succeeded();
/// ```
pub struct AsyncHarness {
    container: Container,
    repository: Arc<InMemoryJobRepository>,
    registry: Arc<CommandRegistry>,
    runner: Arc<JobRunner>,
    /// Ids in the order they ran — the repository offers lookup by id
    /// but no listing, and reaching past its trait for one would be a
    /// harness depending on storage internals.
    ran: Mutex<Vec<String>>,
}

impl AsyncHarness {
    pub fn builder() -> AsyncHarnessBuilder {
        AsyncHarnessBuilder {
            container: None,
            commands: Vec::new(),
            crons: Vec::new(),
        }
    }

    /// The container handlers resolve from — register their
    /// dependencies here before building.
    pub fn container(&self) -> &Container {
        &self.container
    }

    /// Enqueue and run one command through the real runner, returning
    /// the finished job.
    ///
    /// # Panics
    ///
    /// If no handler is registered for the command. A test that runs a
    /// command nothing handles is testing nothing, and in production
    /// that is a boot-time misconfiguration rather than a runtime
    /// branch.
    pub async fn execute<C: CommandData + serde::Serialize>(&self, command: &C) -> FinishedJob {
        let payload = serde_json::to_value(command).expect("a serializable command");
        self.execute_named(C::COMMAND_NAME, payload).await
    }

    /// [`AsyncHarness::execute`] with the command named directly — for
    /// a cron's command, or a payload built as raw JSON.
    pub async fn execute_named(
        &self,
        command_name: &str,
        payload: serde_json::Value,
    ) -> FinishedJob {
        assert!(
            self.registry.get(command_name).is_some(),
            "AsyncHarness: no handler registered for command '{command_name}'. \
             Registered: {:?}",
            self.registry.command_names()
        );

        let job = job::new_job(job::JobData {
            base: Default::default(),
            command_name: command_name.to_string(),
            command_data: payload,
            scheduled_at: Some(chrono::Utc::now().timestamp_millis()),
            started_at: None,
            success_at: None,
            failed_at: None,
            retry_delays_seconds: self
                .registry
                .options_for(command_name)
                .and_then(|o| o.retry_delays_seconds),
            intent_number: None,
            error: None,
            acceptable_running_time_seconds: None,
            stuck_retry_attempts: None,
            dedup_key: None,
            // Whatever attributed the *test's* scope, exactly as an
            // enqueue from a request would capture it.
            actor: wabot_core::audit::audit_actor(),
            request_id: wabot_core::log_context::request_id(),
        });

        self.repository.create(&job).await.expect("stored");
        self.ran.lock().push(job.id().to_string());
        let result = self.runner.run(self.container.clone(), job.clone()).await;
        let stored = self
            .repository
            .find(job.id())
            .await
            .expect("a readable repository")
            .expect("the job it just ran");

        FinishedJob {
            job: stored,
            run_error: result.err(),
        }
    }

    /// Run one cron handler's body immediately — the tick without the
    /// clock.
    pub async fn run_cron(&self, command_name: &str) -> FinishedJob {
        self.execute_named(command_name, serde_json::Value::Null)
            .await
    }

    /// Every job the harness has run, oldest first.
    pub async fn jobs(&self) -> Vec<Job> {
        let ids = self.ran.lock().clone();
        let mut jobs = Vec::with_capacity(ids.len());
        for id in ids {
            if let Ok(Some(job)) = self.repository.find(&id).await {
                jobs.push(job);
            }
        }
        jobs
    }
}

/// A job after the runner finished with it.
pub struct FinishedJob {
    pub job: Job,
    /// Set when the *runner* failed (an unregistered command, a
    /// repository error) — distinct from the handler failing, which is
    /// recorded on the job.
    pub run_error: Option<AsyncError>,
}

impl FinishedJob {
    pub fn succeeded(&self) -> bool {
        job::was_success(&self.job)
    }

    /// The handler's error message, when it failed.
    pub fn error(&self) -> Option<String> {
        self.job.data().error.as_ref().map(|e| e.message.clone())
    }

    /// Attempts so far. Retries increment it, so this is how a test
    /// checks a retry was scheduled rather than a final failure.
    pub fn attempts(&self) -> u32 {
        self.job.data().intent_number.unwrap_or(0)
    }

    /// When the job is queued to run again — `Some` after a failure
    /// with retries configured.
    pub fn retry_at_ms(&self) -> Option<i64> {
        self.job
            .data()
            .failed_at
            .is_none()
            .then(|| self.job.data().scheduled_at)
            .flatten()
    }

    /// Assert the handler succeeded, showing its error if not.
    pub fn assert_succeeded(&self) -> &Self {
        assert!(
            self.succeeded(),
            "expected the job to succeed, but it {}",
            match self.error() {
                Some(message) => format!("failed: {message}"),
                None => "did not finish".to_string(),
            }
        );
        self
    }
}

pub struct AsyncHarnessBuilder {
    container: Option<Container>,
    commands: Vec<CommandHandlerEntry>,
    crons: Vec<CronHandlerEntry>,
}

impl AsyncHarnessBuilder {
    /// The container handlers resolve from. Register their
    /// dependencies (a fake mailer, say) before building.
    pub fn container(mut self, container: Container) -> Self {
        self.container = Some(container);
        self
    }

    /// A `#[command_handler]` entry, exactly as
    /// `run_async_workers` would receive it.
    pub fn command(mut self, entry: CommandHandlerEntry) -> Self {
        self.commands.push(entry);
        self
    }

    /// A `#[cron_handler]` entry.
    pub fn cron(mut self, entry: CronHandlerEntry) -> Self {
        self.crons.push(entry);
        self
    }

    pub fn build(self) -> AsyncHarness {
        let container = self.container.unwrap_or_default();
        let repository = Arc::new(InMemoryJobRepository::new());
        register_job_repository(&container, repository.clone());
        register_async_runtime(&container);

        let registry: Arc<CommandRegistry> = container.resolve();
        for entry in self.commands {
            registry.register(entry);
        }
        for entry in self.crons {
            // The same bridge production installs: a cron tick runs as
            // an ordinary job under the cron's command name.
            registry.register(wabot_feature_async::cron_command_entry(&entry));
        }

        let runner = Arc::new(JobRunner::new(repository.clone(), registry.clone()));

        AsyncHarness {
            container,
            repository,
            registry,
            runner,
            ran: Mutex::new(Vec::new()),
        }
    }
}