confium_log_server/
db.rs

1//! SQLite-backed append-only log storage.
2//!
3//! Schema:
4//!
5//! - `entries` — primary log table. One row per append. Primary key
6//!   is `sequence` (auto-incremented). The Merkle tree is
7//!   materialized in `tree_nodes` for O(log N) incremental
8//!   recomputation.
9//! - `tree_nodes` — cached Merkle tree nodes, indexed by
10//!   `(level, index)`. The root is at the highest level for the
11//!   current size.
12//! - `cert_entries` — join table mapping cert fingerprints to
13//!   log entries. Carries parsed metadata (issuer, subject,
14//!   validity window) so the API can serve cert-specific queries
15//!   without re-parsing.
16//! - `ots_proofs` — Bitcoin OTS proofs keyed by tree head sequence.
17//! - `witness_sigs` — witness countersignatures keyed by tree head
18//!   sequence + witness ID.
19
20use std::path::Path;
21use std::sync::Arc;
22
23use anyhow::{Context, Result, anyhow};
24use confium_transparency::entry::ArtifactType;
25use rusqlite::{Connection, params};
26use serde::{Deserialize, Serialize};
27
28/// One stored entry, parsed into the shape the Merkle rebuild needs.
29/// The sequence is the 0-based leaf index (rowid minus one).
30pub struct RebuildRow {
31    pub sequence: u64,
32    pub timestamp: chrono::DateTime<chrono::Utc>,
33    pub artifact_type: ArtifactType,
34    pub artifact_hash: [u8; 32],
35}
36
37/// OTS proof row: encoded proof bytes, optional Bitcoin block height,
38/// anchor timestamp (ISO 8601).
39pub type OtsProofRow = (Vec<u8>, Option<u64>, String);
40
41/// Wrapper around the SQLite connection. Cheaply cloneable because
42/// `Connection` is wrapped in a `Mutex` inside an `Arc`.
43#[derive(Clone)]
44pub struct Database {
45    conn: Arc<parking_lot::Mutex<Connection>>,
46}
47
48#[derive(Debug, Clone, Serialize, Deserialize)]
49pub struct Entry {
50    pub sequence: u64,
51    pub artifact_type: String,
52    pub artifact_hash: String, // hex
53    pub timestamp: String,     // RFC3339
54    pub issuer_distinguished_name: Option<String>,
55    pub subject_distinguished_name: Option<String>,
56    pub fingerprint_sha256: Option<String>, // hex
57    pub valid_from: Option<String>,
58    pub valid_to: Option<String>,
59}
60
61impl Database {
62    pub fn open(path: &Path) -> Result<Self> {
63        let conn = Connection::open(path)
64            .with_context(|| format!("opening database at {}", path.display()))?;
65        // WAL mode for better concurrency on read-heavy workloads.
66        conn.execute_batch(
67            "PRAGMA journal_mode = WAL;
68             PRAGMA synchronous = NORMAL;
69             PRAGMA foreign_keys = ON;",
70        )?;
71        Ok(Database {
72            conn: Arc::new(parking_lot::Mutex::new(conn)),
73        })
74    }
75
76    pub fn init_schema(&self) -> Result<()> {
77        let conn = self.conn.lock();
78        conn.execute_batch(
79            "CREATE TABLE IF NOT EXISTS entries (
80                sequence           INTEGER PRIMARY KEY AUTOINCREMENT,
81                artifact_type      TEXT NOT NULL,
82                artifact_hash      TEXT NOT NULL,
83                timestamp          TEXT NOT NULL,
84                issuer_dn          TEXT,
85                subject_dn         TEXT,
86                fingerprint_sha256 TEXT,
87                valid_from         TEXT,
88                valid_to           TEXT
89            );
90
91            CREATE INDEX IF NOT EXISTS idx_entries_fingerprint
92                ON entries(fingerprint_sha256);
93            CREATE INDEX IF NOT EXISTS idx_entries_issuer
94                ON entries(issuer_dn);
95            CREATE INDEX IF NOT EXISTS idx_entries_type_ts
96                ON entries(artifact_type, timestamp);
97
98            CREATE TABLE IF NOT EXISTS tree_nodes (
99                level INTEGER NOT NULL,
100                idx   INTEGER NOT NULL,
101                hash  TEXT NOT NULL,
102                PRIMARY KEY (level, idx)
103            );
104
105            CREATE TABLE IF NOT EXISTS tree_meta (
106                key   TEXT PRIMARY KEY,
107                value TEXT NOT NULL
108            );
109
110            CREATE TABLE IF NOT EXISTS ots_proofs (
111                tree_size    INTEGER PRIMARY KEY,
112                root_hash    TEXT NOT NULL,
113                ots_proof    BLOB NOT NULL,
114                bitcoin_height INTEGER,
115                anchor_time  TEXT NOT NULL
116            );
117
118            CREATE TABLE IF NOT EXISTS witness_sigs (
119                tree_size   INTEGER NOT NULL,
120                root_hash   TEXT NOT NULL,
121                witness_id  TEXT NOT NULL,
122                signature   BLOB NOT NULL,
123                timestamp   TEXT NOT NULL,
124                PRIMARY KEY (tree_size, witness_id)
125            );",
126        )?;
127        Ok(())
128    }
129
130    pub fn append(&self, entry: &Entry) -> Result<u64> {
131        let conn = self.conn.lock();
132        conn.execute(
133            "INSERT INTO entries
134                (artifact_type, artifact_hash, timestamp,
135                 issuer_dn, subject_dn, fingerprint_sha256,
136                 valid_from, valid_to)
137             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
138            params![
139                entry.artifact_type,
140                entry.artifact_hash,
141                entry.timestamp,
142                entry.issuer_distinguished_name,
143                entry.subject_distinguished_name,
144                entry.fingerprint_sha256,
145                entry.valid_from,
146                entry.valid_to,
147            ],
148        )?;
149        // Rowids are 1-based; entry sequences are 0-based to match the
150        // Merkle leaf index used by the proof endpoints.
151        Ok((conn.last_insert_rowid() - 1) as u64)
152    }
153
154    pub fn entry_at(&self, sequence: u64) -> Result<Option<Entry>> {
155        let conn = self.conn.lock();
156        let mut stmt = conn.prepare(
157            "SELECT sequence, artifact_type, artifact_hash, timestamp,
158                    issuer_dn, subject_dn, fingerprint_sha256,
159                    valid_from, valid_to
160             FROM entries WHERE sequence = ?1 + 1",
161        )?;
162        let rows = stmt.query_row(params![sequence as i64], |row| {
163            Ok(Entry {
164                sequence: row.get::<_, i64>(0)? as u64,
165                artifact_type: row.get(1)?,
166                artifact_hash: row.get(2)?,
167                timestamp: row.get(3)?,
168                issuer_distinguished_name: row.get(4)?,
169                subject_distinguished_name: row.get(5)?,
170                fingerprint_sha256: row.get(6)?,
171                valid_from: row.get(7)?,
172                valid_to: row.get(8)?,
173            })
174        });
175        match rows {
176            Ok(e) => Ok(Some(e)),
177            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
178            Err(e) => Err(e.into()),
179        }
180    }
181
182    pub fn entry_count(&self) -> Result<u64> {
183        let conn = self.conn.lock();
184        let n: i64 = conn.query_row("SELECT COUNT(*) FROM entries", [], |row| row.get(0))?;
185        Ok(n as u64)
186    }
187
188    pub fn entries_by_fingerprint(&self, fingerprint_hex: &str) -> Result<Vec<Entry>> {
189        let conn = self.conn.lock();
190        let mut stmt = conn.prepare(
191            "SELECT sequence, artifact_type, artifact_hash, timestamp,
192                    issuer_dn, subject_dn, fingerprint_sha256,
193                    valid_from, valid_to
194             FROM entries WHERE fingerprint_sha256 = ?1
195             ORDER BY sequence ASC",
196        )?;
197        let rows = stmt.query_map(params![fingerprint_hex], |row| {
198            Ok(Entry {
199                sequence: row.get::<_, i64>(0)? as u64,
200                artifact_type: row.get(1)?,
201                artifact_hash: row.get(2)?,
202                timestamp: row.get(3)?,
203                issuer_distinguished_name: row.get(4)?,
204                subject_distinguished_name: row.get(5)?,
205                fingerprint_sha256: row.get(6)?,
206                valid_from: row.get(7)?,
207                valid_to: row.get(8)?,
208            })
209        })?;
210        let mut out = Vec::new();
211        for row in rows {
212            out.push(row?);
213        }
214        Ok(out)
215    }
216
217    pub fn entries_by_issuer(&self, issuer_dn: &str, limit: usize) -> Result<Vec<Entry>> {
218        let conn = self.conn.lock();
219        let mut stmt = conn.prepare(
220            "SELECT sequence, artifact_type, artifact_hash, timestamp,
221                    issuer_dn, subject_dn, fingerprint_sha256,
222                    valid_from, valid_to
223             FROM entries WHERE issuer_dn = ?1
224             ORDER BY sequence DESC
225             LIMIT ?2",
226        )?;
227        let rows = stmt.query_map(params![issuer_dn, limit as i64], |row| {
228            Ok(Entry {
229                sequence: row.get::<_, i64>(0)? as u64,
230                artifact_type: row.get(1)?,
231                artifact_hash: row.get(2)?,
232                timestamp: row.get(3)?,
233                issuer_distinguished_name: row.get(4)?,
234                subject_distinguished_name: row.get(5)?,
235                fingerprint_sha256: row.get(6)?,
236                valid_from: row.get(7)?,
237                valid_to: row.get(8)?,
238            })
239        })?;
240        let mut out = Vec::new();
241        for row in rows {
242            out.push(row?);
243        }
244        Ok(out)
245    }
246
247    /// Read every leaf hash in sequence order. Used to rebuild the
248    /// Merkle tree on startup.
249    pub fn all_leaf_hashes(&self) -> Result<Vec<[u8; 32]>> {
250        let conn = self.conn.lock();
251        let mut stmt = conn.prepare("SELECT artifact_hash FROM entries ORDER BY sequence ASC")?;
252        let rows = stmt.query_map([], |row| {
253            let h: String = row.get(0)?;
254            Ok(h)
255        })?;
256        let mut out = Vec::new();
257        for row in rows {
258            let h = row?;
259            let bytes = hex::decode(&h).map_err(|e| anyhow!("hash hex decode: {e}"))?;
260            if bytes.len() != 32 {
261                return Err(anyhow!("hash must be 32 bytes, got {}", bytes.len()));
262            }
263            let mut arr = [0u8; 32];
264            arr.copy_from_slice(&bytes);
265            out.push(arr);
266        }
267        Ok(out)
268    }
269
270    /// Every entry in append order, with the stored timestamp and
271    /// type parsed back into domain types. The Merkle rebuild uses
272    /// exactly these values so that leaf hashes — which cover the
273    /// sequence, timestamp, and artifact hash — are identical before
274    /// and after a restart.
275    pub fn all_entries_for_rebuild(&self) -> Result<Vec<RebuildRow>> {
276        let conn = self.conn.lock();
277        let mut stmt = conn.prepare(
278            "SELECT sequence, artifact_type, artifact_hash, timestamp
279             FROM entries ORDER BY sequence ASC",
280        )?;
281        let rows = stmt.query_map([], |row| {
282            Ok((
283                row.get::<_, i64>(0)?,
284                row.get::<_, String>(1)?,
285                row.get::<_, String>(2)?,
286                row.get::<_, String>(3)?,
287            ))
288        })?;
289        let mut out = Vec::new();
290        for (i, row) in rows.enumerate() {
291            let (rowid, artifact_type, artifact_hash, timestamp) = row?;
292            let bytes = hex::decode(&artifact_hash)
293                .map_err(|e| anyhow!("hash hex decode at row {rowid}: {e}"))?;
294            if bytes.len() != 32 {
295                return Err(anyhow!(
296                    "hash must be 32 bytes at row {rowid}, got {}",
297                    bytes.len()
298                ));
299            }
300            let mut arr = [0u8; 32];
301            arr.copy_from_slice(&bytes);
302            out.push(RebuildRow {
303                // Rowids are 1-based; the tree's leaf index is the
304                // 0-based position in append order.
305                sequence: i as u64,
306                timestamp: chrono::DateTime::parse_from_rfc3339(&timestamp)
307                    .with_context(|| format!("parsing timestamp at row {rowid}"))?
308                    .with_timezone(&chrono::Utc),
309                artifact_type: artifact_type
310                    .parse()
311                    .map_err(|e| anyhow!("artifact type at row {rowid}: {e}"))?,
312                artifact_hash: arr,
313            });
314        }
315        Ok(out)
316    }
317
318    pub fn store_ots_proof(
319        &self,
320        tree_size: u64,
321        root_hash: &[u8; 32],
322        ots_proof: &[u8],
323        bitcoin_height: Option<u64>,
324    ) -> Result<()> {
325        let conn = self.conn.lock();
326        conn.execute(
327            "INSERT OR REPLACE INTO ots_proofs
328                (tree_size, root_hash, ots_proof, bitcoin_height, anchor_time)
329             VALUES (?1, ?2, ?3, ?4, ?5)",
330            params![
331                tree_size as i64,
332                hex::encode(root_hash),
333                ots_proof,
334                bitcoin_height.map(|h| h as i64),
335                chrono::Utc::now().to_rfc3339(),
336            ],
337        )?;
338        Ok(())
339    }
340
341    pub fn get_ots_proof(&self, tree_size: u64) -> Result<Option<OtsProofRow>> {
342        let conn = self.conn.lock();
343        let row = conn.query_row(
344            "SELECT ots_proof, bitcoin_height, anchor_time
345             FROM ots_proofs WHERE tree_size = ?1",
346            params![tree_size as i64],
347            |row| {
348                let proof: Vec<u8> = row.get(0)?;
349                let bh: Option<i64> = row.get(1)?;
350                let at: String = row.get(2)?;
351                Ok((proof, bh.map(|h| h as u64), at))
352            },
353        );
354        match row {
355            Ok(t) => Ok(Some(t)),
356            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
357            Err(e) => Err(e.into()),
358        }
359    }
360
361    pub fn store_witness_sig(
362        &self,
363        tree_size: u64,
364        root_hash: &[u8; 32],
365        witness_id: &str,
366        signature: &[u8],
367    ) -> Result<()> {
368        let conn = self.conn.lock();
369        conn.execute(
370            "INSERT OR REPLACE INTO witness_sigs
371                (tree_size, root_hash, witness_id, signature, timestamp)
372             VALUES (?1, ?2, ?3, ?4, ?5)",
373            params![
374                tree_size as i64,
375                hex::encode(root_hash),
376                witness_id,
377                signature,
378                chrono::Utc::now().to_rfc3339(),
379            ],
380        )?;
381        Ok(())
382    }
383
384    pub fn witness_sigs_for_size(&self, tree_size: u64) -> Result<Vec<(String, Vec<u8>, String)>> {
385        let conn = self.conn.lock();
386        let mut stmt = conn.prepare(
387            "SELECT witness_id, signature, timestamp
388             FROM witness_sigs WHERE tree_size = ?1
389             ORDER BY witness_id ASC",
390        )?;
391        let rows = stmt.query_map(params![tree_size as i64], |row| {
392            let wid: String = row.get(0)?;
393            let sig: Vec<u8> = row.get(1)?;
394            let ts: String = row.get(2)?;
395            Ok((wid, sig, ts))
396        })?;
397        let mut out = Vec::new();
398        for row in rows {
399            out.push(row?);
400        }
401        Ok(out)
402    }
403}
confium_log_server/db.rs

confium_log_server/
db.rs
