use crate::{
    PipelineDefinition, PipelineExecutionError, PipelineExecutionResponse, PipelineStepLogEntry,
    apply_column_aliases, build_fetch_body,
};
use async_trait::async_trait;
use serde_json::{Value, json};
use std::time::Instant;
use uuid::Uuid;

/// Runtime context needed by the portable pipeline runner.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PipelineRunContext {
    pub request_id: Option<String>,
    pub pipeline_name: Option<String>,
    pub request_client: String,
    pub source_client: String,
    pub sink_client: String,
    pub dry_run: bool,
    pub force_camel_case_to_snake_case: bool,
}

/// Structured source-fetch failure returned by the runtime adapter.
#[derive(Debug, Clone, PartialEq)]
pub struct PipelineFetchFailure {
    pub status_code: u16,
    pub response_body: Value,
}

/// Successful pipeline-run result returned by the portable runner.
#[derive(Debug, Clone, PartialEq)]
pub struct PipelineRunSuccess {
    pub response: PipelineExecutionResponse,
    pub step_logs: Vec<PipelineStepLogEntry>,
}

/// Failed pipeline-run result when the source fetch cannot complete.
#[derive(Debug, Clone, PartialEq)]
pub struct PipelineRunFetchFailure {
    pub status_code: u16,
    pub response_body: Value,
    pub step_logs: Vec<PipelineStepLogEntry>,
}

/// Outcome returned by the portable pipeline runner.
#[derive(Debug, Clone, PartialEq)]
pub enum PipelineRunOutcome {
    Success(PipelineRunSuccess),
    FetchFailure(PipelineRunFetchFailure),
}

/// Runtime port used by the portable pipeline runner to fetch source rows and write sink rows.
#[async_trait(?Send)]
pub trait PipelineRuntime {
    async fn fetch_source_rows(
        &self,
        source_client: &str,
        fetch_body: &Value,
    ) -> Result<Vec<Value>, PipelineFetchFailure>;

    async fn insert_sink_row(
        &self,
        sink_client: &str,
        sink_table: &str,
        row: &Value,
    ) -> Result<Value, String>;
}

/// Executes the portable fetch -> transform -> sink pipeline flow using runtime-provided I/O.
pub async fn execute_pipeline_run<R: PipelineRuntime>(
    runtime: &R,
    definition: &PipelineDefinition,
    context: PipelineRunContext,
) -> PipelineRunOutcome {
    let PipelineRunContext {
        request_id,
        pipeline_name,
        request_client,
        source_client,
        sink_client,
        dry_run,
        force_camel_case_to_snake_case,
    } = context;

    let pipeline_run_id = Uuid::new_v4();
    let pipeline_run_id_string = pipeline_run_id.to_string();

    let fetch_body = build_fetch_body(
        &definition.source,
        definition.transform.as_ref(),
        force_camel_case_to_snake_case,
    );
    let source_started = Instant::now();
    let rows = match runtime.fetch_source_rows(&source_client, &fetch_body).await {
        Ok(rows) => rows,
        Err(failure) => {
            let source_duration_ms = source_started.elapsed().as_millis() as i64;
            return PipelineRunOutcome::FetchFailure(PipelineRunFetchFailure {
                status_code: failure.status_code,
                response_body: failure.response_body.clone(),
                step_logs: vec![PipelineStepLogEntry {
                    pipeline_run_id,
                    request_id,
                    pipeline_name,
                    step: "source".to_string(),
                    athena_client: Some(source_client),
                    table_name: Some(definition.source.table_name.clone()),
                    status_code: Some(i32::from(failure.status_code)),
                    duration_ms: Some(source_duration_ms),
                    error_message: Some(format!("fetch failed with HTTP {}", failure.status_code)),
                    details: Some(json!({ "error": failure.response_body })),
                }],
            });
        }
    };

    let source_duration_ms = source_started.elapsed().as_millis() as i64;
    let mut step_logs = vec![PipelineStepLogEntry {
        pipeline_run_id,
        request_id: request_id.clone(),
        pipeline_name: pipeline_name.clone(),
        step: "source".to_string(),
        athena_client: Some(source_client.clone()),
        table_name: Some(definition.source.table_name.clone()),
        status_code: Some(200),
        duration_ms: Some(source_duration_ms),
        error_message: None,
        details: None,
    }];

    let transform_started = Instant::now();
    let transformed_rows = apply_column_aliases(
        rows,
        definition
            .transform
            .as_ref()
            .and_then(|transform| transform.column_aliases.as_ref()),
    );
    let transform_duration_ms = transform_started.elapsed().as_millis() as i64;
    step_logs.push(PipelineStepLogEntry {
        pipeline_run_id,
        request_id: request_id.clone(),
        pipeline_name: pipeline_name.clone(),
        step: "transform".to_string(),
        athena_client: Some(request_client),
        table_name: None,
        status_code: Some(200),
        duration_ms: Some(transform_duration_ms),
        error_message: None,
        details: Some(json!({
            "rows": transformed_rows.len(),
            "column_aliases": definition
                .transform
                .as_ref()
                .and_then(|transform| transform.column_aliases.as_ref()),
        })),
    });

    let sink_table = definition.sink.table_name.clone();
    let mut inserted = Vec::with_capacity(transformed_rows.len());
    let mut errors = Vec::new();

    let sink_started = Instant::now();
    if !dry_run {
        for (index, row) in transformed_rows.iter().enumerate() {
            match runtime
                .insert_sink_row(&sink_client, &sink_table, row)
                .await
            {
                Ok(value) => inserted.push(value),
                Err(error) => errors.push(PipelineExecutionError::new(index, error)),
            }
        }
    }
    let sink_duration_ms = sink_started.elapsed().as_millis() as i64;
    step_logs.push(PipelineStepLogEntry {
        pipeline_run_id,
        request_id,
        pipeline_name,
        step: "sink".to_string(),
        athena_client: Some(sink_client.clone()),
        table_name: Some(sink_table),
        status_code: Some(200),
        duration_ms: Some(sink_duration_ms),
        error_message: if errors.is_empty() {
            None
        } else {
            Some(format!("{} row insert errors", errors.len()))
        },
        details: Some(json!({
            "rows_inserted": inserted.len(),
            "rows_attempted": transformed_rows.len(),
            "error_count": errors.len(),
            "dry_run": dry_run,
        })),
    });

    let response_data = if dry_run {
        transformed_rows
            .iter()
            .take(100)
            .cloned()
            .collect::<Vec<_>>()
    } else {
        inserted.clone()
    };

    PipelineRunOutcome::Success(PipelineRunSuccess {
        response: PipelineExecutionResponse::new(
            response_data,
            transformed_rows.len(),
            inserted.len(),
            errors,
            dry_run,
            source_client,
            sink_client,
            pipeline_run_id_string,
        ),
        step_logs,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{SinkConfig, SourceConfig};
    use futures::executor::block_on;
    use std::collections::VecDeque;
    use std::sync::Mutex;

    fn sample_definition() -> PipelineDefinition {
        PipelineDefinition {
            source: SourceConfig {
                table_name: "events_source".to_string(),
                ..Default::default()
            },
            transform: None,
            sink: SinkConfig {
                table_name: "events_sink".to_string(),
                client: None,
            },
        }
    }

    fn sample_context(dry_run: bool) -> PipelineRunContext {
        PipelineRunContext {
            request_id: Some("req_123".to_string()),
            pipeline_name: Some("copy_events".to_string()),
            request_client: "gateway_client".to_string(),
            source_client: "source_client".to_string(),
            sink_client: "sink_client".to_string(),
            dry_run,
            force_camel_case_to_snake_case: false,
        }
    }

    struct MockRuntime {
        fetch_result: Result<Vec<Value>, PipelineFetchFailure>,
        insert_results: Mutex<VecDeque<Result<Value, String>>>,
        insert_calls: Mutex<Vec<(String, String, Value)>>,
    }

    #[async_trait(?Send)]
    impl PipelineRuntime for MockRuntime {
        async fn fetch_source_rows(
            &self,
            _source_client: &str,
            _fetch_body: &Value,
        ) -> Result<Vec<Value>, PipelineFetchFailure> {
            self.fetch_result.clone()
        }

        async fn insert_sink_row(
            &self,
            sink_client: &str,
            sink_table: &str,
            row: &Value,
        ) -> Result<Value, String> {
            self.insert_calls.lock().expect("insert calls lock").push((
                sink_client.to_string(),
                sink_table.to_string(),
                row.clone(),
            ));
            self.insert_results
                .lock()
                .expect("insert results lock")
                .pop_front()
                .unwrap_or_else(|| Ok(row.clone()))
        }
    }

    #[test]
    fn execute_pipeline_run_supports_dry_runs_without_sink_inserts() {
        let runtime = MockRuntime {
            fetch_result: Ok(vec![json!({ "workspace_id": "acme" })]),
            insert_results: Mutex::new(VecDeque::new()),
            insert_calls: Mutex::new(Vec::new()),
        };

        let outcome = block_on(execute_pipeline_run(
            &runtime,
            &sample_definition(),
            sample_context(true),
        ));

        let PipelineRunOutcome::Success(success) = outcome else {
            panic!("expected successful dry-run outcome");
        };

        assert_eq!(success.response.rows_fetched, 1);
        assert_eq!(success.response.rows_inserted, 0);
        assert_eq!(success.response.would_insert, Some(1));
        assert_eq!(success.step_logs.len(), 3);
        assert!(
            runtime
                .insert_calls
                .lock()
                .expect("insert calls lock")
                .is_empty()
        );
    }

    #[test]
    fn execute_pipeline_run_tracks_sink_insert_errors() {
        let runtime = MockRuntime {
            fetch_result: Ok(vec![
                json!({ "workspace_id": "acme" }),
                json!({ "workspace_id": "other" }),
            ]),
            insert_results: Mutex::new(VecDeque::from([
                Ok(json!({ "id": 1 })),
                Err("duplicate key".to_string()),
            ])),
            insert_calls: Mutex::new(Vec::new()),
        };

        let outcome = block_on(execute_pipeline_run(
            &runtime,
            &sample_definition(),
            sample_context(false),
        ));

        let PipelineRunOutcome::Success(success) = outcome else {
            panic!("expected successful pipeline outcome");
        };

        assert_eq!(success.response.rows_fetched, 2);
        assert_eq!(success.response.rows_inserted, 1);
        assert_eq!(success.response.errors.len(), 1);
        assert_eq!(success.response.errors[0].error, "duplicate key");
        assert_eq!(success.step_logs.len(), 3);
        assert_eq!(success.step_logs[2].step, "sink");
        assert_eq!(
            success.step_logs[2].error_message.as_deref(),
            Some("1 row insert errors")
        );
    }

    #[test]
    fn execute_pipeline_run_returns_fetch_failure_with_source_step_log() {
        let runtime = MockRuntime {
            fetch_result: Err(PipelineFetchFailure {
                status_code: 403,
                response_body: json!({ "error": "denied" }),
            }),
            insert_results: Mutex::new(VecDeque::new()),
            insert_calls: Mutex::new(Vec::new()),
        };

        let outcome = block_on(execute_pipeline_run(
            &runtime,
            &sample_definition(),
            sample_context(false),
        ));

        let PipelineRunOutcome::FetchFailure(failure) = outcome else {
            panic!("expected fetch failure outcome");
        };

        assert_eq!(failure.status_code, 403);
        assert_eq!(failure.response_body, json!({ "error": "denied" }));
        assert_eq!(failure.step_logs.len(), 1);
        assert_eq!(failure.step_logs[0].step, "source");
        assert_eq!(failure.step_logs[0].status_code, Some(403));
        assert!(
            runtime
                .insert_calls
                .lock()
                .expect("insert calls lock")
                .is_empty()
        );
    }
}
