flippico-cache 0.5.0

use std::collections::HashMap;
use std::sync::Arc;

use async_trait::async_trait;
use bullmq::{Queue, QueueOptions};
use tokio::sync::RwLock;

use crate::types::bullmq::BullMqProvider;
use crate::types::queues::{
    BulkJob, BullMqError, EnqueueOptions, JobHandle, JobSnapshot, JobStateFilter, QueueChannel,
    QueueCounts,
};

/// Environment variable overriding the BullMQ key prefix.
pub const PREFIX_ENV: &str = "FLIPPICO_CACHE_BULLMQ_PREFIX";

/// BullMQ's own default key prefix. Keys are `bull:<queue>:*`.
pub const DEFAULT_PREFIX: &str = "bull";

pub(crate) fn resolve_prefix(raw: Option<String>) -> String {
    match raw {
        Some(value) if !value.trim().is_empty() => value.trim().to_string(),
        _ => DEFAULT_PREFIX.to_string(),
    }
}

/// A BullMQ-backed job queue provider.
///
/// Holds its own Redis connection. It deliberately does not implement
/// `Connectable`: `bullmq-official` pins `redis =1.6.0` while this crate uses
/// `redis 0.32.5`, so the two `Client` types are unrelated and cannot be shared.
pub struct BullMq {
    url: String,
    prefix: String,
    queues: RwLock<HashMap<String, Arc<Queue>>>,
}

impl BullMq {
    /// Connect using an explicit Redis URL.
    ///
    /// Prefer `Cache::new().bullmq()` in application code. This constructor
    /// exists so tests can target a throwaway Redis without touching the
    /// production URL configured in `.env`.
    pub async fn connect(url: String) -> Result<Self, BullMqError> {
        Ok(Self {
            url,
            prefix: resolve_prefix(std::env::var(PREFIX_ENV).ok()),
            queues: RwLock::new(HashMap::new()),
        })
    }

    fn queue_options(&self) -> QueueOptions {
        // The upstream default is redis://127.0.0.1:6379 and it reads no
        // environment variable, so the URL must always be set explicitly.
        let mut opts = QueueOptions {
            prefix: self.prefix.clone(),
            ..Default::default()
        };
        opts.connection.url = self.url.clone();
        opts
    }

    /// Fetch the cached `Queue` for this channel, creating it on first use.
    ///
    /// A `Queue` owns a connection pool, so one is reused per queue name rather
    /// than constructed per enqueue.
    ///
    /// The cache never evicts. That is fine for the named variants and the
    /// occasional [`QueueChannel::Custom`], but a caller that mints a new
    /// custom name per request or per tenant would accumulate connection pools
    /// without bound. Custom names are an escape hatch for new queues, not a
    /// dynamic namespace.
    pub(crate) async fn queue_for(
        &self,
        channel: &QueueChannel,
    ) -> Result<Arc<Queue>, BullMqError> {
        channel.validate()?;
        let name = channel.get_queue();

        {
            let cache = self.queues.read().await;
            if let Some(queue) = cache.get(name) {
                return Ok(Arc::clone(queue));
            }
        }

        // Connect WITHOUT holding the lock. `Queue::with_options` performs
        // network I/O, and holding the write guard across it would block every
        // other `queue_for` call — including cache hits on unrelated queues.
        let queue = Arc::new(Queue::with_options(name, self.queue_options()).await?);

        let mut cache = self.queues.write().await;
        // A concurrent caller may have connected the same queue meanwhile.
        // Keep whichever landed first so every caller shares one connection
        // pool; the loser's `Queue` is dropped here.
        Ok(Arc::clone(cache.entry(name.to_string()).or_insert(queue)))
    }
}

fn snapshot(job: &bullmq::Job) -> JobSnapshot {
    JobSnapshot {
        id: job.id().to_string(),
        name: job.name().to_string(),
        data: job.data().clone(),
        attempts_made: job.attempts_made(),
        timestamp: job.timestamp(),
    }
}

#[async_trait]
impl BullMqProvider for BullMq {
    async fn enqueue(
        &self,
        queue: &QueueChannel,
        job_name: &str,
        payload: serde_json::Value,
        options: Option<EnqueueOptions>,
    ) -> Result<JobHandle, BullMqError> {
        let q = self.queue_for(queue).await?;
        let opts = options.unwrap_or_default().into_bullmq();
        // `Queue::add` returns a builder implementing IntoFuture, so awaiting
        // it performs the add.
        let job = q.add(job_name, payload).options(opts).await?;
        Ok(JobHandle {
            id: job.id().to_string(),
            name: job.name().to_string(),
        })
    }

    async fn enqueue_bulk(
        &self,
        queue: &QueueChannel,
        jobs: Vec<BulkJob>,
    ) -> Result<Vec<JobHandle>, BullMqError> {
        let q = self.queue_for(queue).await?;
        // Note the name collision: `BulkJob` here is OUR type from
        // `crate::types::queues`; the upstream one is `bullmq::BulkJob`.
        let bulk: Vec<bullmq::BulkJob> = jobs
            .into_iter()
            .map(|job| {
                bullmq::BulkJob::with_options(
                    job.name,
                    job.payload,
                    job.options.unwrap_or_default().into_bullmq(),
                )
            })
            .collect();

        let added = q.add_bulk(bulk).await?;
        Ok(added
            .iter()
            .map(|job| JobHandle {
                id: job.id().to_string(),
                name: job.name().to_string(),
            })
            .collect())
    }

    async fn job_counts(&self, queue: &QueueChannel) -> Result<QueueCounts, BullMqError> {
        let q = self.queue_for(queue).await?;
        let c = q.get_job_counts().await?;
        Ok(QueueCounts {
            waiting: c.waiting,
            active: c.active,
            delayed: c.delayed,
            prioritized: c.prioritized,
            completed: c.completed,
            failed: c.failed,
            waiting_children: c.waiting_children,
            paused: c.paused,
        })
    }

    async fn get_job(
        &self,
        queue: &QueueChannel,
        job_id: &str,
    ) -> Result<Option<JobSnapshot>, BullMqError> {
        let q = self.queue_for(queue).await?;
        Ok(q.get_job(job_id).await?.as_ref().map(snapshot))
    }

    async fn list_jobs(
        &self,
        queue: &QueueChannel,
        state: JobStateFilter,
        start: i64,
        end: i64,
    ) -> Result<Vec<JobSnapshot>, BullMqError> {
        let q = self.queue_for(queue).await?;
        let jobs = match state {
            JobStateFilter::Waiting => q.get_waiting(start, end).await?,
            JobStateFilter::Active => q.get_active(start, end).await?,
            JobStateFilter::Delayed => q.get_delayed(start, end).await?,
            JobStateFilter::Prioritized => q.get_prioritized(start, end).await?,
            JobStateFilter::Completed => q.get_completed(start, end).await?,
            JobStateFilter::Failed => q.get_failed(start, end).await?,
        };
        Ok(jobs.iter().map(snapshot).collect())
    }

    async fn pause(&self, queue: &QueueChannel) -> Result<(), BullMqError> {
        Ok(self.queue_for(queue).await?.pause().await?)
    }

    async fn resume(&self, queue: &QueueChannel) -> Result<(), BullMqError> {
        Ok(self.queue_for(queue).await?.resume().await?)
    }

    async fn is_paused(&self, queue: &QueueChannel) -> Result<bool, BullMqError> {
        Ok(self.queue_for(queue).await?.is_paused().await?)
    }

    /// The cached `Queue` is deliberately left in place: it holds a connection
    /// pool, not queue state, and upstream recreates the `meta` key on the next
    /// add. Evicting would only discard a usable pool.
    async fn obliterate(&self, queue: &QueueChannel) -> Result<(), BullMqError> {
        // force = true so a queue with active jobs is still removed;
        // count = 1000 is the batch size per Lua invocation.
        Ok(self.queue_for(queue).await?.obliterate(true, 1000).await?)
    }
}

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

    #[test]
    fn prefix_defaults_to_bull_when_unset() {
        // "bull" is BullMQ's own default; keeping it is what makes Bull Board work.
        assert_eq!(resolve_prefix(None), "bull");
    }

    #[test]
    fn prefix_defaults_to_bull_when_blank() {
        assert_eq!(resolve_prefix(Some(String::new())), "bull");
        assert_eq!(resolve_prefix(Some("   ".to_string())), "bull");
    }

    #[test]
    fn prefix_uses_override_when_set() {
        assert_eq!(resolve_prefix(Some("flippico".to_string())), "flippico");
    }

    #[test]
    fn prefix_trims_surrounding_whitespace() {
        assert_eq!(resolve_prefix(Some("  flippico  ".to_string())), "flippico");
    }
}