keepsake_sqlx/

repository.rs

//! `SQLx` repository implementation.

use chrono::{DateTime, Utc};
use sqlx::Pool;
use uuid::Uuid;

#[cfg(feature = "migrations")]
use sqlx::migrate::Migrator;

#[cfg(feature = "postgres")]
mod audit;
mod backend;
mod cache;
#[cfg(feature = "postgres")]
mod expiry;
#[cfg(feature = "postgres")]
mod mutation;
#[cfg(feature = "mysql")]
mod mysql;
#[cfg(feature = "postgres")]
mod query;
#[cfg(feature = "postgres")]
mod relation;
#[cfg(feature = "postgres")]
mod rows;
#[cfg(feature = "sqlite")]
mod sqlite;
#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
mod support;
mod timed;
mod types;

pub use backend::KeepsakeSqlxBackend;
#[cfg(feature = "mysql")]
pub use backend::MySqlBackend;
#[cfg(feature = "postgres")]
pub use backend::PostgresBackend;
#[cfg(feature = "sqlite")]
pub use backend::SqliteBackend;
#[cfg(feature = "cache")]
pub use cache::{LocalRelationCache, LocalRelationCacheConfig};
pub use cache::{NoopRelationCache, RelationCache};
pub use keepsake::ActiveRelation;
#[cfg(feature = "postgres")]
pub use timed::TimedKeepsakeRepository;
#[cfg(feature = "mysql")]
pub use timed::TimedMySqlKeepsakeRepository;
#[cfg(feature = "sqlite")]
pub use timed::TimedSqliteKeepsakeRepository;
pub use timed::TimedSqlxKeepsakeRepository;
pub use types::{
    AppliedKeepsake, AuditCursor, AuditEventRecord, AuditOutboxCursor, AuditOutboxRecord,
    FulfilledExpiryCandidate, MembershipCursor, TimedExpiryCandidate,
};

use backend::BackendMarker;
#[cfg(feature = "postgres")]
use rows::{
    ActiveRelationRow, AppliedKeepsakeRow, AppliedKeepsakeWriteRow, AuditEventRow, RelationRow,
};

#[cfg(all(feature = "migrations", feature = "postgres"))]
static POSTGRES_MIGRATOR: Migrator = sqlx::migrate!("./migrations/postgres");

#[cfg(all(feature = "migrations", feature = "sqlite"))]
static SQLITE_MIGRATOR: Migrator = sqlx::migrate!("./migrations/sqlite");

#[cfg(all(feature = "migrations", feature = "mysql"))]
static MYSQL_MIGRATOR: Migrator = sqlx::migrate!("./migrations/mysql");

#[allow(dead_code)]
const MAX_BATCH_LIMIT: i64 = 10_000;

/// Result alias for SQL repository operations.
pub type RepositoryResult<T> = core::result::Result<T, RepositoryError>;

/// SQL repository errors.
#[non_exhaustive]
#[derive(Debug, thiserror::Error)]
pub enum RepositoryError {
    /// `SQLx` returned an error.
    #[error(transparent)]
    Sqlx(#[from] sqlx::Error),

    /// Migration failed.
    #[cfg(feature = "migrations")]
    #[error(transparent)]
    Migration(#[from] sqlx::migrate::MigrateError),

    /// JSON policy could not be encoded or decoded.
    #[error(transparent)]
    Json(#[from] serde_json::Error),

    /// A Keepsake core model could not be built.
    #[error(transparent)]
    Keepsake(#[from] keepsake::KeepsakeError),

    /// Existing schema metadata belongs to a different backend.
    #[error("schema backend mismatch: expected {expected}, found {actual}")]
    BackendMismatch {
        /// Backend expected by this repository.
        expected: &'static str,
        /// Backend found in schema metadata.
        actual: String,
    },

    /// A command tried to mutate a disabled relation.
    #[error("relation {relation_id} is disabled")]
    RelationDisabled {
        /// Disabled relation id.
        relation_id: Uuid,
    },

    /// A typed relation spec conflicts with an existing natural-key row.
    #[error(
        "relation spec {kind}/{name} expected id {expected_relation_id}, but stored relation uses {stored_relation_id}"
    )]
    RelationSpecIdMismatch {
        /// Relation kind.
        kind: String,
        /// Relation name.
        name: String,
        /// Relation id declared by the typed spec.
        expected_relation_id: Uuid,
        /// Existing stored relation id for the same natural key.
        stored_relation_id: Uuid,
    },

    /// A keepsake row referenced a missing relation definition.
    #[error("relation definition {relation_id} was not found")]
    RelationDefinitionMissing {
        /// Missing relation id.
        relation_id: Uuid,
    },

    /// A batch or scan limit was outside the accepted range.
    #[error("limit {limit} is outside the accepted range 1..={max}")]
    InvalidLimit {
        /// Provided limit.
        limit: i64,
        /// Maximum accepted limit.
        max: i64,
    },

    /// A row contained an unknown lifecycle state.
    #[error("unknown lifecycle state {state}")]
    InvalidLifecycleState {
        /// Stored state value.
        state: String,
    },

    /// A stored audit event carried an unknown event type label.
    #[error("unknown audit event type {event_type}")]
    InvalidAuditEventType {
        /// Stored event type label.
        event_type: String,
    },
}

/// `SQLx`-backed keepsake repository.
#[derive(Debug)]
pub struct SqlxKeepsakeRepository<B, C = NoopRelationCache>
where
    B: KeepsakeSqlxBackend,
{
    pool: Pool<B::Database>,
    #[allow(dead_code)]
    relation_cache: C,
    backend: BackendMarker<B>,
}

impl<B, C> Clone for SqlxKeepsakeRepository<B, C>
where
    B: KeepsakeSqlxBackend,
    C: Clone,
{
    fn clone(&self) -> Self {
        Self {
            pool: self.pool.clone(),
            relation_cache: self.relation_cache.clone(),
            backend: self.backend,
        }
    }
}

/// Postgres-backed keepsake repository.
#[cfg(feature = "postgres")]
pub type PostgresKeepsakeRepository<C = NoopRelationCache> =
    SqlxKeepsakeRepository<PostgresBackend, C>;

/// Default Postgres-backed keepsake repository.
#[cfg(feature = "postgres")]
pub type KeepsakeRepository<C = NoopRelationCache> = PostgresKeepsakeRepository<C>;

/// SQLite-backed keepsake repository.
#[cfg(feature = "sqlite")]
pub type SqliteKeepsakeRepository<C = NoopRelationCache> = SqlxKeepsakeRepository<SqliteBackend, C>;

/// MySQL-backed keepsake repository.
#[cfg(feature = "mysql")]
pub type MySqlKeepsakeRepository<C = NoopRelationCache> = SqlxKeepsakeRepository<MySqlBackend, C>;

#[cfg(feature = "postgres")]
impl PostgresKeepsakeRepository<NoopRelationCache> {
    /// Creates a repository from a Postgres pool.
    #[must_use]
    pub const fn new(pool: sqlx::PgPool) -> Self {
        Self {
            pool,
            relation_cache: NoopRelationCache,
            backend: BackendMarker::new(),
        }
    }
}

#[cfg(feature = "sqlite")]
impl SqliteKeepsakeRepository<NoopRelationCache> {
    /// Creates a repository from a `SQLite` pool.
    #[must_use]
    pub const fn new(pool: sqlx::SqlitePool) -> Self {
        Self {
            pool,
            relation_cache: NoopRelationCache,
            backend: BackendMarker::new(),
        }
    }
}

#[cfg(feature = "mysql")]
impl MySqlKeepsakeRepository<NoopRelationCache> {
    /// Creates a repository from a `MySQL` pool.
    #[must_use]
    pub const fn new(pool: sqlx::MySqlPool) -> Self {
        Self {
            pool,
            relation_cache: NoopRelationCache,
            backend: BackendMarker::new(),
        }
    }
}

impl<B, C> SqlxKeepsakeRepository<B, C>
where
    B: KeepsakeSqlxBackend,
    C: RelationCache,
{
    /// Creates a timestamp-scoped repository view.
    ///
    /// Use this at request or job boundaries to keep one explicit clock read while
    /// avoiding repeated timestamp plumbing through related repository calls.
    pub const fn at(&self, at: DateTime<Utc>) -> TimedSqlxKeepsakeRepository<'_, B, C> {
        TimedSqlxKeepsakeRepository {
            repository: self,
            at,
        }
    }

    /// Enables relation definition caching for read helper methods.
    #[must_use]
    pub fn with_relation_cache<Next>(self, cache: Next) -> SqlxKeepsakeRepository<B, Next>
    where
        Next: RelationCache,
    {
        SqlxKeepsakeRepository {
            pool: self.pool,
            relation_cache: cache,
            backend: self.backend,
        }
    }

    /// Enables local in-process relation definition caching for read helper methods.
    ///
    /// This cache is per-process and has no cross-pod invalidation. Keep the
    /// default [`NoopRelationCache`] when relation definitions change frequently
    /// or when a multi-pod deployment needs invalidation guarantees.
    #[cfg(feature = "cache")]
    #[must_use]
    pub fn with_local_relation_cache(
        self,
        config: LocalRelationCacheConfig,
    ) -> SqlxKeepsakeRepository<B, LocalRelationCache> {
        self.with_relation_cache(LocalRelationCache::new(config))
    }
}

#[cfg(all(feature = "postgres", feature = "migrations"))]
impl<C> PostgresKeepsakeRepository<C>
where
    C: RelationCache,
{
    /// Runs embedded migrations.
    pub async fn migrate(&self) -> RepositoryResult<()> {
        schema::postgres_schema_preflight(&self.pool).await?;
        POSTGRES_MIGRATOR.run(&self.pool).await?;
        Ok(())
    }
}

#[cfg(all(feature = "sqlite", feature = "migrations"))]
impl<C> SqliteKeepsakeRepository<C>
where
    C: RelationCache,
{
    /// Runs embedded `SQLite` migrations.
    pub async fn migrate(&self) -> RepositoryResult<()> {
        schema::sqlite_schema_preflight(&self.pool).await?;
        SQLITE_MIGRATOR.run(&self.pool).await?;
        Ok(())
    }
}

#[cfg(all(feature = "mysql", feature = "migrations"))]
impl<C> MySqlKeepsakeRepository<C>
where
    C: RelationCache,
{
    /// Runs embedded `MySQL` migrations.
    pub async fn migrate(&self) -> RepositoryResult<()> {
        schema::mysql_schema_preflight(&self.pool).await?;
        MYSQL_MIGRATOR.run(&self.pool).await?;
        Ok(())
    }
}

#[allow(dead_code)]
fn validate_limit(limit: i64) -> RepositoryResult<i64> {
    if (1..=MAX_BATCH_LIMIT).contains(&limit) {
        Ok(limit)
    } else {
        Err(RepositoryError::InvalidLimit {
            limit,
            max: MAX_BATCH_LIMIT,
        })
    }
}

#[cfg(feature = "migrations")]
mod schema;
#[cfg(all(test, feature = "postgres"))]
mod tests;