aven_core/
db.rs

1use crate::ids::WorkspaceId;
2use std::fs;
3use std::path::{Path, PathBuf};
4use std::process::Command;
5use std::str::FromStr;
6use std::time::{Duration, SystemTime, UNIX_EPOCH};
7
8use anyhow::{Context, Result, bail};
9use chrono::{NaiveDate, NaiveTime};
10use serde_json::Value;
11use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteRow};
12use sqlx::{Connection as _, Row, Sqlite, SqliteConnection, SqlitePool, Transaction};
13
14use crate::choices::{TaskPriority, TaskSource, TaskStatus};
15use crate::ids::{new_id, now};
16use crate::recurrence::{
17    RecurrenceDuePolicy, RecurrenceFrequency, RecurrenceOutcome, RecurrenceProjectionState,
18    RecurrenceRule, RecurrenceSeriesState, TimeZoneId, WeekdaySet,
19};
20use crate::types::{
21    MutableEntityType, RecurrenceOccurrence, RecurrencePauseInterval, RecurrenceSeries,
22    RecurrenceSeriesLabel, Task,
23};
24use crate::workspaces::ensure_default_workspace;
25
26static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
27const MIGRATION_BACKUP_KEEP: usize = 20;
28
29#[derive(Clone)]
30pub struct Database {
31    pool: SqlitePool,
32    path: PathBuf,
33}
34
35impl Database {
36    pub async fn open(path: &Path) -> Result<Self> {
37        Ok(Self {
38            pool: open_db(path).await?,
39            path: path.to_path_buf(),
40        })
41    }
42
43    pub fn path(&self) -> &Path {
44        &self.path
45    }
46
47    pub fn latest_schema_version() -> Option<i64> {
48        MIGRATOR.iter().map(|migration| migration.version).max()
49    }
50
51    pub async fn meta(&self, key: &str) -> Result<Option<String>> {
52        let mut conn = self.acquire().await?;
53        get_meta(&mut conn, key).await
54    }
55
56    pub async fn conflict_exists(
57        &self,
58        workspace_id: &WorkspaceId,
59        task_id: &crate::ids::TaskId,
60        field: &str,
61    ) -> Result<bool> {
62        let mut conn = self.acquire().await?;
63        conflict_exists(&mut conn, workspace_id, task_id, field).await
64    }
65
66    pub(crate) async fn acquire(&self) -> Result<sqlx::pool::PoolConnection<Sqlite>> {
67        Ok(self.pool.acquire().await?)
68    }
69}
70
71pub(crate) async fn open_db(path: &Path) -> Result<SqlitePool> {
72    let existed_before_open = path.exists();
73    if let Some(parent) = path.parent() {
74        fs::create_dir_all(parent)
75            .with_context(|| format!("could not create {}", parent.display()))?;
76    }
77    let options = SqliteConnectOptions::from_str(&path.display().to_string())?
78        .create_if_missing(true)
79        .foreign_keys(true)
80        .journal_mode(SqliteJournalMode::Wal)
81        .busy_timeout(Duration::from_secs(5));
82    let pool = SqlitePoolOptions::new()
83        .max_connections(1)
84        .connect_with(options)
85        .await
86        .with_context(|| format!("could not open {}", path.display()))?;
87    backup_before_pending_migrations(path, existed_before_open, &pool).await?;
88    MIGRATOR.run(&pool).await?;
89    initialize_meta(&pool).await?;
90    let mut conn = pool.acquire().await?;
91    ensure_default_workspace(&mut conn).await?;
92    Ok(pool)
93}
94
95async fn backup_before_pending_migrations(
96    path: &Path,
97    existed_before_open: bool,
98    pool: &SqlitePool,
99) -> Result<()> {
100    if !migration_backups_enabled() || !existed_before_open || !has_pending_migrations(pool).await?
101    {
102        return Ok(());
103    }
104    let backup_path = migration_backup_path(path)?;
105    run_sqlite_backup(path, &backup_path)?;
106    prune_migration_backups(path)?;
107    Ok(())
108}
109
110fn migration_backups_enabled() -> bool {
111    std::env::var_os("AVEN_DEV_MIGRATION_BACKUPS").is_some()
112}
113
114async fn has_pending_migrations(pool: &SqlitePool) -> Result<bool> {
115    let applied_versions =
116        match sqlx::query_scalar::<_, i64>("SELECT version FROM _sqlx_migrations")
117            .fetch_all(pool)
118            .await
119        {
120            Ok(versions) => versions,
121            Err(error) => {
122                let Some(db_error) = error.as_database_error() else {
123                    return Err(error.into());
124                };
125                if db_error.code().as_deref() == Some("1") {
126                    return Ok(MIGRATOR.iter().next().is_some());
127                }
128                return Err(error.into());
129            }
130        };
131    Ok(MIGRATOR
132        .iter()
133        .any(|migration| !applied_versions.contains(&migration.version)))
134}
135
136fn migration_backup_path(path: &Path) -> Result<PathBuf> {
137    default_sqlite_backup_path(path, "before-migrate")
138}
139
140pub fn default_backup_path(path: &Path, reason: &str) -> Result<PathBuf> {
141    backup_path_with_extension(path, reason, "aven-backup.tar.zst")
142}
143
144pub fn default_sqlite_backup_path(path: &Path, reason: &str) -> Result<PathBuf> {
145    backup_path_with_extension(path, reason, "sqlite")
146}
147
148fn backup_path_with_extension(path: &Path, reason: &str, extension: &str) -> Result<PathBuf> {
149    let parent = path.parent().unwrap_or_else(|| Path::new("."));
150    let backup_dir = parent.join("backups");
151    fs::create_dir_all(&backup_dir)
152        .with_context(|| format!("could not create {}", backup_dir.display()))?;
153    let stem = path
154        .file_name()
155        .and_then(|name| name.to_str())
156        .unwrap_or("db.sqlite");
157    Ok(backup_dir.join(format!(
158        "{stem}.{reason}-{}.{}",
159        backup_timestamp()?,
160        extension
161    )))
162}
163
164pub fn backup_database(source: &Path, backup: &Path) -> Result<()> {
165    if let Some(parent) = backup.parent() {
166        fs::create_dir_all(parent)
167            .with_context(|| format!("could not create {}", parent.display()))?;
168    }
169    run_sqlite_backup(source, backup)
170}
171
172pub fn wal_path(path: &Path) -> PathBuf {
173    PathBuf::from(format!("{}-wal", path.display()))
174}
175
176pub fn shm_path(path: &Path) -> PathBuf {
177    PathBuf::from(format!("{}-shm", path.display()))
178}
179
180pub async fn restore_database_file(target: &Path, source: &Path) -> Result<PathBuf> {
181    validate_sqlite_source(source).await?;
182    let safety = default_sqlite_backup_path(target, "before-restore")?;
183    backup_database(target, &safety)?;
184    let staging = target.with_extension("restore-staging");
185    if staging.exists() {
186        fs::remove_file(&staging)
187            .with_context(|| format!("could not remove {}", staging.display()))?;
188    }
189    fs::copy(source, &staging).with_context(|| {
190        format!(
191            "could not copy {} -> {}",
192            source.display(),
193            staging.display()
194        )
195    })?;
196    for sidecar in [wal_path(target), shm_path(target)] {
197        if sidecar.exists() {
198            fs::remove_file(&sidecar)
199                .with_context(|| format!("could not remove {}", sidecar.display()))?;
200        }
201    }
202    fs::rename(&staging, target)
203        .with_context(|| format!("could not replace {}", target.display()))?;
204    Ok(safety)
205}
206
207async fn validate_sqlite_source(source: &Path) -> Result<()> {
208    let mut conn = sqlx::SqliteConnection::connect_with(
209        &SqliteConnectOptions::new()
210            .filename(source)
211            .read_only(true)
212            .foreign_keys(true),
213    )
214    .await
215    .with_context(|| format!("could not open source {}", source.display()))?;
216    let quick_check: String = sqlx::query_scalar("PRAGMA quick_check")
217        .fetch_one(&mut conn)
218        .await?;
219    if quick_check != "ok" {
220        bail!("error backup-source-corrupt quick_check={quick_check}");
221    }
222    Ok(())
223}
224
225fn backup_timestamp() -> Result<u64> {
226    Ok(SystemTime::now()
227        .duration_since(UNIX_EPOCH)
228        .context("system clock is before unix epoch")?
229        .as_secs())
230}
231
232fn run_sqlite_backup(source: &Path, backup: &Path) -> Result<()> {
233    let backup_sql = format!(".backup '{}'", sqlite_single_quoted(backup));
234    let output = Command::new("sqlite3")
235        .arg(source)
236        .arg(backup_sql)
237        .output()
238        .context("could not run sqlite3 for backup")?;
239    if !output.status.success() {
240        let stderr = String::from_utf8_lossy(&output.stderr);
241        bail!(
242            "sqlite3 backup failed status={} stderr={}",
243            output.status,
244            stderr.trim()
245        );
246    }
247    Ok(())
248}
249
250fn sqlite_single_quoted(path: &Path) -> String {
251    path.display().to_string().replace('\'', "''")
252}
253
254fn prune_migration_backups(path: &Path) -> Result<()> {
255    let Some(parent) = path.parent() else {
256        return Ok(());
257    };
258    let backup_dir = parent.join("backups");
259    let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
260        return Ok(());
261    };
262    let prefix = format!("{file_name}.before-migrate-");
263    let mut backups = fs::read_dir(&backup_dir)
264        .with_context(|| format!("could not read {}", backup_dir.display()))?
265        .filter_map(|entry| entry.ok())
266        .filter(|entry| {
267            entry
268                .file_name()
269                .to_str()
270                .is_some_and(|name| name.starts_with(&prefix) && name.ends_with(".sqlite"))
271        })
272        .collect::<Vec<_>>();
273    backups.sort_by_key(|entry| entry.file_name());
274    let remove_count = backups.len().saturating_sub(MIGRATION_BACKUP_KEEP);
275    for entry in backups.into_iter().take(remove_count) {
276        let path = entry.path();
277        fs::remove_file(&path).with_context(|| format!("could not remove {}", path.display()))?;
278    }
279    Ok(())
280}
281
282async fn initialize_meta(pool: &SqlitePool) -> Result<()> {
283    let mut conn = pool.acquire().await?;
284    insert_meta_if_missing(&mut conn, "client_id", &new_id()).await?;
285    insert_meta_if_missing(&mut conn, "sync_cursor", "0").await?;
286    insert_meta_if_missing(&mut conn, "local_seq", "0").await?;
287    Ok(())
288}
289
290pub(crate) async fn current_schema_version(conn: &mut SqliteConnection) -> Result<i64> {
291    let version: Option<i64> = sqlx::query_scalar("SELECT MAX(version) FROM _sqlx_migrations")
292        .fetch_one(conn)
293        .await?;
294    Ok(version.unwrap_or(0))
295}
296
297pub(crate) async fn get_meta(conn: &mut SqliteConnection, key: &str) -> Result<Option<String>> {
298    Ok(
299        sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?", key)
300            .fetch_optional(&mut *conn)
301            .await?,
302    )
303}
304
305pub(crate) async fn set_meta(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
306    sqlx::query!(
307        "INSERT INTO meta(key, value) VALUES (?, ?)
308         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
309        key,
310        value,
311    )
312    .execute(&mut *conn)
313    .await?;
314    Ok(())
315}
316
317pub(crate) async fn begin_immediate(
318    conn: &mut SqliteConnection,
319) -> sqlx::Result<Transaction<'_, Sqlite>> {
320    conn.begin_with("BEGIN IMMEDIATE").await
321}
322
323async fn insert_meta_if_missing(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
324    sqlx::query!(
325        "INSERT OR IGNORE INTO meta(key, value) VALUES (?, ?)",
326        key,
327        value
328    )
329    .execute(&mut *conn)
330    .await?;
331    Ok(())
332}
333
334async fn next_local_seq(conn: &mut SqliteConnection) -> Result<i64> {
335    let seq = get_meta(conn, "local_seq")
336        .await?
337        .unwrap_or_else(|| "0".to_string())
338        .parse::<i64>()?
339        + 1;
340    set_meta(conn, "local_seq", &seq.to_string()).await?;
341    Ok(seq)
342}
343
344pub(crate) async fn insert_change(
345    conn: &mut SqliteConnection,
346    entity_type: &str,
347    entity_id: &str,
348    field: Option<&str>,
349    op_type: &str,
350    payload: Value,
351    base_version: Option<&str>,
352) -> Result<String> {
353    let change_id = new_id();
354    let client_id = get_meta(conn, "client_id")
355        .await?
356        .context("missing client id")?;
357    let local_seq = next_local_seq(conn).await?;
358    let created_at = now();
359    let payload = payload.to_string();
360    sqlx::query!(
361        "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
362         op_type, payload, base_version, created_at)
363         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
364        change_id,
365        client_id,
366        local_seq,
367        entity_type,
368        entity_id,
369        field,
370        op_type,
371        payload,
372        base_version,
373        created_at,
374    )
375    .execute(&mut *conn)
376    .await?;
377    Ok(change_id)
378}
379
380pub(crate) struct IdentifiedChange<'a> {
381    pub change_id: &'a str,
382    pub entity_type: &'a str,
383    pub entity_id: &'a str,
384    pub field: Option<&'a str>,
385    pub op_type: &'a str,
386    pub payload: Value,
387    pub base_version: Option<&'a str>,
388    pub created_at: &'a str,
389}
390
391pub(crate) async fn insert_change_with_identity(
392    conn: &mut SqliteConnection,
393    change: IdentifiedChange<'_>,
394) -> Result<()> {
395    let IdentifiedChange {
396        change_id,
397        entity_type,
398        entity_id,
399        field,
400        op_type,
401        payload,
402        base_version,
403        created_at,
404    } = change;
405    let payload = payload.to_string();
406    let existing = sqlx::query(
407        "SELECT entity_type, entity_id, field, op_type, payload, base_version, created_at
408         FROM changes WHERE change_id = ?",
409    )
410    .bind(change_id)
411    .fetch_optional(&mut *conn)
412    .await?;
413    if let Some(existing) = existing {
414        let equal = existing.try_get::<String, _>("entity_type")? == entity_type
415            && existing.try_get::<String, _>("entity_id")? == entity_id
416            && existing.try_get::<Option<String>, _>("field")?.as_deref() == field
417            && existing.try_get::<String, _>("op_type")? == op_type
418            && existing.try_get::<String, _>("payload")? == payload
419            && existing
420                .try_get::<Option<String>, _>("base_version")?
421                .as_deref()
422                == base_version
423            && existing.try_get::<String, _>("created_at")? == created_at;
424        if equal {
425            return Ok(());
426        }
427        bail!("error recurrence-generation-conflict change_id={change_id}");
428    }
429
430    let client_id = get_meta(conn, "client_id")
431        .await?
432        .context("missing client id")?;
433    let local_seq = next_local_seq(conn).await?;
434    sqlx::query(
435        "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
436         op_type, payload, base_version, created_at)
437         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
438    )
439    .bind(change_id)
440    .bind(client_id)
441    .bind(local_seq)
442    .bind(entity_type)
443    .bind(entity_id)
444    .bind(field)
445    .bind(op_type)
446    .bind(payload)
447    .bind(base_version)
448    .bind(created_at)
449    .execute(&mut *conn)
450    .await?;
451    Ok(())
452}
453
454fn optional_task_date(value: String) -> Option<String> {
455    (!value.is_empty()).then_some(value)
456}
457
458pub(crate) fn task_from_row(row: &SqliteRow) -> Result<Task> {
459    Ok(Task {
460        id: row.try_get("id")?,
461        workspace_id: row.try_get("workspace_id")?,
462        title: row.try_get("title")?,
463        description: row.try_get("description")?,
464        project_id: row.try_get("project_id")?,
465        project_key: row.try_get("project_key")?,
466        project_prefix: row.try_get("project_prefix")?,
467        status: TaskStatus::parse(row.try_get::<String, _>("status")?.as_str())?,
468        priority: TaskPriority::parse(row.try_get::<String, _>("priority")?.as_str())?,
469        source: TaskSource::parse(row.try_get::<String, _>("source")?.as_str())?,
470        created_at: row.try_get("created_at")?,
471        updated_at: row.try_get("updated_at")?,
472        queue_activity_at: row.try_get("queue_activity_at")?,
473        available_at: optional_task_date(row.try_get("available_at")?),
474        due_on: optional_task_date(row.try_get("due_on")?),
475        deleted: row.try_get::<i64, _>("deleted")? != 0,
476        is_epic: row.try_get::<i64, _>("is_epic")? != 0,
477    })
478}
479
480pub(crate) fn recurrence_series_from_row(row: &SqliteRow) -> Result<RecurrenceSeries> {
481    let frequency = RecurrenceFrequency::parse(&row.try_get::<String, _>("frequency")?)?;
482    let interval = u32::try_from(row.try_get::<i64, _>("interval")?)
483        .context("recurrence interval must fit u32")?;
484    let weekdays = row
485        .try_get::<String, _>("weekdays")?
486        .parse::<WeekdaySet>()
487        .map_err(anyhow::Error::msg)?;
488    let rule = RecurrenceRule::new(frequency, interval, weekdays)?;
489    let start_on = row
490        .try_get::<String, _>("start_on")?
491        .parse::<NaiveDate>()
492        .context("invalid recurrence start date")?;
493    let available_local_time = optional_text(row.try_get("available_local_time")?)
494        .map(|value| {
495            value
496                .parse::<NaiveTime>()
497                .context("invalid recurrence availability time")
498        })
499        .transpose()?;
500    let initial_status = TaskStatus::parse(&row.try_get::<String, _>("initial_status")?)?;
501    if !initial_status.is_open() {
502        bail!("recurrence initial status must be open");
503    }
504    let state = RecurrenceSeriesState::parse(&row.try_get::<String, _>("state")?)?;
505    let stopped_at = optional_text(row.try_get("stopped_at")?);
506    if matches!(state, RecurrenceSeriesState::Stopped) != stopped_at.is_some() {
507        bail!("recurrence stopped state and stop time must agree");
508    }
509    let deleted = row.try_get::<i64, _>("deleted")?;
510    if !matches!(deleted, 0 | 1) {
511        bail!("recurrence deleted value must be zero or one");
512    }
513    Ok(RecurrenceSeries {
514        workspace_id: row.try_get("workspace_id")?,
515        id: row.try_get::<String, _>("id")?.parse()?,
516        title: row.try_get("title")?,
517        description: row.try_get("description")?,
518        project_id: row.try_get("project_id")?,
519        priority: TaskPriority::parse(&row.try_get::<String, _>("priority")?)?,
520        initial_status,
521        rule,
522        timezone: row
523            .try_get::<String, _>("timezone")?
524            .parse::<TimeZoneId>()?,
525        start_on,
526        available_local_time,
527        due_policy: RecurrenceDuePolicy::parse(&row.try_get::<String, _>("due_policy")?)?,
528        state,
529        stopped_at,
530        created_at: row.try_get("created_at")?,
531        updated_at: row.try_get("updated_at")?,
532        deleted: deleted != 0,
533    })
534}
535
536pub(crate) fn recurrence_series_label_from_row(row: &SqliteRow) -> Result<RecurrenceSeriesLabel> {
537    Ok(RecurrenceSeriesLabel {
538        workspace_id: row.try_get("workspace_id")?,
539        series_id: row.try_get::<String, _>("series_id")?.parse()?,
540        label: row.try_get("label")?,
541    })
542}
543
544pub(crate) fn recurrence_occurrence_from_row(row: &SqliteRow) -> Result<RecurrenceOccurrence> {
545    let task_id = optional_text(row.try_get("task_id")?)
546        .map(|value| value.parse())
547        .transpose()?;
548    let outcome = optional_text(row.try_get("outcome")?)
549        .map(|value| RecurrenceOutcome::parse(&value))
550        .transpose()?;
551    let resolved_at = optional_text(row.try_get("resolved_at")?);
552    let outcome_change_id = optional_text(row.try_get("outcome_change_id")?);
553    let projection_state =
554        RecurrenceProjectionState::parse(&row.try_get::<String, _>("projection_state")?)?;
555    let archived_at = optional_text(row.try_get("archived_at")?);
556    let valid_shape = match projection_state {
557        RecurrenceProjectionState::Projected => {
558            task_id.is_some()
559                && outcome.is_none()
560                && resolved_at.is_none()
561                && outcome_change_id.is_none()
562                && archived_at.is_none()
563        }
564        RecurrenceProjectionState::Resolved => {
565            task_id.is_some()
566                && outcome.is_some()
567                && resolved_at.is_some()
568                && outcome_change_id.is_some()
569                && archived_at.is_none()
570        }
571        RecurrenceProjectionState::Archived => {
572            task_id.is_some()
573                && outcome.is_none()
574                && resolved_at.is_none()
575                && outcome_change_id.is_none()
576                && archived_at.is_some()
577        }
578    };
579    if !valid_shape {
580        bail!("recurrence occurrence fields do not match projection state");
581    }
582    Ok(RecurrenceOccurrence {
583        workspace_id: row.try_get("workspace_id")?,
584        series_id: row.try_get::<String, _>("series_id")?.parse()?,
585        slot_on: row
586            .try_get::<String, _>("slot_on")?
587            .parse::<NaiveDate>()
588            .context("invalid recurrence slot date")?,
589        task_id,
590        outcome,
591        resolved_at,
592        outcome_change_id,
593        projection_state,
594        archived_at,
595    })
596}
597
598pub(crate) fn recurrence_pause_interval_from_row(
599    row: &SqliteRow,
600) -> Result<RecurrencePauseInterval> {
601    let suspended_slot_on = optional_text(row.try_get("suspended_slot_on")?)
602        .map(|value| {
603            value
604                .parse::<NaiveDate>()
605                .context("invalid suspended recurrence slot date")
606        })
607        .transpose()?;
608    let suspended_task_id = optional_text(row.try_get("suspended_task_id")?)
609        .map(|value| value.parse())
610        .transpose()?;
611    let resumed_at = optional_text(row.try_get("resumed_at")?);
612    let resolved_by_change_id = optional_text(row.try_get("resolved_by_change_id")?);
613    if resumed_at.is_some() != resolved_by_change_id.is_some() {
614        bail!("recurrence pause resume time and change must agree");
615    }
616    if suspended_slot_on.is_some() != suspended_task_id.is_some() {
617        bail!("recurrence suspended slot and task must agree");
618    }
619    Ok(RecurrencePauseInterval {
620        workspace_id: row.try_get("workspace_id")?,
621        id: row.try_get("id")?,
622        series_id: row.try_get::<String, _>("series_id")?.parse()?,
623        paused_at: row.try_get("paused_at")?,
624        resumed_at,
625        suspended_slot_on,
626        suspended_task_id,
627        created_by_change_id: row.try_get("created_by_change_id")?,
628        resolved_by_change_id,
629    })
630}
631
632fn optional_text(value: String) -> Option<String> {
633    (!value.is_empty()).then_some(value)
634}
635
636pub(crate) async fn entity_field_version(
637    conn: &mut SqliteConnection,
638    workspace_id: &WorkspaceId,
639    entity_type: MutableEntityType,
640    entity_id: &str,
641    field: &str,
642) -> Result<Option<String>> {
643    Ok(sqlx::query_scalar(
644        "SELECT version FROM field_versions
645         WHERE workspace_id = ? AND entity_type = ? AND entity_id = ? AND field = ?",
646    )
647    .bind(workspace_id)
648    .bind(entity_type.as_str())
649    .bind(entity_id)
650    .bind(field)
651    .fetch_optional(&mut *conn)
652    .await?)
653}
654
655pub(crate) async fn set_entity_field_version(
656    conn: &mut SqliteConnection,
657    workspace_id: &WorkspaceId,
658    entity_type: MutableEntityType,
659    entity_id: &str,
660    field: &str,
661    version: &str,
662) -> Result<()> {
663    sqlx::query(
664        "INSERT INTO field_versions(workspace_id, entity_type, entity_id, field, version)
665         VALUES (?, ?, ?, ?, ?)
666         ON CONFLICT(workspace_id, entity_type, entity_id, field)
667         DO UPDATE SET version = excluded.version",
668    )
669    .bind(workspace_id)
670    .bind(entity_type.as_str())
671    .bind(entity_id)
672    .bind(field)
673    .bind(version)
674    .execute(&mut *conn)
675    .await?;
676    Ok(())
677}
678
679pub(crate) async fn field_version(
680    conn: &mut SqliteConnection,
681    entity_id: &str,
682    field: &str,
683) -> Result<Option<String>> {
684    let workspace_id =
685        sqlx::query_scalar::<_, WorkspaceId>("SELECT workspace_id FROM tasks WHERE id = ? LIMIT 1")
686            .bind(entity_id)
687            .fetch_optional(&mut *conn)
688            .await?;
689    let Some(workspace_id) = workspace_id else {
690        return Ok(None);
691    };
692    entity_field_version(
693        conn,
694        &workspace_id,
695        MutableEntityType::Task,
696        entity_id,
697        field,
698    )
699    .await
700}
701
702pub(crate) async fn set_field_version(
703    conn: &mut SqliteConnection,
704    entity_id: &str,
705    field: &str,
706    version: &str,
707) -> Result<()> {
708    let workspace_id =
709        sqlx::query_scalar::<_, WorkspaceId>("SELECT workspace_id FROM tasks WHERE id = ? LIMIT 1")
710            .bind(entity_id)
711            .fetch_optional(&mut *conn)
712            .await?;
713    if let Some(workspace_id) = workspace_id {
714        set_entity_field_version(
715            conn,
716            &workspace_id,
717            MutableEntityType::Task,
718            entity_id,
719            field,
720            version,
721        )
722        .await?;
723    }
724    Ok(())
725}
726
727pub(crate) async fn entity_conflict_exists(
728    conn: &mut SqliteConnection,
729    workspace_id: &WorkspaceId,
730    entity_type: MutableEntityType,
731    entity_id: &str,
732    field: &str,
733) -> Result<bool> {
734    Ok(sqlx::query_scalar::<_, i64>(
735        "SELECT count(*) FROM conflicts
736         WHERE workspace_id = ? AND entity_type = ? AND entity_id = ?
737         AND field = ? AND resolved = 0 LIMIT 1",
738    )
739    .bind(workspace_id)
740    .bind(entity_type.as_str())
741    .bind(entity_id)
742    .bind(field)
743    .fetch_one(&mut *conn)
744    .await?
745        > 0)
746}
747
748pub(crate) async fn conflict_exists(
749    conn: &mut SqliteConnection,
750    workspace_id: &WorkspaceId,
751    task_id: &crate::ids::TaskId,
752    field: &str,
753) -> Result<bool> {
754    entity_conflict_exists(
755        conn,
756        workspace_id,
757        MutableEntityType::Task,
758        task_id.as_str(),
759        field,
760    )
761    .await
762}
763
764#[cfg(test)]
765mod tests {
766    use super::*;
767
768    #[tokio::test]
769    async fn task_from_row_maps_empty_dates_to_absence() {
770        let mut conn = SqliteConnection::connect(":memory:")
771            .await
772            .expect("open db");
773        let row = sqlx::query(
774            "SELECT 'TASK000000000001' AS id,
775                    '0000000000000000' AS workspace_id,
776                    'optional dates' AS title,
777                    '' AS description,
778                    '0000000000000001' AS project_id,
779                    'app' AS project_key,
780                    'APP' AS project_prefix,
781                    'todo' AS status,
782                    'none' AS priority,
783                    'unknown' AS source,
784                    't' AS created_at,
785                    't' AS updated_at,
786                    't' AS queue_activity_at,
787                    '' AS available_at,
788                    '' AS due_on,
789                    0 AS deleted,
790                    0 AS is_epic",
791        )
792        .fetch_one(&mut conn)
793        .await
794        .expect("row");
795
796        let task = task_from_row(&row).unwrap();
797
798        assert_eq!(task.available_at, None);
799        assert_eq!(task.due_on, None);
800    }
801
802    #[test]
803    fn task_date_boundary_preserves_present_values() {
804        assert_eq!(
805            optional_task_date("2099-01-01T00:00:00Z".to_string()).as_deref(),
806            Some("2099-01-01T00:00:00Z")
807        );
808        assert_eq!(
809            optional_task_date("2099-01-01".to_string()).as_deref(),
810            Some("2099-01-01")
811        );
812    }
813
814    #[tokio::test]
815    async fn task_from_row_rejects_invalid_status_and_priority() {
816        let mut conn = SqliteConnection::connect(":memory:")
817            .await
818            .expect("open db");
819        let row = sqlx::query(
820            "SELECT 'TASK000000000001' AS id,
821                    '0000000000000000' AS workspace_id,
822                    'bad status' AS title,
823                    '' AS description,
824                    '0000000000000001' AS project_id,
825                    'app' AS project_key,
826                    'APP' AS project_prefix,
827                    'blocked' AS status,
828                    'none' AS priority,
829                    'unknown' AS source,
830                    't' AS created_at,
831                    't' AS updated_at,
832                    't' AS queue_activity_at,
833                    0 AS deleted,
834                    0 AS is_epic",
835        )
836        .fetch_one(&mut conn)
837        .await
838        .expect("row");
839        assert_eq!(
840            task_from_row(&row).unwrap_err().to_string(),
841            "error invalid-status input=blocked choices=inbox,backlog,todo,active,done,canceled"
842        );
843
844        let row = sqlx::query(
845            "SELECT 'TASK000000000001' AS id,
846                    '0000000000000000' AS workspace_id,
847                    'bad priority' AS title,
848                    '' AS description,
849                    '0000000000000001' AS project_id,
850                    'app' AS project_key,
851                    'APP' AS project_prefix,
852                    'inbox' AS status,
853                    'soon' AS priority,
854                    't' AS created_at,
855                    't' AS updated_at,
856                    't' AS queue_activity_at,
857                    0 AS deleted,
858                    0 AS is_epic",
859        )
860        .fetch_one(&mut conn)
861        .await
862        .expect("row");
863        assert_eq!(
864            task_from_row(&row).unwrap_err().to_string(),
865            "error invalid-priority input=soon choices=none,low,medium,high,urgent"
866        );
867    }
868
869    #[tokio::test]
870    async fn recurrence_migration_enforces_schedule_immutability_and_task_conflict_compatibility() {
871        let (_temp, mut conn) = crate::test_support::test_conn().await;
872        sqlx::query(
873            "INSERT INTO recurrence_series(
874                workspace_id, id, title, description, project_id, priority, initial_status,
875                frequency, interval, weekdays, timezone, start_on, available_local_time,
876                due_policy, state, created_at, updated_at
877             ) VALUES (
878                '0000000000000000', '7KQ9A1X4MV2P8D6R', 'journal', '',
879                '7KQ9A1X4MV2P8D6S', 'none', 'todo', 'daily', 1, '', 'UTC',
880                '2026-07-20', '09:00:00', 'same_day', 'active', 't', 't'
881             )",
882        )
883        .execute(&mut *conn)
884        .await
885        .unwrap();
886
887        sqlx::query(
888            "UPDATE recurrence_series SET title = 'future journal' WHERE id = '7KQ9A1X4MV2P8D6R'",
889        )
890        .execute(&mut *conn)
891        .await
892        .unwrap();
893        let error = sqlx::query(
894            "UPDATE recurrence_series SET frequency = 'weekly', weekdays = 'mon' WHERE id = '7KQ9A1X4MV2P8D6R'",
895        )
896        .execute(&mut *conn)
897        .await
898        .unwrap_err();
899        assert!(
900            error
901                .to_string()
902                .contains("recurrence schedule is immutable")
903        );
904
905        sqlx::query(
906            "INSERT INTO conflicts(task_id, field, local_value, remote_value, remote_change_id, variant_a, variant_b, created_at)
907             VALUES ('7KQ9A1X4MV2P8D6T', 'title', 'a', 'b', 'remote', 'a', 'b', 't')",
908        )
909        .execute(&mut *conn)
910        .await
911        .unwrap();
912        let identity: (String, String) = sqlx::query_as(
913            "SELECT entity_type, entity_id FROM conflicts WHERE task_id = '7KQ9A1X4MV2P8D6T'",
914        )
915        .fetch_one(&mut *conn)
916        .await
917        .unwrap();
918        assert_eq!(
919            identity,
920            ("task".to_string(), "7KQ9A1X4MV2P8D6T".to_string())
921        );
922    }
923
924    #[tokio::test]
925    async fn recurrence_rows_map_through_validated_domain_types() {
926        let mut conn = SqliteConnection::connect(":memory:").await.unwrap();
927        let series_row = sqlx::query(
928            "SELECT '0000000000000000' AS workspace_id,
929                    '7KQ9A1X4MV2P8D6R' AS id, 'journal' AS title, '' AS description,
930                    '7KQ9A1X4MV2P8D6S' AS project_id, 'high' AS priority,
931                    'todo' AS initial_status, 'weekly' AS frequency, 2 AS interval,
932                    'mon,fri' AS weekdays, 'Europe/Stockholm' AS timezone,
933                    '2026-07-20' AS start_on, '09:30:00' AS available_local_time,
934                    'same_day' AS due_policy, 'active' AS state, '' AS stopped_at,
935                    'created' AS created_at, 'updated' AS updated_at, 0 AS deleted",
936        )
937        .fetch_one(&mut conn)
938        .await
939        .unwrap();
940        let series = recurrence_series_from_row(&series_row).unwrap();
941        assert_eq!(series.id.as_str(), "7KQ9A1X4MV2P8D6R");
942        assert_eq!(series.rule.interval(), 2);
943        assert_eq!(series.available_local_time.unwrap().to_string(), "09:30:00");
944
945        let occurrence_row = sqlx::query(
946            "SELECT '0000000000000000' AS workspace_id,
947                    '7KQ9A1X4MV2P8D6R' AS series_id, '2026-07-20' AS slot_on,
948                    '7KQ9A1X4MV2P8D6T' AS task_id, 'completed' AS outcome,
949                    'resolved' AS resolved_at, 'change' AS outcome_change_id,
950                    'resolved' AS projection_state, '' AS archived_at",
951        )
952        .fetch_one(&mut conn)
953        .await
954        .unwrap();
955        let occurrence = recurrence_occurrence_from_row(&occurrence_row).unwrap();
956        assert_eq!(
957            occurrence.task_id.as_ref().map(|task_id| task_id.as_str()),
958            Some("7KQ9A1X4MV2P8D6T")
959        );
960        assert_eq!(occurrence.outcome, Some(RecurrenceOutcome::Completed));
961        assert_eq!(
962            occurrence.projection_state,
963            RecurrenceProjectionState::Resolved
964        );
965
966        let invalid_row = sqlx::query(
967            "SELECT '0000000000000000' AS workspace_id,
968                    '7KQ9A1X4MV2P8D6R' AS id, 'journal' AS title, '' AS description,
969                    '7KQ9A1X4MV2P8D6S' AS project_id, 'none' AS priority,
970                    'todo' AS initial_status, 'weekly' AS frequency, 1 AS interval,
971                    'fri,mon' AS weekdays, 'UTC' AS timezone, '2026-07-20' AS start_on,
972                    '' AS available_local_time, 'none' AS due_policy, 'active' AS state,
973                    '' AS stopped_at, 't' AS created_at, 't' AS updated_at, 0 AS deleted",
974        )
975        .fetch_one(&mut conn)
976        .await
977        .unwrap();
978        assert!(recurrence_series_from_row(&invalid_row).is_err());
979    }
980
981    #[tokio::test]
982    async fn field_versions_support_task_and_recurrence_series_identity() {
983        let (_temp, mut conn) = crate::test_support::test_conn().await;
984        sqlx::query(
985            "INSERT INTO tasks(id, title, description, project_id, status, priority, created_at, updated_at)
986             VALUES ('7KQ9A1X4MV2P8D6T', 'task', '', '7KQ9A1X4MV2P8D6S', 'todo', 'none', 't', 't')",
987        )
988        .execute(&mut *conn)
989        .await
990        .unwrap();
991        set_field_version(&mut conn, "7KQ9A1X4MV2P8D6T", "title", "task-version")
992            .await
993            .unwrap();
994        set_entity_field_version(
995            &mut conn,
996            &crate::workspaces::default_workspace_id(),
997            MutableEntityType::RecurrenceSeries,
998            "7KQ9A1X4MV2P8D6R",
999            "title",
1000            "series-version",
1001        )
1002        .await
1003        .unwrap();
1004
1005        assert_eq!(
1006            field_version(&mut conn, "7KQ9A1X4MV2P8D6T", "title")
1007                .await
1008                .unwrap()
1009                .as_deref(),
1010            Some("task-version")
1011        );
1012        assert_eq!(
1013            entity_field_version(
1014                &mut conn,
1015                &crate::workspaces::default_workspace_id(),
1016                MutableEntityType::RecurrenceSeries,
1017                "7KQ9A1X4MV2P8D6R",
1018                "title",
1019            )
1020            .await
1021            .unwrap()
1022            .as_deref(),
1023            Some("series-version")
1024        );
1025    }
1026}
aven_core/db.rs

aven_core/
db.rs
