Skip to main content

urma_node/
archive.rs

1//! Transactional archive owned by one process. Readers use the owning engine or
2//! its API; consumers never open the service database for independent writes.
3
4use crate::{Error, model::*};
5use bitcoin::{BlockHash, block::Header, consensus::serialize};
6use rusqlite::{Connection, OptionalExtension, params};
7use std::{
8    fs::{File, OpenOptions},
9    path::Path,
10    time::Duration,
11};
12
13mod cache;
14mod names;
15mod objects;
16mod peers;
17mod projections;
18pub(crate) use objects::ObjectMember;
19
20pub struct Archive {
21    // Keep the OS lock until after the connection has closed.
22    db: Connection,
23    _lock: File,
24    config: ArchiveConfig,
25    // One restored registry at a time. SQL remains authoritative; errors drop
26    // the working copy so staged in-memory changes cannot outlive rollback.
27    names_cache: Option<urma_names::engine::Registry>,
28}
29
30impl Archive {
31    /// Opens only local state. The directory is a local-filesystem archive, not
32    /// a network share; SQLite WAL relies on local shared-memory coordination.
33    pub fn open(directory: &Path, config: ArchiveConfig) -> Result<Self, Error> {
34        if config.application_start.height > i64::MAX as u64 {
35            return Err(Error::Invalid(
36                "application start exceeds storage range".into(),
37            ));
38        }
39        match directory.parent() {
40            Some(parent) => std::fs::create_dir_all(parent)?,
41            None => return Err(Error::Invalid("archive directory has no parent".into())),
42        }
43        match urma_io::create_private_directory(directory) {
44            Ok(()) => (),
45            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => (),
46            Err(e) => return Err(e.into()),
47        }
48        if std::fs::symlink_metadata(directory)?
49            .file_type()
50            .is_symlink()
51        {
52            return Err(Error::Invalid(
53                "archive directory must not be a symlink".into(),
54            ));
55        }
56        let lock_path = directory.join("owner.lock");
57        for path in [&lock_path, &directory.join("archive.sqlite3")] {
58            match std::fs::symlink_metadata(path) {
59                Ok(meta) if !meta.is_file() => {
60                    return Err(Error::Invalid("archive files must be regular files".into()));
61                }
62                Ok(_) => (),
63                Err(e) if e.kind() == std::io::ErrorKind::NotFound => (),
64                Err(e) => return Err(e.into()),
65            }
66        }
67        let lock = OpenOptions::new()
68            .create(true)
69            .truncate(false)
70            .read(true)
71            .write(true)
72            .open(lock_path)?;
73        lock.try_lock().map_err(|e| Error::Busy(e.to_string()))?;
74        let mut db = Connection::open(directory.join("archive.sqlite3"))?;
75        db.busy_timeout(Duration::from_secs(5))?;
76        db.execute_batch(
77            "PRAGMA foreign_keys=ON; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;",
78        )?;
79        let version: u32 = db.pragma_query_value(None, "user_version", |r| r.get(0))?;
80        let encoded = serde_json::to_string(&config)?;
81        if version == 0 {
82            let tx = db.transaction()?;
83            tx.execute_batch(include_str!("schema.sql"))?;
84            tx.execute("INSERT INTO config VALUES(1,?1)", [&encoded])?;
85            tx.commit()?;
86        } else if !(1..=5).contains(&version) {
87            return Err(Error::Invalid(format!(
88                "unsupported archive schema {version}"
89            )));
90        }
91        let saved: String =
92            db.query_row("SELECT value FROM config WHERE singleton=1", [], |r| {
93                r.get(0)
94            })?;
95        if saved != encoded {
96            return Err(Error::Invalid(
97                "archive configuration differs; explicit migration required".into(),
98            ));
99        }
100        // Configuration identity is checked before any existing archive migration.
101        // Schema 1 remains intact if the additive migration fails.
102        migrate(&mut db, version)?;
103        Ok(Self {
104            db,
105            _lock: lock,
106            config,
107            names_cache: None,
108        })
109    }
110
111    pub fn config(&self) -> &ArchiveConfig {
112        &self.config
113    }
114
115    pub fn status(&self) -> Result<ArchiveStatus, Error> {
116        let generation = generation(&self.db)?;
117        let selected_tip = self
118            .db
119            .query_row(
120                "SELECT height,hash FROM selected ORDER BY height DESC LIMIT 1",
121                [],
122                point,
123            )
124            .optional()?;
125        let indexed_through = self.indexed_through()?;
126        let scanned_blocks = self.db.query_row(
127            "SELECT count(*) FROM scans c JOIN selected s ON s.hash=c.block_hash WHERE c.rules=?1",
128            [SCAN_RULES_VERSION],
129            |r| unsigned(r, 0),
130        )?;
131        let unresolved_dependencies = self.db.query_row(
132            "SELECT count(*) FROM dependencies d JOIN selected s ON s.hash=d.block_hash",
133            [],
134            |r| unsigned(r, 0),
135        )?;
136        let stored_transactions =
137            self.db
138                .query_row("SELECT count(*) FROM transactions", [], |r| unsigned(r, 0))?;
139        let mut status = ArchiveStatus::default();
140        status.api_version = API_VERSION;
141        status.chain_genesis = self
142            .config
143            .chain
144            .genesis()
145            .map_err(|e| Error::Invalid(e.to_string()))?
146            .0;
147        status.application_start = self.config.application_start;
148        status.retention = self.config.retention;
149        status.generation = generation;
150        status.selected_tip = selected_tip;
151        status.indexed_through = indexed_through;
152        status.scanned_blocks = scanned_blocks;
153        status.unresolved_dependencies = unresolved_dependencies;
154        status.stored_transactions = stored_transactions;
155        status.scan_rules_version = SCAN_RULES_VERSION;
156        status.peer_synced_through =
157            peers::synced_through(&self.db, self.config.application_start.height)?;
158        status.peer_omissions = peers::omission_count(&self.db)?;
159        status.unaudited_records = peers::unaudited_count(&self.db)?;
160        Ok(status)
161    }
162
163    /// Contiguous fully scanned, dependency-complete coverage from application start.
164    fn indexed_through(&self) -> Result<Option<ChainPoint>, Error> {
165        let start = self.config.application_start.height;
166        let mut next = start;
167        let mut indexed_through = None;
168        let mut rows = self.db.prepare(
169            "SELECT s.height,s.hash FROM selected s JOIN scans c ON c.block_hash=s.hash
170             WHERE s.height>=?1 AND c.rules=?2 AND c.unresolved=0 ORDER BY s.height",
171        )?;
172        for row in rows.query_map(params![sql_integer(start)?, SCAN_RULES_VERSION], point)? {
173            let current = row?;
174            if current.height != next {
175                break;
176            }
177            next += 1;
178            indexed_through = Some(current);
179        }
180        Ok(indexed_through)
181    }
182
183    pub fn ticket(&self, height: u64) -> Result<ScanTicket, Error> {
184        let point = self
185            .db
186            .query_row(
187                "SELECT height,hash FROM selected WHERE height=?1",
188                [sql_integer(height)?],
189                point,
190            )
191            .optional()?
192            .ok_or_else(|| Error::Invalid("block is not in the selected view".into()))?;
193        Ok(ScanTicket {
194            generation: generation(&self.db)?,
195            point,
196        })
197    }
198
199    pub(crate) fn header_at(&self, height: u64) -> Result<Option<Header>, Error> {
200        let raw: Option<Vec<u8>> = self
201            .db
202            .query_row(
203                "SELECT h.bytes FROM selected s JOIN headers h ON h.hash=s.hash WHERE s.height=?1",
204                [sql_integer(height)?],
205                |r| r.get(0),
206            )
207            .optional()?;
208        raw.map(|raw| {
209            bitcoin::consensus::deserialize(&raw)
210                .map_err(|e| Error::Invalid(format!("stored header: {e}")))
211        })
212        .transpose()
213    }
214
215    /// Only the chain coordinator calls this, after the header engine selected
216    /// the branch. This validates archive continuity, not difficulty/chainwork.
217    /// Untrusted peers and API clients cannot supply a branch to this method.
218    pub(crate) fn select(&mut self, start: u64, headers: &[Header]) -> Result<(), Error> {
219        if headers.is_empty() || start < self.config.application_start.height {
220            return Err(Error::Invalid("empty or out-of-domain chain update".into()));
221        }
222        let end = start
223            .checked_add(headers.len() as u64 - 1)
224            .filter(|h| *h <= i64::MAX as u64)
225            .ok_or_else(|| Error::Invalid("chain update height overflow".into()))?;
226        if start == self.config.application_start.height {
227            if headers[0].block_hash() != self.config.application_start.hash {
228                return Err(Error::Invalid(
229                    "selected chain does not contain configured application start".into(),
230                ));
231            }
232        } else {
233            let previous = self
234                .header_at(start - 1)?
235                .ok_or_else(|| Error::Invalid("chain update has a gap".into()))?;
236            if headers[0].prev_blockhash != previous.block_hash() {
237                return Err(Error::Invalid(
238                    "chain update does not join selected parent".into(),
239                ));
240            }
241        }
242        for pair in headers.windows(2) {
243            if pair[1].prev_blockhash != pair[0].block_hash() {
244                return Err(Error::Invalid("noncontiguous selected headers".into()));
245            }
246        }
247        let mut changed_at = None;
248        for (offset, header) in headers.iter().enumerate() {
249            let height = start + offset as u64;
250            if self.header_at(height)?.is_some_and(|old| old != *header) {
251                changed_at = Some(height);
252                break;
253            }
254        }
255        let tx = self.db.transaction()?;
256        if let Some(height) = changed_at {
257            tx.execute(
258                "DELETE FROM selected WHERE height>=?1",
259                [sql_integer(height)?],
260            )?;
261            tx.execute(
262                "UPDATE state SET generation=generation+1 WHERE singleton=1",
263                [],
264            )?;
265            tx.execute("INSERT INTO events(generation,kind,height) SELECT generation,'reorg',?1 FROM state", [sql_integer(height)?])?;
266            peers::reorg(&tx, height)?;
267        }
268        for (offset, header) in headers.iter().enumerate() {
269            let hash = header.block_hash().to_string();
270            tx.execute(
271                "INSERT OR IGNORE INTO headers VALUES(?1,?2)",
272                params![hash, serialize(header)],
273            )?;
274            tx.execute("INSERT INTO selected VALUES(?1,?2) ON CONFLICT(height) DO UPDATE SET hash=excluded.hash",
275                params![sql_integer(start + offset as u64)?, hash])?;
276        }
277        tx.execute("INSERT INTO events(generation,kind,height,block_hash) SELECT generation,'chain',?1,?2 FROM state",
278            params![sql_integer(end)?, headers.last().expect("nonempty checked").block_hash().to_string()])?;
279        tx.commit()?;
280        Ok(())
281    }
282
283    pub(crate) fn check_ticket(&self, ticket: ScanTicket) -> Result<(), Error> {
284        if generation(&self.db)? != ticket.generation
285            || self.ticket(ticket.point.height)?.point != ticket.point
286        {
287            return Err(Error::Stale);
288        }
289        Ok(())
290    }
291
292    pub(crate) fn next_scan(&self) -> Result<Option<ScanTicket>, Error> {
293        let height: Option<u64> = self
294            .db
295            .query_row(
296                "SELECT s.height FROM selected s LEFT JOIN scans c ON c.block_hash=s.hash
297             WHERE c.block_hash IS NULL OR c.rules!=?1 ORDER BY s.height LIMIT 1",
298                [SCAN_RULES_VERSION],
299                |r| unsigned(r, 0),
300            )
301            .optional()?;
302        height.map(|h| self.ticket(h)).transpose()
303    }
304
305    pub(crate) fn transaction(&self, txid: bitcoin::Txid) -> Result<Option<Vec<u8>>, Error> {
306        Ok(self
307            .db
308            .query_row(
309                "SELECT raw FROM transactions WHERE txid=?1 LIMIT 1",
310                [txid.to_string()],
311                |r| r.get(0),
312            )
313            .optional()?)
314    }
315
316    pub(crate) fn location(&self, txid: bitcoin::Txid) -> Result<Option<ChainPoint>, Error> {
317        Ok(self.db.query_row(
318            "SELECT s.height,s.hash FROM locators l JOIN selected s ON s.hash=l.block_hash WHERE l.txid=?1 ORDER BY s.height LIMIT 1",
319            [txid.to_string()], point,
320        ).optional()?)
321    }
322
323    /// Placements of records this node holds in fully scanned selected blocks
324    /// within [from, to], strictly after `after` as (height, position),
325    /// ascending; `limit` rows at most. Rows never exceed `scanned_through`,
326    /// this node's contiguous indexed prefix: a scanned block still waiting on
327    /// a dependency sits above that ceiling and is not listed until it joins
328    /// the prefix. Pending reveals appear once their parent resolves. Evicted records drop out; only LeafManifest/DataPart
329    /// are ever evicted, and those are exempt from omission detection.
330    pub(crate) fn inventory_page(
331        &self,
332        from: u64,
333        to: u64,
334        after: Option<(u64, u32)>,
335        limit: u32,
336    ) -> Result<InventoryRows, Error> {
337        let scanned_from = self.config.application_start.height;
338        let Some(ceiling) = self.indexed_through()? else {
339            return Ok(InventoryRows {
340                scanned_from,
341                scanned_through: None,
342                entries: Vec::new(),
343                more: false,
344            });
345        };
346        let (after_height, after_position) = match after {
347            Some((height, position)) => (sql_integer(height)?, i64::from(position)),
348            None => (-1, -1),
349        };
350        let mut statement = self.db.prepare(
351            "SELECT s.height,s.hash,l.position,l.txid,l.wtxid FROM records p
352             JOIN locators l ON l.block_hash=p.block_hash AND l.position=p.position
353             JOIN selected s ON s.hash=p.block_hash
354             JOIN scans c ON c.block_hash=p.block_hash AND c.rules=?5
355             WHERE s.height>=?1 AND s.height<=?2 AND s.height<=?7
356               AND (s.height>?3 OR (s.height=?3 AND l.position>?4))
357             ORDER BY s.height,l.position LIMIT ?6",
358        )?;
359        let wanted = usize::try_from(limit).map_err(|e| Error::Invalid(e.to_string()))?;
360        let bound = params![
361            sql_integer(from)?,
362            sql_integer(to)?,
363            after_height,
364            after_position,
365            SCAN_RULES_VERSION,
366            i64::from(limit) + 1,
367            sql_integer(ceiling.height)?
368        ];
369        let mut entries = Vec::new();
370        for row in statement.query_map(bound, |row| {
371            Ok((
372                point(row)?,
373                row.get::<_, i64>(2)?,
374                row.get::<_, String>(3)?,
375                row.get::<_, String>(4)?,
376            ))
377        })? {
378            let (block, position, txid, wtxid) = row?;
379            let invalid = |e: String| Error::Invalid(format!("archived locator: {e}"));
380            entries.push((
381                block,
382                u32::try_from(position).map_err(|e| invalid(e.to_string()))?,
383                txid.parse()
384                    .map_err(|e: bitcoin::hex::HexToArrayError| invalid(e.to_string()))?,
385                wtxid
386                    .parse()
387                    .map_err(|e: bitcoin::hex::HexToArrayError| invalid(e.to_string()))?,
388            ));
389        }
390        let more = entries.len() > wanted;
391        entries.truncate(wanted);
392        Ok(InventoryRows {
393            scanned_from,
394            scanned_through: Some(ceiling.height),
395            entries,
396            more,
397        })
398    }
399
400    /// Orphan bytes remain archived but are never returned as selected evidence.
401    pub(crate) fn record(&self, txid: bitcoin::Txid) -> Result<Option<ArchivedRecord>, Error> {
402        let row: Option<(ChainPoint, u32, String, Vec<u8>, Vec<u8>)> = self
403            .db
404            .query_row(
405                "SELECT s.height,s.hash,r.position,r.wtxid,r.proof,t.raw FROM records r
406             JOIN selected s ON s.hash=r.block_hash
407             JOIN transactions reveal ON reveal.wtxid=r.wtxid
408             JOIN transactions t ON t.wtxid=r.commit_wtxid
409             WHERE reveal.txid=?1 ORDER BY s.height,r.position LIMIT 1",
410                [txid.to_string()],
411                |r| Ok((point(r)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?)),
412            )
413            .optional()?;
414        let Some((block, position, wtxid, proof, commit)) = row else {
415            return Ok(None);
416        };
417        Ok(Some(ArchivedRecord {
418            receipt: RecordReceipt {
419                generation: generation(&self.db)?,
420                block,
421                position,
422                txid,
423                wtxid: wtxid
424                    .parse()
425                    .map_err(|e| Error::Invalid(format!("stored WTXID: {e}")))?,
426            },
427            package: RecordPackage {
428                version: API_VERSION,
429                chain_genesis: self
430                    .config
431                    .chain
432                    .genesis()
433                    .map_err(|e| Error::Invalid(e.to_string()))?
434                    .0,
435                block,
436                reveal: serde_json::from_slice(&proof)?,
437                commit,
438            },
439        }))
440    }
441
442    /// One distinct selected parent absent from every cached witness variant.
443    /// Cursor comparison uses persisted display hex, not Txid's byte ordering.
444    pub(crate) fn next_missing_dependency(
445        &self,
446        after: Option<bitcoin::Txid>,
447    ) -> Result<Option<(bitcoin::Txid, u64)>, Error> {
448        let id: Option<String> = self
449            .db
450            .query_row(
451                "SELECT DISTINCT d.commit_txid FROM dependencies d
452             JOIN selected s ON s.hash=d.block_hash
453             WHERE NOT EXISTS(SELECT 1 FROM transactions t WHERE t.txid=d.commit_txid)
454             AND (?1 IS NULL OR d.commit_txid>?1)
455             ORDER BY d.commit_txid LIMIT 1",
456                [after.map(|id| id.to_string())],
457                |row| row.get(0),
458            )
459            .optional()?;
460        id.map(|id| {
461            let id = id
462                .parse()
463                .map_err(|error| Error::Invalid(format!("stored parent TXID: {error}")))?;
464            Ok((id, generation(&self.db)?))
465        })
466        .transpose()
467    }
468
469    pub(crate) fn check_generation(&self, expected: u64) -> Result<(), Error> {
470        if generation(&self.db)? != expected {
471            return Err(Error::Stale);
472        }
473        Ok(())
474    }
475
476    pub(crate) fn pending_for(
477        &self,
478        txid: bitcoin::Txid,
479    ) -> Result<Vec<(ScanTicket, u32, urma_chain::inclusion::TransactionProof)>, Error> {
480        let mut stmt = self.db.prepare(
481            "SELECT s.height,s.hash,d.position,d.proof FROM dependencies d
482             JOIN selected s ON s.hash=d.block_hash WHERE d.commit_txid=?1
483             ORDER BY s.height,d.position LIMIT 128",
484        )?;
485        let rows = stmt.query_map([txid.to_string()], |r| {
486            Ok((point(r)?, r.get::<_, u32>(2)?, r.get::<_, Vec<u8>>(3)?))
487        })?;
488        let epoch = generation(&self.db)?;
489        rows.map(|row| {
490            let (point, position, proof) = row?;
491            Ok((
492                ScanTicket {
493                    generation: epoch,
494                    point,
495                },
496                position,
497                serde_json::from_slice(&proof)?,
498            ))
499        })
500        .collect()
501    }
502
503    pub(crate) fn cached_dependency(&self) -> Result<Option<Vec<u8>>, Error> {
504        Ok(self
505            .db
506            .query_row(
507                "SELECT t.raw FROM dependencies d JOIN selected s ON s.hash=d.block_hash
508             JOIN transactions t ON t.txid=d.commit_txid ORDER BY s.height,d.position LIMIT 1",
509                [],
510                |row| row.get(0),
511            )
512            .optional()?)
513    }
514
515    /// Completion changes only evidence from blocks already fully scanned.
516    /// A supplied parent cannot create a scan mark or a transaction placement.
517    pub(crate) fn finish_dependencies(
518        &mut self,
519        completed: Vec<(ScanTicket, u32, Option<crate::scan::Record>)>,
520    ) -> Result<usize, Error> {
521        for (ticket, _, _) in &completed {
522            self.check_ticket(*ticket)?;
523        }
524        let tx = self.db.transaction()?;
525        let mut count = 0;
526        for (ticket, position, record) in completed {
527            let hash = ticket.point.hash.to_string();
528            let removed = tx.execute(
529                "DELETE FROM dependencies WHERE block_hash=?1 AND position=?2",
530                params![hash, position],
531            )?;
532            if removed == 0 {
533                continue;
534            }
535            if let Some(record) = record {
536                store_record(&tx, &hash, &record)?;
537            }
538            tx.execute("UPDATE scans SET unresolved=(SELECT count(*) FROM dependencies WHERE block_hash=?1) WHERE block_hash=?1", [&hash])?;
539            tx.execute("INSERT INTO events(generation,kind,height,block_hash) VALUES(?1,'dependency',?2,?3)",
540                params![sql_integer(ticket.generation)?, sql_integer(ticket.point.height)?, hash])?;
541            count += 1;
542        }
543        tx.commit()?;
544        Ok(count)
545    }
546
547    pub(crate) fn admit_record(
548        &mut self,
549        ticket: ScanTicket,
550        record: crate::scan::Record,
551    ) -> Result<RecordReceipt, Error> {
552        self.check_ticket(ticket)?;
553        let receipt = RecordReceipt {
554            generation: ticket.generation,
555            block: ticket.point,
556            position: record.position,
557            txid: record.verified.txid(),
558            wtxid: record.wtxid,
559        };
560        let tx = self.db.transaction()?;
561        store_record(&tx, &ticket.point.hash.to_string(), &record)?;
562        tx.execute(
563            "INSERT INTO events(generation,kind,height,block_hash) VALUES(?1,'record',?2,?3)",
564            params![
565                sql_integer(ticket.generation)?,
566                sql_integer(ticket.point.height)?,
567                ticket.point.hash.to_string()
568            ],
569        )?;
570        // No locators, scan marks, dependency resolution, or completeness claims
571        // are inferred from a selective package.
572        tx.commit()?;
573        Ok(receipt)
574    }
575
576    pub(crate) fn commit_scan(&mut self, batch: crate::scan::ScanBatch) -> Result<(), Error> {
577        self.check_ticket(batch.ticket)?;
578        let point = batch.ticket.point;
579        let hash = point.hash.to_string();
580        let tx = self.db.transaction()?;
581        for (position, (txid, wtxid)) in batch.locators.iter().enumerate() {
582            tx.execute("INSERT INTO locators VALUES(?1,?2,?3,?4) ON CONFLICT(block_hash,position) DO UPDATE SET txid=excluded.txid,wtxid=excluded.wtxid",
583                params![hash, sql_integer(position as u64)?, txid.to_string(), wtxid.to_string()])?;
584        }
585        // One transaction includes all records, dependencies, progress and event.
586        // An I/O/constraint failure rolls it all back, including progress.
587        tx.execute("DELETE FROM dependencies WHERE block_hash=?1", [&hash])?;
588        peers::record_omissions(&tx, &batch)?;
589        for record in batch.records {
590            store_record(&tx, &hash, &record)?;
591        }
592        let unresolved = batch.dependencies.len() as u64;
593        for dependency in batch.dependencies {
594            tx.execute(
595                "INSERT INTO dependencies VALUES(?1,?2,?3,?4)",
596                params![
597                    hash,
598                    dependency.position,
599                    dependency.commit.to_string(),
600                    serde_json::to_vec(&dependency.proof)?
601                ],
602            )?;
603        }
604        tx.execute("INSERT INTO scans VALUES(?1,?2,?3) ON CONFLICT(block_hash) DO UPDATE SET rules=excluded.rules,unresolved=excluded.unresolved",
605            params![hash, SCAN_RULES_VERSION, sql_integer(unresolved)?])?;
606        tx.execute("INSERT INTO events(generation,kind,height,block_hash) SELECT generation,'scan',?1,?2 FROM state",
607            params![sql_integer(point.height)?, hash])?;
608        tx.commit()?;
609        Ok(())
610    }
611}
612
613fn migrate(db: &mut Connection, version: u32) -> Result<(), Error> {
614    if version >= 5 {
615        return Ok(());
616    }
617    let tx = db.transaction()?;
618    if version < 2 {
619        tx.execute_batch(objects::SCHEMA)?;
620    } else if version == 2 {
621        objects::migrate_v2(&tx)?;
622    }
623    if version < 4 {
624        tx.execute_batch(names::SCHEMA)?;
625    }
626    tx.execute_batch(peers::SCHEMA)?;
627    tx.commit()?;
628    Ok(())
629}
630
631/// One inventory page as stored: (block, position, txid, wtxid) rows.
632pub(crate) struct InventoryRows {
633    pub(crate) scanned_from: u64,
634    pub(crate) scanned_through: Option<u64>,
635    pub(crate) entries: Vec<(ChainPoint, u32, bitcoin::Txid, bitcoin::Wtxid)>,
636    pub(crate) more: bool,
637}
638
639fn generation(db: &Connection) -> Result<u64, Error> {
640    Ok(
641        db.query_row("SELECT generation FROM state WHERE singleton=1", [], |r| {
642            unsigned(r, 0)
643        })?,
644    )
645}
646
647fn point(row: &rusqlite::Row<'_>) -> rusqlite::Result<ChainPoint> {
648    let hash: String = row.get(1)?;
649    Ok(ChainPoint {
650        height: unsigned(row, 0)?,
651        hash: hash.parse::<BlockHash>().map_err(|e| {
652            rusqlite::Error::FromSqlConversionFailure(1, rusqlite::types::Type::Text, Box::new(e))
653        })?,
654    })
655}
656
657fn sql_integer(value: u64) -> Result<i64, Error> {
658    i64::try_from(value).map_err(|_| Error::Invalid("integer exceeds archive range".into()))
659}
660
661fn unsigned(row: &rusqlite::Row<'_>, column: usize) -> rusqlite::Result<u64> {
662    let value: i64 = row.get(column)?;
663    u64::try_from(value).map_err(|e| {
664        rusqlite::Error::FromSqlConversionFailure(
665            column,
666            rusqlite::types::Type::Integer,
667            Box::new(e),
668        )
669    })
670}
671
672fn store_record(db: &Connection, hash: &str, record: &crate::scan::Record) -> Result<(), Error> {
673    let wtxid = record.wtxid.to_string();
674    let commit_wtxid = record.commit.compute_wtxid().to_string();
675    db.execute(
676        "INSERT OR IGNORE INTO transactions VALUES(?1,?2,?3)",
677        params![
678            wtxid,
679            record.verified.txid().to_string(),
680            record.proof.transaction
681        ],
682    )?;
683    db.execute(
684        "INSERT OR IGNORE INTO transactions VALUES(?1,?2,?3)",
685        params![
686            commit_wtxid,
687            record.commit.compute_txid().to_string(),
688            record.raw_commit
689        ],
690    )?;
691    db.execute("INSERT INTO records VALUES(?1,?2,?3,?4,?5,?6,?7) ON CONFLICT(block_hash,position) DO UPDATE SET wtxid=excluded.wtxid,author=excluded.author,record=excluded.record,commit_wtxid=excluded.commit_wtxid,proof=excluded.proof",
692        params![hash, record.position, wtxid, record.verified.author().to_string(), record.verified.record_bytes(), commit_wtxid, serde_json::to_vec(&record.proof)?])?;
693    Ok(())
694}