dpm_store/sqlite/
history.rs

1use super::{
2    SqliteStore, lineage,
3    snapshot::{column, decode, load_cached_blocking, revision_from_sql},
4};
5use crate::{
6    HistoryEntry, HistoryPage, RecordedOperation, StoreError, StoredRecord, error::database_error,
7};
8use dpm_engine::Operation;
9use dpm_model::OperationId;
10use rusqlite::{Connection, OptionalExtension, params};
11use std::path::Path;
12
13impl SqliteStore {
14    /// Read at most 1000 operations after a local sequence cursor without modifying history.
15    pub fn history_blocking(
16        &self,
17        after_sequence: u64,
18        limit: u16,
19    ) -> Result<HistoryPage, StoreError> {
20        let transaction = self
21            .connection
22            .unchecked_transaction()
23            .map_err(database_error(&self.path, "begin history query"))?;
24        // The page's revision comes from the snapshot in this transaction; the decode cache makes
25        // that a byte comparison while the snapshot is unchanged.
26        let plan = load_cached_blocking(&transaction, &self.path, &self.cache)?
27            .ok_or_else(|| StoreError::NotInitialized(self.path.clone()))?;
28        let lineage = lineage::read_blocking(&transaction, &self.path)?;
29        let mut entries = Vec::new();
30        if let Ok(cursor) = i64::try_from(after_sequence) {
31            let mut statement = transaction
32                .prepare(&format!(
33                    "{OPERATION_COLUMNS} WHERE sequence > ?1 ORDER BY sequence LIMIT ?2"
34                ))
35                .map_err(database_error(&self.path, "prepare history query"))?;
36            let mut rows = statement
37                .query(params![cursor, limit.min(1000)])
38                .map_err(database_error(&self.path, "query history"))?;
39            while let Some(row) = rows
40                .next()
41                .map_err(database_error(&self.path, "read history"))?
42            {
43                entries.push(read_entry(row, &self.path)?);
44            }
45        }
46        transaction
47            .commit()
48            .map_err(database_error(&self.path, "finish history query"))?;
49        Ok(HistoryPage {
50            revision: plan.revision,
51            lineage_id: Some(lineage.lineage_id),
52            next_after_sequence: entries
53                .last()
54                .map_or(after_sequence, |entry| entry.sequence),
55            entries,
56        })
57    }
58
59    /// The operation recorded under this identity, if any, whichever lineage recorded it.
60    pub fn recorded_operation_blocking(
61        &self,
62        id: OperationId,
63    ) -> Result<Option<HistoryEntry>, StoreError> {
64        recorded_blocking(&self.connection, &self.path, id)
65    }
66}
67
68/// Look an operation identity up through the unique index on `operation_id`.
69pub(super) fn recorded_blocking(
70    connection: &Connection,
71    path: &Path,
72    id: OperationId,
73) -> Result<Option<HistoryEntry>, StoreError> {
74    let mut statement = connection
75        .prepare(&format!("{OPERATION_COLUMNS} WHERE operation_id = ?1"))
76        .map_err(database_error(path, "prepare operation lookup"))?;
77    statement
78        .query_row(params![id.to_string()], |row| Ok(read_entry(row, path)))
79        .optional()
80        .map_err(database_error(path, "look up operation"))?
81        .transpose()
82}
83
84/// Selects the columns [`read_entry`] expects, in its order.
85pub(crate) const OPERATION_COLUMNS: &str = "SELECT sequence, operation_id, base_revision, \
86     resulting_revision, actor_json, timestamp, command_json, workspace_id, lineage_id \
87     FROM operations";
88
89/// Decode one row selected with [`OPERATION_COLUMNS`], attributing damage to its sequence.
90pub(crate) fn read_entry(row: &rusqlite::Row<'_>, path: &Path) -> Result<HistoryEntry, StoreError> {
91    let sequence: u64 = row
92        .get(0)
93        .map_err(database_error(path, "read history cursor"))?;
94    let record = StoredRecord::Operation { sequence };
95    let text = |index, field| column::<String>(row, index, path, record, field);
96    let revision =
97        |index, field| column::<i64>(row, index, path, record, field).map(revision_from_sql);
98    // Identifier and timestamp columns hold bare strings; decoding them as JSON strings reuses
99    // the serde formats the operation itself declares.
100    let scalar =
101        |index, field| text(index, field).map(|value| serde_json::Value::String(value).to_string());
102    Ok(HistoryEntry {
103        sequence,
104        operation: RecordedOperation {
105            operation: Operation {
106                id: decode(&scalar(1, "operation_id")?, path, record, "operation_id")?,
107                base_revision: revision(2, "base_revision")?,
108                resulting_revision: revision(3, "resulting_revision")?,
109                actor: decode(&text(4, "actor_json")?, path, record, "actor_json")?,
110                timestamp: decode(&scalar(5, "timestamp")?, path, record, "timestamp")?,
111                command: decode(&text(6, "command_json")?, path, record, "command_json")?,
112            },
113            workspace_id: decode(&scalar(7, "workspace_id")?, path, record, "workspace_id")?,
114            lineage_id: decode(&scalar(8, "lineage_id")?, path, record, "lineage_id")?,
115        },
116    })
117}
118
119#[cfg(test)]
120mod tests;
dpm_store/sqlite/history.rs

dpm_store/sqlite/
history.rs
