laterite_admin/import/
mod.rs

1//! Bulk import: a file's rows written through a resource's form, with a report.
2//!
3//! A resource declaring `import: true` gets the pages under its own path: a
4//! file taken in, its columns shown beside the form's fields with a match
5//! proposed for each, a check that writes nothing, then the run and its report
6//! as it fills. Each row is written the way the form's own submit would: the
7//! same rules, field types, listeners, persister and audit entry, one commit a
8//! row. The run is detached from the request that started it and keeps its
9//! counts and notes in a table the report polls, so a long file finishes
10//! whether or not the browser waits, and a refused row is a line in the report,
11//! never a stopped import. From code, [`start`] and [`find`] do the same
12//! without the pages.
13
14pub(crate) mod file;
15pub(crate) mod migrations;
16mod pages;
17pub mod run;
18pub mod schema;
19mod store;
20
21pub use run::{
22    find, start, start_inline, Duplicates, ImportJob, Match, Note, NoteKind, OnMatch, Run,
23    RunError, RunSpec, Status,
24};
25
26pub(crate) use pages::ImportFeature;
27
28use laterite_core::queue::{Job, JobFailed};
29use laterite_core::{EventCx, EventError, Listener, MigrationSet, Module, ModuleId};
30
31/// The name of the disk uploaded files are stored on. The application may
32/// register a `laterite_media::DiskReg` under it; without one, the files go to
33/// `private` under the configured storage directory.
34pub const DISK: &str = "private";
35
36/// The `laterite.import` module: the run table. Built in.
37pub struct ImportModule;
38
39impl Module for ImportModule {
40    fn id(&self) -> ModuleId {
41        ModuleId::new(migrations::MODULE_ID)
42    }
43
44    fn migrations(&self) -> MigrationSet {
45        migrations::migrations()
46    }
47
48    /// A run whose job gave up (a worker lost mid-run, a file that would not
49    /// read) is marked failed from the queue's own record of it.
50    fn register(&self, registry: &mut laterite_core::Registry) {
51        registry.listen::<JobFailed>(RunFailed);
52    }
53}
54
55struct RunFailed;
56
57#[laterite_core::strata::async_trait]
58impl Listener<JobFailed> for RunFailed {
59    async fn handle(&self, cx: &EventCx<'_>, event: &JobFailed) -> Result<(), EventError> {
60        if event.name != run::ImportJob::NAME {
61            return Ok(());
62        }
63        let Some(run_id) = event.payload.get("run_id").and_then(|v| v.as_i64()) else {
64            return Ok(());
65        };
66        if let Ok(Some(found)) = store::find(cx.db(), run_id).await {
67            if matches!(found.status, run::Status::Queued | run::Status::Running) {
68                store::fail(cx.db(), run_id, &event.error).await?;
69            }
70        }
71        Ok(())
72    }
73}
laterite_admin/import/mod.rs

laterite_admin/import/
mod.rs
