cloacina 0.10.0

/*
 *  Copyright 2025-2026 Colliery Software
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

//! Unified Schedule Execution DAL with runtime backend selection
//!
//! This module provides operations for the unified `schedule_executions` table
//! that replaces the separate `cron_executions` and `trigger_executions` tables.
//! Works with both PostgreSQL and SQLite backends, selecting the appropriate
//! implementation at runtime based on the database connection type.

use diesel::prelude::*;

use super::DAL;
use crate::dal::unified::models::{NewUnifiedScheduleExecution, UnifiedScheduleExecution};
use crate::database::schema::unified::schedule_executions;
use crate::database::universal_types::{UniversalTimestamp, UniversalUuid};
use crate::error::ValidationError;
use crate::models::schedule::{NewScheduleExecution, ScheduleExecution};
use chrono::{DateTime, Duration, Utc};

/// Statistics about schedule execution performance
#[derive(Debug)]
pub struct ScheduleExecutionStats {
    /// Total number of executions attempted
    pub total_executions: i64,
    /// Number of executions that successfully handed off to workflow executor
    pub successful_executions: i64,
    /// Number of executions that were lost (started but never completed within expected time)
    pub lost_executions: i64,
    /// Success rate as a percentage
    pub success_rate: f64,
}

/// Data access layer for unified schedule execution operations with runtime backend selection.
#[derive(Clone)]
pub struct ScheduleExecutionDAL<'a> {
    dal: &'a DAL,
}

impl<'a> ScheduleExecutionDAL<'a> {
    /// Creates a new ScheduleExecutionDAL instance.
    pub fn new(dal: &'a DAL) -> Self {
        Self { dal }
    }

    /// Creates a new schedule execution record in the database.
    pub async fn create(
        &self,
        new_execution: NewScheduleExecution,
    ) -> Result<ScheduleExecution, ValidationError> {
        let id = UniversalUuid::new_v4();
        let now = UniversalTimestamp::now();

        let new_unified = NewUnifiedScheduleExecution {
            id,
            schedule_id: new_execution.schedule_id,
            workflow_execution_id: new_execution.workflow_execution_id,
            scheduled_time: new_execution.scheduled_time,
            claimed_at: new_execution.claimed_at,
            context_hash: new_execution.context_hash,
            started_at: now,
            created_at: now,
            updated_at: now,
        };

        crate::interact_on_backend!(self.dal, |conn| {
            diesel::insert_into(schedule_executions::table)
                .values(&new_unified)
                .execute(conn)
        })?;

        let result: UnifiedScheduleExecution = crate::interact_on_backend!(self.dal, |conn| {
            schedule_executions::table.find(id).first(conn)
        })?;

        Ok(result.into())
    }

    /// Retrieves a schedule execution by its ID.
    pub async fn get_by_id(&self, id: UniversalUuid) -> Result<ScheduleExecution, ValidationError> {
        let result: UnifiedScheduleExecution = crate::interact_on_backend!(self.dal, |conn| {
            schedule_executions::table.find(id).first(conn)
        })?;

        Ok(result.into())
    }

    /// Lists schedule executions for a given schedule.
    pub async fn list_by_schedule(
        &self,
        schedule_id: UniversalUuid,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<ScheduleExecution>, ValidationError> {
        let results: Vec<UnifiedScheduleExecution> =
            crate::interact_on_backend!(self.dal, |conn| {
                schedule_executions::table
                    .filter(schedule_executions::schedule_id.eq(schedule_id))
                    .order(schedule_executions::created_at.desc())
                    .limit(limit)
                    .offset(offset)
                    .load(conn)
            })?;

        Ok(results.into_iter().map(|r| r.into()).collect())
    }

    /// Marks a schedule execution as completed.
    pub async fn complete(
        &self,
        id: UniversalUuid,
        completed_at: DateTime<Utc>,
    ) -> Result<(), ValidationError> {
        let completed_ts = UniversalTimestamp::from(completed_at);
        let now = UniversalTimestamp::now();

        crate::interact_on_backend!(self.dal, |conn| {
            diesel::update(schedule_executions::table.find(id))
                .set((
                    schedule_executions::completed_at.eq(Some(completed_ts)),
                    schedule_executions::updated_at.eq(now),
                ))
                .execute(conn)
        })?;

        Ok(())
    }

    /// Checks if there is an active (uncompleted) execution for a schedule with the given context hash.
    pub async fn has_active_execution(
        &self,
        schedule_id: UniversalUuid,
        context_hash: &str,
    ) -> Result<bool, ValidationError> {
        let context_hash_owned = context_hash.to_string();

        let count: i64 = crate::interact_on_backend!(self.dal, |conn| {
            schedule_executions::table
                .filter(schedule_executions::schedule_id.eq(schedule_id))
                .filter(schedule_executions::context_hash.eq(context_hash_owned))
                .filter(schedule_executions::completed_at.is_null())
                .count()
                .get_result(conn)
        })?;

        Ok(count > 0)
    }

    /// Updates the workflow execution ID for a schedule execution.
    pub async fn update_workflow_execution_id(
        &self,
        id: UniversalUuid,
        workflow_execution_id: UniversalUuid,
    ) -> Result<(), ValidationError> {
        let now = UniversalTimestamp::now();

        crate::interact_on_backend!(self.dal, |conn| {
            diesel::update(schedule_executions::table.find(id))
                .set((
                    schedule_executions::workflow_execution_id.eq(Some(workflow_execution_id)),
                    schedule_executions::updated_at.eq(now),
                ))
                .execute(conn)
        })?;

        Ok(())
    }

    /// Finds lost executions (started but not completed) older than the specified minutes.
    pub async fn find_lost_executions(
        &self,
        older_than_minutes: i32,
    ) -> Result<Vec<ScheduleExecution>, ValidationError> {
        let cutoff = Utc::now() - Duration::minutes(older_than_minutes as i64);
        let cutoff_ts = UniversalTimestamp::from(cutoff);

        let results: Vec<UnifiedScheduleExecution> =
            crate::interact_on_backend!(self.dal, |conn| {
                schedule_executions::table
                    .filter(schedule_executions::completed_at.is_null())
                    .filter(schedule_executions::started_at.lt(cutoff_ts))
                    .order(schedule_executions::started_at.asc())
                    .load(conn)
            })?;

        Ok(results.into_iter().map(|r| r.into()).collect())
    }

    /// Gets the latest execution for a given schedule.
    pub async fn get_latest_by_schedule(
        &self,
        schedule_id: UniversalUuid,
    ) -> Result<Option<ScheduleExecution>, ValidationError> {
        let result: Option<UnifiedScheduleExecution> =
            crate::interact_on_backend!(self.dal, |conn| {
                schedule_executions::table
                    .filter(schedule_executions::schedule_id.eq(schedule_id))
                    .order(schedule_executions::created_at.desc())
                    .first(conn)
                    .optional()
            })?;

        Ok(result.map(|r| r.into()))
    }

    /// Gets execution statistics for monitoring and alerting.
    pub async fn get_execution_stats(
        &self,
        since: DateTime<Utc>,
    ) -> Result<ScheduleExecutionStats, ValidationError> {
        use crate::database::schema::unified::workflow_executions;

        let since_ts = UniversalTimestamp::from(since);
        let lost_cutoff = UniversalTimestamp::from(Utc::now() - Duration::minutes(10));

        let (total_executions, successful_executions, lost_executions) =
            crate::interact_on_backend!(self.dal, |conn| {
                let total_executions: i64 = schedule_executions::table
                    .filter(schedule_executions::started_at.ge(since_ts))
                    .count()
                    .first(conn)?;

                let successful_executions: i64 = schedule_executions::table
                    .filter(schedule_executions::started_at.ge(since_ts))
                    .filter(schedule_executions::workflow_execution_id.is_not_null())
                    .count()
                    .first(conn)?;

                let lost_executions: i64 = schedule_executions::table
                    .left_join(
                        workflow_executions::table.on(schedule_executions::workflow_execution_id
                            .eq(workflow_executions::id.nullable())),
                    )
                    .filter(workflow_executions::id.is_null())
                    .filter(schedule_executions::started_at.ge(since_ts))
                    .filter(schedule_executions::started_at.lt(lost_cutoff))
                    .count()
                    .first(conn)?;

                Ok::<(i64, i64, i64), diesel::result::Error>((
                    total_executions,
                    successful_executions,
                    lost_executions,
                ))
            })?;

        Ok(ScheduleExecutionStats {
            total_executions,
            successful_executions,
            lost_executions,
            success_rate: if total_executions > 0 {
                (successful_executions as f64 / total_executions as f64) * 100.0
            } else {
                0.0
            },
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::database::universal_types::UniversalTimestamp;
    use crate::database::Database;
    use crate::models::schedule::{NewSchedule, NewScheduleExecution};

    #[cfg(feature = "sqlite")]
    async fn unique_dal() -> DAL {
        let url = format!(
            "file:sched_exec_test_{}?mode=memory&cache=shared",
            uuid::Uuid::new_v4()
        );
        let db = Database::new(&url, "", 5);
        db.run_migrations()
            .await
            .expect("migrations should succeed");
        DAL::new(db)
    }

    /// Helper: create a cron schedule and return its ID.
    #[cfg(feature = "sqlite")]
    async fn create_schedule(dal: &DAL) -> UniversalUuid {
        let next_run = UniversalTimestamp::now();
        let schedule = dal
            .schedule()
            .create(NewSchedule::cron("test_wf", "0 * * * *", next_run))
            .await
            .unwrap();
        schedule.id
    }

    /// Helper: build a NewScheduleExecution for a given schedule.
    #[cfg(feature = "sqlite")]
    fn new_exec(schedule_id: UniversalUuid) -> NewScheduleExecution {
        NewScheduleExecution {
            schedule_id,
            workflow_execution_id: None,
            scheduled_time: Some(UniversalTimestamp::now()),
            claimed_at: Some(UniversalTimestamp::now()),
            context_hash: Some(uuid::Uuid::new_v4().to_string()),
        }
    }

    // ── create + get_by_id ──────────────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_create_execution() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;

        let exec = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();

        assert_eq!(exec.schedule_id, sched_id);
        assert!(exec.workflow_execution_id.is_none());
        assert!(exec.completed_at.is_none());
        assert!(exec.scheduled_time.is_some());
        assert!(exec.context_hash.is_some());
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_get_by_id() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;
        let created = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();

        let fetched = dal
            .schedule_execution()
            .get_by_id(created.id)
            .await
            .unwrap();
        assert_eq!(fetched.id, created.id);
        assert_eq!(fetched.schedule_id, sched_id);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_get_by_id_not_found() {
        let dal = unique_dal().await;
        let result = dal
            .schedule_execution()
            .get_by_id(UniversalUuid::new_v4())
            .await;
        assert!(result.is_err());
    }

    // ── list_by_schedule ────────────────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_list_by_schedule() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;
        let other_sched_id = create_schedule(&dal).await;

        // Create 3 executions for sched_id, 1 for other
        for _ in 0..3 {
            dal.schedule_execution()
                .create(new_exec(sched_id))
                .await
                .unwrap();
        }
        dal.schedule_execution()
            .create(new_exec(other_sched_id))
            .await
            .unwrap();

        let list = dal
            .schedule_execution()
            .list_by_schedule(sched_id, 100, 0)
            .await
            .unwrap();
        assert_eq!(list.len(), 3);

        // With limit
        let limited = dal
            .schedule_execution()
            .list_by_schedule(sched_id, 2, 0)
            .await
            .unwrap();
        assert_eq!(limited.len(), 2);

        // With offset
        let offset = dal
            .schedule_execution()
            .list_by_schedule(sched_id, 100, 2)
            .await
            .unwrap();
        assert_eq!(offset.len(), 1);
    }

    // ── complete ────────────────────────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_complete_execution() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;
        let exec = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();
        assert!(exec.completed_at.is_none());

        let completed_at = Utc::now();
        dal.schedule_execution()
            .complete(exec.id, completed_at)
            .await
            .unwrap();

        let updated = dal.schedule_execution().get_by_id(exec.id).await.unwrap();
        assert!(updated.completed_at.is_some());
    }

    // ── has_active_execution ────────────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_has_active_execution() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;

        // No executions yet
        let active = dal
            .schedule_execution()
            .has_active_execution(sched_id, "hash1")
            .await
            .unwrap();
        assert!(!active);

        // Create an uncompleted execution
        let mut ne = new_exec(sched_id);
        ne.context_hash = Some("hash1".to_string());
        dal.schedule_execution().create(ne).await.unwrap();

        let active = dal
            .schedule_execution()
            .has_active_execution(sched_id, "hash1")
            .await
            .unwrap();
        assert!(active);

        // Different hash should not match
        let active_other = dal
            .schedule_execution()
            .has_active_execution(sched_id, "hash_other")
            .await
            .unwrap();
        assert!(!active_other);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_has_active_execution_completed_not_active() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;

        let mut ne = new_exec(sched_id);
        ne.context_hash = Some("hash_done".to_string());
        let exec = dal.schedule_execution().create(ne).await.unwrap();

        // Complete it
        dal.schedule_execution()
            .complete(exec.id, Utc::now())
            .await
            .unwrap();

        let active = dal
            .schedule_execution()
            .has_active_execution(sched_id, "hash_done")
            .await
            .unwrap();
        assert!(!active);
    }

    // ── update_workflow_execution_id ────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_update_workflow_execution_id() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;
        let exec = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();
        assert!(exec.workflow_execution_id.is_none());

        // Create a real workflow execution so the FK constraint is satisfied
        use crate::models::workflow_execution::NewWorkflowExecution;
        let wf_exec = dal
            .workflow_execution()
            .create(NewWorkflowExecution {
                workflow_name: "fk-test".to_string(),
                workflow_version: "1.0".to_string(),
                status: "Running".to_string(),
                context_id: None,
            })
            .await
            .unwrap();

        dal.schedule_execution()
            .update_workflow_execution_id(exec.id, wf_exec.id)
            .await
            .unwrap();

        let updated = dal.schedule_execution().get_by_id(exec.id).await.unwrap();
        assert_eq!(updated.workflow_execution_id, Some(wf_exec.id));
    }

    // ── get_latest_by_schedule ──────────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_get_latest_by_schedule() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;

        // No executions => None
        let latest = dal
            .schedule_execution()
            .get_latest_by_schedule(sched_id)
            .await
            .unwrap();
        assert!(latest.is_none());

        // Create two executions; the second is "latest" by created_at
        let _first = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();
        let second = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();

        let latest = dal
            .schedule_execution()
            .get_latest_by_schedule(sched_id)
            .await
            .unwrap();
        assert!(latest.is_some());
        assert_eq!(latest.unwrap().id, second.id);
    }

    // ── find_lost_executions ────────────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_find_lost_executions_none_lost() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;

        // Create a fresh (just-started) execution — not lost yet
        dal.schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();

        // Looking for executions older than 60 minutes — our fresh one should not appear
        let lost = dal
            .schedule_execution()
            .find_lost_executions(60)
            .await
            .unwrap();
        assert!(lost.is_empty());
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_find_lost_executions_completed_not_lost() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;

        let exec = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();

        // Complete it so it should never be considered "lost"
        dal.schedule_execution()
            .complete(exec.id, Utc::now())
            .await
            .unwrap();

        let lost = dal
            .schedule_execution()
            .find_lost_executions(0)
            .await
            .unwrap();
        assert!(lost.is_empty());
    }

    // ── get_execution_stats ─────────────────────────────────────────

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_get_execution_stats_empty() {
        let dal = unique_dal().await;
        let since = Utc::now() - chrono::Duration::hours(1);

        let stats = dal
            .schedule_execution()
            .get_execution_stats(since)
            .await
            .unwrap();

        assert_eq!(stats.total_executions, 0);
        assert_eq!(stats.successful_executions, 0);
        assert_eq!(stats.lost_executions, 0);
        assert_eq!(stats.success_rate, 0.0);
    }

    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn test_get_execution_stats_with_data() {
        let dal = unique_dal().await;
        let sched_id = create_schedule(&dal).await;
        let since = Utc::now() - chrono::Duration::hours(1);

        // Create two executions
        let exec1 = dal
            .schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();
        dal.schedule_execution()
            .create(new_exec(sched_id))
            .await
            .unwrap();

        // Link one to a real workflow execution (FK constraint requires it to exist)
        use crate::models::workflow_execution::NewWorkflowExecution;
        let wf_exec = dal
            .workflow_execution()
            .create(NewWorkflowExecution {
                workflow_name: "stats-test".to_string(),
                workflow_version: "1.0".to_string(),
                status: "Completed".to_string(),
                context_id: None,
            })
            .await
            .unwrap();
        dal.schedule_execution()
            .update_workflow_execution_id(exec1.id, wf_exec.id)
            .await
            .unwrap();

        let stats = dal
            .schedule_execution()
            .get_execution_stats(since)
            .await
            .unwrap();

        assert_eq!(stats.total_executions, 2);
        assert_eq!(stats.successful_executions, 1);
        assert_eq!(stats.success_rate, 50.0);
    }
}