aven_core/

db.rs

use crate::ids::WorkspaceId;
use std::fs;
use std::ops::{Deref, DerefMut};
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result, bail};
use chrono::{NaiveDate, NaiveTime};
use serde_json::Value;
use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteRow};
use sqlx::{Connection as _, Row, Sqlite, SqliteConnection, SqlitePool, Transaction};
use tokio::sync::{Mutex, OwnedMutexGuard};

use crate::choices::{TaskPriority, TaskSource, TaskStatus};
use crate::error::CoreError;
use crate::ids::{new_id, now};
use crate::recurrence::{
    RecurrenceDuePolicy, RecurrenceFrequency, RecurrenceOutcome, RecurrenceProjectionState,
    RecurrenceRule, RecurrenceSeriesState, TimeZoneId, WeekdaySet,
};
use crate::types::{
    MutableEntityType, RecurrenceOccurrence, RecurrencePauseInterval, RecurrenceSeries,
    RecurrenceSeriesLabel, Task,
};
use crate::workspaces::ensure_default_workspace;

static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");
const MIGRATION_BACKUP_KEEP: usize = 20;
const FILE_DATABASE_CONNECTIONS: u32 = 5;

#[derive(Clone)]
pub struct Database {
    pool: SqlitePool,
    writer: Arc<Mutex<()>>,
    path: PathBuf,
}

pub(crate) struct WriterConnection {
    connection: sqlx::pool::PoolConnection<Sqlite>,
    _guard: OwnedMutexGuard<()>,
}

impl Deref for WriterConnection {
    type Target = SqliteConnection;

    fn deref(&self) -> &Self::Target {
        &self.connection
    }
}

impl DerefMut for WriterConnection {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.connection
    }
}

impl Database {
    pub async fn open(path: &Path) -> Result<Self> {
        Ok(Self {
            pool: open_db(path).await?,
            writer: Arc::new(Mutex::new(())),
            path: path.to_path_buf(),
        })
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    pub fn latest_schema_version() -> Option<i64> {
        MIGRATOR.iter().map(|migration| migration.version).max()
    }

    pub async fn meta(&self, key: &str) -> Result<Option<String>> {
        let mut conn = self.acquire_reader().await?;
        get_meta(&mut conn, key).await
    }

    pub async fn conflict_exists(
        &self,
        workspace_id: &WorkspaceId,
        task_id: &crate::ids::TaskId,
        field: &str,
    ) -> Result<bool> {
        let mut conn = self.acquire_reader().await?;
        conflict_exists(&mut conn, workspace_id, task_id, field).await
    }

    pub(crate) async fn acquire_reader(&self) -> Result<sqlx::pool::PoolConnection<Sqlite>> {
        Ok(self.pool.acquire().await?)
    }

    pub(crate) async fn acquire_writer(&self) -> Result<WriterConnection> {
        let guard = self.writer.clone().lock_owned().await;
        let connection = self.pool.acquire().await?;
        Ok(WriterConnection {
            connection,
            _guard: guard,
        })
    }
}

pub(crate) async fn open_db(path: &Path) -> Result<SqlitePool> {
    let connection_input = path.to_string_lossy();
    let mut options = SqliteConnectOptions::from_str(&connection_input)?;
    let storage = database_storage(&connection_input, &options);
    let existed_before_open = storage == DatabaseStorage::File && path.exists();
    if storage == DatabaseStorage::File
        && let Some(parent) = path.parent()
    {
        fs::create_dir_all(parent)
            .with_context(|| format!("could not create {}", parent.display()))?;
    }
    options = options
        .create_if_missing(true)
        .foreign_keys(true)
        .busy_timeout(Duration::from_secs(5));
    let pool_options = match storage {
        DatabaseStorage::InMemory => SqlitePoolOptions::new()
            .min_connections(1)
            .max_connections(1)
            .idle_timeout(None)
            .max_lifetime(None),
        DatabaseStorage::File => {
            options = options.journal_mode(SqliteJournalMode::Wal);
            SqlitePoolOptions::new().max_connections(FILE_DATABASE_CONNECTIONS)
        }
    };
    let pool = pool_options
        .connect_with(options)
        .await
        .with_context(|| format!("could not open {}", path.display()))?;
    backup_before_pending_migrations(path, existed_before_open, &pool).await?;
    MIGRATOR.run(&pool).await?;
    initialize_meta(&pool).await?;
    let mut conn = pool.acquire().await?;
    ensure_default_workspace(&mut conn).await?;
    Ok(pool)
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum DatabaseStorage {
    InMemory,
    File,
}

fn database_storage(connection_input: &str, options: &SqliteConnectOptions) -> DatabaseStorage {
    let connection_input = connection_input
        .trim_start_matches("sqlite://")
        .trim_start_matches("sqlite:");
    let mut database_and_params = connection_input.splitn(2, '?');
    let database = database_and_params.next().unwrap_or_default();
    let uses_memory_mode = database_and_params.next().is_some_and(|params| {
        url::form_urlencoded::parse(params.as_bytes())
            .any(|(key, value)| key == "mode" && value == "memory")
    });
    let filename = options.get_filename();

    if database == ":memory:"
        || database == "file::memory:"
        || filename == Path::new(":memory:")
        || filename == Path::new("file::memory:")
        || uses_memory_mode
    {
        DatabaseStorage::InMemory
    } else {
        DatabaseStorage::File
    }
}

async fn backup_before_pending_migrations(
    path: &Path,
    existed_before_open: bool,
    pool: &SqlitePool,
) -> Result<()> {
    if !migration_backups_enabled() || !existed_before_open || !has_pending_migrations(pool).await?
    {
        return Ok(());
    }
    let backup_path = migration_backup_path(path)?;
    let mut conn = pool.acquire().await?;
    backup_database_with_connection(&mut conn, &backup_path).await?;
    prune_migration_backups(path)?;
    Ok(())
}

fn migration_backups_enabled() -> bool {
    std::env::var_os("AVEN_DEV_MIGRATION_BACKUPS").is_some()
}

async fn has_pending_migrations(pool: &SqlitePool) -> Result<bool> {
    let applied_versions =
        match sqlx::query_scalar::<_, i64>("SELECT version FROM _sqlx_migrations")
            .fetch_all(pool)
            .await
        {
            Ok(versions) => versions,
            Err(error) => {
                let Some(db_error) = error.as_database_error() else {
                    return Err(error.into());
                };
                if db_error.code().as_deref() == Some("1") {
                    return Ok(MIGRATOR.iter().next().is_some());
                }
                return Err(error.into());
            }
        };
    Ok(MIGRATOR
        .iter()
        .any(|migration| !applied_versions.contains(&migration.version)))
}

fn migration_backup_path(path: &Path) -> Result<PathBuf> {
    default_sqlite_backup_path(path, "before-migrate")
}

pub fn default_backup_path(path: &Path, reason: &str) -> Result<PathBuf> {
    backup_path_with_extension(path, reason, "aven-backup.tar.zst")
}

pub fn default_sqlite_backup_path(path: &Path, reason: &str) -> Result<PathBuf> {
    backup_path_with_extension(path, reason, "sqlite")
}

fn backup_path_with_extension(path: &Path, reason: &str, extension: &str) -> Result<PathBuf> {
    let parent = path.parent().unwrap_or_else(|| Path::new("."));
    let backup_dir = parent.join("backups");
    fs::create_dir_all(&backup_dir)
        .with_context(|| format!("could not create {}", backup_dir.display()))?;
    let stem = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or("db.sqlite");
    Ok(backup_dir.join(format!(
        "{stem}.{reason}-{}.{}",
        backup_timestamp()?,
        extension
    )))
}

pub async fn backup_database(source: &Path, backup: &Path) -> Result<()> {
    if !source.is_file() {
        bail!("could not open source {}", source.display());
    }
    if let Some(parent) = backup.parent() {
        fs::create_dir_all(parent)
            .with_context(|| format!("could not create {}", parent.display()))?;
    }
    let mut conn = SqliteConnection::connect_with(
        &SqliteConnectOptions::new()
            .filename(source)
            .read_only(true)
            .foreign_keys(true)
            .busy_timeout(Duration::from_secs(5)),
    )
    .await
    .with_context(|| format!("could not open source {}", source.display()))?;
    backup_database_with_connection(&mut conn, backup).await
}

pub fn wal_path(path: &Path) -> PathBuf {
    PathBuf::from(format!("{}-wal", path.display()))
}

pub fn shm_path(path: &Path) -> PathBuf {
    PathBuf::from(format!("{}-shm", path.display()))
}

pub async fn restore_database_file(target: &Path, source: &Path) -> Result<PathBuf> {
    validate_sqlite_source(source).await?;
    let safety = create_restore_safety_backup(target).await?;
    let staging = target.with_extension("restore-staging");
    if staging.exists() {
        fs::remove_file(&staging)
            .with_context(|| format!("could not remove {}", staging.display()))?;
    }
    fs::copy(source, &staging).with_context(|| {
        format!(
            "could not copy {} -> {}",
            source.display(),
            staging.display()
        )
    })?;
    for sidecar in [wal_path(target), shm_path(target)] {
        if sidecar.exists() {
            fs::remove_file(&sidecar)
                .with_context(|| format!("could not remove {}", sidecar.display()))?;
        }
    }
    fs::rename(&staging, target)
        .with_context(|| format!("could not replace {}", target.display()))?;
    Ok(safety)
}

pub(crate) async fn create_restore_safety_backup(target: &Path) -> Result<PathBuf> {
    let safety = default_sqlite_backup_path(target, "before-restore")?;
    if target.exists() {
        backup_database(target, &safety).await?;
    } else {
        SqliteConnection::connect_with(
            &SqliteConnectOptions::new()
                .filename(&safety)
                .create_if_missing(true),
        )
        .await
        .with_context(|| format!("could not create {}", safety.display()))?
        .close()
        .await?;
    }
    Ok(safety)
}

async fn validate_sqlite_source(source: &Path) -> Result<()> {
    let mut conn = sqlx::SqliteConnection::connect_with(
        &SqliteConnectOptions::new()
            .filename(source)
            .read_only(true)
            .foreign_keys(true),
    )
    .await
    .with_context(|| format!("could not open source {}", source.display()))?;
    let quick_check: String = sqlx::query_scalar("PRAGMA quick_check")
        .fetch_one(&mut conn)
        .await?;
    if quick_check != "ok" {
        bail!("error backup-source-corrupt quick_check={quick_check}");
    }
    Ok(())
}

fn backup_timestamp() -> Result<u64> {
    Ok(SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .context("system clock is before unix epoch")?
        .as_secs())
}

pub(crate) async fn backup_database_with_connection(
    conn: &mut SqliteConnection,
    backup: &Path,
) -> Result<()> {
    let parent = backup.parent().unwrap_or_else(|| Path::new("."));
    fs::create_dir_all(parent).with_context(|| format!("could not create {}", parent.display()))?;
    let staging_dir = tempfile::Builder::new()
        .prefix(".aven-sqlite-backup-")
        .tempdir_in(parent)
        .with_context(|| format!("could not create backup staging in {}", parent.display()))?;
    let staging = staging_dir.path().join("database.sqlite");
    sqlx::query("VACUUM INTO ?")
        .bind(staging.display().to_string())
        .execute(&mut *conn)
        .await
        .with_context(|| format!("could not back up database to {}", backup.display()))?;
    fs::rename(&staging, backup)
        .with_context(|| format!("could not replace {}", backup.display()))?;
    Ok(())
}

fn prune_migration_backups(path: &Path) -> Result<()> {
    let Some(parent) = path.parent() else {
        return Ok(());
    };
    let backup_dir = parent.join("backups");
    let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
        return Ok(());
    };
    let prefix = format!("{file_name}.before-migrate-");
    let mut backups = fs::read_dir(&backup_dir)
        .with_context(|| format!("could not read {}", backup_dir.display()))?
        .filter_map(|entry| entry.ok())
        .filter(|entry| {
            entry
                .file_name()
                .to_str()
                .is_some_and(|name| name.starts_with(&prefix) && name.ends_with(".sqlite"))
        })
        .collect::<Vec<_>>();
    backups.sort_by_key(|entry| entry.file_name());
    let remove_count = backups.len().saturating_sub(MIGRATION_BACKUP_KEEP);
    for entry in backups.into_iter().take(remove_count) {
        let path = entry.path();
        fs::remove_file(&path).with_context(|| format!("could not remove {}", path.display()))?;
    }
    Ok(())
}

async fn initialize_meta(pool: &SqlitePool) -> Result<()> {
    let mut conn = pool.acquire().await?;
    insert_meta_if_missing(&mut conn, "client_id", &new_id()).await?;
    insert_meta_if_missing(&mut conn, "sync_cursor", "0").await?;
    insert_meta_if_missing(&mut conn, "local_seq", "0").await?;
    Ok(())
}

pub(crate) async fn current_schema_version(conn: &mut SqliteConnection) -> Result<i64> {
    let version: Option<i64> = sqlx::query_scalar("SELECT MAX(version) FROM _sqlx_migrations")
        .fetch_one(conn)
        .await?;
    Ok(version.unwrap_or(0))
}

pub(crate) async fn get_meta(conn: &mut SqliteConnection, key: &str) -> Result<Option<String>> {
    Ok(
        sqlx::query_scalar!("SELECT value FROM meta WHERE key = ?", key)
            .fetch_optional(&mut *conn)
            .await?,
    )
}

pub(crate) async fn set_meta(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
    sqlx::query!(
        "INSERT INTO meta(key, value) VALUES (?, ?)
         ON CONFLICT(key) DO UPDATE SET value = excluded.value",
        key,
        value,
    )
    .execute(&mut *conn)
    .await?;
    Ok(())
}

pub(crate) async fn begin_immediate(
    conn: &mut SqliteConnection,
) -> sqlx::Result<Transaction<'_, Sqlite>> {
    conn.begin_with("BEGIN IMMEDIATE").await
}

async fn insert_meta_if_missing(conn: &mut SqliteConnection, key: &str, value: &str) -> Result<()> {
    sqlx::query!(
        "INSERT OR IGNORE INTO meta(key, value) VALUES (?, ?)",
        key,
        value
    )
    .execute(&mut *conn)
    .await?;
    Ok(())
}

async fn next_local_seq(conn: &mut SqliteConnection) -> Result<i64> {
    let seq = get_meta(conn, "local_seq")
        .await?
        .unwrap_or_else(|| "0".to_string())
        .parse::<i64>()?
        + 1;
    set_meta(conn, "local_seq", &seq.to_string()).await?;
    Ok(seq)
}

pub(crate) async fn insert_change(
    conn: &mut SqliteConnection,
    entity_type: &str,
    entity_id: &str,
    field: Option<&str>,
    op_type: &str,
    payload: Value,
    base_version: Option<&str>,
) -> Result<String> {
    let change_id = new_id();
    let client_id = get_meta(conn, "client_id")
        .await?
        .context("missing client id")?;
    let local_seq = next_local_seq(conn).await?;
    let created_at = now();
    let payload = payload.to_string();
    sqlx::query!(
        "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
         op_type, payload, base_version, created_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
        change_id,
        client_id,
        local_seq,
        entity_type,
        entity_id,
        field,
        op_type,
        payload,
        base_version,
        created_at,
    )
    .execute(&mut *conn)
    .await?;
    Ok(change_id)
}

pub(crate) struct IdentifiedChange<'a> {
    pub change_id: &'a str,
    pub entity_type: &'a str,
    pub entity_id: &'a str,
    pub field: Option<&'a str>,
    pub op_type: &'a str,
    pub payload: Value,
    pub base_version: Option<&'a str>,
    pub created_at: &'a str,
}

pub(crate) async fn insert_change_with_identity(
    conn: &mut SqliteConnection,
    change: IdentifiedChange<'_>,
) -> Result<()> {
    let IdentifiedChange {
        change_id,
        entity_type,
        entity_id,
        field,
        op_type,
        payload,
        base_version,
        created_at,
    } = change;
    let payload = payload.to_string();
    let existing = sqlx::query(
        "SELECT entity_type, entity_id, field, op_type, payload, base_version, created_at
         FROM changes WHERE change_id = ?",
    )
    .bind(change_id)
    .fetch_optional(&mut *conn)
    .await?;
    if let Some(existing) = existing {
        let equal = existing.try_get::<String, _>("entity_type")? == entity_type
            && existing.try_get::<String, _>("entity_id")? == entity_id
            && existing.try_get::<Option<String>, _>("field")?.as_deref() == field
            && existing.try_get::<String, _>("op_type")? == op_type
            && existing.try_get::<String, _>("payload")? == payload
            && existing
                .try_get::<Option<String>, _>("base_version")?
                .as_deref()
                == base_version
            && existing.try_get::<String, _>("created_at")? == created_at;
        if equal {
            return Ok(());
        }
        return Err(CoreError::generation_conflict(format!(
            "error recurrence-generation-conflict change_id={change_id}"
        ))
        .into());
    }

    let client_id = get_meta(conn, "client_id")
        .await?
        .context("missing client id")?;
    let local_seq = next_local_seq(conn).await?;
    sqlx::query(
        "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
         op_type, payload, base_version, created_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
    )
    .bind(change_id)
    .bind(client_id)
    .bind(local_seq)
    .bind(entity_type)
    .bind(entity_id)
    .bind(field)
    .bind(op_type)
    .bind(payload)
    .bind(base_version)
    .bind(created_at)
    .execute(&mut *conn)
    .await?;
    Ok(())
}

fn optional_task_date(value: String) -> Option<String> {
    (!value.is_empty()).then_some(value)
}

pub(crate) fn task_from_row(row: &SqliteRow) -> Result<Task> {
    Ok(Task {
        id: row.try_get("id")?,
        workspace_id: row.try_get("workspace_id")?,
        title: row.try_get("title")?,
        description: row.try_get("description")?,
        project_id: row.try_get("project_id")?,
        project_key: row.try_get("project_key")?,
        project_prefix: row.try_get("project_prefix")?,
        status: TaskStatus::parse(row.try_get::<String, _>("status")?.as_str())?,
        priority: TaskPriority::parse(row.try_get::<String, _>("priority")?.as_str())?,
        source: TaskSource::parse(row.try_get::<String, _>("source")?.as_str())?,
        created_at: row.try_get("created_at")?,
        updated_at: row.try_get("updated_at")?,
        queue_activity_at: row.try_get("queue_activity_at")?,
        available_at: optional_task_date(row.try_get("available_at")?),
        due_on: optional_task_date(row.try_get("due_on")?),
        deleted: row.try_get::<i64, _>("deleted")? != 0,
        is_epic: row.try_get::<i64, _>("is_epic")? != 0,
    })
}

pub(crate) fn recurrence_series_from_row(row: &SqliteRow) -> Result<RecurrenceSeries> {
    let frequency = RecurrenceFrequency::parse(&row.try_get::<String, _>("frequency")?)?;
    let interval = u32::try_from(row.try_get::<i64, _>("interval")?)
        .context("recurrence interval must fit u32")?;
    let weekdays = row
        .try_get::<String, _>("weekdays")?
        .parse::<WeekdaySet>()
        .map_err(anyhow::Error::msg)?;
    let rule = RecurrenceRule::new(frequency, interval, weekdays)?;
    let start_on = row
        .try_get::<String, _>("start_on")?
        .parse::<NaiveDate>()
        .context("invalid recurrence start date")?;
    let available_local_time = optional_text(row.try_get("available_local_time")?)
        .map(|value| {
            value
                .parse::<NaiveTime>()
                .context("invalid recurrence availability time")
        })
        .transpose()?;
    let initial_status = TaskStatus::parse(&row.try_get::<String, _>("initial_status")?)?;
    if !initial_status.is_open() {
        bail!("recurrence initial status must be open");
    }
    let state = RecurrenceSeriesState::parse(&row.try_get::<String, _>("state")?)?;
    let stopped_at = optional_text(row.try_get("stopped_at")?);
    if matches!(state, RecurrenceSeriesState::Stopped) != stopped_at.is_some() {
        bail!("recurrence stopped state and stop time must agree");
    }
    let deleted = row.try_get::<i64, _>("deleted")?;
    if !matches!(deleted, 0 | 1) {
        bail!("recurrence deleted value must be zero or one");
    }
    Ok(RecurrenceSeries {
        workspace_id: row.try_get("workspace_id")?,
        id: row.try_get::<String, _>("id")?.parse()?,
        title: row.try_get("title")?,
        description: row.try_get("description")?,
        project_id: row.try_get("project_id")?,
        priority: TaskPriority::parse(&row.try_get::<String, _>("priority")?)?,
        initial_status,
        rule,
        timezone: row
            .try_get::<String, _>("timezone")?
            .parse::<TimeZoneId>()?,
        start_on,
        available_local_time,
        due_policy: RecurrenceDuePolicy::parse(&row.try_get::<String, _>("due_policy")?)?,
        state,
        stopped_at,
        created_at: row.try_get("created_at")?,
        updated_at: row.try_get("updated_at")?,
        deleted: deleted != 0,
    })
}

pub(crate) fn recurrence_series_label_from_row(row: &SqliteRow) -> Result<RecurrenceSeriesLabel> {
    Ok(RecurrenceSeriesLabel {
        workspace_id: row.try_get("workspace_id")?,
        series_id: row.try_get::<String, _>("series_id")?.parse()?,
        label: row.try_get("label")?,
    })
}

pub(crate) fn recurrence_occurrence_from_row(row: &SqliteRow) -> Result<RecurrenceOccurrence> {
    let task_id = optional_text(row.try_get("task_id")?)
        .map(|value| value.parse())
        .transpose()?;
    let outcome = optional_text(row.try_get("outcome")?)
        .map(|value| RecurrenceOutcome::parse(&value))
        .transpose()?;
    let resolved_at = optional_text(row.try_get("resolved_at")?);
    let outcome_change_id = optional_text(row.try_get("outcome_change_id")?);
    let projection_state =
        RecurrenceProjectionState::parse(&row.try_get::<String, _>("projection_state")?)?;
    let archived_at = optional_text(row.try_get("archived_at")?);
    let valid_shape = match projection_state {
        RecurrenceProjectionState::Projected => {
            task_id.is_some()
                && outcome.is_none()
                && resolved_at.is_none()
                && outcome_change_id.is_none()
                && archived_at.is_none()
        }
        RecurrenceProjectionState::Resolved => {
            task_id.is_some()
                && outcome.is_some()
                && resolved_at.is_some()
                && outcome_change_id.is_some()
                && archived_at.is_none()
        }
        RecurrenceProjectionState::Archived => {
            task_id.is_some()
                && outcome.is_none()
                && resolved_at.is_none()
                && outcome_change_id.is_none()
                && archived_at.is_some()
        }
    };
    if !valid_shape {
        bail!("recurrence occurrence fields do not match projection state");
    }
    Ok(RecurrenceOccurrence {
        workspace_id: row.try_get("workspace_id")?,
        series_id: row.try_get::<String, _>("series_id")?.parse()?,
        slot_on: row
            .try_get::<String, _>("slot_on")?
            .parse::<NaiveDate>()
            .context("invalid recurrence slot date")?,
        task_id,
        outcome,
        resolved_at,
        outcome_change_id,
        projection_state,
        archived_at,
    })
}

pub(crate) fn recurrence_pause_interval_from_row(
    row: &SqliteRow,
) -> Result<RecurrencePauseInterval> {
    let suspended_slot_on = optional_text(row.try_get("suspended_slot_on")?)
        .map(|value| {
            value
                .parse::<NaiveDate>()
                .context("invalid suspended recurrence slot date")
        })
        .transpose()?;
    let suspended_task_id = optional_text(row.try_get("suspended_task_id")?)
        .map(|value| value.parse())
        .transpose()?;
    let resumed_at = optional_text(row.try_get("resumed_at")?);
    let resolved_by_change_id = optional_text(row.try_get("resolved_by_change_id")?);
    if resumed_at.is_some() != resolved_by_change_id.is_some() {
        bail!("recurrence pause resume time and change must agree");
    }
    if suspended_slot_on.is_some() != suspended_task_id.is_some() {
        bail!("recurrence suspended slot and task must agree");
    }
    Ok(RecurrencePauseInterval {
        workspace_id: row.try_get("workspace_id")?,
        id: row.try_get("id")?,
        series_id: row.try_get::<String, _>("series_id")?.parse()?,
        paused_at: row.try_get("paused_at")?,
        resumed_at,
        suspended_slot_on,
        suspended_task_id,
        created_by_change_id: row.try_get("created_by_change_id")?,
        resolved_by_change_id,
    })
}

fn optional_text(value: String) -> Option<String> {
    (!value.is_empty()).then_some(value)
}

pub(crate) async fn entity_field_version(
    conn: &mut SqliteConnection,
    workspace_id: &WorkspaceId,
    entity_type: MutableEntityType,
    entity_id: &str,
    field: &str,
) -> Result<Option<String>> {
    Ok(sqlx::query_scalar(
        "SELECT version FROM field_versions
         WHERE workspace_id = ? AND entity_type = ? AND entity_id = ? AND field = ?",
    )
    .bind(workspace_id)
    .bind(entity_type.as_str())
    .bind(entity_id)
    .bind(field)
    .fetch_optional(&mut *conn)
    .await?)
}

pub(crate) async fn set_entity_field_version(
    conn: &mut SqliteConnection,
    workspace_id: &WorkspaceId,
    entity_type: MutableEntityType,
    entity_id: &str,
    field: &str,
    version: &str,
) -> Result<()> {
    sqlx::query(
        "INSERT INTO field_versions(workspace_id, entity_type, entity_id, field, version)
         VALUES (?, ?, ?, ?, ?)
         ON CONFLICT(workspace_id, entity_type, entity_id, field)
         DO UPDATE SET version = excluded.version",
    )
    .bind(workspace_id)
    .bind(entity_type.as_str())
    .bind(entity_id)
    .bind(field)
    .bind(version)
    .execute(&mut *conn)
    .await?;
    Ok(())
}

pub(crate) async fn field_version(
    conn: &mut SqliteConnection,
    entity_id: &str,
    field: &str,
) -> Result<Option<String>> {
    let workspace_id =
        sqlx::query_scalar::<_, WorkspaceId>("SELECT workspace_id FROM tasks WHERE id = ? LIMIT 1")
            .bind(entity_id)
            .fetch_optional(&mut *conn)
            .await?;
    let Some(workspace_id) = workspace_id else {
        return Ok(None);
    };
    entity_field_version(
        conn,
        &workspace_id,
        MutableEntityType::Task,
        entity_id,
        field,
    )
    .await
}

pub(crate) async fn set_field_version(
    conn: &mut SqliteConnection,
    entity_id: &str,
    field: &str,
    version: &str,
) -> Result<()> {
    let workspace_id =
        sqlx::query_scalar::<_, WorkspaceId>("SELECT workspace_id FROM tasks WHERE id = ? LIMIT 1")
            .bind(entity_id)
            .fetch_optional(&mut *conn)
            .await?;
    if let Some(workspace_id) = workspace_id {
        set_entity_field_version(
            conn,
            &workspace_id,
            MutableEntityType::Task,
            entity_id,
            field,
            version,
        )
        .await?;
    }
    Ok(())
}

pub(crate) async fn entity_conflict_exists(
    conn: &mut SqliteConnection,
    workspace_id: &WorkspaceId,
    entity_type: MutableEntityType,
    entity_id: &str,
    field: &str,
) -> Result<bool> {
    Ok(sqlx::query_scalar::<_, i64>(
        "SELECT count(*) FROM conflicts
         WHERE workspace_id = ? AND entity_type = ? AND entity_id = ?
         AND field = ? AND resolved = 0 LIMIT 1",
    )
    .bind(workspace_id)
    .bind(entity_type.as_str())
    .bind(entity_id)
    .bind(field)
    .fetch_one(&mut *conn)
    .await?
        > 0)
}

pub(crate) async fn conflict_exists(
    conn: &mut SqliteConnection,
    workspace_id: &WorkspaceId,
    task_id: &crate::ids::TaskId,
    field: &str,
) -> Result<bool> {
    entity_conflict_exists(
        conn,
        workspace_id,
        MutableEntityType::Task,
        task_id.as_str(),
        field,
    )
    .await
}

#[cfg(test)]
mod tests {
    use super::*;

    const PRIVATE_IN_MEMORY_INPUTS: &[&str] = &[
        ":memory:",
        "sqlite::memory:",
        "sqlite://:memory:",
        "file::memory:",
        "sqlite:file::memory:",
        "sqlite://file::memory:",
        "%3Amemory%3A",
        "sqlite:%3Amemory%3A",
        "sqlite://%3Amemory%3A",
        "file:%3Amemory%3A",
        "sqlite:file:%3Amemory%3A",
        "sqlite://file:%3Amemory%3A",
        "file::memory:?cache=private",
        "sqlite://?mode=memory&cache=private",
        "named?mode=memory&cache=private",
        "sqlite://named?mode=memory&cache=private",
        "sqlite://named?cache=private&mode=memory",
        "sqlite://named?mode=mem%6Fry&cache=private",
    ];

    #[tokio::test]
    async fn wal_readers_progress_while_writes_remain_serialized() {
        let temp = tempfile::tempdir().unwrap();
        let database = Database::open(&temp.path().join("concurrency.sqlite"))
            .await
            .unwrap();
        let mut writer = database.acquire_writer().await.unwrap();
        let mut tx = begin_immediate(&mut writer).await.unwrap();
        set_meta(&mut tx, "local_seq", "1").await.unwrap();

        let reader_database = database.clone();
        let reader = tokio::spawn(async move {
            let mut conn = reader_database.acquire_reader().await.unwrap();
            get_meta(&mut conn, "local_seq").await.unwrap()
        });
        let observed = tokio::time::timeout(Duration::from_secs(1), reader)
            .await
            .expect("reader should not wait for the writer")
            .unwrap();
        assert_eq!(observed.as_deref(), Some("0"));

        let second_writer_database = database.clone();
        let second_writer = tokio::spawn(async move {
            let mut conn = second_writer_database.acquire_writer().await.unwrap();
            let mut tx = begin_immediate(&mut conn).await.unwrap();
            set_meta(&mut tx, "local_seq", "2").await.unwrap();
            tx.commit().await.unwrap();
        });
        tokio::task::yield_now().await;
        assert!(!second_writer.is_finished());

        tx.commit().await.unwrap();
        drop(writer);
        tokio::time::timeout(Duration::from_secs(1), second_writer)
            .await
            .expect("second writer should proceed after the first commits")
            .unwrap();
        assert_eq!(
            database.meta("local_seq").await.unwrap().as_deref(),
            Some("2")
        );
    }

    #[test]
    fn database_storage_follows_sqlx_connection_input_semantics() {
        for &input in PRIVATE_IN_MEMORY_INPUTS {
            let options = SqliteConnectOptions::from_str(input).unwrap();
            assert_eq!(
                database_storage(input, &options),
                DatabaseStorage::InMemory,
                "{input}"
            );
        }

        for input in [
            "mode=memory.sqlite",
            "ordinary-file::memory:.sqlite",
            "sqlite::memory:.sqlite",
            "/tmp/directory-mode=memory/database.sqlite",
        ] {
            let options = SqliteConnectOptions::from_str(input).unwrap();
            assert_eq!(
                database_storage(input, &options),
                DatabaseStorage::File,
                "{input}"
            );
        }
    }

    #[tokio::test]
    async fn private_in_memory_inputs_keep_one_connection_and_one_schema() {
        for &input in PRIVATE_IN_MEMORY_INPUTS {
            let pool = open_db(Path::new(input)).await.unwrap();
            let pool_options = pool.options();
            assert_eq!(pool_options.get_min_connections(), 1, "{input}");
            assert_eq!(pool_options.get_max_connections(), 1, "{input}");
            assert_eq!(pool_options.get_idle_timeout(), None, "{input}");
            assert_eq!(pool_options.get_max_lifetime(), None, "{input}");
            assert_eq!(pool.size(), 1, "{input}");
            let mut first = pool.acquire().await.unwrap();
            let second_pool = pool.clone();
            let mut second = tokio::spawn(async move {
                let mut connection = second_pool.acquire().await.unwrap();
                sqlx::query_scalar::<_, i64>(
                    "SELECT count(*) FROM sqlite_schema WHERE name = 'acquisition_visibility'",
                )
                .fetch_one(&mut *connection)
                .await
                .unwrap()
            });
            assert!(
                tokio::time::timeout(Duration::from_millis(50), &mut second)
                    .await
                    .is_err(),
                "{input} opened another connection"
            );

            sqlx::query("CREATE TABLE acquisition_visibility(id INTEGER PRIMARY KEY)")
                .execute(&mut *first)
                .await
                .unwrap();
            drop(first);

            let visible = tokio::time::timeout(Duration::from_secs(1), second)
                .await
                .unwrap()
                .unwrap();
            assert_eq!(visible, 1, "{input}");
        }
    }

    #[tokio::test]
    async fn filesystem_lookalike_keeps_wal_and_concurrent_connections() {
        let temp = tempfile::tempdir().unwrap();
        let path = temp
            .path()
            .join("ordinary-file::memory:-mode=memory.sqlite");
        let pool = open_db(&path).await.unwrap();
        let first = pool.acquire().await.unwrap();
        let mut second = tokio::time::timeout(Duration::from_secs(1), pool.acquire())
            .await
            .expect("file pool should allow a concurrent acquisition")
            .unwrap();
        let journal_mode: String = sqlx::query_scalar("PRAGMA journal_mode")
            .fetch_one(&mut *second)
            .await
            .unwrap();

        assert_eq!(
            pool.options().get_max_connections(),
            FILE_DATABASE_CONNECTIONS
        );
        assert_eq!(journal_mode, "wal");
        drop(first);
        drop(second);
        pool.close().await;
    }

    #[tokio::test]
    async fn in_process_backup_captures_wal_and_replaces_destination() {
        let temp = tempfile::tempdir().unwrap();
        let source = temp.path().join("source.sqlite");
        let backup = temp.path().join("backup.sqlite");
        let database = Database::open(&source).await.unwrap();
        let mut writer = database.acquire_writer().await.unwrap();
        let mut tx = begin_immediate(&mut writer).await.unwrap();
        set_meta(&mut tx, "backup-test", "first").await.unwrap();
        tx.commit().await.unwrap();
        drop(writer);
        assert!(wal_path(&source).exists());

        fs::write(&backup, b"existing destination").unwrap();
        backup_database(&source, &backup).await.unwrap();
        let mut backup_conn = SqliteConnection::connect_with(
            &SqliteConnectOptions::new()
                .filename(&backup)
                .read_only(true),
        )
        .await
        .unwrap();
        assert_eq!(
            get_meta(&mut backup_conn, "backup-test")
                .await
                .unwrap()
                .as_deref(),
            Some("first")
        );
        drop(backup_conn);

        let mut writer = database.acquire_writer().await.unwrap();
        set_meta(&mut writer, "backup-test", "second")
            .await
            .unwrap();
        drop(writer);
        backup_database(&source, &backup).await.unwrap();
        let mut backup_conn = SqliteConnection::connect_with(
            &SqliteConnectOptions::new()
                .filename(&backup)
                .read_only(true),
        )
        .await
        .unwrap();
        assert_eq!(
            get_meta(&mut backup_conn, "backup-test")
                .await
                .unwrap()
                .as_deref(),
            Some("second")
        );
    }

    #[tokio::test]
    async fn in_process_backup_rejects_missing_source() {
        let temp = tempfile::tempdir().unwrap();
        let backup = temp.path().join("backup.sqlite");
        let error = backup_database(&temp.path().join("missing.sqlite"), &backup)
            .await
            .unwrap_err();
        assert!(error.to_string().contains("could not open source"));
        assert!(!backup.exists());
    }

    #[tokio::test]
    async fn restore_replaces_sidecars_and_preserves_safety_copy() {
        let temp = tempfile::tempdir().unwrap();
        let target = temp.path().join("target.sqlite");
        let source = temp.path().join("source.sqlite");
        let target_database = Database::open(&target).await.unwrap();
        let source_database = Database::open(&source).await.unwrap();
        let mut target_writer = target_database.acquire_writer().await.unwrap();
        set_meta(&mut target_writer, "restore-test", "target")
            .await
            .unwrap();
        drop(target_writer);
        let mut source_writer = source_database.acquire_writer().await.unwrap();
        set_meta(&mut source_writer, "restore-test", "source")
            .await
            .unwrap();
        drop(source_writer);
        target_database.pool.close().await;
        source_database.pool.close().await;
        fs::write(wal_path(&target), b"stale wal").unwrap();
        fs::write(shm_path(&target), b"stale shm").unwrap();

        let safety = restore_database_file(&target, &source).await.unwrap();
        assert!(!wal_path(&target).exists());
        assert!(!shm_path(&target).exists());

        let mut restored = SqliteConnection::connect_with(
            &SqliteConnectOptions::new()
                .filename(&target)
                .read_only(true),
        )
        .await
        .unwrap();
        assert_eq!(
            get_meta(&mut restored, "restore-test")
                .await
                .unwrap()
                .as_deref(),
            Some("source")
        );
        let mut preserved = SqliteConnection::connect_with(
            &SqliteConnectOptions::new()
                .filename(&safety)
                .read_only(true),
        )
        .await
        .unwrap();
        assert_eq!(
            get_meta(&mut preserved, "restore-test")
                .await
                .unwrap()
                .as_deref(),
            Some("target")
        );
    }

    #[tokio::test]
    async fn task_from_row_maps_empty_dates_to_absence() {
        let mut conn = SqliteConnection::connect(":memory:")
            .await
            .expect("open db");
        let row = sqlx::query(
            "SELECT 'TASK000000000001' AS id,
                    '0000000000000000' AS workspace_id,
                    'optional dates' AS title,
                    '' AS description,
                    '0000000000000001' AS project_id,
                    'app' AS project_key,
                    'APP' AS project_prefix,
                    'todo' AS status,
                    'none' AS priority,
                    'unknown' AS source,
                    't' AS created_at,
                    't' AS updated_at,
                    't' AS queue_activity_at,
                    '' AS available_at,
                    '' AS due_on,
                    0 AS deleted,
                    0 AS is_epic",
        )
        .fetch_one(&mut conn)
        .await
        .expect("row");

        let task = task_from_row(&row).unwrap();

        assert_eq!(task.available_at, None);
        assert_eq!(task.due_on, None);
    }

    #[test]
    fn task_date_boundary_preserves_present_values() {
        assert_eq!(
            optional_task_date("2099-01-01T00:00:00Z".to_string()).as_deref(),
            Some("2099-01-01T00:00:00Z")
        );
        assert_eq!(
            optional_task_date("2099-01-01".to_string()).as_deref(),
            Some("2099-01-01")
        );
    }

    #[tokio::test]
    async fn task_from_row_rejects_invalid_status_and_priority() {
        let mut conn = SqliteConnection::connect(":memory:")
            .await
            .expect("open db");
        let row = sqlx::query(
            "SELECT 'TASK000000000001' AS id,
                    '0000000000000000' AS workspace_id,
                    'bad status' AS title,
                    '' AS description,
                    '0000000000000001' AS project_id,
                    'app' AS project_key,
                    'APP' AS project_prefix,
                    'blocked' AS status,
                    'none' AS priority,
                    'unknown' AS source,
                    't' AS created_at,
                    't' AS updated_at,
                    't' AS queue_activity_at,
                    0 AS deleted,
                    0 AS is_epic",
        )
        .fetch_one(&mut conn)
        .await
        .expect("row");
        assert_eq!(
            task_from_row(&row).unwrap_err().to_string(),
            "error invalid-status input=blocked choices=inbox,backlog,todo,active,done,canceled"
        );

        let row = sqlx::query(
            "SELECT 'TASK000000000001' AS id,
                    '0000000000000000' AS workspace_id,
                    'bad priority' AS title,
                    '' AS description,
                    '0000000000000001' AS project_id,
                    'app' AS project_key,
                    'APP' AS project_prefix,
                    'inbox' AS status,
                    'soon' AS priority,
                    't' AS created_at,
                    't' AS updated_at,
                    't' AS queue_activity_at,
                    0 AS deleted,
                    0 AS is_epic",
        )
        .fetch_one(&mut conn)
        .await
        .expect("row");
        assert_eq!(
            task_from_row(&row).unwrap_err().to_string(),
            "error invalid-priority input=soon choices=none,low,medium,high,urgent"
        );
    }

    #[tokio::test]
    async fn recurrence_migration_enforces_schedule_immutability_and_task_conflict_compatibility() {
        let (_temp, mut conn) = crate::test_support::test_conn().await;
        sqlx::query(
            "INSERT INTO recurrence_series(
                workspace_id, id, title, description, project_id, priority, initial_status,
                frequency, interval, weekdays, timezone, start_on, available_local_time,
                due_policy, state, created_at, updated_at
             ) VALUES (
                '0000000000000000', '7KQ9A1X4MV2P8D6R', 'journal', '',
                '7KQ9A1X4MV2P8D6S', 'none', 'todo', 'daily', 1, '', 'UTC',
                '2026-07-20', '09:00:00', 'same_day', 'active', 't', 't'
             )",
        )
        .execute(&mut *conn)
        .await
        .unwrap();

        sqlx::query(
            "UPDATE recurrence_series SET title = 'future journal' WHERE id = '7KQ9A1X4MV2P8D6R'",
        )
        .execute(&mut *conn)
        .await
        .unwrap();
        let error = sqlx::query(
            "UPDATE recurrence_series SET frequency = 'weekly', weekdays = 'mon' WHERE id = '7KQ9A1X4MV2P8D6R'",
        )
        .execute(&mut *conn)
        .await
        .unwrap_err();
        assert!(
            error
                .to_string()
                .contains("recurrence schedule is immutable")
        );

        sqlx::query(
            "INSERT INTO conflicts(task_id, field, local_value, remote_value, remote_change_id, variant_a, variant_b, created_at)
             VALUES ('7KQ9A1X4MV2P8D6T', 'title', 'a', 'b', 'remote', 'a', 'b', 't')",
        )
        .execute(&mut *conn)
        .await
        .unwrap();
        let identity: (String, String) = sqlx::query_as(
            "SELECT entity_type, entity_id FROM conflicts WHERE task_id = '7KQ9A1X4MV2P8D6T'",
        )
        .fetch_one(&mut *conn)
        .await
        .unwrap();
        assert_eq!(
            identity,
            ("task".to_string(), "7KQ9A1X4MV2P8D6T".to_string())
        );
    }

    #[tokio::test]
    async fn recurrence_rows_map_through_validated_domain_types() {
        let mut conn = SqliteConnection::connect(":memory:").await.unwrap();
        let series_row = sqlx::query(
            "SELECT '0000000000000000' AS workspace_id,
                    '7KQ9A1X4MV2P8D6R' AS id, 'journal' AS title, '' AS description,
                    '7KQ9A1X4MV2P8D6S' AS project_id, 'high' AS priority,
                    'todo' AS initial_status, 'weekly' AS frequency, 2 AS interval,
                    'mon,fri' AS weekdays, 'Europe/Stockholm' AS timezone,
                    '2026-07-20' AS start_on, '09:30:00' AS available_local_time,
                    'same_day' AS due_policy, 'active' AS state, '' AS stopped_at,
                    'created' AS created_at, 'updated' AS updated_at, 0 AS deleted",
        )
        .fetch_one(&mut conn)
        .await
        .unwrap();
        let series = recurrence_series_from_row(&series_row).unwrap();
        assert_eq!(series.id.as_str(), "7KQ9A1X4MV2P8D6R");
        assert_eq!(series.rule.interval(), 2);
        assert_eq!(series.available_local_time.unwrap().to_string(), "09:30:00");

        let occurrence_row = sqlx::query(
            "SELECT '0000000000000000' AS workspace_id,
                    '7KQ9A1X4MV2P8D6R' AS series_id, '2026-07-20' AS slot_on,
                    '7KQ9A1X4MV2P8D6T' AS task_id, 'completed' AS outcome,
                    'resolved' AS resolved_at, 'change' AS outcome_change_id,
                    'resolved' AS projection_state, '' AS archived_at",
        )
        .fetch_one(&mut conn)
        .await
        .unwrap();
        let occurrence = recurrence_occurrence_from_row(&occurrence_row).unwrap();
        assert_eq!(
            occurrence.task_id.as_ref().map(|task_id| task_id.as_str()),
            Some("7KQ9A1X4MV2P8D6T")
        );
        assert_eq!(occurrence.outcome, Some(RecurrenceOutcome::Completed));
        assert_eq!(
            occurrence.projection_state,
            RecurrenceProjectionState::Resolved
        );

        let invalid_row = sqlx::query(
            "SELECT '0000000000000000' AS workspace_id,
                    '7KQ9A1X4MV2P8D6R' AS id, 'journal' AS title, '' AS description,
                    '7KQ9A1X4MV2P8D6S' AS project_id, 'none' AS priority,
                    'todo' AS initial_status, 'weekly' AS frequency, 1 AS interval,
                    'fri,mon' AS weekdays, 'UTC' AS timezone, '2026-07-20' AS start_on,
                    '' AS available_local_time, 'none' AS due_policy, 'active' AS state,
                    '' AS stopped_at, 't' AS created_at, 't' AS updated_at, 0 AS deleted",
        )
        .fetch_one(&mut conn)
        .await
        .unwrap();
        assert!(recurrence_series_from_row(&invalid_row).is_err());
    }

    #[tokio::test]
    async fn field_versions_support_task_and_recurrence_series_identity() {
        let (_temp, mut conn) = crate::test_support::test_conn().await;
        sqlx::query(
            "INSERT INTO tasks(id, title, description, project_id, status, priority, created_at, updated_at)
             VALUES ('7KQ9A1X4MV2P8D6T', 'task', '', '7KQ9A1X4MV2P8D6S', 'todo', 'none', 't', 't')",
        )
        .execute(&mut *conn)
        .await
        .unwrap();
        set_field_version(&mut conn, "7KQ9A1X4MV2P8D6T", "title", "task-version")
            .await
            .unwrap();
        set_entity_field_version(
            &mut conn,
            &crate::workspaces::default_workspace_id(),
            MutableEntityType::RecurrenceSeries,
            "7KQ9A1X4MV2P8D6R",
            "title",
            "series-version",
        )
        .await
        .unwrap();

        assert_eq!(
            field_version(&mut conn, "7KQ9A1X4MV2P8D6T", "title")
                .await
                .unwrap()
                .as_deref(),
            Some("task-version")
        );
        assert_eq!(
            entity_field_version(
                &mut conn,
                &crate::workspaces::default_workspace_id(),
                MutableEntityType::RecurrenceSeries,
                "7KQ9A1X4MV2P8D6R",
                "title",
            )
            .await
            .unwrap()
            .as_deref(),
            Some("series-version")
        );
    }
}