gunnar_server/
sso_import_queue.rs

1//! **The directory import, one per account at a time, on its own jera lane**
2//! (A34-17).
3//!
4//! `Sso.ImportUsers` used to run up to 10 000 store writes inside the gRPC
5//! call, on the async runtime, and a failure half way left half the file
6//! applied. Now the call checks the caller and the size and enqueues; this
7//! kind runs [`gunnar_grpc::sso::scim::run_import`] — edda's plan, applied in
8//! ONE identity-table write or not at all — and the console polls
9//! `Sso.ImportStatus`. Its own lane: an import waits behind nobody's gc or
10//! purge, and nobody's gc waits behind it.
11
12use gunnar_grpc::sso::scim::{ImportJob, ImportOrder, Imported};
13
14use crate::job_queue::{Job, JobKind, JobQueue, JobState, Reporter};
15
16/// What an import came to (the shared crate's counts and ended sessions).
17#[derive(Debug, Clone, Default)]
18pub struct ImportReport {
19    pub outcome: Imported,
20}
21
22/// **The import kind.**
23pub struct SsoImport;
24
25impl JobKind for SsoImport {
26    const NAME: &'static str = "sso-import";
27
28    type Order = ImportOrder;
29    type Report = ImportReport;
30
31    /// The account: one import per account at a time.
32    fn key(order: &ImportOrder) -> &str {
33        &order.account
34    }
35
36    fn run(&self, order: &ImportOrder, report: &Reporter<SsoImport>) -> Result<(), String> {
37        let progress = |done: u64, total: u64| report.progress(done, total);
38        let done = gunnar_grpc::sso::scim::run_import(order, crate::job_queue::now_ms(), &progress);
39        (order.finished)(&done);
40        let done = done?;
41        report.report(|r| r.outcome = done);
42        Ok(())
43    }
44}
45
46/// The queue the control plane holds.
47pub type SsoImportQueue = JobQueue<SsoImport>;
48
49/// A job as the seam answers it.
50pub fn job_out(job: Job<SsoImport>) -> ImportJob {
51    let (done_rows, total_rows) = job.progress.unwrap_or((0, 0));
52    let mut outcome = job.report.outcome.clone();
53    outcome.dry_run = job.order.dry_run;
54    ImportJob {
55        job_id: job.job_id.clone(),
56        state: match job.state {
57            JobState::Queued => "queued",
58            JobState::Running => "running",
59            JobState::Done => "done",
60            JobState::Failed => "failed",
61        },
62        error: job.error.clone().unwrap_or_default(),
63        done_rows,
64        total_rows,
65        outcome,
66    }
67}
gunnar_server/sso_import_queue.rs

gunnar_server/
sso_import_queue.rs
