aven_core/sync/seed_claim/peer/
persistence.rs

1use super::*;
2use crate::db::{self, Database, begin_immediate, installation::InstallationGuard};
3use anyhow::Context;
4use sqlx::SqliteConnection;
5use std::collections::HashMap;
6
7impl Database {
8    pub async fn peer_target_preflight(&self) -> Result<String> {
9        let mut conn = self.acquire_writer().await?;
10        let mut tx = begin_immediate(&mut conn).await?;
11        fresh(&mut tx).await?;
12        let client = db::get_meta(&mut tx, "client_id")
13            .await?
14            .context("error enrollment-client-missing")?;
15        tx.commit().await?;
16        Ok(client)
17    }
18    /// Rechecks that this pinned peer join is unfinished: nothing installed,
19    /// associated or set up, and the domain still empty. The enrollment pin
20    /// and installation fence stay in place.
21    pub async fn peer_retry_preflight(
22        &self,
23        identity: [u8; 32],
24        client: &str,
25        guard: &InstallationGuard,
26    ) -> Result<()> {
27        check(self.file_identity() == Some(guard.identity()))?;
28        let mut conn = self.acquire_writer().await?;
29        let mut tx = begin_immediate(&mut conn).await?;
30        let pinned: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_enrollment WHERE identity=? AND client_id=? AND role='peer' AND client_id=(SELECT value FROM meta WHERE key='client_id'))").bind(identity.as_slice()).bind(client).fetch_one(&mut *tx).await?;
31        ensure!(pinned, "error enrollment-pin");
32        let occupied: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_snapshot_install) OR EXISTS(SELECT 1 FROM local_seed_source) OR EXISTS(SELECT 1 FROM local_seed_publication_intent) OR EXISTS(SELECT 1 FROM local_seed_genesis_pin) OR EXISTS(SELECT 1 FROM server_seed_claim) OR EXISTS(SELECT 1 FROM meta WHERE key IN ('sync_server_url','e2ee_association'))").fetch_one(&mut *tx).await?;
33        ensure!(!occupied, "error enrollment-retry-unavailable");
34        crate::sync::shared_state::ensure_empty_domain(&mut tx).await?;
35        tx.commit().await?;
36        Ok(())
37    }
38    pub async fn enrollment_pin(&self) -> Result<Option<([u8; 32], String, String)>> {
39        let mut conn = self.acquire_reader().await?;
40        let row: Option<(Vec<u8>, String, String)> = sqlx::query_as(
41            "SELECT identity,client_id,role FROM local_peer_enrollment WHERE singleton=1",
42        )
43        .fetch_optional(&mut *conn)
44        .await?;
45        row.map(|(id, client, role)| {
46            Ok((
47                id.try_into()
48                    .map_err(|_| anyhow::anyhow!("error enrollment-pin"))?,
49                client,
50                role,
51            ))
52        })
53        .transpose()
54    }
55    pub async fn pin_enrollment(
56        &self,
57        identity: [u8; 32],
58        client: &str,
59        role: &str,
60        guard: &InstallationGuard,
61    ) -> Result<()> {
62        check(self.file_identity() == Some(guard.identity()))?;
63        let mut conn = self.acquire_writer().await?;
64        let mut tx = begin_immediate(&mut conn).await?;
65        let stored: Option<(Vec<u8>, String, String)> = sqlx::query_as(
66            "SELECT identity,client_id,role FROM local_peer_enrollment WHERE singleton=1",
67        )
68        .fetch_optional(&mut *tx)
69        .await?;
70        check(db::get_meta(&mut tx, "client_id").await?.as_deref() == Some(client))?;
71        if let Some((id, c, r)) = stored {
72            check(id == identity && c == client && r == role)?;
73        } else {
74            if role == "peer" {
75                fresh(&mut tx).await?;
76            } else {
77                ensure_adopted(&mut tx).await?;
78            }
79            sqlx::query("INSERT INTO local_peer_enrollment(singleton,identity,client_id,role) VALUES(1,?,?,?)").bind(identity.as_slice()).bind(client).bind(role).execute(&mut *tx).await?;
80        }
81        tx.commit().await?;
82        Ok(())
83    }
84    pub async fn adopted_enrollment_client(&self) -> Result<String> {
85        let mut conn = self.acquire_reader().await?;
86        ensure_adopted(&mut conn).await?;
87        db::get_meta(&mut conn, "client_id")
88            .await?
89            .context("error enrollment-client-missing")
90    }
91    pub async fn enrollment_artifact(&self, kind: &str) -> Result<Option<Vec<u8>>> {
92        let mut conn = self.acquire_reader().await?;
93        Ok(sqlx::query_scalar(
94            "SELECT commitment FROM local_peer_enrollment_artifacts WHERE kind=?",
95        )
96        .bind(kind)
97        .fetch_optional(&mut *conn)
98        .await?)
99    }
100    pub async fn enrollment_artifacts(&self) -> Result<HashMap<String, Vec<u8>>> {
101        let mut conn = self.acquire_reader().await?;
102        let rows: Vec<(String, Vec<u8>)> =
103            sqlx::query_as("SELECT kind, commitment FROM local_peer_enrollment_artifacts")
104                .fetch_all(&mut *conn)
105                .await?;
106        Ok(rows.into_iter().collect())
107    }
108    /// Monotonic, nonsecret marker for write-once enrollment state changes.
109    pub async fn enrollment_artifact_marker(&self) -> Result<i64> {
110        let mut conn = self.acquire_reader().await?;
111        Ok(
112            sqlx::query_scalar("SELECT count(*) FROM local_peer_enrollment_artifacts")
113                .fetch_one(&mut *conn)
114                .await?,
115        )
116    }
117    pub async fn pin_enrollment_artifact(
118        &self,
119        identity: [u8; 32],
120        kind: &str,
121        digest: [u8; 32],
122    ) -> Result<()> {
123        let mut conn = self.acquire_writer().await?;
124        let mut tx = begin_immediate(&mut conn).await?;
125        let valid: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_peer_enrollment WHERE identity=? AND client_id=(SELECT value FROM meta WHERE key='client_id'))").bind(identity.as_slice()).fetch_one(&mut *tx).await?;
126        check(valid)?;
127        sqlx::query(
128            "INSERT OR IGNORE INTO local_peer_enrollment_artifacts(kind,commitment) VALUES(?,?)",
129        )
130        .bind(kind)
131        .bind(digest.as_slice())
132        .execute(&mut *tx)
133        .await?;
134        let saved: Vec<u8> = sqlx::query_scalar(
135            "SELECT commitment FROM local_peer_enrollment_artifacts WHERE kind=?",
136        )
137        .bind(kind)
138        .fetch_one(&mut *tx)
139        .await?;
140        check(saved == digest)?;
141        tx.commit().await?;
142        Ok(())
143    }
144}
145async fn ensure_adopted(conn: &mut SqliteConnection) -> Result<()> {
146    let ready: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_publication_intent WHERE state='adopted' AND association=(SELECT value FROM meta WHERE key='e2ee_association') AND association_generation=CAST((SELECT value FROM meta WHERE key='sync_generation') AS INTEGER))").fetch_one(conn).await?;
147    ensure!(ready, "error enrollment-inviter-not-adopted");
148    Ok(())
149}
150async fn fresh(conn: &mut SqliteConnection) -> Result<()> {
151    crate::sync::shared_state::ensure_empty_target(conn).await?;
152    let occupied: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_genesis_pin) OR EXISTS(SELECT 1 FROM server_seed_claim) OR EXISTS(SELECT 1 FROM local_peer_enrollment) OR EXISTS(SELECT 1 FROM meta WHERE key IN ('sync_server_url','e2ee_association'))").fetch_one(conn).await?;
153    ensure!(!occupied, "error enrollment-target-not-fresh");
154    Ok(())
155}
aven_core/sync/seed_claim/peer/persistence.rs

aven_core/sync/seed_claim/peer/
persistence.rs
