nordisk/gunnar

crates/gunnar-server/src/sso_import_queue.rs

//! **The directory import, one per account at a time, on its own jera lane**
//! (A34-17).
//!
//! `Sso.ImportUsers` used to run up to 10 000 store writes inside the gRPC
//! call, on the async runtime, and a failure half way left half the file
//! applied. Now the call checks the caller and the size and enqueues; this
//! kind runs [`gunnar_grpc::sso::scim::run_import`] — edda's plan, applied in
//! ONE identity-table write or not at all — and the console polls
//! `Sso.ImportStatus`. Its own lane: an import waits behind nobody's gc or
//! purge, and nobody's gc waits behind it.

use gunnar_grpc::sso::scim::{ImportJob, ImportOrder, Imported};

use crate::job_queue::{Job, JobKind, JobQueue, JobState, Reporter};

/// What an import came to (the shared crate's counts and ended sessions).
#[derive(Debug, Clone, Default)]
pub struct ImportReport {
    pub outcome: Imported,
}

/// **The import kind.**
pub struct SsoImport;

impl JobKind for SsoImport {
    const NAME: &'static str = "sso-import";

    type Order = ImportOrder;
    type Report = ImportReport;

    /// The account: one import per account at a time.
    fn key(order: &ImportOrder) -> &str {
        &order.account
    }

    fn run(&self, order: &ImportOrder, report: &Reporter<SsoImport>) -> Result<(), String> {
        let progress = |done: u64, total: u64| report.progress(done, total);
        let done = gunnar_grpc::sso::scim::run_import(order, crate::job_queue::now_ms(), &progress);
        (order.finished)(&done);
        let done = done?;
        report.report(|r| r.outcome = done);
        Ok(())
    }
}

/// The queue the control plane holds.
pub type SsoImportQueue = JobQueue<SsoImport>;

/// A job as the seam answers it.
pub fn job_out(job: Job<SsoImport>) -> ImportJob {
    let (done_rows, total_rows) = job.progress.unwrap_or((0, 0));
    let mut outcome = job.report.outcome.clone();
    outcome.dry_run = job.order.dry_run;
    ImportJob {
        job_id: job.job_id.clone(),
        state: match job.state {
            JobState::Queued => "queued",
            JobState::Running => "running",
            JobState::Done => "done",
            JobState::Failed => "failed",
        },
        error: job.error.clone().unwrap_or_default(),
        done_rows,
        total_rows,
        outcome,
    }
}