use futures::future::join_all;
use tokio::task::JoinHandle;

use crate::chopper::error::Result;
use crate::chopper::types::TimestampRange;
use crate::driver::util::header::get_header_from_sources;
use crate::driver::util::sink::drive_output;
use crate::driver::util::source::{
    drive_sources, inputs_to_sources_with_factory, inputs_to_sources_with_ts_config,
};
use crate::input::factory::ChopperInputFactory;
use crate::source::csv_timestamp_config::TimestampConfig;
use crate::source::source::Source;

pub async fn drive_with_ts_config(
    input: &str,
    output: &str,
    ts_config: TimestampConfig,
    timestamp_range: TimestampRange,
) -> Result<()> {
    let sources =
        inputs_to_sources_with_ts_config(ts_config, &vec![input.into()], timestamp_range)?;
    drive_with_sources(sources, output, timestamp_range).await
}

pub async fn drive_with_input_factory(
    input: &str,
    output: &str,
    input_factory: &ChopperInputFactory,
    timestamp_range: TimestampRange,
) -> Result<()> {
    let sources =
        inputs_to_sources_with_factory(input_factory, &vec![input.into()], &timestamp_range)?;
    drive_with_sources(sources, output, timestamp_range).await
}

pub async fn drive_with_sources(
    sources: Vec<Box<dyn Source>>,
    output: &str,
    timestamp_range: TimestampRange,
) -> Result<()> {
    let header = get_header_from_sources(&sources)?;

    let mut drivers: Vec<JoinHandle<()>> = Vec::new();

    let (output_tx, output_handle) = drive_output(&output.into(), header).await?;
    drivers.push(output_handle);

    drivers.append(
        drive_sources(sources, output_tx, timestamp_range)
            .await
            .as_mut(),
    );

    join_all(drivers).await;
    Ok(())
}
