use std::sync::Arc;
use std::time::Duration;

use anyhow::Context;
use entertainarr_domain::media::prelude::MediaService;
use entertainarr_domain::podcast::prelude::PodcastService;
use entertainarr_domain::task::entity::{Task, TaskParams, TaskPayload};
use entertainarr_domain::task::prelude::TaskService;
use entertainarr_domain::tvshow::prelude::TvShowService;
use tokio::sync::Notify;
use tokio_util::sync::CancellationToken;

mod media;
mod podcast;
mod tvshow;

#[derive(serde::Deserialize)]
pub struct Config {
    #[serde(default = "Config::default_heartbeat")]
    pub heartbeat: Duration,
}

impl Default for Config {
    fn default() -> Self {
        Self {
            heartbeat: Self::default_heartbeat(),
        }
    }
}

impl Config {
    pub const fn default_heartbeat() -> Duration {
        Duration::from_secs(10)
    }

    pub fn build(self) -> Runner<(), (), (), ()> {
        let notifier = Arc::new(Notify::new());
        Runner {
            media_service: (),
            podcast_service: (),
            task_service: (),
            tvshow_service: (),
            heartbeat: self.heartbeat,
            notifier,
        }
    }
}

pub struct Runner<MS, PS, TS, TSS> {
    media_service: MS,
    podcast_service: PS,
    task_service: TS,
    tvshow_service: TSS,
    heartbeat: Duration,
    notifier: Arc<Notify>,
}

impl<MS, PS, TS, TSS> Runner<MS, PS, TS, TSS> {
    pub fn publisher(&self, pool: entertainarr_adapter_sqlite::Pool) -> Publisher {
        Publisher {
            notifier: self.notifier.clone(),
            sqlite: pool,
        }
    }

    pub fn with_media_service<MS2>(self, media_service: MS2) -> Runner<MS2, PS, TS, TSS> {
        Runner {
            media_service,
            podcast_service: self.podcast_service,
            task_service: self.task_service,
            tvshow_service: self.tvshow_service,
            heartbeat: self.heartbeat,
            notifier: self.notifier,
        }
    }

    pub fn with_podcast_service<PS2>(self, podcast_service: PS2) -> Runner<MS, PS2, TS, TSS> {
        Runner {
            media_service: self.media_service,
            podcast_service,
            task_service: self.task_service,
            tvshow_service: self.tvshow_service,
            heartbeat: self.heartbeat,
            notifier: self.notifier,
        }
    }

    pub fn with_task_service<TS2>(self, task_service: TS2) -> Runner<MS, PS, TS2, TSS> {
        Runner {
            media_service: self.media_service,
            podcast_service: self.podcast_service,
            task_service,
            tvshow_service: self.tvshow_service,
            heartbeat: self.heartbeat,
            notifier: self.notifier,
        }
    }

    pub fn with_tvshow_service<TSS2>(self, tvshow_service: TSS2) -> Runner<MS, PS, TS, TSS2> {
        Runner {
            media_service: self.media_service,
            podcast_service: self.podcast_service,
            task_service: self.task_service,
            tvshow_service,
            heartbeat: self.heartbeat,
            notifier: self.notifier,
        }
    }
}

impl<MS, PS, TS, TSS> Runner<MS, PS, TS, TSS>
where
    MS: MediaService,
    PS: PodcastService,
    TS: TaskService,
    TSS: TvShowService,
{
    #[tracing::instrument(
        parent = None,
        skip_all,
        fields(
            error.type = tracing::field::Empty,
            error.message = tracing::field::Empty,
            error.stacktrace = tracing::field::Empty,
            messaging.batch.message_count = 1,
            messaging.destination.name = "worker",
            messaging.message.attributes = ?task,
            messaging.message.id = task.id,
            messaging.message.kind = "task",
            messaging.operation.type = "receive",
            otel.kind = "consumer",
            otel.status_code = tracing::field::Empty,
            span.type = "queue",
        ),
        err(Debug)
    )]
    async fn handle(&self, task: Task) -> anyhow::Result<()> {
        let span = tracing::Span::current();
        let params = task.params();
        match task.payload {
            TaskPayload::Media(inner) => self.handle_media(inner, &params).await,
            TaskPayload::Noop => Ok(()),
            TaskPayload::Podcast(inner) => self.handle_podcast(inner, &params).await,
            TaskPayload::TvShow(inner) => self.handle_tvshow(inner, &params).await,
        }
        .inspect(|_| {
            span.record("otel.status_code", "ok");
        })
        .inspect_err(|err| {
            span.record("otel.status_code", "err");
            span.record("error.message", err.to_string());
        })
    }

    async fn iterate(&self) -> anyhow::Result<()> {
        while let Some(task) = self.task_service.start_task().await? {
            let task_id = task.id;
            match self.handle(task).await {
                Ok(_) => {
                    self.task_service.complete_task(task_id).await?;
                }
                Err(err) => {
                    self.task_service.fail_task(task_id, err).await?;
                }
            }
        }
        Ok(())
    }

    async fn run(self, token: CancellationToken) {
        while !token.is_cancelled() {
            tokio::select! {
                _ = self.notifier.notified() => {
                    match self.iterate().await {
                        Ok(_) => {
                            tracing::debug!("nothing left in the queue, waiting for the next heartbeat");
                        }
                        Err(err) => {
                            tracing::error!(error = ?err, "failed iterating over tasks");
                        }
                    }
                }
            }
        }
    }
}

impl<MS, PS, TS, TSS> Runner<MS, PS, TS, TSS>
where
    MS: MediaService,
    PS: PodcastService,
    TS: TaskService,
    TSS: TvShowService,
{
    pub fn start(self) -> Worker {
        let token = CancellationToken::new();
        Worker {
            handler: tokio::task::spawn(self.run(token.clone())),
            token,
        }
    }
}

#[derive(Clone, Debug)]
pub struct Publisher {
    notifier: Arc<Notify>,
    sqlite: entertainarr_adapter_sqlite::Pool,
}

impl entertainarr_domain::task::prelude::TaskPublisher for Publisher {
    #[tracing::instrument(
        skip_all,
        fields(
            otel.kind = "client",
            messaging.batch.message_count = tracing::field::Empty,
            messaging.destination.name = "tasks",
            messaging.operation.type = "send",
            error.message = tracing::field::Empty,
        )
    )]
    async fn publish<I>(&self, tasks: I, params: &TaskParams) -> anyhow::Result<()>
    where
        I: Iterator<Item = TaskPayload> + Send + 'static,
    {
        let span = tracing::Span::current();
        self.sqlite
            .insert_tasks(tasks, params)
            .await
            .inspect(|count| {
                span.record("messaging.batch.message_count", count);
            })
            .inspect_err(|err| {
                span.record("error.message", err.to_string());
            })?;
        self.notifier.notify_one();

        Ok(())
    }
}

pub struct Worker {
    handler: tokio::task::JoinHandle<()>,
    token: CancellationToken,
}

impl Worker {
    pub async fn shutdown(self) -> anyhow::Result<()> {
        self.token.cancel();
        self.handler.await.context("worker failed shutting down")
    }
}
