use tokio::sync::mpsc::Sender;

use crate::chopper::types::{Nanos, Row, TimestampRange};
use crate::source::source::Source;

pub struct SourceDriver {
    source: Box<dyn Source>,
    tx: Sender<Box<Row>>,
    timestamp_range: TimestampRange,
}

impl SourceDriver {
    pub fn new(
        source: Box<dyn Source>,
        tx: Sender<Box<Row>>,
        timestamp_range: TimestampRange,
    ) -> SourceDriver {
        SourceDriver {
            source,
            tx,
            timestamp_range,
        }
    }

    pub async fn run(&mut self) {
        loop {
            let row = match self.source.next() {
                None => return,
                Some(row) => row.unwrap(),
            };

            match filter_data_range(&self.timestamp_range, row.timestamp) {
                Action::Stop => return,
                Action::Write => self.tx.send(row).await.unwrap(),
                Action::Skip => {}
            }
        }
    }
}

#[derive(PartialEq)]
enum Action {
    Stop,
    Write,
    Skip,
}

fn filter_data_range(timestamp_range: &TimestampRange, timestamp: Nanos) -> Action {
    if timestamp_range.end.is_some() && timestamp >= timestamp_range.end.unwrap() {
        return Action::Stop;
    }
    match timestamp_range.begin.is_none() || timestamp >= timestamp_range.begin.unwrap() {
        true => Action::Write,
        false => Action::Skip,
    }
}
