1use std::{
2 collections::{HashMap, HashSet},
3 path::Path,
4 sync::Arc,
5};
6
7use chrono::Utc;
8use sqlx::{
9 Executor, QueryBuilder, Row, Sqlite, SqlitePool, Transaction,
10 sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteRow},
11};
12use tokio::sync::Mutex;
13
14use crate::{
15 codec::{Codec, compress, decompress},
16 error::{Error, Result},
17 hash::Hash32,
18 object::{ObjectKind, ObjectRecord, ObjectSource, VerifiedObject},
19 pack::{MerkleProof, MerkleSummary, merkle_root_from_hashes},
20};
21
22const SQLITE_PARAM_LIMIT: usize = 999;
23const OBJECT_COLUMNS: usize = 7;
24const MAX_OBJECTS_PER_BATCH: usize = SQLITE_PARAM_LIMIT / OBJECT_COLUMNS;
25const FILE_RECIPE_COLUMNS: usize = 3;
26const MAX_FILE_RECIPE_PER_BATCH: usize = SQLITE_PARAM_LIMIT / FILE_RECIPE_COLUMNS;
27
28pub type StoreWriteTx<'a> = Transaction<'a, Sqlite>;
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
31pub enum MerkleMode {
32 Disabled,
33 #[default]
34 Enabled,
35}
36
37#[derive(Debug, Clone, Default)]
38pub struct SqliteStoreOptions {
39 pub namespace: Option<String>,
40 pub merkle: MerkleMode,
41}
42
43#[derive(Default)]
44struct MerkleCache {
45 summary: Option<MerkleSummary>,
46 leaf_pos: HashMap<Hash32, u64>,
47 leaf_hashes: HashMap<u64, Hash32>,
48 node_hashes: HashMap<i64, HashMap<u64, Hash32>>,
49}
50
51#[derive(Debug, Clone)]
52struct TableNames {
53 namespace: Option<String>,
54 objects: String,
55 file_recipe_cache: String,
56 merkle_leaves: String,
57 merkle_nodes: String,
58 merkle_meta: String,
59 merkle_leaves_pos_index: String,
60}
61
62impl TableNames {
63 fn new(namespace: Option<&str>) -> Result<Self> {
64 let namespace = normalize_namespace(namespace)?;
65 let prefix = namespace.as_ref().map(|value| format!("{value}_")).unwrap_or_default();
66 let index_prefix = namespace
67 .as_ref()
68 .map(|value| format!("idx_{value}_"))
69 .unwrap_or_else(|| "idx_".into());
70
71 Ok(Self {
72 namespace,
73 objects: format!("{prefix}objects"),
74 file_recipe_cache: format!("{prefix}file_recipe_cache"),
75 merkle_leaves: format!("{prefix}merkle_leaves"),
76 merkle_nodes: format!("{prefix}merkle_nodes"),
77 merkle_meta: format!("{prefix}merkle_meta"),
78 merkle_leaves_pos_index: format!("{index_prefix}merkle_leaves_pos"),
79 })
80 }
81}
82
83#[derive(Clone)]
84pub struct SqliteStore {
85 pool: SqlitePool,
86 tables: TableNames,
87 merkle: MerkleMode,
88 cache: Arc<Mutex<MerkleCache>>,
89}
90
91impl SqliteStore {
92 pub async fn open(path: impl AsRef<Path>) -> Result<Self> {
93 Self::open_with_options(path, SqliteStoreOptions::default()).await
94 }
95
96 pub async fn open_with_options(path: impl AsRef<Path>, options: SqliteStoreOptions) -> Result<Self> {
97 let path = path.as_ref().to_path_buf();
98 if let Some(parent) = path.parent() {
99 tokio::fs::create_dir_all(parent).await?;
100 }
101 let pool = SqlitePoolOptions::new()
102 .max_connections(1)
103 .connect_with(SqliteConnectOptions::new().filename(&path).create_if_missing(true))
104 .await?;
105 Self::from_pool_with_options(pool, options).await
106 }
107
108 pub async fn from_pool(pool: SqlitePool) -> Result<Self> {
109 Self::from_pool_with_options(pool, SqliteStoreOptions::default()).await
110 }
111
112 pub async fn from_pool_with_namespace(pool: SqlitePool, namespace: impl AsRef<str>) -> Result<Self> {
113 Self::from_pool_with_options(
114 pool,
115 SqliteStoreOptions {
116 namespace: Some(namespace.as_ref().to_string()),
117 ..Default::default()
118 },
119 )
120 .await
121 }
122
123 pub async fn from_pool_with_options(pool: SqlitePool, options: SqliteStoreOptions) -> Result<Self> {
124 let tables = TableNames::new(options.namespace.as_deref())?;
125 Self::ensure_schema_with_options(&pool, &options).await?;
126 Ok(Self {
127 pool,
128 tables,
129 merkle: options.merkle,
130 cache: Arc::new(Mutex::new(MerkleCache::default())),
131 })
132 }
133
134 pub async fn ensure_schema(pool: &SqlitePool) -> Result<()> {
135 Self::ensure_schema_with_options(pool, &SqliteStoreOptions::default()).await
136 }
137
138 pub async fn ensure_schema_with_namespace(pool: &SqlitePool, namespace: impl AsRef<str>) -> Result<()> {
139 Self::ensure_schema_with_options(
140 pool,
141 &SqliteStoreOptions {
142 namespace: Some(namespace.as_ref().to_string()),
143 ..Default::default()
144 },
145 )
146 .await
147 }
148
149 pub async fn ensure_schema_with_options(pool: &SqlitePool, options: &SqliteStoreOptions) -> Result<()> {
150 let tables = TableNames::new(options.namespace.as_deref())?;
151 for statement in schema_statements(&tables, options.merkle) {
152 pool.execute(statement.as_str()).await?;
153 }
154 Ok(())
155 }
156
157 pub fn pool(&self) -> &SqlitePool {
158 &self.pool
159 }
160
161 pub fn namespace(&self) -> Option<&str> {
162 self.tables.namespace.as_deref()
163 }
164
165 pub fn merkle_mode(&self) -> MerkleMode {
166 self.merkle
167 }
168
169 pub async fn begin_write_tx(&self) -> Result<StoreWriteTx<'_>> {
170 Ok(self.pool.begin().await?)
171 }
172
173 pub async fn put_objects_batch_tx(&self, tx: &mut StoreWriteTx<'_>, objects: &[ObjectRecord]) -> Result<()> {
174 if objects.is_empty() {
175 return Ok(());
176 }
177
178 let now = Utc::now().timestamp();
179 for chunk in objects.chunks(MAX_OBJECTS_PER_BATCH.max(1)) {
180 let mut builder = QueryBuilder::<Sqlite>::new(format!(
181 "INSERT OR IGNORE INTO {} (hash, kind, size, stored_size, codec, content, created_at) ",
182 self.tables.objects
183 ));
184 builder.push_values(chunk, |mut b, obj| {
185 b.push_bind(obj.hash.as_bytes().as_ref())
186 .push_bind(obj.kind as i64)
187 .push_bind(obj.decoded_len as i64)
188 .push_bind(obj.stored_bytes.len() as i64)
189 .push_bind(codec_to_str(obj.codec))
190 .push_bind(obj.stored_bytes.as_slice())
191 .push_bind(now);
192 });
193 builder.build().execute(&mut **tx).await?;
194 }
195 Ok(())
196 }
197
198 pub async fn put_file_recipe_cache_batch_tx(&self, tx: &mut StoreWriteTx<'_>, entries: &[(Hash32, Hash32)]) -> Result<()> {
199 if entries.is_empty() {
200 return Ok(());
201 }
202
203 let now = Utc::now().timestamp();
204 for chunk in entries.chunks(MAX_FILE_RECIPE_PER_BATCH.max(1)) {
205 let mut builder = QueryBuilder::<Sqlite>::new(format!(
206 "INSERT INTO {} (file_hash, recipe_hash, updated_at) ",
207 self.tables.file_recipe_cache
208 ));
209 builder.push_values(chunk, |mut b, (file_hash, recipe_hash)| {
210 b.push_bind(file_hash.as_bytes().as_ref())
211 .push_bind(recipe_hash.as_bytes().as_ref())
212 .push_bind(now);
213 });
214 builder.push(" ON CONFLICT(file_hash) DO UPDATE SET recipe_hash=excluded.recipe_hash, updated_at=excluded.updated_at");
215 builder.build().execute(&mut **tx).await?;
216 }
217 Ok(())
218 }
219
220 pub async fn put_chunk(&self, hash: Hash32, content: &[u8], codec: Codec) -> Result<()> {
221 self.put_object(hash, ObjectKind::Chunk, content, codec).await
222 }
223
224 pub async fn put_recipe(&self, hash: Hash32, content: &[u8], codec: Codec) -> Result<()> {
225 self.put_object(hash, ObjectKind::Recipe, content, codec).await
226 }
227
228 async fn put_object(&self, hash: Hash32, kind: ObjectKind, content: &[u8], codec: Codec) -> Result<()> {
229 let calc = Hash32::sha3_256(content);
230 if calc != hash {
231 return Err(Error::HashMismatch);
232 }
233
234 let size = content.len() as i64;
235 let mut final_codec = codec;
236 let compressed = if codec == Codec::Raw {
237 content.to_vec()
238 } else {
239 let candidate = compress(codec, content)?;
240 let ratio = candidate.len() as f64 / size.max(1) as f64;
241 if ratio >= 0.98 {
242 final_codec = Codec::Raw;
243 content.to_vec()
244 } else {
245 candidate
246 }
247 };
248
249 let query = format!(
250 "INSERT OR IGNORE INTO {} (hash, kind, size, stored_size, codec, content, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
251 self.tables.objects
252 );
253 sqlx::query(&query)
254 .bind(hash.as_bytes().as_ref())
255 .bind(kind as i64)
256 .bind(size)
257 .bind(compressed.len() as i64)
258 .bind(codec_to_str(final_codec))
259 .bind(compressed)
260 .bind(Utc::now().timestamp())
261 .execute(&self.pool)
262 .await?;
263 Ok(())
264 }
265
266 async fn read_verified_object(&self, hash: &Hash32) -> Result<Option<VerifiedObject>> {
267 let query = format!("SELECT kind, size, codec, content FROM {} WHERE hash = ?1", self.tables.objects);
268 let row = sqlx::query(&query)
269 .bind(hash.as_bytes().as_ref())
270 .fetch_optional(&self.pool)
271 .await?;
272 let Some(row) = row else {
273 return Ok(None);
274 };
275 Ok(Some(verified_object_from_row(row, *hash)?))
276 }
277
278 pub async fn existing_hashes(&self, hashes: &[Hash32]) -> Result<HashSet<Hash32>> {
279 let mut found = HashSet::new();
280 if hashes.is_empty() {
281 return Ok(found);
282 }
283
284 for chunk in hashes.chunks(SQLITE_PARAM_LIMIT.max(1)) {
285 let mut builder = QueryBuilder::<Sqlite>::new(format!("SELECT hash FROM {} WHERE hash IN (", self.tables.objects));
286 let mut separated = builder.separated(", ");
287 for hash in chunk {
288 separated.push_bind(hash.as_bytes().as_ref());
289 }
290 separated.push_unseparated(")");
291 let rows = builder.build().fetch_all(&self.pool).await?;
292 for row in rows {
293 let bytes: Vec<u8> = row.get("hash");
294 found.insert(Hash32::from_bytes(&bytes)?);
295 }
296 }
297 Ok(found)
298 }
299
300 pub async fn chunk_sizes(&self) -> Result<Vec<(Hash32, u64)>> {
301 let query = format!("SELECT hash, stored_size FROM {} WHERE kind=?1", self.tables.objects);
302 let rows = sqlx::query(&query).bind(ObjectKind::Chunk as i64).fetch_all(&self.pool).await?;
303 let mut items = Vec::with_capacity(rows.len());
304 for row in rows {
305 let hash_bytes: Vec<u8> = row.get("hash");
306 let hash = Hash32::from_bytes(&hash_bytes)?;
307 let stored_size: i64 = row.get("stored_size");
308 items.push((hash, stored_size as u64));
309 }
310 Ok(items)
311 }
312
313 pub async fn file_recipe_cache(&self) -> Result<HashMap<Hash32, Hash32>> {
314 let query = format!("SELECT file_hash, recipe_hash FROM {}", self.tables.file_recipe_cache);
315 let rows = sqlx::query(&query).fetch_all(&self.pool).await?;
316 let mut map = HashMap::with_capacity(rows.len());
317 for row in rows {
318 let file_bytes: Vec<u8> = row.get("file_hash");
319 let recipe_bytes: Vec<u8> = row.get("recipe_hash");
320 map.insert(Hash32::from_bytes(&file_bytes)?, Hash32::from_bytes(&recipe_bytes)?);
321 }
322 Ok(map)
323 }
324
325 pub async fn recipe_for_file_hash(&self, file_hash: &Hash32) -> Result<Option<Hash32>> {
326 let query = format!("SELECT recipe_hash FROM {} WHERE file_hash=?1", self.tables.file_recipe_cache);
327 let row = sqlx::query(&query)
328 .bind(file_hash.as_bytes().as_ref())
329 .fetch_optional(&self.pool)
330 .await?;
331 let Some(row) = row else {
332 return Ok(None);
333 };
334 let bytes: Vec<u8> = row.get("recipe_hash");
335 Ok(Some(Hash32::from_bytes(&bytes)?))
336 }
337
338 pub async fn all_objects(&self) -> Result<Vec<VerifiedObject>> {
339 let query = format!(
340 "SELECT hash, kind, size, codec, content FROM {} ORDER BY hash ASC",
341 self.tables.objects
342 );
343 let rows = sqlx::query(&query).fetch_all(&self.pool).await?;
344 let mut objects = Vec::with_capacity(rows.len());
345 for row in rows {
346 let hash_bytes: Vec<u8> = row.get("hash");
347 let hash = Hash32::from_bytes(&hash_bytes)?;
348 objects.push(verified_object_from_row(row, hash)?);
349 }
350 Ok(objects)
351 }
352
353 pub async fn integrity_check(&self) -> Result<()> {
354 let row = sqlx::query("PRAGMA integrity_check;").fetch_one(&self.pool).await?;
355 let status: String = row.get(0);
356 if status.trim() == "ok" {
357 Ok(())
358 } else {
359 Err(Error::Integrity(status))
360 }
361 }
362
363 pub async fn rebuild_merkle_index(&self) -> Result<MerkleSummary> {
364 self.ensure_merkle_enabled()?;
365
366 let mut tx = self.pool.begin().await?;
367 sqlx::query(&format!("DELETE FROM {}", self.tables.merkle_leaves))
368 .execute(&mut *tx)
369 .await?;
370 sqlx::query(&format!("DELETE FROM {}", self.tables.merkle_nodes))
371 .execute(&mut *tx)
372 .await?;
373 sqlx::query(&format!("DELETE FROM {}", self.tables.merkle_meta))
374 .execute(&mut *tx)
375 .await?;
376
377 let rows = sqlx::query(&format!("SELECT hash FROM {} ORDER BY hash ASC", self.tables.objects))
378 .fetch_all(&mut *tx)
379 .await?;
380 let mut leaves = Vec::with_capacity(rows.len());
381 for row in rows {
382 let bytes: Vec<u8> = row.get("hash");
383 leaves.push(Hash32::from_bytes(&bytes)?);
384 }
385
386 for (idx, hash) in leaves.iter().enumerate() {
387 sqlx::query(&format!("INSERT INTO {} (hash, pos) VALUES (?1, ?2)", self.tables.merkle_leaves))
388 .bind(hash.as_bytes().as_ref())
389 .bind(idx as i64)
390 .execute(&mut *tx)
391 .await?;
392 }
393
394 let summary = merkle_root_from_hashes(&leaves);
395 let mut current = leaves;
396 let mut next_level = Vec::new();
397 let mut level = 0i64;
398 while current.len() > 1 {
399 next_level.clear();
400 for (idx, chunk) in current.chunks(2).enumerate() {
401 let left = chunk[0];
402 let right = if chunk.len() > 1 { chunk[1] } else { chunk[0] };
403 let parent = hash_pair(&left, &right);
404 sqlx::query(&format!(
405 "INSERT INTO {} (level, pos, hash) VALUES (?1, ?2, ?3) ON CONFLICT(level, pos) DO UPDATE SET hash=excluded.hash",
406 self.tables.merkle_nodes
407 ))
408 .bind(level)
409 .bind(idx as i64)
410 .bind(parent.as_bytes().as_ref())
411 .execute(&mut *tx)
412 .await?;
413 next_level.push(parent);
414 }
415 current.clear();
416 current.extend_from_slice(&next_level);
417 level += 1;
418 }
419
420 for (key, value) in [
421 ("leaf_count", summary.leaf_count.to_string()),
422 ("root_hash", summary.root.to_hex()),
423 ("algo", "sha3-256".to_string()),
424 ("tree", "merkle-set-v1".to_string()),
425 ("sorted", "byte-lex".to_string()),
426 ] {
427 sqlx::query(&format!(
428 "INSERT INTO {} (key, value) VALUES (?1, ?2) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
429 self.tables.merkle_meta
430 ))
431 .bind(key)
432 .bind(value)
433 .execute(&mut *tx)
434 .await?;
435 }
436
437 tx.commit().await?;
438 let mut cache = self.cache.lock().await;
439 cache.summary = Some(summary.clone());
440 cache.leaf_pos.clear();
441 cache.leaf_hashes.clear();
442 cache.node_hashes.clear();
443 Ok(summary)
444 }
445
446 pub async fn merkle_summary(&self) -> Result<MerkleSummary> {
447 self.ensure_merkle_enabled()?;
448
449 if let Some(summary) = self.cache.lock().await.summary.clone() {
450 return Ok(summary);
451 }
452
453 let leaf_count = read_meta_u64(&self.pool, &self.tables.merkle_meta, "leaf_count").await?;
454 let root_hex = read_meta_value(&self.pool, &self.tables.merkle_meta, "root_hash")
455 .await?
456 .ok_or_else(|| Error::MissingMeta("root_hash".into()))?;
457 let root = Hash32::from_hex(&root_hex).map_err(|_| Error::MissingMeta("root_hash".into()))?;
458 let summary = MerkleSummary { root, leaf_count };
459 self.cache.lock().await.summary = Some(summary.clone());
460 Ok(summary)
461 }
462
463 pub async fn prove_membership(&self, hash: &Hash32) -> Result<Option<MerkleProof>> {
464 self.ensure_merkle_enabled()?;
465
466 let summary = self.merkle_summary().await?;
467 if summary.leaf_count == 0 {
468 return Ok(None);
469 }
470
471 let row = sqlx::query(&format!("SELECT pos FROM {} WHERE hash=?1", self.tables.merkle_leaves))
472 .bind(hash.as_bytes().as_ref())
473 .fetch_optional(&self.pool)
474 .await?;
475 let Some(row) = row else {
476 return Ok(None);
477 };
478
479 let mut pos = row.get::<i64, _>("pos") as u64;
480 let leaf_pos = pos;
481 let mut count = summary.leaf_count;
482 let mut siblings = Vec::new();
483 let mut current_hash = *hash;
484 let mut node_level = 0i64;
485 let mut on_leaf = true;
486
487 while count > 1 {
488 let sibling_pos = pos ^ 1;
489 let sibling_hash = if on_leaf {
490 if sibling_pos >= count {
491 current_hash
492 } else {
493 fetch_hash_by_pos(&self.pool, &self.tables.merkle_leaves, None, sibling_pos).await?
494 }
495 } else if sibling_pos >= count {
496 current_hash
497 } else {
498 fetch_hash_by_pos(&self.pool, &self.tables.merkle_nodes, Some(node_level), sibling_pos).await?
499 };
500
501 siblings.push(sibling_hash);
502 let (left, right) = if pos.is_multiple_of(2) {
503 (current_hash, sibling_hash)
504 } else {
505 (sibling_hash, current_hash)
506 };
507 current_hash = hash_pair(&left, &right);
508 pos /= 2;
509 count = count.div_ceil(2);
510 if on_leaf {
511 on_leaf = false;
512 } else {
513 node_level += 1;
514 }
515 }
516
517 Ok(Some(MerkleProof { leaf_pos, siblings }))
518 }
519
520 pub fn verify_proof(target: &Hash32, proof: &MerkleProof, expected_root: &Hash32, leaf_count: u64) -> bool {
521 if leaf_count == 0 {
522 return false;
523 }
524 let mut hash = *target;
525 let mut pos = proof.leaf_pos;
526 let mut count = leaf_count;
527 for sibling in &proof.siblings {
528 let (left, right) = if pos.is_multiple_of(2) {
529 (hash, *sibling)
530 } else {
531 (*sibling, hash)
532 };
533 hash = hash_pair(&left, &right);
534 pos /= 2;
535 count = count.div_ceil(2);
536 if count == 1 {
537 break;
538 }
539 }
540 hash == *expected_root
541 }
542
543 fn ensure_merkle_enabled(&self) -> Result<()> {
544 if self.merkle == MerkleMode::Enabled {
545 Ok(())
546 } else {
547 Err(Error::Integrity("merkle disabled for sqlite store".into()))
548 }
549 }
550}
551
552fn normalize_namespace(namespace: Option<&str>) -> Result<Option<String>> {
553 let Some(namespace) = namespace else {
554 return Ok(None);
555 };
556 if namespace.is_empty() {
557 return Err(Error::InvalidConfig("store namespace cannot be empty".into()));
558 }
559 if namespace.chars().all(|ch| ch.is_ascii_alphanumeric() || ch == '_') {
560 Ok(Some(namespace.to_string()))
561 } else {
562 Err(Error::InvalidConfig(format!(
563 "store namespace must contain only ASCII letters, digits, or underscores: {namespace}"
564 )))
565 }
566}
567
568fn schema_statements(tables: &TableNames, merkle: MerkleMode) -> Vec<String> {
569 let mut statements = vec![
570 format!(
571 "CREATE TABLE IF NOT EXISTS {} (hash BLOB PRIMARY KEY, kind INTEGER NOT NULL, size INTEGER NOT NULL, stored_size INTEGER NOT NULL, \
572 codec TEXT NOT NULL, content BLOB NOT NULL, created_at INTEGER NOT NULL);",
573 tables.objects
574 ),
575 format!(
576 "CREATE TABLE IF NOT EXISTS {} (file_hash BLOB PRIMARY KEY, recipe_hash BLOB NOT NULL, updated_at INTEGER NOT NULL);",
577 tables.file_recipe_cache
578 ),
579 ];
580
581 if merkle == MerkleMode::Enabled {
582 statements.extend([
583 format!(
584 "CREATE TABLE IF NOT EXISTS {} (hash BLOB PRIMARY KEY, pos INTEGER NOT NULL);",
585 tables.merkle_leaves
586 ),
587 format!(
588 "CREATE UNIQUE INDEX IF NOT EXISTS {} ON {}(pos);",
589 tables.merkle_leaves_pos_index, tables.merkle_leaves
590 ),
591 format!(
592 "CREATE TABLE IF NOT EXISTS {} (level INTEGER NOT NULL, pos INTEGER NOT NULL, hash BLOB NOT NULL, PRIMARY KEY(level, pos));",
593 tables.merkle_nodes
594 ),
595 format!(
596 "CREATE TABLE IF NOT EXISTS {} (key TEXT PRIMARY KEY, value TEXT NOT NULL);",
597 tables.merkle_meta
598 ),
599 ]);
600 }
601
602 statements
603}
604
605fn verified_object_from_row(row: SqliteRow, hash: Hash32) -> Result<VerifiedObject> {
606 let kind = ObjectKind::from_i64(row.get::<i64, _>("kind")).ok_or_else(|| Error::Integrity("invalid kind".into()))?;
607 let size: i64 = row.get("size");
608 let codec = codec_from_str(&row.get::<String, _>("codec"))?;
609 let stored: Vec<u8> = row.get("content");
610 let content = decompress(codec, &stored)?;
611 let expected = u64::try_from(size).map_err(|_| Error::Integrity("negative object size".into()))?;
612 let actual = content.len() as u64;
613 if actual != expected {
614 return Err(Error::ObjectLengthMismatch { expected, actual });
615 }
616 let actual_hash = Hash32::sha3_256(&content);
617 if actual_hash != hash {
618 return Err(Error::ObjectHashMismatch {
619 expected: hash,
620 actual: actual_hash,
621 });
622 }
623 Ok(VerifiedObject {
624 hash,
625 kind,
626 bytes: content,
627 })
628}
629
630impl ObjectSource for SqliteStore {
631 async fn read_object(&self, hash: &Hash32) -> Result<Option<VerifiedObject>> {
632 self.read_verified_object(hash).await
633 }
634}
635
636async fn read_meta_value(pool: &SqlitePool, table: &str, key: &str) -> Result<Option<String>> {
637 let row = sqlx::query(&format!("SELECT value FROM {table} WHERE key=?1"))
638 .bind(key)
639 .fetch_optional(pool)
640 .await?;
641 Ok(row.map(|row| row.get::<String, _>("value")))
642}
643
644async fn read_meta_u64(pool: &SqlitePool, table: &str, key: &str) -> Result<u64> {
645 let value = read_meta_value(pool, table, key)
646 .await?
647 .ok_or_else(|| Error::MissingMeta(key.to_string()))?;
648 value.parse::<u64>().map_err(|_| Error::MissingMeta(key.to_string()))
649}
650
651async fn fetch_hash_by_pos(pool: &SqlitePool, table: &str, level: Option<i64>, pos: u64) -> Result<Hash32> {
652 let row = match level {
653 Some(level) => {
654 sqlx::query(&format!("SELECT hash FROM {table} WHERE level=?1 AND pos=?2"))
655 .bind(level)
656 .bind(pos as i64)
657 .fetch_one(pool)
658 .await?
659 }
660 None => {
661 sqlx::query(&format!("SELECT hash FROM {table} WHERE pos=?1"))
662 .bind(pos as i64)
663 .fetch_one(pool)
664 .await?
665 }
666 };
667 let bytes: Vec<u8> = row.get("hash");
668 Hash32::from_bytes(&bytes)
669}
670
671fn codec_to_str(codec: Codec) -> &'static str {
672 match codec {
673 Codec::Raw => "raw",
674 Codec::Zstd => "zstd",
675 Codec::Brotli => "brotli",
676 }
677}
678
679fn codec_from_str(value: &str) -> Result<Codec> {
680 match value {
681 "raw" => Ok(Codec::Raw),
682 "zstd" => Ok(Codec::Zstd),
683 "brotli" => Ok(Codec::Brotli),
684 other => Err(Error::InvalidCodec(other.to_string())),
685 }
686}
687
688fn hash_pair(left: &Hash32, right: &Hash32) -> Hash32 {
689 let mut bytes = Vec::with_capacity(64);
690 bytes.extend_from_slice(left.as_bytes());
691 bytes.extend_from_slice(right.as_bytes());
692 Hash32::sha3_256(bytes)
693}
694
695#[cfg(test)]
696mod tests {
697 use sqlx::sqlite::SqliteConnectOptions;
698
699 use super::*;
700
701 async fn memory_pool() -> SqlitePool {
702 SqlitePoolOptions::new()
703 .max_connections(1)
704 .connect_with(SqliteConnectOptions::new().filename(":memory:").create_if_missing(true))
705 .await
706 .unwrap()
707 }
708
709 async fn table_exists(pool: &SqlitePool, table: &str) -> bool {
710 sqlx::query("SELECT name FROM sqlite_master WHERE type='table' AND name=?1")
711 .bind(table)
712 .fetch_optional(pool)
713 .await
714 .unwrap()
715 .is_some()
716 }
717
718 #[tokio::test]
719 async fn namespaced_store_creates_prefixed_tables_and_roundtrips_objects() {
720 let pool = memory_pool().await;
721 let store = SqliteStore::from_pool_with_namespace(pool.clone(), "ipg").await.unwrap();
722 assert_eq!(store.namespace(), Some("ipg"));
723 assert!(table_exists(&pool, "ipg_objects").await);
724 assert!(table_exists(&pool, "ipg_file_recipe_cache").await);
725 assert!(!table_exists(&pool, "objects").await);
726
727 let data = b"hello embedded store".to_vec();
728 let hash = Hash32::sha3_256(&data);
729 store.put_chunk(hash, &data, Codec::Zstd).await.unwrap();
730 let loaded = store.read_object(&hash).await.unwrap().unwrap();
731 assert_eq!(loaded.hash, hash);
732 assert_eq!(loaded.kind, ObjectKind::Chunk);
733 assert_eq!(loaded.bytes, data);
734
735 sqlx::query("UPDATE ipg_objects SET size = size + 1 WHERE hash = ?1")
736 .bind(hash.as_bytes().as_ref())
737 .execute(&pool)
738 .await
739 .unwrap();
740 assert!(matches!(store.read_object(&hash).await, Err(Error::ObjectLengthMismatch { .. })));
741 sqlx::query("UPDATE ipg_objects SET size = ?1, codec = 'raw', content = ?2 WHERE hash = ?3")
742 .bind(data.len() as i64)
743 .bind(vec![0_u8; data.len()])
744 .bind(hash.as_bytes().as_ref())
745 .execute(&pool)
746 .await
747 .unwrap();
748 assert!(matches!(store.read_object(&hash).await, Err(Error::ObjectHashMismatch { expected, .. }) if expected == hash));
749 sqlx::query("UPDATE ipg_objects SET kind = 99 WHERE hash = ?1")
750 .bind(hash.as_bytes().as_ref())
751 .execute(&pool)
752 .await
753 .unwrap();
754 assert!(matches!(store.read_object(&hash).await, Err(Error::Integrity(_))));
755 }
756
757 #[tokio::test]
758 async fn store_batch_writes_can_share_host_transaction_boundary() {
759 let pool = memory_pool().await;
760 sqlx::query("CREATE TABLE app_history (id INTEGER PRIMARY KEY, label TEXT NOT NULL);")
761 .execute(&pool)
762 .await
763 .unwrap();
764 let store = SqliteStore::from_pool(pool.clone()).await.unwrap();
765
766 let data = b"tx-object".to_vec();
767 let hash = Hash32::sha3_256(&data);
768 let object = ObjectRecord {
769 hash,
770 kind: ObjectKind::Chunk,
771 decoded_len: data.len() as u64,
772 codec: Codec::Raw,
773 stored_bytes: data.clone(),
774 };
775
776 let mut tx = pool.begin().await.unwrap();
777 sqlx::query("INSERT INTO app_history (label) VALUES (?1)")
778 .bind("draft")
779 .execute(&mut *tx)
780 .await
781 .unwrap();
782 store.put_objects_batch_tx(&mut tx, &[object]).await.unwrap();
783 tx.rollback().await.unwrap();
784
785 let app_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM app_history")
786 .fetch_one(&pool)
787 .await
788 .unwrap();
789 let object_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM objects").fetch_one(&pool).await.unwrap();
790 assert_eq!(app_count, 0);
791 assert_eq!(object_count, 0);
792 }
793
794 #[tokio::test]
795 async fn merkle_can_be_disabled_for_embedded_store() {
796 let pool = memory_pool().await;
797 let store = SqliteStore::from_pool_with_options(
798 pool.clone(),
799 SqliteStoreOptions {
800 namespace: Some("ipg".into()),
801 merkle: MerkleMode::Disabled,
802 },
803 )
804 .await
805 .unwrap();
806 assert_eq!(store.merkle_mode(), MerkleMode::Disabled);
807 assert!(!table_exists(&pool, "ipg_merkle_leaves").await);
808 assert!(!table_exists(&pool, "ipg_merkle_nodes").await);
809 assert!(!table_exists(&pool, "ipg_merkle_meta").await);
810
811 let error = store.rebuild_merkle_index().await.unwrap_err();
812 assert!(error.to_string().contains("merkle disabled"));
813 }
814
815 #[tokio::test]
816 async fn merkle_enabled_store_can_build_and_verify_proof() {
817 let pool = memory_pool().await;
818 let store = SqliteStore::from_pool_with_options(
819 pool,
820 SqliteStoreOptions {
821 namespace: Some("ipg".into()),
822 merkle: MerkleMode::Enabled,
823 },
824 )
825 .await
826 .unwrap();
827
828 for item in [b"foo".as_slice(), b"bar".as_slice(), b"baz".as_slice()] {
829 let hash = Hash32::sha3_256(item);
830 store.put_chunk(hash, item, Codec::Raw).await.unwrap();
831 }
832
833 let target = Hash32::sha3_256(b"bar");
834 let summary = store.rebuild_merkle_index().await.unwrap();
835 let proof = store.prove_membership(&target).await.unwrap().unwrap();
836 assert!(SqliteStore::verify_proof(&target, &proof, &summary.root, summary.leaf_count));
837 }
838}