Archify Sync server, Dockerfile and Helm chart
Rust server (axum, yrs, sqlx/PostgreSQL): live shared models over the y-websocket protocol, versions, browser sign-in with PKCE, invites, single sign-on through OIDC, metrics. Docker image on distroless, compose file, and a Helm chart with one pod, optional bundled PostgreSQL and secrets from existing Secrets.
This commit is contained in:
Oleg Lukasonok committed
commit
99778d7489
67 files changed
+8362
No files matched your search
@@ -0,0 +1,726 @@
|
||||
//! PostgreSQL, the server's only state: accounts and sessions, spaces and
|
||||
//! their members, the Yjs update log and the versions.
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use sqlx::postgres::{PgPool, PgPoolOptions};
|
||||
use sqlx::{FromRow, Postgres, Transaction};
|
||||
use uuid::Uuid;
|
||||
|
||||
pub use sqlx::Error as SqlError;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum StoreError {
|
||||
#[error(transparent)]
|
||||
Sql(#[from] sqlx::Error),
|
||||
#[error(transparent)]
|
||||
Migrate(#[from] sqlx::migrate::MigrateError),
|
||||
#[error("{0} already exists")]
|
||||
Conflict(&'static str),
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, StoreError>;
|
||||
|
||||
#[derive(Debug, Clone, FromRow)]
|
||||
pub struct AccountRow {
|
||||
pub id: Uuid,
|
||||
pub organisation_id: Uuid,
|
||||
pub name: String,
|
||||
pub display_name: String,
|
||||
pub role: String,
|
||||
pub password_hash: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, FromRow)]
|
||||
pub struct SpaceRow {
|
||||
pub id: Uuid,
|
||||
pub organisation_id: Uuid,
|
||||
pub name: String,
|
||||
pub created_at: DateTime<Utc>,
|
||||
/// The asking account's role in the space.
|
||||
pub role: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, FromRow)]
|
||||
pub struct MemberRow {
|
||||
pub account_id: Uuid,
|
||||
pub name: String,
|
||||
pub display_name: String,
|
||||
pub role: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, FromRow)]
|
||||
pub struct VersionRow {
|
||||
pub id: i64,
|
||||
pub major: i32,
|
||||
pub minor: i32,
|
||||
pub kind: String,
|
||||
pub name: Option<String>,
|
||||
pub up_to: i64,
|
||||
pub restored_from: Option<i64>,
|
||||
pub created_by: Option<Uuid>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
/// What a space's document is made of: the latest version's state and the updates after it.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct StoredDoc {
|
||||
pub state: Option<Vec<u8>>,
|
||||
pub updates: Vec<Vec<u8>>,
|
||||
/// Id of the last update in the log, or the latest version's `up_to` when the log is empty.
|
||||
pub last_update: i64,
|
||||
pub latest: Option<VersionRow>,
|
||||
}
|
||||
|
||||
/// family, account, expires_at, used_at
|
||||
type RefreshRow = (Uuid, Uuid, DateTime<Utc>, bool);
|
||||
|
||||
pub enum RefreshOutcome {
|
||||
Valid {
|
||||
account: AccountRow,
|
||||
family: Uuid,
|
||||
},
|
||||
/// Unknown or expired.
|
||||
Invalid,
|
||||
/// Used before: someone holds a copy. The whole family is revoked.
|
||||
Reused,
|
||||
}
|
||||
|
||||
pub enum InviteOutcome {
|
||||
Joined { space_id: Uuid },
|
||||
Invalid,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Store {
|
||||
pool: PgPool,
|
||||
}
|
||||
|
||||
// A macro, not a const: sqlx takes only static SQL, which `concat!` keeps.
|
||||
macro_rules! version_columns {
|
||||
() => {
|
||||
"id, major, minor, kind, name, up_to, restored_from, created_by, created_at"
|
||||
};
|
||||
}
|
||||
|
||||
impl Store {
|
||||
pub async fn connect(url: &str, max_connections: u32) -> Result<Self> {
|
||||
let pool = PgPoolOptions::new()
|
||||
.max_connections(max_connections)
|
||||
.connect(url)
|
||||
.await?;
|
||||
Ok(Self { pool })
|
||||
}
|
||||
|
||||
pub async fn migrate(&self) -> Result<()> {
|
||||
sqlx::migrate!("./migrations").run(&self.pool).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn ping(&self) -> Result<()> {
|
||||
sqlx::query("SELECT 1").execute(&self.pool).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ settings
|
||||
|
||||
pub async fn setting(&self, key: &str) -> Result<Option<serde_json::Value>> {
|
||||
Ok(
|
||||
sqlx::query_scalar("SELECT value FROM settings WHERE key = $1")
|
||||
.bind(key)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?,
|
||||
)
|
||||
}
|
||||
|
||||
/// Stores the value unless the key exists; returns the value that is stored.
|
||||
pub async fn setting_or_insert(
|
||||
&self,
|
||||
key: &str,
|
||||
value: serde_json::Value,
|
||||
) -> Result<serde_json::Value> {
|
||||
sqlx::query(
|
||||
"INSERT INTO settings (key, value) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING",
|
||||
)
|
||||
.bind(key)
|
||||
.bind(&value)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(self.setting(key).await?.unwrap_or(value))
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ accounts
|
||||
|
||||
pub async fn account_count(&self) -> Result<i64> {
|
||||
Ok(sqlx::query_scalar("SELECT count(*) FROM accounts")
|
||||
.fetch_one(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// The first organisation and its admin, only while there are no accounts.
|
||||
pub async fn create_first_admin(
|
||||
&self,
|
||||
organisation: &str,
|
||||
name: &str,
|
||||
display_name: &str,
|
||||
password_hash: &str,
|
||||
) -> Result<Option<AccountRow>> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
// Serialises concurrent setup attempts.
|
||||
sqlx::query("LOCK TABLE accounts IN EXCLUSIVE MODE")
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
let count: i64 = sqlx::query_scalar("SELECT count(*) FROM accounts")
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
if count > 0 {
|
||||
return Ok(None);
|
||||
}
|
||||
let org = Uuid::now_v7();
|
||||
sqlx::query("INSERT INTO organisations (id, name) VALUES ($1, $2)")
|
||||
.bind(org)
|
||||
.bind(organisation)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
let account = insert_account(
|
||||
&mut tx,
|
||||
org,
|
||||
name,
|
||||
display_name,
|
||||
"admin",
|
||||
Some(password_hash),
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(Some(account))
|
||||
}
|
||||
|
||||
pub async fn create_account(
|
||||
&self,
|
||||
organisation_id: Uuid,
|
||||
name: &str,
|
||||
display_name: &str,
|
||||
password_hash: &str,
|
||||
) -> Result<AccountRow> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let account = insert_account(
|
||||
&mut tx,
|
||||
organisation_id,
|
||||
name,
|
||||
display_name,
|
||||
"member",
|
||||
Some(password_hash),
|
||||
)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(account)
|
||||
}
|
||||
|
||||
pub async fn account_by_name(&self, name: &str) -> Result<Option<AccountRow>> {
|
||||
Ok(sqlx::query_as(
|
||||
"SELECT id, organisation_id, name, display_name, role, password_hash FROM accounts WHERE lower(name) = lower($1)",
|
||||
)
|
||||
.bind(name)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
pub async fn account(&self, id: Uuid) -> Result<Option<AccountRow>> {
|
||||
Ok(sqlx::query_as(
|
||||
"SELECT id, organisation_id, name, display_name, role, password_hash FROM accounts WHERE id = $1",
|
||||
)
|
||||
.bind(id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
// ------------------------------------------------------ SSO identities
|
||||
|
||||
/// The account an identity provider's person signs in as, if any.
|
||||
pub async fn account_by_identity(
|
||||
&self,
|
||||
issuer: &str,
|
||||
subject: &str,
|
||||
) -> Result<Option<AccountRow>> {
|
||||
Ok(sqlx::query_as(
|
||||
"UPDATE identities i SET last_seen_at = now() FROM accounts a
|
||||
WHERE i.issuer = $1 AND i.subject = $2 AND a.id = i.account_id
|
||||
RETURNING a.id, a.organisation_id, a.name, a.display_name, a.role, a.password_hash",
|
||||
)
|
||||
.bind(issuer)
|
||||
.bind(subject)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// A new account without a password, signed in through an identity
|
||||
/// provider. Takes `name`, or `name-2`, `name-3`, … when it is taken: an
|
||||
/// SSO person never takes over an existing account by its name.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn create_sso_account(
|
||||
&self,
|
||||
organisation_id: Uuid,
|
||||
name: &str,
|
||||
display_name: &str,
|
||||
role: &str,
|
||||
issuer: &str,
|
||||
subject: &str,
|
||||
email: Option<&str>,
|
||||
) -> Result<AccountRow> {
|
||||
for n in 1..=20 {
|
||||
let candidate = if n == 1 {
|
||||
name.to_owned()
|
||||
} else {
|
||||
format!("{name}-{n}")
|
||||
};
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let account = match insert_account(
|
||||
&mut tx,
|
||||
organisation_id,
|
||||
&candidate,
|
||||
display_name,
|
||||
role,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(account) => account,
|
||||
Err(StoreError::Conflict(_)) => continue,
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
sqlx::query("INSERT INTO identities (issuer, subject, account_id, email) VALUES ($1, $2, $3, $4)")
|
||||
.bind(issuer)
|
||||
.bind(subject)
|
||||
.bind(account.id)
|
||||
.bind(email)
|
||||
.execute(&mut *tx)
|
||||
.await
|
||||
.map_err(|e| match &e {
|
||||
sqlx::Error::Database(d) if d.is_unique_violation() => StoreError::Conflict("identity"),
|
||||
_ => StoreError::Sql(e),
|
||||
})?;
|
||||
tx.commit().await?;
|
||||
return Ok(account);
|
||||
}
|
||||
Err(StoreError::Conflict("account"))
|
||||
}
|
||||
|
||||
pub async fn set_role(&self, account: Uuid, role: &str) -> Result<()> {
|
||||
sqlx::query("UPDATE accounts SET role = $2 WHERE id = $1")
|
||||
.bind(account)
|
||||
.bind(role)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The server's organisation: the one made at setup.
|
||||
pub async fn first_organisation(&self) -> Result<Option<Uuid>> {
|
||||
Ok(
|
||||
sqlx::query_scalar("SELECT id FROM organisations ORDER BY created_at LIMIT 1")
|
||||
.fetch_optional(&self.pool)
|
||||
.await?,
|
||||
)
|
||||
}
|
||||
|
||||
// ------------------------------------------------------ refresh tokens
|
||||
|
||||
pub async fn insert_refresh(
|
||||
&self,
|
||||
hash: &[u8],
|
||||
family: Uuid,
|
||||
account: Uuid,
|
||||
expires_at: DateTime<Utc>,
|
||||
) -> Result<()> {
|
||||
sqlx::query("INSERT INTO refresh_tokens (token_hash, family, account_id, expires_at) VALUES ($1, $2, $3, $4)")
|
||||
.bind(hash)
|
||||
.bind(family)
|
||||
.bind(account)
|
||||
.bind(expires_at)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Marks the token used. A token used again later than `grace_secs` after its
|
||||
/// first use revokes every token of its family; within the grace, two windows
|
||||
/// of the same editor that refreshed at the same moment both get through.
|
||||
pub async fn use_refresh(&self, hash: &[u8], grace_secs: u32) -> Result<RefreshOutcome> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let row: Option<RefreshRow> = sqlx::query_as(
|
||||
"SELECT family, account_id, expires_at,
|
||||
used_at IS NOT NULL AND used_at < now() - make_interval(secs => $2)
|
||||
FROM refresh_tokens WHERE token_hash = $1 FOR UPDATE",
|
||||
)
|
||||
.bind(hash)
|
||||
.bind(f64::from(grace_secs))
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some((family, account_id, expires_at, reused)) = row else {
|
||||
return Ok(RefreshOutcome::Invalid);
|
||||
};
|
||||
if reused {
|
||||
sqlx::query("DELETE FROM refresh_tokens WHERE family = $1")
|
||||
.bind(family)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
return Ok(RefreshOutcome::Reused);
|
||||
}
|
||||
if expires_at < Utc::now() {
|
||||
return Ok(RefreshOutcome::Invalid);
|
||||
}
|
||||
sqlx::query(
|
||||
"UPDATE refresh_tokens SET used_at = coalesce(used_at, now()) WHERE token_hash = $1",
|
||||
)
|
||||
.bind(hash)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
let account: Option<AccountRow> = sqlx::query_as(
|
||||
"SELECT id, organisation_id, name, display_name, role, password_hash FROM accounts WHERE id = $1",
|
||||
)
|
||||
.bind(account_id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(match account {
|
||||
Some(account) => RefreshOutcome::Valid { account, family },
|
||||
None => RefreshOutcome::Invalid,
|
||||
})
|
||||
}
|
||||
|
||||
/// Signs out the session the token belongs to.
|
||||
pub async fn revoke_refresh(&self, hash: &[u8]) -> Result<()> {
|
||||
sqlx::query("DELETE FROM refresh_tokens WHERE family = (SELECT family FROM refresh_tokens WHERE token_hash = $1)")
|
||||
.bind(hash)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn delete_expired_refresh(&self) -> Result<u64> {
|
||||
// Used tokens are kept until expiry so a reuse is still recognised.
|
||||
let done = sqlx::query("DELETE FROM refresh_tokens WHERE expires_at < now()")
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(done.rows_affected())
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------- spaces
|
||||
|
||||
pub async fn create_space(
|
||||
&self,
|
||||
organisation_id: Uuid,
|
||||
name: &str,
|
||||
owner: Uuid,
|
||||
) -> Result<SpaceRow> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let id = Uuid::now_v7();
|
||||
let created_at: DateTime<Utc> = sqlx::query_scalar(
|
||||
"INSERT INTO spaces (id, organisation_id, name, created_by) VALUES ($1, $2, $3, $4) RETURNING created_at",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(organisation_id)
|
||||
.bind(name)
|
||||
.bind(owner)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
sqlx::query("INSERT INTO members (space_id, account_id, role) VALUES ($1, $2, 'owner')")
|
||||
.bind(id)
|
||||
.bind(owner)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(SpaceRow {
|
||||
id,
|
||||
organisation_id,
|
||||
name: name.to_owned(),
|
||||
created_at,
|
||||
role: "owner".into(),
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn spaces_of(&self, account: Uuid) -> Result<Vec<SpaceRow>> {
|
||||
Ok(sqlx::query_as(
|
||||
"SELECT s.id, s.organisation_id, s.name, s.created_at, m.role
|
||||
FROM spaces s JOIN members m ON m.space_id = s.id
|
||||
WHERE m.account_id = $1 ORDER BY s.name",
|
||||
)
|
||||
.bind(account)
|
||||
.fetch_all(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// The space as the account sees it; None when it is not a member.
|
||||
pub async fn space_for(&self, space: Uuid, account: Uuid) -> Result<Option<SpaceRow>> {
|
||||
Ok(sqlx::query_as(
|
||||
"SELECT s.id, s.organisation_id, s.name, s.created_at, m.role
|
||||
FROM spaces s JOIN members m ON m.space_id = s.id
|
||||
WHERE s.id = $1 AND m.account_id = $2",
|
||||
)
|
||||
.bind(space)
|
||||
.bind(account)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
pub async fn members(&self, space: Uuid) -> Result<Vec<MemberRow>> {
|
||||
Ok(sqlx::query_as(
|
||||
"SELECT a.id AS account_id, a.name, a.display_name, m.role
|
||||
FROM members m JOIN accounts a ON a.id = m.account_id
|
||||
WHERE m.space_id = $1 ORDER BY a.display_name",
|
||||
)
|
||||
.bind(space)
|
||||
.fetch_all(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
pub async fn remove_member(&self, space: Uuid, account: Uuid) -> Result<bool> {
|
||||
let done = sqlx::query(
|
||||
"DELETE FROM members WHERE space_id = $1 AND account_id = $2 AND role <> 'owner'",
|
||||
)
|
||||
.bind(space)
|
||||
.bind(account)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(done.rows_affected() > 0)
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------- invites
|
||||
|
||||
pub async fn create_invite(
|
||||
&self,
|
||||
hash: &[u8],
|
||||
space: Uuid,
|
||||
created_by: Uuid,
|
||||
expires_at: DateTime<Utc>,
|
||||
max_uses: Option<i32>,
|
||||
) -> Result<()> {
|
||||
sqlx::query("INSERT INTO invites (token_hash, space_id, created_by, expires_at, max_uses) VALUES ($1, $2, $3, $4, $5)")
|
||||
.bind(hash)
|
||||
.bind(space)
|
||||
.bind(created_by)
|
||||
.bind(expires_at)
|
||||
.bind(max_uses)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The organisation an invite belongs to, if it is still usable; for the join page.
|
||||
pub async fn invite_organisation(&self, hash: &[u8]) -> Result<Option<(Uuid, String)>> {
|
||||
Ok(sqlx::query_as(
|
||||
"SELECT s.organisation_id, s.name FROM invites i JOIN spaces s ON s.id = i.space_id
|
||||
WHERE i.token_hash = $1 AND i.expires_at > now() AND (i.max_uses IS NULL OR i.uses < i.max_uses)",
|
||||
)
|
||||
.bind(hash)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// Adds the account to the invite's space as an editor. Accounts of other
|
||||
/// organisations cannot join.
|
||||
pub async fn accept_invite(&self, hash: &[u8], account: &AccountRow) -> Result<InviteOutcome> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
// (space, organisation, still usable, already a member)
|
||||
let row: Option<(Uuid, Uuid, bool, bool)> = sqlx::query_as(
|
||||
"SELECT i.space_id, s.organisation_id,
|
||||
i.expires_at > now() AND (i.max_uses IS NULL OR i.uses < i.max_uses),
|
||||
EXISTS (SELECT 1 FROM members m WHERE m.space_id = i.space_id AND m.account_id = $2)
|
||||
FROM invites i JOIN spaces s ON s.id = i.space_id
|
||||
WHERE i.token_hash = $1
|
||||
FOR UPDATE OF i",
|
||||
)
|
||||
.bind(hash)
|
||||
.bind(account.id)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
let Some((space_id, organisation_id, usable, member)) = row else {
|
||||
return Ok(InviteOutcome::Invalid);
|
||||
};
|
||||
// Signing up through the invite already joined: accepting again only says where.
|
||||
if member {
|
||||
return Ok(InviteOutcome::Joined { space_id });
|
||||
}
|
||||
if !usable || organisation_id != account.organisation_id {
|
||||
return Ok(InviteOutcome::Invalid);
|
||||
}
|
||||
let added = sqlx::query(
|
||||
"INSERT INTO members (space_id, account_id, role) VALUES ($1, $2, 'editor') ON CONFLICT DO NOTHING",
|
||||
)
|
||||
.bind(space_id)
|
||||
.bind(account.id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
if added.rows_affected() > 0 {
|
||||
sqlx::query("UPDATE invites SET uses = uses + 1 WHERE token_hash = $1")
|
||||
.bind(hash)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
}
|
||||
tx.commit().await?;
|
||||
Ok(InviteOutcome::Joined { space_id })
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ the CRDT
|
||||
|
||||
/// Appends one Yjs update; it is durable when this returns.
|
||||
pub async fn append_update(
|
||||
&self,
|
||||
space: Uuid,
|
||||
data: &[u8],
|
||||
account: Option<Uuid>,
|
||||
) -> Result<i64> {
|
||||
Ok(sqlx::query_scalar(
|
||||
"INSERT INTO updates (space_id, data, account_id) VALUES ($1, $2, $3) RETURNING id",
|
||||
)
|
||||
.bind(space)
|
||||
.bind(data)
|
||||
.bind(account)
|
||||
.fetch_one(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
pub async fn load_doc(&self, space: Uuid) -> Result<StoredDoc> {
|
||||
let latest = self.latest_version(space).await?;
|
||||
let state: Option<Vec<u8>> = match &latest {
|
||||
Some(v) => {
|
||||
sqlx::query_scalar("SELECT state FROM versions WHERE id = $1")
|
||||
.bind(v.id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?
|
||||
}
|
||||
None => None,
|
||||
};
|
||||
let from = latest.as_ref().map_or(0, |v| v.up_to);
|
||||
let rows: Vec<(i64, Vec<u8>)> = sqlx::query_as(
|
||||
"SELECT id, data FROM updates WHERE space_id = $1 AND id > $2 ORDER BY id",
|
||||
)
|
||||
.bind(space)
|
||||
.bind(from)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
let last_update = rows.last().map_or(from, |(id, _)| *id);
|
||||
Ok(StoredDoc {
|
||||
state,
|
||||
updates: rows.into_iter().map(|(_, d)| d).collect(),
|
||||
last_update,
|
||||
latest,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn latest_version(&self, space: Uuid) -> Result<Option<VersionRow>> {
|
||||
Ok(sqlx::query_as(concat!(
|
||||
"SELECT ",
|
||||
version_columns!(),
|
||||
" FROM versions WHERE space_id = $1 ORDER BY id DESC LIMIT 1"
|
||||
))
|
||||
.bind(space)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// Stores a version and drops the updates it contains from the log.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn insert_version(
|
||||
&self,
|
||||
space: Uuid,
|
||||
major: i32,
|
||||
minor: i32,
|
||||
kind: &str,
|
||||
name: Option<&str>,
|
||||
state: &[u8],
|
||||
up_to: i64,
|
||||
restored_from: Option<i64>,
|
||||
created_by: Option<Uuid>,
|
||||
) -> Result<VersionRow> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let row: VersionRow = sqlx::query_as(concat!("INSERT INTO versions (space_id, major, minor, kind, name, state, up_to, restored_from, created_by)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING ", version_columns!(), ""))
|
||||
.bind(space)
|
||||
.bind(major)
|
||||
.bind(minor)
|
||||
.bind(kind)
|
||||
.bind(name)
|
||||
.bind(state)
|
||||
.bind(up_to)
|
||||
.bind(restored_from)
|
||||
.bind(created_by)
|
||||
.fetch_one(&mut *tx)
|
||||
.await
|
||||
.map_err(|e| match &e {
|
||||
sqlx::Error::Database(d) if d.is_unique_violation() => StoreError::Conflict("version"),
|
||||
_ => StoreError::Sql(e),
|
||||
})?;
|
||||
sqlx::query("DELETE FROM updates WHERE space_id = $1 AND id <= $2")
|
||||
.bind(space)
|
||||
.bind(up_to)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(row)
|
||||
}
|
||||
|
||||
pub async fn versions(&self, space: Uuid, limit: i64) -> Result<Vec<VersionRow>> {
|
||||
Ok(sqlx::query_as(concat!(
|
||||
"SELECT ",
|
||||
version_columns!(),
|
||||
" FROM versions WHERE space_id = $1 ORDER BY id DESC LIMIT $2"
|
||||
))
|
||||
.bind(space)
|
||||
.bind(limit)
|
||||
.fetch_all(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
pub async fn version(&self, space: Uuid, id: i64) -> Result<Option<(VersionRow, Vec<u8>)>> {
|
||||
let row: Option<VersionRow> = sqlx::query_as(concat!(
|
||||
"SELECT ",
|
||||
version_columns!(),
|
||||
" FROM versions WHERE space_id = $1 AND id = $2"
|
||||
))
|
||||
.bind(space)
|
||||
.bind(id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
let Some(row) = row else { return Ok(None) };
|
||||
let state: Vec<u8> = sqlx::query_scalar("SELECT state FROM versions WHERE id = $1")
|
||||
.bind(id)
|
||||
.fetch_one(&self.pool)
|
||||
.await?;
|
||||
Ok(Some((row, state)))
|
||||
}
|
||||
}
|
||||
|
||||
async fn insert_account(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
organisation_id: Uuid,
|
||||
name: &str,
|
||||
display_name: &str,
|
||||
role: &str,
|
||||
password_hash: Option<&str>,
|
||||
) -> Result<AccountRow> {
|
||||
let id = Uuid::now_v7();
|
||||
sqlx::query(
|
||||
"INSERT INTO accounts (id, organisation_id, name, display_name, role, password_hash) VALUES ($1, $2, $3, $4, $5, $6)",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(organisation_id)
|
||||
.bind(name)
|
||||
.bind(display_name)
|
||||
.bind(role)
|
||||
.bind(password_hash)
|
||||
.execute(&mut **tx)
|
||||
.await
|
||||
.map_err(|e| match &e {
|
||||
sqlx::Error::Database(d) if d.is_unique_violation() => StoreError::Conflict("account"),
|
||||
_ => StoreError::Sql(e),
|
||||
})?;
|
||||
Ok(AccountRow {
|
||||
id,
|
||||
organisation_id,
|
||||
name: name.to_owned(),
|
||||
display_name: display_name.to_owned(),
|
||||
role: role.to_owned(),
|
||||
password_hash: password_hash.map(str::to_owned),
|
||||
})
|
||||
}
|
||||
Reference in new issue
Block a user