pito_sync/
book.rs

1use std::time::{Duration, SystemTime, UNIX_EPOCH};
2
3use rusqlite::{Connection, OptionalExtension, Transaction, TransactionBehavior};
4
5use crate::Failure;
6
7pub(crate) fn now_ms() -> i64 {
8    SystemTime::now()
9        .duration_since(UNIX_EPOCH)
10        .map(|d| d.as_millis() as i64)
11        .unwrap_or(0)
12}
13
14#[derive(Debug, Clone, Default, PartialEq, Eq)]
15pub struct Sync {
16    pub source: String,
17    pub started_at: Option<i64>,
18    pub finished_at: Option<i64>,
19    pub ok: Option<bool>,
20    pub error: Option<String>,
21    pub cursor: Option<String>,
22    pub rows: Option<i64>,
23}
24
25pub fn sync(conn: &Connection, source: &str) -> rusqlite::Result<Option<Sync>> {
26    conn.query_row(
27        "SELECT source, started_at, finished_at, ok, error, cursor, rows FROM sync WHERE source = ?1",
28        [source],
29        |row| {
30            Ok(Sync {
31                source: row.get(0)?,
32                started_at: row.get(1)?,
33                finished_at: row.get(2)?,
34                ok: row.get(3)?,
35                error: row.get(4)?,
36                cursor: row.get(5)?,
37                rows: row.get(6)?,
38            })
39        },
40    )
41    .optional()
42}
43
44pub fn retry_after(conn: &Connection, source: &str) -> rusqlite::Result<Option<i64>> {
45    conn.query_row(
46        "SELECT retry_after FROM sync WHERE source = ?1",
47        [source],
48        |row| row.get(0),
49    )
50    .optional()
51    .map(Option::flatten)
52}
53
54pub fn started(conn: &Connection, source: &str, at: i64) -> rusqlite::Result<()> {
55    conn.execute(
56        "INSERT INTO sync (source, started_at) VALUES (?1, ?2)
57         ON CONFLICT (source) DO UPDATE SET started_at = excluded.started_at",
58        (source, at),
59    )
60    .map(|_| ())
61}
62
63pub fn failed(conn: &Connection, source: &str, failure: &Failure) -> rusqlite::Result<()> {
64    conn.execute(
65        "INSERT INTO sync (source, finished_at, ok, error, retry_after) VALUES (?1, ?2, 0, ?3, ?4)
66         ON CONFLICT (source) DO UPDATE SET
67           finished_at = max(excluded.finished_at, coalesce(started_at, excluded.finished_at)),
68           ok = 0, error = excluded.error, retry_after = excluded.retry_after",
69        (source, now_ms(), failure.text(), failure.retry_after),
70    )
71    .map(|_| ())
72}
73
74pub struct Pass<'a> {
75    pub source: &'a str,
76    pub started_at: i64,
77    pub finished_at: i64,
78    pub failure: Option<Failure>,
79    pub cursor: Option<String>,
80}
81
82pub fn commit(
83    conn: &mut Connection,
84    pass: Pass,
85    write: impl FnOnce(&Transaction) -> rusqlite::Result<usize>,
86) -> rusqlite::Result<usize> {
87    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
88    let rows = write(&tx)?;
89    tx.execute(
90        "INSERT INTO sync (source, started_at, finished_at, ok, error, cursor, rows, retry_after)
91         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
92         ON CONFLICT (source) DO UPDATE SET started_at = excluded.started_at,
93           finished_at = excluded.finished_at, ok = excluded.ok, error = excluded.error,
94           cursor = excluded.cursor, rows = excluded.rows, retry_after = excluded.retry_after",
95        (
96            pass.source,
97            pass.started_at,
98            pass.finished_at,
99            pass.failure.is_none(),
100            pass.failure.as_ref().map(Failure::text),
101            pass.cursor,
102            rows as i64,
103            pass.failure.as_ref().and_then(|f| f.retry_after),
104        ),
105    )?;
106    tx.commit()?;
107    Ok(rows)
108}
109
110pub fn due(conn: &Connection, source: &str, every: Duration, now: i64) -> bool {
111    if retry_after(conn, source)
112        .ok()
113        .flatten()
114        .is_some_and(|at| now < at)
115    {
116        return false;
117    }
118    let finished = sync(conn, source)
119        .ok()
120        .flatten()
121        .filter(|s| s.ok != Some(false))
122        .and_then(|s| s.finished_at);
123    let every = every.as_millis() as i64;
124    finished.is_none_or(|at| now - at >= every)
125}
pito_sync/book.rs

pito_sync/
book.rs
