laterite_admin/import/
run.rs

1//! The run: a file's rows through a form, detached from the request, with a
2//! report a screen polls.
3
4use std::collections::HashMap;
5use std::sync::Arc;
6
7use crate::form::{FormHandle, Forms};
8use crate::persist::SaveError;
9use futures::TryStreamExt;
10use laterite_core::query::{bind_values, build, text_cast};
11use laterite_core::queue::{Job, JobCx, JobError, JobReg, Pushed, Queue};
12use laterite_core::strata::async_trait;
13use laterite_core::{t, Actor, AnyRowExt, Db, Text};
14use laterite_media::{blob_path, ByteStream, StorageDriver};
15use sea_query::{Alias, Expr, Query};
16use serde::{Deserialize, Serialize};
17
18use crate::import::store;
19
20/// How one run reads its file and writes its rows.
21#[derive(Debug, Clone, Serialize, Deserialize)]
22#[non_exhaustive]
23pub struct RunSpec {
24    /// The resource whose form the rows go through, as its descriptor declares
25    /// the path (`/products`). Recorded on the run.
26    pub resource: String,
27    /// The stored payload's media record, kept on the run for the report.
28    pub media_id: Option<i64>,
29    /// Which file column feeds which field. A column may feed several fields.
30    pub matches: Vec<Match>,
31    /// What a row does when its key matches a record already there. `None`
32    /// creates every row.
33    pub duplicates: Option<Duplicates>,
34    /// Whether the first row names the columns and is passed over. Default
35    /// `true`.
36    pub first_row_titles: bool,
37    /// The byte between values. Default `,`.
38    pub delimiter: u8,
39    /// Run every row through the rules and conversions with nothing written,
40    /// so the report says what an import would do. Default `false`.
41    #[serde(default)]
42    pub dry_run: bool,
43    /// On an update, a matched cell that is blank leaves the record's value
44    /// as it is. Default `true`; `false` clears the field.
45    #[serde(default = "yes")]
46    pub keep_blank: bool,
47    /// The file's column titles, kept so a later file with the same titles
48    /// is proposed the same matching.
49    #[serde(default)]
50    pub titles: Vec<String>,
51}
52
53fn yes() -> bool {
54    true
55}
56
57impl RunSpec {
58    /// A run into `resource` with these column matches.
59    pub fn new(resource: impl Into<String>, matches: Vec<Match>) -> Self {
60        Self {
61            resource: resource.into(),
62            media_id: None,
63            matches,
64            duplicates: None,
65            first_row_titles: true,
66            delimiter: b',',
67            dry_run: false,
68            keep_blank: true,
69            titles: Vec::new(),
70        }
71    }
72
73    /// Run every row with nothing written; the report says what would happen.
74    pub fn dry_run(mut self, dry: bool) -> Self {
75        self.dry_run = dry;
76        self
77    }
78
79    /// Whether a blank matched cell leaves an updated record's value as it is.
80    pub fn keep_blank(mut self, keep: bool) -> Self {
81        self.keep_blank = keep;
82        self
83    }
84
85    /// The file's column titles, for a later file to be matched the same way.
86    pub fn titles(mut self, titles: Vec<String>) -> Self {
87        self.titles = titles;
88        self
89    }
90
91    /// The stored payload's media record.
92    pub fn media(mut self, id: i64) -> Self {
93        self.media_id = Some(id);
94        self
95    }
96
97    /// What a row does when its key matches a record already there.
98    pub fn duplicates(mut self, duplicates: Duplicates) -> Self {
99        self.duplicates = Some(duplicates);
100        self
101    }
102
103    /// Whether the first row names the columns.
104    pub fn first_row_titles(mut self, titles: bool) -> Self {
105        self.first_row_titles = titles;
106        self
107    }
108
109    /// The byte between values.
110    pub fn delimiter(mut self, delimiter: u8) -> Self {
111        self.delimiter = delimiter;
112        self
113    }
114}
115
116/// One file column feeding one field.
117#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
118#[non_exhaustive]
119pub struct Match {
120    /// The column's position in the file, from 0.
121    pub column: usize,
122    /// The form field it fills.
123    pub field: String,
124}
125
126impl Match {
127    pub fn new(column: usize, field: impl Into<String>) -> Self {
128        Self {
129            column,
130            field: field.into(),
131        }
132    }
133}
134
135/// What a row does when a record already carries its key.
136#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
137#[non_exhaustive]
138pub struct Duplicates {
139    /// The matched field whose value identifies a record.
140    pub key: String,
141    pub on_match: OnMatch,
142}
143
144impl Duplicates {
145    /// A row whose `key` is already there is passed over and noted.
146    pub fn skip(key: impl Into<String>) -> Self {
147        Self {
148            key: key.into(),
149            on_match: OnMatch::Skip,
150        }
151    }
152
153    /// A row whose `key` is already there updates that record: the file's
154    /// columns over the record as it stands.
155    pub fn update(key: impl Into<String>) -> Self {
156        Self {
157            key: key.into(),
158            on_match: OnMatch::Update,
159        }
160    }
161}
162
163#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
164#[serde(rename_all = "lowercase")]
165pub enum OnMatch {
166    Skip,
167    Update,
168}
169
170/// Where a run is.
171#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
172#[serde(rename_all = "lowercase")]
173pub enum Status {
174    /// Pushed on the queue; a worker has not taken it yet.
175    Queued,
176    /// Rows are still being written; the counts move.
177    Running,
178    /// Every row was read; the counts and notes are final.
179    Done,
180    /// The run itself stopped (the file could not be read, the run table could
181    /// not be written); [`Run::failure`] says why. Rows written before it stay.
182    Failed,
183}
184
185impl Status {
186    pub(crate) fn as_str(self) -> &'static str {
187        match self {
188            Status::Queued => "queued",
189            Status::Running => "running",
190            Status::Done => "done",
191            Status::Failed => "failed",
192        }
193    }
194
195    pub(crate) fn parse(s: &str) -> Option<Self> {
196        match s {
197            "queued" => Some(Status::Queued),
198            "running" => Some(Status::Running),
199            "done" => Some(Status::Done),
200            "failed" => Some(Status::Failed),
201            _ => None,
202        }
203    }
204}
205
206/// One line of the report: what happened to one row.
207#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
208#[non_exhaustive]
209pub struct Note {
210    /// The row's line in the file, from 1, the title row counted.
211    pub row: u64,
212    pub kind: NoteKind,
213    /// The field the message is about, when it is about one.
214    pub field: Option<String>,
215    /// Localized where the report renders, like a validation message.
216    pub message: Text,
217}
218
219impl Note {
220    fn new(kind: NoteKind, row: u64, field: Option<String>, message: Text) -> Self {
221        Self {
222            row,
223            kind,
224            field,
225            message,
226        }
227    }
228}
229
230#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
231#[serde(rename_all = "lowercase")]
232pub enum NoteKind {
233    /// The row was not written.
234    Error,
235    /// The row was passed over by the duplicates policy.
236    Skipped,
237}
238
239/// A run as the table holds it: what it ran, where it is, what it did.
240#[derive(Debug, Clone)]
241#[non_exhaustive]
242pub struct Run {
243    pub id: i64,
244    pub resource: String,
245    pub media_id: Option<i64>,
246    /// What it ran: the matches, the policy, and whether anything was written.
247    pub spec: RunSpec,
248    pub actor_label: String,
249    pub status: Status,
250    /// Rows read, blank rows left out.
251    pub processed: i64,
252    pub created: i64,
253    pub updated: i64,
254    pub skipped: i64,
255    /// Rows refused.
256    pub errors: i64,
257    pub notes: Vec<Note>,
258    pub failure: Option<String>,
259    pub started_at: String,
260    pub finished_at: Option<String>,
261}
262
263#[derive(Debug, thiserror::Error)]
264#[non_exhaustive]
265pub enum RunError {
266    #[error("database error: {0}")]
267    Database(#[from] sqlx::Error),
268    /// The spec does not fit the form: a match naming a field it lacks, no
269    /// matches at all, a duplicates key no column feeds.
270    #[error("the run cannot start: {0}")]
271    Spec(String),
272    #[error("the file could not be read: {0}")]
273    Read(#[from] std::io::Error),
274}
275
276/// Rows counted as the run goes.
277#[derive(Debug, Default, Clone, Copy)]
278pub(crate) struct Counts {
279    pub processed: i64,
280    pub created: i64,
281    pub updated: i64,
282    pub skipped: i64,
283    pub errors: i64,
284}
285
286/// How many rows between writes of the counts, so a poll sees movement.
287const PROGRESS_EVERY: i64 = 25;
288
289/// The job an import runs as: the run's id and everything the worker needs
290/// to pick the file up and write the rows. Pushed by [`start`]; the handler
291/// [`job_reg`] builds reads the file from the private disk.
292#[derive(Debug, Clone, Serialize, Deserialize)]
293pub struct ImportJob {
294    pub run_id: i64,
295    /// The resource's path as its descriptor declares it (`/products`).
296    pub resource: String,
297    pub spec: RunSpec,
298    pub actor: Actor,
299    /// The stored file's hash on the private disk.
300    pub hash: String,
301}
302
303#[async_trait]
304impl Job for ImportJob {
305    const NAME: &'static str = "laterite.import::run";
306
307    /// Never called: the handler registered by the admin router runs the
308    /// job, since it needs the resource's prepared form and the disk.
309    async fn run(&self, _cx: &JobCx) -> Result<(), JobError> {
310        Err(JobError::fail(
311            "the import job runs through the admin's handler",
312        ))
313    }
314
315    /// One attempt: rows written before a failure stay, so a second pass
316    /// would write them again.
317    fn tries(&self) -> u32 {
318        1
319    }
320
321    /// A large file takes as long as it takes.
322    fn timeout(&self) -> Option<std::time::Duration> {
323        None
324    }
325
326    fn unique_key(&self) -> Option<String> {
327        Some(format!("import:{}", self.run_id))
328    }
329}
330
331/// The job's handler, registered by the admin router once the forms and the
332/// private disk exist: reads the file and runs the rows through the form.
333pub(crate) fn job_reg(forms: Forms, disk: Arc<dyn StorageDriver>) -> JobReg {
334    JobReg::with_handler::<ImportJob, _, _>(move |job: ImportJob, cx: JobCx| {
335        let forms = Arc::clone(&forms);
336        let disk = Arc::clone(&disk);
337        async move {
338            let db = cx.db().clone();
339            let outcome = async {
340                let prepared = forms.get(&job.resource).ok_or_else(|| {
341                    RunError::Spec(format!("no resource imports at `{}`", job.resource))
342                })?;
343                let form = FormHandle::new(db.clone(), cx.cache().clone(), prepared.clone());
344                store::started(&db, job.run_id).await?;
345                let source = disk
346                    .get(&blob_path(&job.hash))
347                    .await
348                    .map_err(|e| RunError::Read(std::io::Error::other(e.to_string())))?;
349                run(&db, job.run_id, &form, &job.actor, &job.spec, source).await
350            }
351            .await;
352            if let Err(e) = outcome {
353                tracing::error!(run = job.run_id, error = %e, "import run stopped");
354                if let Err(e2) = store::fail(&db, job.run_id, &e.to_string()).await {
355                    tracing::error!(run = job.run_id, error = %e2, "import run could not be marked failed");
356                }
357                return Err(JobError::fail(e.to_string()));
358            }
359            Ok(())
360        }
361    })
362}
363
364/// Starts a run: records it as queued, pushes its job and returns the run's
365/// id for [`find`] to poll. In `inline` mode the job runs here and the run
366/// is settled on return; otherwise a worker takes it.
367///
368/// `resource` is the path the form is registered under; `hash` names the
369/// stored file; `actor` is who the rows are written as, on every record and
370/// audit entry. The spec is checked against the form first, so a mistaken
371/// screen fails here and no run row is left behind.
372pub async fn start(
373    db: &Db,
374    queue: &Queue,
375    resource: &str,
376    form: &FormHandle,
377    actor: Actor,
378    spec: RunSpec,
379    hash: &str,
380) -> Result<i64, RunError> {
381    check(&spec, form)?;
382    let id = store::insert(db, &spec, &actor).await?;
383    let job = ImportJob {
384        run_id: id,
385        resource: resource.to_string(),
386        spec,
387        actor,
388        hash: hash.to_string(),
389    };
390    match queue.push(&job).await {
391        Ok(Pushed::Queued(_)) | Ok(Pushed::Ran) | Ok(Pushed::Duplicate) => Ok(id),
392        Err(e) => {
393            // An inline failure already marked the run; a push that was
394            // refused leaves a queued row nothing will take.
395            if let Ok(Some(run)) = store::find(db, id).await {
396                if run.status == Status::Queued {
397                    let _ = store::fail(db, id, &e.to_string()).await;
398                }
399            }
400            Ok(id)
401        }
402    }
403}
404
405/// Runs a file through a form on this task, to completion: what a test or a
406/// tool with the bytes in hand calls. Returns the run's id.
407pub async fn start_inline(
408    db: &Db,
409    form: FormHandle,
410    actor: Actor,
411    spec: RunSpec,
412    source: ByteStream,
413) -> Result<i64, RunError> {
414    check(&spec, &form)?;
415    let id = store::insert(db, &spec, &actor).await?;
416    store::started(db, id).await?;
417    if let Err(e) = run(db, id, &form, &actor, &spec, source).await {
418        tracing::error!(run = id, error = %e, "import run stopped");
419        store::fail(db, id, &e.to_string()).await?;
420    }
421    Ok(id)
422}
423
424fn check(spec: &RunSpec, form: &FormHandle) -> Result<(), RunError> {
425    if spec.matches.is_empty() {
426        return Err(RunError::Spec(
427            "no column is matched to a field".to_string(),
428        ));
429    }
430    let known = |name: &str| form.fields().iter().any(|f| f.name == name);
431    for m in &spec.matches {
432        if !known(&m.field) {
433            return Err(RunError::Spec(format!(
434                "column {} is matched to `{}`, which is not a field of this form",
435                m.column, m.field
436            )));
437        }
438    }
439    if let Some(duplicates) = &spec.duplicates {
440        if !spec.matches.iter().any(|m| m.field == duplicates.key) {
441            return Err(RunError::Spec(format!(
442                "duplicates are keyed on `{}`, which no column feeds",
443                duplicates.key
444            )));
445        }
446    }
447    Ok(())
448}
449
450/// One run by id, or `None` when there is no such run.
451pub async fn find(db: &Db, id: i64) -> Result<Option<Run>, RunError> {
452    Ok(store::find(db, id).await?)
453}
454
455async fn run(
456    db: &Db,
457    id: i64,
458    form: &FormHandle,
459    actor: &Actor,
460    spec: &RunSpec,
461    source: ByteStream,
462) -> Result<(), RunError> {
463    let bytes: Vec<u8> = source
464        .try_fold(Vec::new(), |mut all, chunk| async move {
465            all.extend_from_slice(&chunk);
466            Ok(all)
467        })
468        .await?;
469    let mut reader = csv::ReaderBuilder::new()
470        .delimiter(spec.delimiter)
471        .has_headers(spec.first_row_titles)
472        .flexible(true)
473        .from_reader(bytes.as_slice());
474
475    let mut counts = Counts::default();
476    let mut notes = Vec::new();
477    for record in reader.records() {
478        let record = match record {
479            Ok(record) => record,
480            Err(e) => {
481                let row = e.position().map(|p| p.line()).unwrap_or(0);
482                counts.processed += 1;
483                counts.errors += 1;
484                notes.push(Note::new(
485                    NoteKind::Error,
486                    row,
487                    None,
488                    t!(
489                        "The row could not be read: {reason}",
490                        reason = e.to_string()
491                    ),
492                ));
493                continue;
494            }
495        };
496        // A blank line is nothing to import, so it is neither a row nor a note.
497        if record.iter().all(|v| v.trim().is_empty()) {
498            continue;
499        }
500        let row = record.position().map(|p| p.line()).unwrap_or(0);
501        let mut data = HashMap::new();
502        for m in &spec.matches {
503            data.insert(
504                m.field.clone(),
505                record.get(m.column).unwrap_or_default().to_string(),
506            );
507        }
508        counts.processed += 1;
509        write_row(db, form, actor, spec, row, data, &mut counts, &mut notes).await;
510        if counts.processed % PROGRESS_EVERY == 0 {
511            store::progress(db, id, &counts).await?;
512        }
513    }
514    store::finish(db, id, &counts, &notes).await?;
515    Ok(())
516}
517
518/// Writes one row as its spec says, counting what happened and noting anything
519/// the operator should hear about.
520#[allow(clippy::too_many_arguments)]
521async fn write_row(
522    db: &Db,
523    form: &FormHandle,
524    actor: &Actor,
525    spec: &RunSpec,
526    row: u64,
527    data: HashMap<String, String>,
528    counts: &mut Counts,
529    notes: &mut Vec<Note>,
530) {
531    if let Some(duplicates) = &spec.duplicates {
532        let key = data.get(&duplicates.key).cloned().unwrap_or_default();
533        if !key.is_empty() {
534            let found = match existing(db, form, &duplicates.key, &key).await {
535                Ok(found) => found,
536                Err(e) => {
537                    tracing::error!(row, error = %e, "import duplicate lookup failed");
538                    counts.errors += 1;
539                    notes.push(Note::new(
540                        NoteKind::Error,
541                        row,
542                        Some(duplicates.key.clone()),
543                        t!("The existing record could not be looked up."),
544                    ));
545                    return;
546                }
547            };
548            if let Some(id) = found {
549                match duplicates.on_match {
550                    OnMatch::Skip => {
551                        counts.skipped += 1;
552                        notes.push(Note::new(
553                            NoteKind::Skipped,
554                            row,
555                            Some(duplicates.key.clone()),
556                            t!(
557                                "A record with this {field} is already there.",
558                                field = duplicates.key.clone()
559                            ),
560                        ));
561                    }
562                    OnMatch::Update => {
563                        // The file's columns over the record as it stands, so a
564                        // field the file does not carry keeps its value, and so
565                        // does one whose cell is blank when the spec says so.
566                        let current = match form.load(id).await {
567                            Ok(current) => current.unwrap_or_default(),
568                            Err(e) => {
569                                refused(row, e, counts, notes);
570                                return;
571                            }
572                        };
573                        let mut merged = current;
574                        for (field, value) in data {
575                            if spec.keep_blank && value.trim().is_empty() {
576                                continue;
577                            }
578                            merged.insert(field, value);
579                        }
580                        let outcome = if spec.dry_run {
581                            form.check(Some(id), merged).await
582                        } else {
583                            form.update(id, merged, actor).await
584                        };
585                        match outcome {
586                            Ok(()) => counts.updated += 1,
587                            Err(e) => refused(row, e, counts, notes),
588                        }
589                    }
590                }
591                return;
592            }
593        }
594    }
595    let outcome = if spec.dry_run {
596        form.check(None, data).await
597    } else {
598        form.create(data, actor).await.map(|_| ())
599    };
600    match outcome {
601        Ok(()) => counts.created += 1,
602        Err(e) => refused(row, e, counts, notes),
603    }
604}
605
606/// Notes a row the form would not take: one line per message, as the form
607/// would show them, or the reason a write failed.
608fn refused(row: u64, error: SaveError, counts: &mut Counts, notes: &mut Vec<Note>) {
609    counts.errors += 1;
610    match error {
611        SaveError::Invalid(bag) => {
612            for (field, messages) in bag.iter() {
613                for message in messages {
614                    notes.push(Note::new(
615                        NoteKind::Error,
616                        row,
617                        Some(field.to_string()),
618                        message.clone(),
619                    ));
620                }
621            }
622        }
623        SaveError::Failed(reason) => {
624            tracing::warn!(row, error = %reason, "import row could not be saved");
625            notes.push(Note::new(
626                NoteKind::Error,
627                row,
628                None,
629                t!("Could not save: {reason}", reason = reason),
630            ));
631        }
632    }
633}
634
635/// The id of the record whose `key` column holds `value`, if one does.
636async fn existing(
637    db: &Db,
638    form: &FormHandle,
639    key: &str,
640    value: &str,
641) -> Result<Option<i64>, sqlx::Error> {
642    // Scoped so the builder is dropped before the await: its identifiers are
643    // reference-counted, and a live one across the await would make the run's
644    // future non-`Send`.
645    let (sql, values) = {
646        let cast = text_cast(db.backend);
647        build(
648            db.backend,
649            Query::select()
650                .column(Alias::new(form.id_field()))
651                .from(Alias::new(form.entity()))
652                .and_where(
653                    Expr::col(Alias::new(key))
654                        .cast_as(Alias::new(cast))
655                        .eq(value),
656                )
657                .limit(1)
658                .to_owned(),
659        )
660    };
661    let row = bind_values(sqlx::query(&sql), values)
662        .fetch_optional(&db.pool)
663        .await?;
664    row.map(|r| r.get_int(form.id_field())).transpose()
665}
laterite_admin/import/run.rs

laterite_admin/import/
run.rs
