// file: crates/ksp-app-backfill-desk/src/backfill_run.rs
// version: 4

//! Single-active-run admission and launch ownership for Backfill Desk.

/// Fully prepared Backfill execution moved into the Tauri async runtime after admission.
pub(crate) struct BackfillRunLaunch {
    /// Backend-generated logical Job identity.
    pub(crate) job_id: ksp_job_api::JobId,
    /// Concrete Backfill runtime owning discovery, hydration, persistence and latest-value publication.
    pub(crate) runtime: ksp_job_backfill_lib::BackfillJobRuntime,
    /// Independent latest-value source retained by the monitoring bridge.
    pub(crate) snapshots: ksp_job_backfill_lib::BackfillSnapshotSource,
    /// Store facade temporarily borrowed from the application Store runtime for the duration of this run.
    pub(crate) store: ksp_store_lib::Store,
    /// Shareable HTTP Transport pool selected by the active composite profile.
    pub(crate) transport: ksp_onchain_transport_lib::HttpTransportPool,
}

impl crate::BackfillRunLaunch {
    /// Builds the safe acknowledgement returned immediately after the non-blocking Start admission.
    #[must_use]
    pub(crate) fn response(&self) -> crate::BackfillStartResponseDto {
        return crate::BackfillStartResponseDto { job_id: self.job_id.as_str().to_owned(), state: ksp_job_api::JobState::Created.code().to_owned() };
    }
}

/// Single-run control slot retained by the application while one Backfill runtime is active.
pub(crate) struct BackfillRunState {
    active: std::sync::Mutex<std::option::Option<ActiveBackfillRun>>,
    last_terminal: std::sync::Mutex<std::option::Option<TerminalBackfillRun>>,
    next_sequence: std::sync::atomic::AtomicU64,
}

impl crate::BackfillRunState {
    /// Creates an empty single-run state for one desktop application session.
    #[must_use]
    pub(crate) const fn new() -> Self {
        return Self {
            active: std::sync::Mutex::new(std::option::Option::None),
            last_terminal: std::sync::Mutex::new(std::option::Option::None),
            next_sequence: std::sync::atomic::AtomicU64::new(1),
        };
    }

    /// Allocates one bounded backend-owned Job identity without accepting caller-supplied identity material.
    pub(crate) fn next_job_id(&self) -> ksp_core_lib::Result<ksp_job_api::JobId> {
        let sequence = self
            .next_sequence
            .fetch_update(std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire, |current| return current.checked_add(1));
        let sequence = match sequence {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_INVALID,
                    "Backfill Desk Job identity sequence is exhausted",
                ));
            },
        };
        return ksp_job_api::JobId::new(format!("backfill-desk-{sequence}"));
    }

    /// Installs one control handle atomically and rejects concurrent Starts while another run owns the slot.
    pub(crate) fn install(
        &self,
        job_id: ksp_job_api::JobId,
        handle: ksp_job_backfill_lib::BackfillJobHandle,
        request: ksp_job_backfill_lib::BackfillRequest,
    ) -> ksp_core_lib::Result<()> {
        let active = self.active.lock();
        let mut active = match active {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk active-run state lock is poisoned",
                ));
            },
        };
        if active.is_some() {
            return std::result::Result::Err(ksp_core_lib::Error::new(
                crate::ERROR_CODE_BACKFILL_RUN_ACTIVE,
                "Backfill Desk already has an active Backfill run",
            ));
        }
        *active = std::option::Option::Some(ActiveBackfillRun { handle, job_id, request });
        return std::result::Result::Ok(());
    }

    /// Clears the matching terminal run and reports whether cancellation had been requested on its retained handle.
    pub(crate) fn finish(&self, job_id: &ksp_job_api::JobId) -> ksp_core_lib::Result<bool> {
        let active = self.active.lock();
        let mut active = match active {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk active-run state lock is poisoned",
                ));
            },
        };
        let current = active.as_ref();
        let current = match current {
            std::option::Option::Some(value) if &value.job_id == job_id => value,
            std::option::Option::Some(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_INVALID,
                    "Backfill Desk terminal run does not match the active Job identity",
                ));
            },
            std::option::Option::None => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_INVALID,
                    "Backfill Desk terminal run has no active admission slot",
                ));
            },
        };
        let cancellation_requested = current.handle.is_cancellation_requested();
        let request = current.request.clone();
        let source = current.handle.snapshots();
        let notification = <ksp_job_backfill_lib::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&source);
        if notification.state().is_terminal() {
            let terminal = self.last_terminal.lock();
            let mut terminal = match terminal {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(_) => {
                    return std::result::Result::Err(ksp_core_lib::Error::new(
                        crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                        "Backfill Desk terminal monitoring state lock is poisoned",
                    ));
                },
            };
            *terminal = std::option::Option::Some(TerminalBackfillRun { notification, request });
        }
        *active = std::option::Option::None;
        return std::result::Result::Ok(cancellation_requested);
    }

    /// Requests cooperative cancellation for the exact backend-issued active Job identity.
    pub(crate) fn cancel(&self, job_id: &str) -> ksp_core_lib::Result<crate::BackfillCancelResponseDto> {
        let active = self.active.lock();
        let active = match active {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk active-run cancellation state lock is poisoned",
                ));
            },
        };
        if let std::option::Option::Some(current) = active.as_ref() {
            if current.job_id.as_str() != job_id {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_BACKFILL_RUN_MISMATCH,
                    "Backfill Desk cancellation target does not match the active Job",
                ));
            }
            let source = current.handle.snapshots();
            let notification = <ksp_job_backfill_lib::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&source);
            let state = notification.state();
            if state.is_terminal() {
                return std::result::Result::Ok(crate::BackfillCancelResponseDto {
                    accepted: false,
                    job_id: current.job_id.as_str().to_owned(),
                    state: state.code().to_owned(),
                });
            }
            let accepted = current.handle.cancel();
            let state_code = if current.handle.is_cancellation_requested() { "cancelling" } else { state.code() };
            return std::result::Result::Ok(crate::BackfillCancelResponseDto {
                accepted,
                job_id: current.job_id.as_str().to_owned(),
                state: state_code.to_owned(),
            });
        }
        drop(active);
        let terminal = self.last_terminal.lock();
        let terminal = match terminal {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk terminal cancellation state lock is poisoned",
                ));
            },
        };
        if let std::option::Option::Some(terminal) = terminal.as_ref() {
            let notification = &terminal.notification;
            if notification.id().as_str() != job_id {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_BACKFILL_RUN_MISMATCH,
                    "Backfill Desk cancellation target does not match the retained terminal Job",
                ));
            }
            return std::result::Result::Ok(crate::BackfillCancelResponseDto {
                accepted: false,
                job_id: notification.id().as_str().to_owned(),
                state: notification.state().code().to_owned(),
            });
        }
        return std::result::Result::Err(ksp_core_lib::Error::new(
            crate::ERROR_CODE_BACKFILL_RUN_NOT_ACTIVE,
            "Backfill Desk has no active Backfill run to cancel",
        ));
    }

    /// Requests best-effort cooperative cancellation of whichever run is active after shutdown wins admission.
    pub(crate) fn cancel_active_for_shutdown(&self) -> ksp_core_lib::Result<bool> {
        let active = self.active.lock();
        let active = match active {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk shutdown cancellation state lock is poisoned",
                ));
            },
        };
        return match active.as_ref() {
            std::option::Option::Some(current) => std::result::Result::Ok(current.handle.cancel()),
            std::option::Option::None => std::result::Result::Ok(false),
        };
    }

    /// Returns the complete current active snapshot or the retained latest terminal snapshot for resynchronization.
    pub(crate) fn current_notification(
        &self,
    ) -> ksp_core_lib::Result<std::option::Option<ksp_job_api::JobNotification<ksp_job_backfill_lib::BackfillJobSnapshot>>> {
        let active = self.active.lock();
        let active = match active {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk active-run monitoring state lock is poisoned",
                ));
            },
        };
        if let std::option::Option::Some(current) = active.as_ref() {
            let source = current.handle.snapshots();
            let notification = <ksp_job_backfill_lib::BackfillSnapshotSource as ksp_job_api::JobSnapshotSource>::current(&source);
            return std::result::Result::Ok(std::option::Option::Some(notification));
        }
        drop(active);
        let terminal = self.last_terminal.lock();
        let terminal = match terminal {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk terminal monitoring state lock is poisoned",
                ));
            },
        };
        return std::result::Result::Ok(terminal.as_ref().map(|terminal| return terminal.notification.clone()));
    }

    /// Rebuilds the last terminal request with its opaque checkpoint reissued onto one new Job identity.
    pub(crate) fn resume_request(&self, job_id: ksp_job_api::JobId) -> ksp_core_lib::Result<ksp_job_backfill_lib::BackfillRequest> {
        let active = self.active.lock();
        let active = match active {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk active-run resume state lock is poisoned",
                ));
            },
        };
        if active.is_some() {
            return std::result::Result::Err(ksp_core_lib::Error::new(
                crate::ERROR_CODE_BACKFILL_RUN_ACTIVE,
                "Backfill Desk cannot resume while another Backfill run owns the single-run slot",
            ));
        }
        drop(active);
        let terminal = self.last_terminal.lock();
        let terminal = match terminal {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(_) => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_APP_STATE_LOCK_FAILED,
                    "Backfill Desk terminal resume state lock is poisoned",
                ));
            },
        };
        let terminal = match terminal.as_ref() {
            std::option::Option::Some(value) => value,
            std::option::Option::None => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_BACKFILL_RESUME_UNAVAILABLE,
                    "Backfill Desk has no retained terminal run available for in-session Resume",
                ));
            },
        };
        let checkpoint = terminal.notification.snapshot().checkpoint();
        let checkpoint = match checkpoint {
            std::option::Option::Some(value) => value.clone(),
            std::option::Option::None => {
                return std::result::Result::Err(ksp_core_lib::Error::new(
                    crate::ERROR_CODE_BACKFILL_RESUME_UNAVAILABLE,
                    "Backfill Desk retained terminal run has no safe checkpoint for in-session Resume",
                ));
            },
        };
        return terminal.request.resume_for_job(job_id, checkpoint);
    }

    /// Rolls back a just-installed admission when execution resources cannot be acquired before spawn.
    pub(crate) fn rollback(&self, job_id: &ksp_job_api::JobId) -> ksp_core_lib::Result<()> {
        let finished = self.finish(job_id);
        return match finished {
            std::result::Result::Ok(_) => std::result::Result::Ok(()),
            std::result::Result::Err(error) => std::result::Result::Err(error),
        };
    }
}

struct ActiveBackfillRun {
    handle: ksp_job_backfill_lib::BackfillJobHandle,
    job_id: ksp_job_api::JobId,
    request: ksp_job_backfill_lib::BackfillRequest,
}

#[derive(Clone)]
struct TerminalBackfillRun {
    notification: ksp_job_api::JobNotification<ksp_job_backfill_lib::BackfillJobSnapshot>,
    request: ksp_job_backfill_lib::BackfillRequest,
}

#[cfg(test)]
#[path = "../unit_tests/backfill_run.rs"]
mod tests;