keepsake-sqlx 0.5.0

SQLx adapter for keepsake lifecycle storage
Documentation
#![allow(missing_docs)]
#![cfg(feature = "sqlite-tests")]

use chrono::{DateTime, Utc};
use keepsake::{
    ActorRef, ApplyKeepsake, CommandContext, ExpiryPolicy, FulfillmentPolicy, RelationDefinition,
    RelationKey, SubjectRef,
};
use keepsake_sqlx::{RepositoryError, SqliteKeepsakeRepository};
use sqlx::{Row, sqlite::SqlitePoolOptions};
use uuid::Uuid;

type TestResult<T> = Result<T, TestError>;

#[derive(Debug, thiserror::Error)]
enum TestError {
    #[error(transparent)]
    Chrono(#[from] chrono::ParseError),
    #[error(transparent)]
    Keepsake(#[from] keepsake::KeepsakeError),
    #[error(transparent)]
    Repository(#[from] RepositoryError),
    #[error(transparent)]
    SerdeJson(#[from] serde_json::Error),
    #[error(transparent)]
    Sqlx(#[from] sqlx::Error),
}

fn ts(value: &str) -> Result<DateTime<Utc>, chrono::ParseError> {
    DateTime::parse_from_rfc3339(value).map(|timestamp| timestamp.with_timezone(&Utc))
}

async fn repo() -> TestResult<(SqliteKeepsakeRepository, sqlx::SqlitePool)> {
    let pool = SqlitePoolOptions::new()
        .max_connections(1)
        .connect("sqlite::memory:")
        .await?;
    let repo = SqliteKeepsakeRepository::new(pool.clone());
    repo.migrate().await?;
    Ok((repo, pool))
}

fn context() -> TestResult<CommandContext> {
    Ok(CommandContext::new(ActorRef::new("test", "worker")?))
}

async fn upsert_relation(
    repo: &SqliteKeepsakeRepository,
    expiry: ExpiryPolicy,
) -> TestResult<RelationDefinition> {
    let relation = RelationDefinition::enabled(
        Uuid::now_v7(),
        RelationKey::new("tag", format!("sqlite-{}", Uuid::now_v7()))?,
        expiry,
    )?;
    Ok(repo
        .upsert_relation(&relation, ts("2026-01-01T00:00:00Z")?)
        .await?)
}

#[tokio::test]
async fn sqlite_migration_initializes_backend_marker() -> TestResult<()> {
    let (_repo, pool) = repo().await?;
    let marker = sqlx::query("select value from keepsake_schema_metadata where key = 'backend'")
        .fetch_one(&pool)
        .await?
        .try_get::<String, _>("value")?;

    assert_eq!(marker, "sqlite");
    Ok(())
}

#[tokio::test]
async fn sqlite_apply_duplicate_and_active_read() -> TestResult<()> {
    let (repo, _pool) = repo().await?;
    let relation = upsert_relation(&repo, ExpiryPolicy::ManualOnly).await?;
    let subject = SubjectRef::new("account", "acct_123")?;
    let command = ApplyKeepsake::new(
        subject.clone(),
        relation.id,
        ts("2026-01-01T00:01:00Z")?,
        context()?,
    );

    let first = repo.apply(&command).await?;
    let second = repo
        .apply(&ApplyKeepsake::new(
            subject.clone(),
            relation.id,
            ts("2026-01-01T00:02:00Z")?,
            context()?,
        ))
        .await?;
    let active = repo.active_relations_for_subject(&subject).await?;

    assert!(!first.duplicate_prevented);
    assert!(second.duplicate_prevented);
    assert_eq!(first.keepsake.id(), second.keepsake.id());
    assert_eq!(active.len(), 1);
    assert_eq!(active[0].relation().id, relation.id);
    Ok(())
}

#[tokio::test]
async fn sqlite_timed_expiry_expires_due_keepsake() -> TestResult<()> {
    let (repo, _pool) = repo().await?;
    let relation = upsert_relation(
        &repo,
        ExpiryPolicy::At {
            timestamp: ts("2026-01-01T00:02:00Z")?,
        },
    )
    .await?;
    let subject = SubjectRef::new("account", "acct_expiring")?;
    let applied = repo
        .apply(&ApplyKeepsake::new(
            subject,
            relation.id,
            ts("2026-01-01T00:01:00Z")?,
            context()?,
        ))
        .await?;

    let expired = repo
        .expire_due_timed(ts("2026-01-01T00:02:00Z")?, 10)
        .await?;
    let keepsake = repo.active_for_subject(applied.keepsake.subject()).await?;

    assert_eq!(expired, 1);
    assert!(keepsake.is_empty());
    Ok(())
}

#[tokio::test]
async fn sqlite_lifecycle_invariants_reject_invalid_rows() -> TestResult<()> {
    let (repo, pool) = repo().await?;
    let relation = upsert_relation(&repo, ExpiryPolicy::ManualOnly).await?;
    let result = sqlx::query(
        r"
        insert into keepsakes
            (id, subject_kind, subject_id, relation_id, state, expiry_policy, applied_at,
             expires_at, fulfilled_at, revoked_at, metadata, created_at, updated_at)
        values (?1, 'account', 'invalid', ?2, 'applied', ?3, ?4, null, null, ?4, '{}', ?4, ?4)
        ",
    )
    .bind(Uuid::now_v7().to_string())
    .bind(relation.id.to_string())
    .bind(serde_json::to_string(&ExpiryPolicy::ManualOnly)?)
    .bind("2026-01-01T00:00:00.000000Z")
    .execute(&pool)
    .await;

    assert!(matches!(result, Err(sqlx::Error::Database(_))));
    Ok(())
}

#[tokio::test]
async fn sqlite_lifecycle_invariants_reject_malformed_policy_rows() -> TestResult<()> {
    let (repo, pool) = repo().await?;
    let relation = upsert_relation(&repo, ExpiryPolicy::ManualOnly).await?;
    let result = sqlx::query(
        r"
        insert into keepsakes
            (id, subject_kind, subject_id, relation_id, state, expiry_policy, applied_at,
             expires_at, fulfilled_at, revoked_at, metadata, created_at, updated_at)
        values (?1, 'account', 'malformed', ?2, 'applied', '{}', ?3, null, null, null, '{}', ?3, ?3)
        ",
    )
    .bind(Uuid::now_v7().to_string())
    .bind(relation.id.to_string())
    .bind("2026-01-01T00:00:00.000000Z")
    .execute(&pool)
    .await;

    assert!(matches!(result, Err(sqlx::Error::Database(_))));
    Ok(())
}

#[tokio::test]
async fn sqlite_projection_invariant_rejects_fractional_expiry_mismatch() -> TestResult<()> {
    let (repo, pool) = repo().await?;
    let relation = upsert_relation(&repo, ExpiryPolicy::ManualOnly).await?;
    let policy = serde_json::json!({
        "type": "at",
        "timestamp": "2026-01-01T00:00:00.123456Z"
    });
    let result = sqlx::query(
        r"
        insert into keepsakes
            (id, subject_kind, subject_id, relation_id, state, expiry_policy, applied_at,
             expires_at, fulfilled_at, revoked_at, metadata, created_at, updated_at)
        values (?1, 'account', 'fractional', ?2, 'applied', ?3, ?4, ?5, null, null, '{}', ?4, ?4)
        ",
    )
    .bind(Uuid::now_v7().to_string())
    .bind(relation.id.to_string())
    .bind(policy.to_string())
    .bind("2026-01-01T00:00:00.000000Z")
    .bind("2026-01-01T00:00:00.654321Z")
    .execute(&pool)
    .await;

    assert!(matches!(result, Err(sqlx::Error::Database(_))));
    Ok(())
}

#[tokio::test]
async fn sqlite_fulfilled_expiry_uses_counter_snapshot() -> TestResult<()> {
    let (repo, _pool) = repo().await?;
    let relation = upsert_relation(
        &repo,
        ExpiryPolicy::WhenFulfilled {
            policy: FulfillmentPolicy::CounterAtLeast {
                key: "steps".to_owned(),
                threshold: 3,
            },
        },
    )
    .await?;
    let subject = SubjectRef::new("account", "acct_steps")?;
    let applied = repo
        .apply(&ApplyKeepsake::new(
            subject,
            relation.id,
            ts("2026-01-01T00:01:00Z")?,
            context()?,
        ))
        .await?;

    assert_eq!(
        repo.expire_due_fulfilled(ts("2026-01-01T00:02:00Z")?, 10)
            .await?,
        0
    );
    repo.upsert_counter_projection(
        applied.keepsake.id(),
        "steps",
        3,
        ts("2026-01-01T00:03:00Z")?,
    )
    .await?;

    assert_eq!(
        repo.expire_due_fulfilled(ts("2026-01-01T00:04:00Z")?, 10)
            .await?,
        1
    );
    Ok(())
}

#[tokio::test]
async fn sqlite_migration_rejects_wrong_backend_marker() -> TestResult<()> {
    let pool = SqlitePoolOptions::new()
        .max_connections(1)
        .connect("sqlite::memory:")
        .await?;
    sqlx::query(
        "create table keepsake_schema_metadata (key text primary key, value text not null)",
    )
    .execute(&pool)
    .await?;
    sqlx::query("insert into keepsake_schema_metadata (key, value) values ('backend', 'postgres')")
        .execute(&pool)
        .await?;

    let repo = SqliteKeepsakeRepository::new(pool);
    let result = repo.migrate().await;

    assert!(matches!(
        result,
        Err(RepositoryError::BackendMismatch {
            expected: "sqlite",
            actual
        }) if actual == "postgres"
    ));
    Ok(())
}