//! End-to-end schedule lifecycle integration tests over `InMemoryStore`.

mod common;

use std::sync::Arc;
use std::time::Duration;

use aion::schedule::TimerEvaluationOutcome;
use aion::{Engine, EngineBuilder};
use aion_core::{CatchUpPolicy, Event, OverlapPolicy, ScheduleConfig, TriggerSpec};
use aion_store::EventStore;
use serde_json::json;

#[tokio::test]
async fn schedule_lifecycle_fires_skips_overlap_and_recovers()
-> Result<(), Box<dyn std::error::Error>> {
    let store: Arc<dyn EventStore> = Arc::new(aion_store::InMemoryStore::default());
    let engine = engine_with_store(Arc::clone(&store)).await?;
    let input = common::payload(&json!({ "scheduled": true }))?;
    let config = ScheduleConfig {
        trigger: TriggerSpec::Interval {
            period: Duration::from_millis(25),
        },
        overlap_policy: OverlapPolicy::Skip,
        catch_up_policy: CatchUpPolicy::Skip,
        workflow_type: common::FIXTURE_MODULE.to_owned(),
        input: input.clone(),
        search_attributes: std::collections::HashMap::new(),
    };

    let schedule_id = engine.create_schedule(config.clone()).await?;
    let created_state = engine.describe_schedule(&schedule_id).await?;
    let first_fire_at = created_state.next_trigger_at;

    let first = engine
        .handle_schedule_timer_fired(&schedule_id, first_fire_at)
        .await?;
    let TimerEvaluationOutcome::Started(first_execution) = first else {
        return Err("expected first schedule timer to start workflow".into());
    };
    let first_history = store.read_history(&first_execution.workflow_id).await?;
    assert!(first_history.iter().any(|event| matches!(
        event,
        Event::WorkflowStarted { input, .. } if input == &config.input
    )));
    assert_schedule_triggered(&engine, &schedule_id, &first_execution.workflow_id).await?;

    assert!(
        engine
            .registry()
            .get(&first_execution.workflow_id, &first_execution.run_id)?
            .is_some(),
        "scheduled workflow must be registered as a live run"
    );

    let state_after_first = engine.describe_schedule(&schedule_id).await?;
    let skipped = engine
        .handle_schedule_timer_fired(&schedule_id, state_after_first.next_trigger_at)
        .await?;
    assert_eq!(skipped, TimerEvaluationOutcome::Skipped);
    let after_skip_triggers = schedule_trigger_count(&engine, &schedule_id).await?;
    assert_eq!(after_skip_triggers, 1);

    engine.shutdown()?;
    // The restarted engine recovers the active scheduled workflow through
    // the production AD seam — re-spawned from durable history, no
    // synthetic process metadata.
    let recovered = engine_with_store(Arc::clone(&store)).await?;

    let recovered_schedules = recovered.list_schedules().await?;
    let [listed_state] = recovered_schedules.as_slice() else {
        return Err(format!(
            "expected exactly one recovered schedule, found {}",
            recovered_schedules.len()
        )
        .into());
    };
    assert_eq!(listed_state.schedule_id, schedule_id);
    assert_eq!(listed_state.config, config);
    assert!(!listed_state.is_paused);
    assert!(!listed_state.is_deleted);
    assert!(listed_state.current_execution.is_none());
    assert!(listed_state.last_triggered_at.is_some());

    let recovered_state = recovered.describe_schedule(&schedule_id).await?;
    let recovered_fire = recovered
        .handle_schedule_timer_fired(&schedule_id, recovered_state.next_trigger_at)
        .await?;
    let TimerEvaluationOutcome::Started(_) = recovered_fire else {
        return Err("expected recovered schedule to re-arm and fire again".into());
    };
    let after_recovery_triggers = schedule_trigger_count(&recovered, &schedule_id).await?;
    assert_eq!(after_recovery_triggers, 2);

    Ok(())
}

#[tokio::test]
async fn schedule_uses_common_helpers_for_engine_and_payload()
-> Result<(), Box<dyn std::error::Error>> {
    let (engine, _store) = common::engine_with_fixture("wait").await?;
    let input = common::input_payload()?;
    let config = ScheduleConfig {
        trigger: TriggerSpec::Interval {
            period: Duration::from_secs(3600),
        },
        overlap_policy: OverlapPolicy::Skip,
        catch_up_policy: CatchUpPolicy::Skip,
        workflow_type: common::FIXTURE_MODULE.to_owned(),
        input,
        search_attributes: std::collections::HashMap::new(),
    };
    let schedule_id = engine.create_schedule(config).await?;
    let state = engine.describe_schedule(&schedule_id).await?;
    assert!(!state.is_paused);
    Ok(())
}

async fn engine_with_store(
    store: Arc<dyn EventStore>,
) -> Result<Engine, Box<dyn std::error::Error>> {
    Ok(EngineBuilder::new()
        .store_arc(store)
        .in_memory_visibility()
        .scheduler_threads(1)
        .load_workflows(common::fixture_package("wait")?)
        .build()
        .await?)
}

async fn assert_schedule_triggered(
    engine: &Engine,
    schedule_id: &aion_core::ScheduleId,
    workflow_id: &aion_core::WorkflowId,
) -> Result<(), Box<dyn std::error::Error>> {
    let history = engine
        .store()
        .read_history(engine.schedule_coordinator_workflow_id())
        .await?;
    assert!(history.iter().any(|event| matches!(
        event,
        Event::ScheduleTriggered { schedule_id: recorded_schedule_id, workflow_id: recorded_workflow_id, .. }
            if recorded_schedule_id == schedule_id && recorded_workflow_id == workflow_id
    )));
    Ok(())
}

async fn schedule_trigger_count(
    engine: &Engine,
    schedule_id: &aion_core::ScheduleId,
) -> Result<usize, Box<dyn std::error::Error>> {
    let history = engine
        .store()
        .read_history(engine.schedule_coordinator_workflow_id())
        .await?;
    Ok(history
        .iter()
        .filter(|event| {
            matches!(
                event,
                Event::ScheduleTriggered { schedule_id: recorded_schedule_id, .. }
                    if recorded_schedule_id == schedule_id
            )
        })
        .count())
}
