//! [`TriggerStore`](crate::TriggerStore) conformance: trigger
//! subscriptions and idempotent occurrence reservations.

use super::*;

// ---------------------------------------------------------------------------
// TriggerStore conformance
// ---------------------------------------------------------------------------

/// Run the full [`TriggerStore`](crate::TriggerStore) conformance suite
/// against the backend produced by `make`. `make` must return a fresh, empty
/// store on each call.
pub async fn trigger_store<F>(make: F, expected_tier: DurabilityTier)
where
    F: Fn() -> Arc<dyn crate::TriggerStore>,
{
    trigger_store_reports_declared_tier(make(), expected_tier);
    trigger_source_key_is_stable(make()).await;
    trigger_store_registers_lists_and_cancels(make()).await;
    trigger_store_lists_agent_frame_registrations_by_session(make()).await;
    trigger_store_handles_host_scoped_lifecycle(make()).await;
    trigger_store_records_and_reserves_idempotently(make()).await;
}

/// Run the full [`TriggerStore`](crate::TriggerStore) suite plus durable
/// reopen checks.
pub async fn trigger_store_reopenable<F>(make: F, expected_tier: DurabilityTier)
where
    F: Fn() -> ReopenableTriggerStore,
{
    trigger_store(|| make().open, expected_tier).await;
    trigger_store_survives_reopen(make()).await;
}

fn sample_trigger_subscription_draft(
    session_id: &str,
    source_key: &str,
    process_name: &str,
) -> crate::TriggerSubscriptionDraft {
    let mut inputs = BTreeMap::new();
    inputs.insert("event".to_string(), crate::TriggerInputBinding::Event);
    let registrant_scope = crate::SessionScope::new(session_id);
    crate::TriggerSubscriptionDraft {
        registrant: crate::ProcessOriginator::session(registrant_scope.clone()),
        env_ref: crate::ProcessExecutionEnvRef::new(format!("process-env:{session_id}")),
        wake_target: Some(registrant_scope),
        name: Some(process_name.to_string()),
        source_type: "ui.button.pressed".to_string(),
        source_key: source_key.to_string(),
        source: serde_json::json!({}),
        payload_schema: crate::LashSchema::new(serde_json::json!({
            "type": "object",
            "properties": {
                "button": { "type": "string" }
            },
            "required": ["button"],
            "additionalProperties": false
        })),
        target: crate::ProcessInput::Engine {
            kind: "test".to_string(),
            payload: serde_json::json!({ "process": process_name }),
        },
        target_identity: crate::ProcessIdentity::new("test")
            .with_label(Some(process_name.to_string()))
            .with_definition(Some(serde_json::json!({ "process_name": process_name }))),
        event_types: Vec::new(),
        input_template: inputs,
        target_label: Some(process_name.to_string()),
    }
}

fn sample_host_trigger_subscription_draft(
    originator: crate::ProcessOriginator,
    source_key: &str,
    process_name: &str,
) -> crate::TriggerSubscriptionDraft {
    let mut draft = sample_trigger_subscription_draft("host-template", source_key, process_name);
    draft.registrant = originator;
    draft.wake_target = None;
    draft.env_ref = crate::ProcessExecutionEnvRef::new(format!("process-env:{process_name}"));
    draft
}

fn session_scope_id(session_id: &str) -> String {
    crate::ProcessOriginator::session(crate::SessionScope::new(session_id)).scope_id()
}

fn button_occurrence_request(
    source_key: impl Into<String>,
    idempotency_key: impl Into<String>,
) -> crate::TriggerOccurrenceRequest {
    crate::TriggerOccurrenceRequest::new(
        "ui.button.pressed",
        source_key,
        serde_json::json!({ "button": "Blue" }),
        idempotency_key,
    )
    .with_source(serde_json::json!({}))
}

fn trigger_store_reports_declared_tier(
    store: Arc<dyn crate::TriggerStore>,
    expected: DurabilityTier,
) {
    assert_eq!(
        store.durability_tier(),
        expected,
        "durability tier must match the backend"
    );
}

async fn trigger_source_key_is_stable(store: Arc<dyn crate::TriggerStore>) {
    let source = serde_json::json!({ "button": "Blue" });
    let first = store
        .source_key_for_subscription("ui.button.pressed", &source)
        .await
        .expect("first source key");
    let second = store
        .source_key_for_subscription("ui.button.pressed", &source)
        .await
        .expect("second source key");
    assert_eq!(first, second, "source keys must be stable");
    assert!(!first.is_empty(), "source keys must be non-empty");
}

async fn trigger_store_registers_lists_and_cancels(store: Arc<dyn crate::TriggerStore>) {
    let source_key = store
        .source_key_for_subscription("ui.button.pressed", &serde_json::json!({}))
        .await
        .expect("source key");
    let first = store
        .register_subscription(sample_trigger_subscription_draft(
            "session-a",
            &source_key,
            "first",
        ))
        .await
        .expect("register first subscription");
    let second = store
        .register_subscription(sample_trigger_subscription_draft(
            "session-b",
            &source_key,
            "second",
        ))
        .await
        .expect("register second subscription");

    assert!(!first.subscription_id.is_empty());
    assert!(!first.handle.is_empty());
    assert_ne!(first.handle, second.handle);

    let by_session = store
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_session("session-a"))
        .await
        .expect("list by session");
    assert_eq!(by_session.len(), 1);
    assert_eq!(by_session[0].handle, first.handle);

    let mut by_source = crate::TriggerSubscriptionFilter::for_source_type("ui.button.pressed");
    by_source.source_key = Some(source_key.clone());
    by_source.enabled = Some(true);
    let source_matches = store
        .list_subscriptions(by_source)
        .await
        .expect("list by source");
    assert_eq!(source_matches.len(), 2);

    assert!(
        !store
            .cancel_subscription(&session_scope_id("session-b"), &first.handle)
            .await
            .expect("wrong-session cancel"),
        "cancel must be scoped by session"
    );
    assert!(
        store
            .cancel_subscription(&session_scope_id("session-a"), &first.handle)
            .await
            .expect("cancel first")
    );

    let mut disabled_filter = crate::TriggerSubscriptionFilter::for_session("session-a");
    disabled_filter.handle = Some(first.handle.clone());
    let disabled = store
        .list_subscriptions(disabled_filter)
        .await
        .expect("list disabled");
    assert_eq!(disabled.len(), 1);
    assert!(!disabled[0].enabled);
}

async fn trigger_store_lists_agent_frame_registrations_by_session(
    store: Arc<dyn crate::TriggerStore>,
) {
    let source_key = store
        .source_key_for_subscription("ui.button.pressed", &serde_json::json!({}))
        .await
        .expect("source key");
    let frame_scope = crate::SessionScope::for_agent_frame("session-a", "frame-a");
    let mut draft = sample_trigger_subscription_draft("session-a", &source_key, "frame-route");
    draft.registrant = crate::ProcessOriginator::session(frame_scope.clone());
    draft.wake_target = Some(frame_scope);
    let registration = store
        .register_subscription(draft)
        .await
        .expect("register agent-frame subscription");

    let by_session = store
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_session("session-a"))
        .await
        .expect("list session-wide registrations");
    assert_eq!(by_session.len(), 1);
    assert_eq!(by_session[0].handle, registration.handle);

    let by_root_scope = store
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_registrant_scope(
            session_scope_id("session-a"),
        ))
        .await
        .expect("list exact root session scope");
    assert!(
        by_root_scope.is_empty(),
        "exact root session scope must not match agent-frame registrations"
    );

    let by_frame_scope = store
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_registrant_scope(
            registration.registrant_scope_id(),
        ))
        .await
        .expect("list exact agent-frame scope");
    assert_eq!(by_frame_scope.len(), 1);
    assert_eq!(by_frame_scope[0].handle, registration.handle);

    assert_eq!(
        store
            .delete_session_subscriptions("session-a")
            .await
            .expect("delete session registrations"),
        1,
        "session deletion must include agent-frame registrations"
    );

    let special_session_id = "session-%_\\special";
    let guard_session_id = "session-abcX\\special";
    let special_frame_scope = crate::SessionScope::for_agent_frame(special_session_id, "frame-a");
    let mut special_draft =
        sample_trigger_subscription_draft(special_session_id, &source_key, "special-frame-route");
    special_draft.registrant = crate::ProcessOriginator::session(special_frame_scope.clone());
    special_draft.wake_target = Some(special_frame_scope);
    store
        .register_subscription(special_draft)
        .await
        .expect("register special-character session subscription");

    let guard_frame_scope = crate::SessionScope::for_agent_frame(guard_session_id, "frame-a");
    let mut guard_draft =
        sample_trigger_subscription_draft(guard_session_id, &source_key, "guard-frame-route");
    guard_draft.registrant = crate::ProcessOriginator::session(guard_frame_scope.clone());
    guard_draft.wake_target = Some(guard_frame_scope);
    let guard = store
        .register_subscription(guard_draft)
        .await
        .expect("register wildcard guard subscription");

    assert_eq!(
        store
            .delete_session_subscriptions(special_session_id)
            .await
            .expect("delete special-character session registrations"),
        1,
        "session deletion must treat LIKE metacharacters in session ids literally"
    );
    let guard_by_session = store
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_session(
            guard_session_id,
        ))
        .await
        .expect("list wildcard guard session registrations");
    assert_eq!(guard_by_session.len(), 1);
    assert_eq!(guard_by_session[0].handle, guard.handle);
}

async fn trigger_store_handles_host_scoped_lifecycle(store: Arc<dyn crate::TriggerStore>) {
    let source_key = store
        .source_key_for_subscription("ui.button.pressed", &serde_json::json!({}))
        .await
        .expect("source key");
    let scoped = store
        .register_subscription(sample_host_trigger_subscription_draft(
            crate::ProcessOriginator::host_scoped("automation-a"),
            &source_key,
            "scoped-host",
        ))
        .await
        .expect("register scoped host subscription");
    let scopeless = store
        .register_subscription(sample_host_trigger_subscription_draft(
            crate::ProcessOriginator::host(),
            &source_key,
            "scopeless-host",
        ))
        .await
        .expect("register scopeless host subscription");

    let scoped_matches = store
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_registrant_scope(
            "host:automation-a",
        ))
        .await
        .expect("list scoped host subscriptions");
    assert_eq!(scoped_matches.len(), 1);
    assert_eq!(scoped_matches[0].handle, scoped.handle);

    let scopeless_matches = store
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_registrant_scope(
            "host",
        ))
        .await
        .expect("list scopeless host subscriptions");
    assert_eq!(scopeless_matches.len(), 1);
    assert_eq!(scopeless_matches[0].handle, scopeless.handle);

    assert!(
        !store
            .cancel_subscription("host", &scoped.handle)
            .await
            .expect("wrong-host-scope cancel"),
        "cancel must be scoped by host scope id"
    );
    assert!(
        store
            .cancel_subscription("host:automation-a", &scoped.handle)
            .await
            .expect("cancel scoped host subscription")
    );
    assert!(
        store
            .cancel_subscription("host", &scopeless.handle)
            .await
            .expect("cancel scopeless host subscription")
    );
}

async fn trigger_store_records_and_reserves_idempotently(store: Arc<dyn crate::TriggerStore>) {
    let source_key = store
        .source_key_for_subscription("ui.button.pressed", &serde_json::json!({}))
        .await
        .expect("source key");
    let subscription = store
        .register_subscription(sample_trigger_subscription_draft(
            "session-a",
            &source_key,
            "on_button",
        ))
        .await
        .expect("register subscription");

    let occurrence = store
        .record_occurrence(button_occurrence_request(
            source_key.clone(),
            "button-blue-1",
        ))
        .await
        .expect("record occurrence");
    assert!(!occurrence.occurrence_id.is_empty());
    assert_eq!(occurrence.source_type, "ui.button.pressed");
    assert_eq!(occurrence.source_key, source_key);

    let first = store
        .reserve_matching_deliveries(&occurrence.occurrence_id)
        .await
        .expect("reserve first delivery");
    assert_eq!(first.len(), 1);
    assert_eq!(first[0].subscription.handle, subscription.handle);
    assert_eq!(first[0].occurrence.occurrence_id, occurrence.occurrence_id);
    assert_eq!(
        first[0].reservation_status,
        crate::TriggerDeliveryReservationStatus::Reserved
    );
    assert_eq!(
        first[0].process_id,
        crate::deterministic_delivery_process_id(
            &occurrence.occurrence_id,
            &subscription.subscription_id
        )
        .expect("deterministic delivery process id")
    );

    let occurrence_matches = store
        .list_occurrences({
            let mut filter =
                crate::TriggerOccurrenceFilter::for_source("ui.button.pressed", &source_key);
            filter.occurred_at_start_ms = Some(occurrence.occurred_at_ms);
            filter.occurred_at_end_ms = Some(occurrence.occurred_at_ms.saturating_add(1));
            filter
        })
        .await
        .expect("list occurrences");
    assert_eq!(occurrence_matches, vec![occurrence.clone()]);

    let by_occurrence = store
        .list_deliveries_by_occurrence_id(&occurrence.occurrence_id)
        .await
        .expect("list deliveries by occurrence");
    assert_eq!(by_occurrence.len(), 1);
    assert_eq!(by_occurrence[0].process_id, first[0].process_id);
    let by_subscription = store
        .list_deliveries_by_subscription_id(&subscription.subscription_id)
        .await
        .expect("list deliveries by subscription");
    assert_eq!(by_subscription.len(), 1);
    assert_eq!(by_subscription[0].process_id, first[0].process_id);
    let by_process = store
        .list_deliveries_by_process_id(&first[0].process_id)
        .await
        .expect("list deliveries by process");
    assert_eq!(by_process.len(), 1);
    assert_eq!(
        by_process[0].subscription.subscription_id,
        subscription.subscription_id
    );

    let duplicate = store
        .reserve_matching_deliveries(&occurrence.occurrence_id)
        .await
        .expect("reserve duplicate delivery");
    assert_eq!(duplicate.len(), 1);
    assert_eq!(
        duplicate[0].reservation_status,
        crate::TriggerDeliveryReservationStatus::AlreadyReserved
    );

    let replayed = store
        .record_occurrence(button_occurrence_request(
            source_key.clone(),
            "button-blue-1",
        ))
        .await
        .expect("replay occurrence");
    assert_eq!(replayed.occurrence_id, occurrence.occurrence_id);
    let replayed_delivery = store
        .reserve_matching_deliveries(&replayed.occurrence_id)
        .await
        .expect("reserve replayed delivery");
    assert_eq!(replayed_delivery.len(), 1);
    assert_eq!(
        replayed_delivery[0].reservation_status,
        crate::TriggerDeliveryReservationStatus::AlreadyReserved
    );

    assert!(
        store
            .cancel_subscription(&session_scope_id("session-a"), &subscription.handle)
            .await
            .expect("cancel subscription")
    );
    let disabled = store
        .record_occurrence(button_occurrence_request(source_key, "button-blue-2"))
        .await
        .expect("record disabled occurrence");
    let disabled_deliveries = store
        .reserve_matching_deliveries(&disabled.occurrence_id)
        .await
        .expect("reserve disabled occurrence");
    assert!(disabled_deliveries.is_empty());
}

async fn trigger_store_survives_reopen(factory: ReopenableTriggerStore) {
    let source_key = factory
        .open
        .source_key_for_subscription("ui.button.pressed", &serde_json::json!({}))
        .await
        .expect("source key");
    let subscription = factory
        .open
        .register_subscription(sample_trigger_subscription_draft(
            "session-a",
            &source_key,
            "on_button",
        ))
        .await
        .expect("register subscription before reopen");
    let occurrence = factory
        .open
        .record_occurrence(button_occurrence_request(
            source_key.clone(),
            "button-blue-1",
        ))
        .await
        .expect("record occurrence before reopen");
    let first_delivery = factory
        .open
        .reserve_matching_deliveries(&occurrence.occurrence_id)
        .await
        .expect("reserve before reopen");
    assert_eq!(first_delivery.len(), 1);

    let reopened_subscriptions = factory
        .reopen
        .list_subscriptions(crate::TriggerSubscriptionFilter::for_session("session-a"))
        .await
        .expect("list subscriptions after reopen");
    assert_eq!(reopened_subscriptions.len(), 1);
    assert_eq!(reopened_subscriptions[0].handle, subscription.handle);

    let replayed = factory
        .reopen
        .record_occurrence(button_occurrence_request(
            source_key.clone(),
            "button-blue-1",
        ))
        .await
        .expect("replay after reopen");
    assert_eq!(replayed.occurrence_id, occurrence.occurrence_id);
    let replayed_delivery = factory
        .reopen
        .reserve_matching_deliveries(&replayed.occurrence_id)
        .await
        .expect("reserve replay after reopen");
    assert_eq!(replayed_delivery.len(), 1);
    assert_eq!(
        replayed_delivery[0].reservation_status,
        crate::TriggerDeliveryReservationStatus::AlreadyReserved
    );
    let reopened_deliveries = factory
        .reopen
        .list_deliveries_by_process_id(&first_delivery[0].process_id)
        .await
        .expect("list delivery after reopen");
    assert_eq!(reopened_deliveries.len(), 1);
    assert_eq!(
        reopened_deliveries[0].subscription.subscription_id,
        subscription.subscription_id
    );

    let next = factory
        .reopen
        .record_occurrence(button_occurrence_request(source_key, "button-blue-2"))
        .await
        .expect("record new occurrence after reopen");
    let next_delivery = factory
        .reopen
        .reserve_matching_deliveries(&next.occurrence_id)
        .await
        .expect("reserve new occurrence after reopen");
    assert_eq!(next_delivery.len(), 1);
    assert_eq!(next_delivery[0].subscription.handle, subscription.handle);
}
