flippico-cache 0.5.0

/// Errors produced by the BullMQ capability.
///
/// Unlike the older sync traits, which log and return `()`, every BullMQ
/// operation returns a `Result`: a silently dropped enqueue is lost work.
#[derive(Debug, thiserror::Error)]
pub enum BullMqError {
    /// The queue name is not usable as a BullMQ queue name.
    #[error("invalid queue name {name:?}: {reason}")]
    InvalidQueueName {
        /// The offending name.
        name: String,
        /// Why it was rejected.
        reason: &'static str,
    },

    /// An error surfaced by the underlying `bullmq` crate.
    #[error("bullmq error: {0}")]
    BullMq(#[from] bullmq::Error),
}

/// A BullMQ queue.
///
/// Mirrors the `ListChannel` / `CacheSpace` pattern used elsewhere in this
/// crate so shared queues cannot be typo'd at call sites, with `Custom` as an
/// escape hatch so a new queue needs no crate release.
///
/// Names are lowercase-kebab because they surface in Bull Board and must match
/// any non-Rust consumer exactly.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum QueueChannel {
    /// The Amadeus queue.
    Amadeus,
    /// The Notifica queue.
    Notifica,
    /// The Bajkomat API queue.
    BajkomatApi,
    /// The KSeF GPT queue.
    KsefGpt,
    /// The n8n queue.
    N8n,
    /// The Shopify queue.
    Shopify,
    /// An arbitrary queue name not covered by the variants above.
    ///
    /// An escape hatch for queues that do not warrant a crate release — not a
    /// dynamic namespace. The provider caches one connection pool per distinct
    /// queue name and never evicts, so minting a fresh name per request or per
    /// tenant would accumulate pools without bound.
    Custom(String),
}

impl QueueChannel {
    /// The queue name as BullMQ sees it. Keys become `bull:<name>:*`.
    pub fn get_queue(&self) -> &str {
        match self {
            QueueChannel::Amadeus => "amadeus",
            QueueChannel::Notifica => "notifica",
            QueueChannel::BajkomatApi => "bajkomat-api",
            QueueChannel::KsefGpt => "ksef-gpt",
            QueueChannel::N8n => "n8n",
            QueueChannel::Shopify => "shopify",
            QueueChannel::Custom(name) => name.as_str(),
        }
    }

    /// Reject names BullMQ cannot use, before any Redis round-trip.
    pub fn validate(&self) -> Result<(), BullMqError> {
        let name = self.get_queue();
        if name.is_empty() {
            return Err(BullMqError::InvalidQueueName {
                name: name.to_string(),
                reason: "must not be empty",
            });
        }
        if name.contains(':') {
            return Err(BullMqError::InvalidQueueName {
                name: name.to_string(),
                reason: "must not contain ':' — it would corrupt the bull:<queue>:* key layout",
            });
        }
        Ok(())
    }
}

/// Retry backoff strategy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Backoff {
    /// Wait a constant number of milliseconds between attempts.
    Fixed {
        /// Delay in milliseconds.
        delay_ms: u64,
    },
    /// Double the wait after each attempt, starting from `delay_ms`.
    Exponential {
        /// Initial delay in milliseconds.
        delay_ms: u64,
    },
}

/// What to do with a job once it finishes.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RemovePolicy {
    /// Remove the job immediately.
    Always,
    /// Keep the job.
    Never,
    /// Keep only the most recent `n` jobs.
    KeepLast(usize),
}

/// Per-job options applied at enqueue time.
///
/// A thin, owned mirror of the subset of `bullmq::JobOptions` this crate
/// supports, so upstream types do not leak into the public API and an upstream
/// version bump cannot silently change this crate's surface.
#[derive(Debug, Clone, Default)]
pub struct EnqueueOptions {
    /// Delay before the job becomes available, in milliseconds.
    pub delay_ms: Option<u64>,
    /// Priority; lower runs sooner.
    pub priority: Option<u32>,
    /// Total attempts before the job is considered failed.
    pub attempts: Option<u32>,
    /// Backoff between attempts.
    pub backoff: Option<Backoff>,
    /// Removal policy on success.
    pub remove_on_complete: Option<RemovePolicy>,
    /// Removal policy on permanent failure.
    pub remove_on_fail: Option<RemovePolicy>,
    /// Explicit job id. Reusing an id makes the enqueue idempotent.
    pub job_id: Option<String>,
    /// Deduplication key; further jobs with the same key are dropped.
    pub deduplication_id: Option<String>,
}

impl EnqueueOptions {
    pub(crate) fn into_bullmq(self) -> bullmq::JobOptions {
        bullmq::JobOptions {
            delay: self.delay_ms,
            priority: self.priority,
            attempts: self.attempts,
            job_id: self.job_id,
            backoff: self.backoff.map(|b| match b {
                Backoff::Fixed { delay_ms } => bullmq::types::BackoffStrategy::Fixed(delay_ms),
                Backoff::Exponential { delay_ms } => {
                    bullmq::types::BackoffStrategy::Exponential(delay_ms)
                }
            }),
            remove_on_complete: self.remove_on_complete.map(into_remove_on_finish),
            remove_on_fail: self.remove_on_fail.map(into_remove_on_finish),
            deduplication: self
                .deduplication_id
                .map(|id| bullmq::DeduplicationOptions {
                    id,
                    ttl: None,
                    extend: None,
                    replace: None,
                    keep_last_if_active: None,
                }),
            ..Default::default()
        }
    }
}

fn into_remove_on_finish(policy: RemovePolicy) -> bullmq::types::RemoveOnFinish {
    match policy {
        RemovePolicy::Always => bullmq::types::RemoveOnFinish::Bool(true),
        RemovePolicy::Never => bullmq::types::RemoveOnFinish::Bool(false),
        RemovePolicy::KeepLast(n) => bullmq::types::RemoveOnFinish::Count(n),
    }
}

/// One job in a bulk enqueue.
///
/// A named struct rather than a tuple so that call sites read unambiguously:
/// `name` and `payload` are both easy to transpose in a positional tuple, and
/// the compiler could not catch it.
#[derive(Debug, Clone)]
pub struct BulkJob {
    /// The job name workers may dispatch on.
    pub name: String,
    /// The job payload.
    pub payload: serde_json::Value,
    /// Per-job options, if any.
    pub options: Option<EnqueueOptions>,
}

/// Identifies a job that was successfully enqueued.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JobHandle {
    /// The BullMQ-assigned job id.
    pub id: String,
    /// The job name supplied at enqueue time.
    pub name: String,
}

/// A point-in-time view of a job.
#[derive(Debug, Clone, PartialEq)]
pub struct JobSnapshot {
    /// The BullMQ-assigned job id.
    pub id: String,
    /// The job name.
    pub name: String,
    /// The job payload.
    pub data: serde_json::Value,
    /// How many attempts have been made so far.
    pub attempts_made: u32,
    /// Creation timestamp, milliseconds since the Unix epoch.
    pub timestamp: u64,
}

/// Job totals per queue state.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct QueueCounts {
    /// Jobs waiting to be processed.
    pub waiting: u64,
    /// Jobs currently being processed.
    pub active: u64,
    /// Jobs scheduled for the future.
    pub delayed: u64,
    /// Jobs ordered by priority.
    pub prioritized: u64,
    /// Jobs that completed successfully.
    pub completed: u64,
    /// Jobs that permanently failed.
    pub failed: u64,
    /// Jobs waiting on children.
    pub waiting_children: u64,
    /// Jobs held because the queue is paused.
    pub paused: u64,
}

/// Which queue state to list jobs from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JobStateFilter {
    /// Waiting jobs.
    Waiting,
    /// Active jobs.
    Active,
    /// Delayed jobs.
    Delayed,
    /// Prioritized jobs.
    Prioritized,
    /// Completed jobs.
    Completed,
    /// Failed jobs.
    Failed,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn maps_known_queues_to_kebab_case() {
        assert_eq!(QueueChannel::Amadeus.get_queue(), "amadeus");
        assert_eq!(QueueChannel::Notifica.get_queue(), "notifica");
        assert_eq!(QueueChannel::BajkomatApi.get_queue(), "bajkomat-api");
        assert_eq!(QueueChannel::KsefGpt.get_queue(), "ksef-gpt");
        assert_eq!(QueueChannel::N8n.get_queue(), "n8n");
        assert_eq!(QueueChannel::Shopify.get_queue(), "shopify");
    }

    #[test]
    fn custom_queue_returns_its_own_name() {
        let channel = QueueChannel::Custom("my-queue".to_string());
        assert_eq!(channel.get_queue(), "my-queue");
    }

    #[test]
    fn validate_accepts_known_queues() {
        assert!(QueueChannel::KsefGpt.validate().is_ok());
    }

    #[test]
    fn validate_rejects_empty_custom_name() {
        let err = QueueChannel::Custom(String::new()).validate().unwrap_err();
        assert!(matches!(err, BullMqError::InvalidQueueName { .. }));
    }

    #[test]
    fn validate_rejects_colon_in_custom_name() {
        // BullMQ builds keys as `bull:<queue>:*`, so a colon corrupts the key layout.
        let err = QueueChannel::Custom("a:b".to_string())
            .validate()
            .unwrap_err();
        assert!(matches!(err, BullMqError::InvalidQueueName { .. }));
    }

    #[test]
    fn default_options_map_to_empty_bullmq_options() {
        let mapped = EnqueueOptions::default().into_bullmq();
        assert!(mapped.delay.is_none());
        assert!(mapped.priority.is_none());
        assert!(mapped.attempts.is_none());
        assert!(mapped.job_id.is_none());
    }

    #[test]
    fn maps_scalar_options() {
        let mapped = EnqueueOptions {
            delay_ms: Some(5_000),
            priority: Some(3),
            attempts: Some(4),
            job_id: Some("job-1".to_string()),
            ..Default::default()
        }
        .into_bullmq();

        assert_eq!(mapped.delay, Some(5_000));
        assert_eq!(mapped.priority, Some(3));
        assert_eq!(mapped.attempts, Some(4));
        assert_eq!(mapped.job_id.as_deref(), Some("job-1"));
    }

    #[test]
    fn maps_backoff_variants() {
        use bullmq::types::BackoffStrategy;

        let fixed = EnqueueOptions {
            backoff: Some(Backoff::Fixed { delay_ms: 1_000 }),
            ..Default::default()
        }
        .into_bullmq();
        assert!(matches!(fixed.backoff, Some(BackoffStrategy::Fixed(1_000))));

        let exp = EnqueueOptions {
            backoff: Some(Backoff::Exponential { delay_ms: 250 }),
            ..Default::default()
        }
        .into_bullmq();
        assert!(matches!(
            exp.backoff,
            Some(BackoffStrategy::Exponential(250))
        ));
    }

    #[test]
    fn maps_remove_policies() {
        use bullmq::types::RemoveOnFinish;

        let mapped = EnqueueOptions {
            remove_on_complete: Some(RemovePolicy::Always),
            remove_on_fail: Some(RemovePolicy::KeepLast(50)),
            ..Default::default()
        }
        .into_bullmq();

        assert!(matches!(
            mapped.remove_on_complete,
            Some(RemoveOnFinish::Bool(true))
        ));
        assert!(matches!(
            mapped.remove_on_fail,
            Some(RemoveOnFinish::Count(50))
        ));

        let never = EnqueueOptions {
            remove_on_complete: Some(RemovePolicy::Never),
            ..Default::default()
        }
        .into_bullmq();

        assert!(matches!(
            never.remove_on_complete,
            Some(RemoveOnFinish::Bool(false))
        ));
    }

    #[test]
    fn maps_deduplication_id() {
        let mapped = EnqueueOptions {
            deduplication_id: Some("dedup-key".to_string()),
            ..Default::default()
        }
        .into_bullmq();

        assert_eq!(
            mapped.deduplication.map(|d| d.id).as_deref(),
            Some("dedup-key")
        );
    }
}