#![deny(missing_docs)]

//! # An Actix-based Jobs Processor
//!
//! This library will spin up as many actors as requested for each processor to process jobs
//! concurrently. Keep in mind that, by default, spawned actors run on the same Arbiter, so in
//! order to achieve parallel execution, multiple Arbiters must be in use.
//!
//! The thread count is used to spawn Synchronous Actors to handle the storage of job
//! information. For storage backends that cannot be parallelized, a thread-count of 1 should be
//! used. By default, the number of cores of the running system is used.
//!
//! ### Example
//! ```rust
//! use background_jobs_core::{Backoff, Job, MaxRetries, BoxError};
//! use background_jobs_actix::{ActixTimer, WorkerConfig};
//! use std::future::{ready, Ready};
//!
//! const DEFAULT_QUEUE: &'static str = "default";
//!
//! #[derive(Clone, Debug)]
//! pub struct MyState {
//!     pub app_name: String,
//! }
//!
//! #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
//! pub struct MyJob {
//!     some_usize: usize,
//!     other_usize: usize,
//! }
//!
//! #[actix_rt::main]
//! async fn main() -> Result<(), BoxError> {
//!     // Set up our Storage
//!     // For this example, we use the default in-memory storage mechanism
//!     use background_jobs_core::memory_storage::Storage;
//!     let storage = Storage::new(ActixTimer);
//!
//!     // Configure and start our workers
//!     let queue_handle = WorkerConfig::new(storage, move |_| MyState::new("My App"))
//!         .register::<MyJob>()
//!         .set_worker_count(DEFAULT_QUEUE, 16)
//!         .start();
//!
//!     // Queue our jobs
//!     queue_handle.queue(MyJob::new(1, 2)).await?;
//!     queue_handle.queue(MyJob::new(3, 4)).await?;
//!     queue_handle.queue(MyJob::new(5, 6)).await?;
//!
//!     // actix_rt::signal::ctrl_c().await?;
//!
//!     Ok(())
//! }
//!
//! impl MyState {
//!     pub fn new(app_name: &str) -> Self {
//!         MyState {
//!             app_name: app_name.to_owned(),
//!         }
//!     }
//! }
//!
//! impl MyJob {
//!     pub fn new(some_usize: usize, other_usize: usize) -> Self {
//!         MyJob {
//!             some_usize,
//!             other_usize,
//!         }
//!     }
//! }
//!
//! impl Job for MyJob {
//!     type State = MyState;
//!     type Future = Ready<Result<(), BoxError>>;
//!
//!     // The name of the job. It is super important that each job has a unique name,
//!     // because otherwise one job will overwrite another job when they're being
//!     // registered.
//!     const NAME: &'static str = "MyJob";
//!
//!     // The queue that this processor belongs to
//!     //
//!     // Workers have the option to subscribe to specific queues, so this is important to
//!     // determine which worker will call the processor
//!     //
//!     // Jobs can optionally override the queue they're spawned on
//!     const QUEUE: &'static str = DEFAULT_QUEUE;
//!
//!     // The number of times background-jobs should try to retry a job before giving up
//!     //
//!     // This value defaults to MaxRetries::Count(5)
//!     // Jobs can optionally override this value
//!     const MAX_RETRIES: MaxRetries = MaxRetries::Count(1);
//!
//!     // The logic to determine how often to retry this job if it fails
//!     //
//!     // This value defaults to Backoff::Exponential(2)
//!     // Jobs can optionally override this value
//!     const BACKOFF: Backoff = Backoff::Exponential(2);
//!
//!     // This is important for allowing the job server to reap processes that were started but never
//!     // completed.
//!     //
//!     // Defaults to 5 seconds
//!     const HEARTBEAT_INTERVAL: u64 = 5_000;
//!
//!     fn run(self, state: MyState) -> Self::Future {
//!         println!("{}: args, {:?}", state.app_name, self);
//!
//!         ready(Ok(()))
//!     }
//! }
//! ```

use actix_rt::{Arbiter, ArbiterHandle};
use background_jobs_core::{
    memory_storage::Timer, new_job, new_scheduled_job, BoxError, Job, ProcessorMap, Storage,
};
use std::{
    collections::BTreeMap,
    marker::PhantomData,
    num::NonZeroUsize,
    ops::Deref,
    sync::Arc,
    time::{Duration, SystemTime},
};
use tokio::sync::Notify;

mod actix_job;
mod every;
mod server;
mod spawn;
mod storage;
mod worker;

use self::{every::every, server::Server};

pub use actix_job::ActixSpawner;

/// A timer implementation for the Memory Storage backend
#[derive(Debug, Clone)]
pub struct ActixTimer;

#[async_trait::async_trait]
impl Timer for ActixTimer {
    async fn timeout<F>(&self, duration: Duration, future: F) -> Result<F::Output, ()>
    where
        F: std::future::Future + Send + Sync,
    {
        tokio::time::timeout(duration, future).await.map_err(|_| ())
    }
}

/// Manager for worker threads
///
/// Manager attempts to restart workers as their arbiters die. Dropping the manager kills the
/// workers
pub struct Manager {
    // the manager arbiter
    arbiter: Option<Arbiter>,

    // handle for queueing
    queue_handle: QueueHandle,
}

impl Manager {
    /// Create a new manager to keep jobs alive
    ///
    /// Manager works by startinng a new Arbiter to run jobs, and if that arbiter ever dies, it
    /// spins up another one and spawns the workers again
    fn new<State>(worker_config: WorkerConfig<State, Managed>, thread_count: NonZeroUsize) -> Self
    where
        State: Clone,
    {
        let manager_arbiter = Arbiter::new();
        let queue_handle = worker_config.queue_handle.clone();

        for i in 0..thread_count.into() {
            let worker_config = worker_config.clone();

            manager_arbiter.spawn(async move {
                let mut worker_arbiter = ArbiterDropper::new();

                loop {
                    let notifier = DropNotifier::default();
                    worker_config.start_managed(&worker_arbiter.handle(), &());

                    let notified = notifier.notify.notified();

                    let drop_notifier = notifier.clone();
                    worker_arbiter.spawn(async move {
                        std::future::pending::<()>().await;
                        drop(drop_notifier);
                    });

                    notified.await;

                    metrics::counter!("background-jobs.actix.worker-arbiter.restart", "number" => i.to_string()).increment(1);
                    tracing::warn!("Recovering from dead worker arbiter");

                    drop(worker_arbiter);

                    worker_arbiter = ArbiterDropper::new();
                }
            });
        }

        Manager {
            arbiter: Some(manager_arbiter),
            queue_handle,
        }
    }

    /// Retrieve the QueueHandle for the managed workers
    pub fn queue_handle(&self) -> &QueueHandle {
        &self.queue_handle
    }
}

impl Deref for Manager {
    type Target = QueueHandle;

    fn deref(&self) -> &Self::Target {
        &self.queue_handle
    }
}

impl Drop for Manager {
    fn drop(&mut self) {
        tracing::warn!("Dropping manager, tearing down workers");
        if let Some(arbiter) = self.arbiter.take() {
            arbiter.stop();
            let _ = arbiter.join();
        }
    }
}

#[derive(Clone, Default)]
struct DropNotifier {
    notify: Arc<Notify>,
}

impl Drop for DropNotifier {
    fn drop(&mut self) {
        tracing::warn!("DropNotifier dropped - Arbiter tearing down");
        self.notify.notify_waiters();
    }
}

struct ArbiterDropper {
    arbiter: Option<Arbiter>,
}

impl ArbiterDropper {
    fn new() -> Self {
        Self {
            arbiter: Some(Arbiter::new()),
        }
    }
}

impl Deref for ArbiterDropper {
    type Target = Arbiter;

    fn deref(&self) -> &Self::Target {
        self.arbiter.as_ref().unwrap()
    }
}

impl Drop for ArbiterDropper {
    fn drop(&mut self) {
        tracing::warn!("Stopping and joining arbiter");
        let arbiter = self.arbiter.take().unwrap();
        arbiter.stop();
        let _ = arbiter.join();
        tracing::warn!("Joined");
    }
}

/// Create a new managed Server
///
/// In previous versions of this library, the server itself was run on it's own dedicated threads
/// and guarded access to jobs via messages. Since we now have futures-aware synchronization
/// primitives, the Server has become an object that gets shared between client threads.
fn create_server_managed<S>(storage: S) -> QueueHandle
where
    S: Storage + Sync + 'static,
{
    QueueHandle {
        inner: Server::new(storage),
    }
}

/// Marker type for Unmanaged workers
#[derive(Clone)]
pub struct Unmanaged;
/// Marker type for Managed workers
#[derive(Clone)]
pub struct Managed;

/// Worker Configuration
///
/// This type is used for configuring and creating workers to process jobs. Before starting the
/// workers, register `Job` types with this struct. This worker registration allows for
/// different worker processes to handle different sets of workers.
#[derive(Clone)]
pub struct WorkerConfig<State, M>
where
    State: Clone + 'static,
{
    processors: ProcessorMap<State>,
    queues: BTreeMap<String, u64>,
    arbiter: Option<ArbiterHandle>,
    queue_handle: QueueHandle,
    managed: PhantomData<M>,
}

impl<State> WorkerConfig<State, Managed>
where
    State: Clone + 'static,
{
    /// Create a new managed WorkerConfig
    ///
    /// The supplied function should return the State required by the jobs intended to be
    /// processed. The function must be sharable between threads, but the state itself does not
    /// have this requirement.
    pub fn new_managed<S: Storage + Send + Sync + 'static>(
        storage: S,
        state_fn: impl Fn(QueueHandle) -> State + Send + Sync + 'static,
    ) -> Self {
        let queue_handle = create_server_managed(storage);
        let q2 = queue_handle.clone();

        WorkerConfig {
            processors: ProcessorMap::new(Arc::new(move || state_fn(q2.clone()))),
            queues: BTreeMap::new(),
            arbiter: None,
            queue_handle,
            managed: PhantomData,
        }
    }

    /// Start the workers on a managed thread, returning the manager struct
    pub fn start(self) -> Manager {
        Self::start_with_threads(self, NonZeroUsize::try_from(1).expect("nonzero"))
    }

    /// Start the workers on the specified number of managed threads, returning the Manager struct
    pub fn start_with_threads(self, thread_count: NonZeroUsize) -> Manager {
        Manager::new(self, thread_count)
    }
}

impl<State> WorkerConfig<State, Unmanaged>
where
    State: Clone + 'static,
{
    /// Create a new WorkerConfig in the current arbiter
    ///
    /// The supplied function should return the State required by the jobs intended to be
    /// processed. The function must be sharable between threads, but the state itself does not
    /// have this requirement.
    pub fn new<S: Storage + Send + Sync + 'static>(
        storage: S,
        state_fn: impl Fn(QueueHandle) -> State + Send + Sync + 'static,
    ) -> Self {
        Self::new_in_arbiter(Arbiter::current(), storage, state_fn)
    }

    /// Create a new WorkerConfig in the provided arbiter
    ///
    /// The supplied function should return the State required by the jobs intended to be
    /// processed. The function must be sharable between threads, but the state itself does not
    /// have this requirement.
    pub fn new_in_arbiter<S: Storage + Send + Sync + 'static>(
        arbiter: ArbiterHandle,
        storage: S,
        state_fn: impl Fn(QueueHandle) -> State + Send + Sync + 'static,
    ) -> Self {
        let queue_handle = create_server_managed(storage);
        let q2 = queue_handle.clone();

        WorkerConfig {
            processors: ProcessorMap::new(Arc::new(move || state_fn(q2.clone()))),
            queues: BTreeMap::new(),
            arbiter: Some(arbiter),
            queue_handle,
            managed: PhantomData,
        }
    }

    /// Start the workers in the provided arbiter
    pub fn start(self) -> QueueHandle {
        self.start_managed(self.arbiter.as_ref().unwrap(), &());

        self.queue_handle
    }
}

impl<State, M> WorkerConfig<State, M>
where
    State: Clone + 'static,
{
    /// Register a `Job` with the worker
    ///
    /// This enables the worker to handle jobs associated with this processor. If a processor is
    /// not registered, none of it's jobs will be run, even if another processor handling the same
    /// job queue is registered.
    pub fn register<J>(mut self) -> Self
    where
        J: Job<State = State>,
    {
        self.queues.insert(J::QUEUE.to_owned(), 4);
        self.processors.register::<J>();
        self
    }

    /// Set the number of workers to run for a given queue
    ///
    /// This does not spin up any additional threads. The `Arbiter` the workers are spawned onto
    /// will handle processing all workers, regardless of how many are configured.
    ///
    /// By default, 4 workers are spawned
    pub fn set_worker_count(mut self, queue: &str, count: u64) -> Self {
        self.queues.insert(queue.to_owned(), count);
        self
    }

    /// Start a workers in a managed way
    fn start_managed<Extras: Clone + Send + 'static>(
        &self,
        arbiter: &ArbiterHandle,
        extras: &Extras,
    ) {
        for (key, count) in self.queues.iter() {
            for _ in 0..*count {
                let queue = key.clone();
                let processors = self.processors.clone();
                let server = self.queue_handle.inner.clone();

                let extras_2 = extras.clone();

                arbiter.spawn_fn(move || {
                    if let Err(e) = spawn::spawn(
                        "local-worker",
                        worker::local_worker(queue, processors.cached(), server, extras_2),
                    ) {
                        tracing::error!("Failed to spawn worker {e}");
                    }
                });
            }
        }
    }
}

/// A handle to the job server, used for queuing new jobs
///
/// `QueueHandle` should be stored in your application's state in order to allow all parts of your
/// application to spawn jobs.
#[derive(Clone)]
pub struct QueueHandle {
    inner: Server,
}

impl QueueHandle {
    /// Queues a job for execution
    ///
    /// This job will be sent to the server for storage, and will execute whenever a worker for the
    /// job's queue is free to do so.
    pub async fn queue<J>(&self, job: J) -> Result<(), BoxError>
    where
        J: Job,
    {
        let job = new_job(job)?;
        self.inner.push(job).await?;
        Ok(())
    }

    /// Schedule a job for execution later
    ///
    /// This job will be sent to the server for storage, and will execute after the specified time
    /// and when a worker for the job's queue is free to do so.
    pub async fn schedule<J>(&self, job: J, after: SystemTime) -> Result<(), BoxError>
    where
        J: Job,
    {
        let job = new_scheduled_job(job, after)?;
        self.inner.push(job).await?;
        Ok(())
    }

    /// Queues a job for recurring execution
    ///
    /// This job will be added to it's queue on the server once every `Duration`. It will be
    /// processed whenever workers are free to do so.
    pub fn every<J>(&self, duration: Duration, job: J) -> std::io::Result<()>
    where
        J: Job + Clone + Send + 'static,
    {
        spawn::spawn("every", every(self.clone(), duration, job)).map(|_| ())
    }
}