use anyhow::Result;
use log::*;
use model::meta::CollectionMeta;
use sqlx::mysql::MySqlPool;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::time::Instant;

use crate::pipeline::pager::Pager;

use super::Source;
use super::SourceReport;

pub struct CollectionSource {
    db: Arc<MySqlPool>,
}

impl Source for CollectionSource {
    type Item = CollectionMeta;
    async fn send(
        &mut self,
        sender: mpsc::Sender<Self::Item>,
        start_version: u64,
        end_versin: u64,
    ) -> Result<SourceReport> {
        let instant = Instant::now();
        let mut report = SourceReport::default();
        let pagers =
            Pager::collections_by_version(&self.db, 100, start_version, end_versin).await?;
        let totals = pagers.iter().map(|pager| pager.page_size).sum::<i64>();
        if totals == 0 {
            drop(sender);
            return Ok(SourceReport::default());
        }

        info!(
            "Reading: {} collections, version: {} -> {}",
            totals, start_version, end_versin
        );
        for pager in pagers {
            info!("Reading: page {} collections", pager.page_number,);
            let updated = CollectionMeta::updated_between_all(
                &self.db,
                pager.start_version,
                pager.end_version,
            )
            .await?;
            if let Some(updated) = updated {
                for collection in updated {
                    match sender.send(collection.clone()).await {
                        Ok(_) => {
                            report.successes += 1;
                        }
                        Err(e) => {
                            report.faileds += 1;
                            error!(
                                "Failed to send collection: {:?}, error: {:?}",
                                collection, e
                            );
                        }
                    };
                }
            }
        }

        drop(sender);
        debug!("CollectionSource finished");

        report.cost = instant.elapsed();
        Ok(report)
    }
}

impl CollectionSource {
    pub fn new(db: Arc<MySqlPool>) -> Self {
        Self { db }
    }
}