use tokio::spawn;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::task::JoinHandle;

use crate::chopper::error::Result;
use crate::chopper::types::{Row, TimestampRange};
use crate::driver::joiner::Joiner;
use crate::driver::source::SourceDriver;
use crate::input::builder::chopper::ChopperInputFactoryBuilder;
use crate::input::factory::ChopperInputFactory;
use crate::input::input::Input;
use crate::source::csv_input_config::CSVInputConfig;
use crate::source::csv_timestamp_config::TimestampConfig;
use crate::source::source::Source;

pub fn inputs_to_sources_with_factory(
    factory: &ChopperInputFactory,
    inputs: &Vec<Input>,
    timestamp_range: &TimestampRange,
) -> Result<Vec<Box<dyn Source>>> {
    let mut sources: Vec<Box<dyn Source>> = Vec::new();
    for input in inputs {
        let source = factory.create_source(input, timestamp_range, ())?;
        sources.push(source);
    }
    Ok(sources)
}

pub fn inputs_to_sources_with_ts_config(
    ts_config: TimestampConfig,
    inputs: &Vec<Input>,
    timestamp_range: TimestampRange,
) -> Result<Vec<Box<dyn Source>>> {
    let csv_input_config = CSVInputConfig::new(ts_config);
    let input_factory = ChopperInputFactoryBuilder::new()
        .with_csv_input_config(csv_input_config)
        .build()?;

    inputs_to_sources_with_factory(&input_factory, inputs, &timestamp_range)
}

pub async fn drive_sources(
    mut sources: Vec<Box<dyn Source>>,
    output_tx: Sender<Box<Row>>,
    timestamp_range: TimestampRange,
) -> Vec<JoinHandle<()>> {
    let mut handles: Vec<JoinHandle<()>> = Vec::new();

    if sources.len() == 1 {
        handles.push(drive_source(sources.remove(0), output_tx, timestamp_range).await);
        return handles;
    }

    let mut source_rxs: Vec<Receiver<Box<Row>>> = Vec::new();
    let mut source_drivers: Vec<SourceDriver> = Vec::new();

    while !sources.is_empty() {
        let source = sources.remove(0);
        let (source_tx, source_rx) = channel(256);
        let source_driver = SourceDriver::new(source, source_tx, timestamp_range);
        source_rxs.push(source_rx);
        source_drivers.push(source_driver);
    }

    let mut joiner = Joiner::new(source_rxs, output_tx);
    handles.push(spawn(async move { joiner.run().await }));

    for mut source_driver in source_drivers {
        handles.push(spawn(async move { source_driver.run().await }));
    }

    handles
}

pub async fn drive_source(
    source: Box<dyn Source>,
    output_tx: Sender<Box<Row>>,
    timestamp_range: TimestampRange,
) -> JoinHandle<()> {
    let mut driver = SourceDriver::new(source, output_tx, timestamp_range);
    spawn(async move { driver.run().await })
}
