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

laterite_admin/import/
run.rs
