use chrono::{DateTime, Utc};
use keepsake::{RelationDefinition, RelationId, RelationKey, RelationSpec};
use uuid::Uuid;
use super::{KeepsakeRepository, RelationCache, RelationRow, RepositoryError, RepositoryResult};
impl<C> KeepsakeRepository<C>
where
C: RelationCache,
{
pub async fn upsert_relation(
&self,
relation: &RelationDefinition,
at: DateTime<Utc>,
) -> RepositoryResult<RelationDefinition> {
let expiry_policy = serde_json::to_value(&relation.expiry)?;
let row = sqlx::query_as::<_, RelationRow>(
r"
insert into keepsake_relation_definitions
(id, kind, key, enabled, expiry_policy, created_at, updated_at)
values ($1, $2, $3, $4, $5, $6, $6)
on conflict (kind, key) do update set
enabled = excluded.enabled,
expiry_policy = excluded.expiry_policy,
updated_at = $6
returning id, kind, key, enabled, expiry_policy
",
)
.bind(relation.id)
.bind(relation.key.kind())
.bind(relation.key.name())
.bind(relation.enabled)
.bind(expiry_policy)
.bind(at)
.fetch_one(&self.pool)
.await?;
let relation = row.try_into_relation()?;
self.relation_cache.remove_by_id(relation.id).await;
Ok(relation)
}
pub async fn upsert_relation_spec<Spec>(
&self,
at: DateTime<Utc>,
) -> RepositoryResult<RelationDefinition>
where
Spec: RelationSpec,
{
let relation = RelationDefinition::from_spec::<Spec>(at)?;
let expiry_policy = serde_json::to_value(&relation.expiry)?;
let mut tx = self.pool.begin().await?;
let row = sqlx::query_as::<_, RelationRow>(
r"
insert into keepsake_relation_definitions
(id, kind, key, enabled, expiry_policy, created_at, updated_at)
values ($1, $2, $3, $4, $5, $6, $6)
on conflict (kind, key) do update set
enabled = excluded.enabled,
expiry_policy = excluded.expiry_policy,
updated_at = $6
where keepsake_relation_definitions.id = excluded.id
returning id, kind, key, enabled, expiry_policy
",
)
.bind(relation.id)
.bind(relation.key.kind())
.bind(relation.key.name())
.bind(relation.enabled)
.bind(expiry_policy)
.bind(at)
.fetch_optional(&mut *tx)
.await?;
let Some(row) = row else {
let stored_relation_id = sqlx::query_scalar::<_, Uuid>(
r"
select id
from keepsake_relation_definitions
where kind = $1 and key = $2
",
)
.bind(relation.key.kind())
.bind(relation.key.name())
.fetch_one(&mut *tx)
.await?;
return Err(RepositoryError::RelationSpecIdMismatch {
kind: relation.key.kind().to_owned(),
name: relation.key.name().to_owned(),
expected_relation_id: relation.id,
stored_relation_id,
});
};
tx.commit().await?;
let relation = row.try_into_relation()?;
self.relation_cache.remove_by_id(relation.id).await;
Ok(relation)
}
pub async fn relation_by_id(
&self,
relation_id: RelationId,
) -> RepositoryResult<Option<RelationDefinition>> {
if let Some(relation) = self.relation_cache.get_by_id(relation_id).await {
return Ok(Some(relation));
}
let relation = self.fetch_relation_by_id(relation_id).await?;
if let Some(relation) = &relation {
self.relation_cache.store(relation).await;
}
Ok(relation)
}
pub async fn relation_by_key(
&self,
key: &RelationKey,
) -> RepositoryResult<Option<RelationDefinition>> {
if let Some(relation) = self.relation_cache.get_by_key(key).await {
return Ok(Some(relation));
}
let relation = self.fetch_relation_by_key(key).await?;
if let Some(relation) = &relation {
self.relation_cache.store(relation).await;
}
Ok(relation)
}
pub async fn set_relation_enabled(
&self,
relation_id: RelationId,
enabled: bool,
at: DateTime<Utc>,
) -> RepositoryResult<bool> {
let result = sqlx::query(
r"
update keepsake_relation_definitions
set enabled = $2, updated_at = $3
where id = $1
",
)
.bind(relation_id)
.bind(enabled)
.bind(at)
.execute(&self.pool)
.await?;
let changed = result.rows_affected() == 1;
if changed {
self.relation_cache.remove_by_id(relation_id).await;
}
Ok(changed)
}
async fn fetch_relation_by_id(
&self,
relation_id: RelationId,
) -> RepositoryResult<Option<RelationDefinition>> {
let row = sqlx::query_as::<_, RelationRow>(
r"
select id, kind, key, enabled, expiry_policy
from keepsake_relation_definitions
where id = $1
",
)
.bind(relation_id)
.fetch_optional(&self.pool)
.await?;
row.map(RelationRow::try_into_relation).transpose()
}
async fn fetch_relation_by_key(
&self,
key: &RelationKey,
) -> RepositoryResult<Option<RelationDefinition>> {
let row = sqlx::query_as::<_, RelationRow>(
r"
select id, kind, key, enabled, expiry_policy
from keepsake_relation_definitions
where kind = $1 and key = $2
",
)
.bind(key.kind())
.bind(key.name())
.fetch_optional(&self.pool)
.await?;
row.map(RelationRow::try_into_relation).transpose()
}
}