coven_database/
circle_operation_records.rs

1use super::*;
2
3use coven_protocol::circle_journal::{
4    CircleOperationIntent, CircleOperationJournal, CircleOperationProgress, PreparedCircleOperation,
5};
6
7/// What `circle_operations.prepared` holds: the operation as prepared and the
8/// intent that named it.
9///
10/// These two travel together because neither changes while the operation
11/// publishes — the phase beside them and the upload rows below them are what
12/// move. The objects appear here as references; their bytes are in the payload
13/// spool.
14#[derive(serde::Serialize, serde::Deserialize)]
15#[serde(deny_unknown_fields)]
16struct PreparedCircleOperationPayload {
17    operation_id: coven_protocol::circle::CircleOperationId,
18    circle_id: coven_protocol::circle::CircleId,
19    intent: CircleOperationIntent,
20    operation: PreparedCircleOperation,
21}
22
23pub struct PreparedCircleOperationRow {
24    pub operation_id: String,
25    pub circle_id: String,
26    pub prepared: Vec<u8>,
27    pub phase: String,
28}
29
30impl PreparedCircleOperationRow {
31    pub fn from_journal(journal: &CircleOperationJournal) -> Result<Self, DbError> {
32        journal.validate_identity().map_err(DbError::from)?;
33        Ok(Self {
34            operation_id: journal.operation_id.as_str().to_string(),
35            circle_id: journal.circle_id.to_string(),
36            prepared: prepared_circle_operation_payload(journal)?,
37            phase: circle_operation_phase_json(&journal.progress)?,
38        })
39    }
40}
41
42/// The bytes `circle_operations.prepared` holds for one operation.
43pub(crate) fn prepared_circle_operation_payload(
44    journal: &CircleOperationJournal,
45) -> Result<Vec<u8>, DbError> {
46    serde_json::to_vec(&PreparedCircleOperationPayload {
47        operation_id: journal.operation_id.clone(),
48        circle_id: journal.circle_id,
49        intent: journal.intent.clone(),
50        operation: journal.operation.clone(),
51    })
52    .map_err(|error| DbError::context("serialize prepared circle operation", error))
53}
54
55pub(crate) fn circle_operation_phase_json(
56    progress: &CircleOperationProgress,
57) -> Result<String, DbError> {
58    serde_json::to_string(progress)
59        .map_err(|error| DbError::context("serialize circle operation phase", error))
60}
61
62/// Rebuild one operation from the three places it is stored.
63///
64/// The stored ids are checked against the ones inside `prepared` rather than
65/// trusted: the columns are what queries dispatch on, so a row whose payload
66/// names a different operation or circle would route work at one identity and
67/// perform it at another.
68pub fn parse_circle_operation_row(
69    stored_operation_id: &str,
70    stored_circle_id: &str,
71    prepared: &[u8],
72    phase: &str,
73    uploaded: BTreeSet<String>,
74) -> Result<CircleOperationJournal, DbError> {
75    let payload: PreparedCircleOperationPayload = serde_json::from_slice(prepared)
76        .map_err(|error| DbError::context("parse prepared circle operation", error))?;
77    let progress: CircleOperationProgress = serde_json::from_str(phase)
78        .map_err(|error| DbError::context("parse circle operation phase", error))?;
79    if payload.operation_id.as_str() != stored_operation_id {
80        return Err(DbError::Message(format!(
81            "circle operation id row names {stored_operation_id} but its payload operation id is {}",
82            payload.operation_id
83        )));
84    }
85    if payload.circle_id.to_string() != stored_circle_id {
86        return Err(DbError::Message(format!(
87            "circle operation {stored_operation_id} row names circle {stored_circle_id} but its payload circle id is {}",
88            payload.circle_id
89        )));
90    }
91    let journal = CircleOperationJournal {
92        operation_id: payload.operation_id,
93        circle_id: payload.circle_id,
94        intent: payload.intent,
95        operation: payload.operation,
96        progress,
97        uploaded,
98    };
99    journal.validate_identity().map_err(DbError::from)?;
100    journal.validate_uploaded().map_err(DbError::from)?;
101    Ok(journal)
102}
103
104pub(crate) fn circle_operation_uploaded_steps_on(
105    conn: &Connection,
106    operation_id: &str,
107) -> Result<BTreeSet<String>, DbError> {
108    crate::query_mapped_rows(
109        conn,
110        "SELECT step FROM circle_operation_uploads WHERE operation_id = ?1 ORDER BY step",
111        [operation_id],
112        |row| row.get::<_, String>(0),
113    )
114    .map_err(DbError::from)
115    .map(BTreeSet::from_iter)
116}
117
118pub(crate) fn load_circle_operation_on(
119    conn: &Connection,
120    operation_id: &str,
121) -> Result<Option<CircleOperationJournal>, DbError> {
122    let Some((stored_operation_id, circle_id, prepared, phase)) = conn
123        .query_row(
124            "SELECT operation_id, circle_id, prepared, phase
125             FROM circle_operations
126             WHERE operation_id = ?1",
127            [operation_id],
128            |row| {
129                Ok((
130                    row.get::<_, String>(0)?,
131                    row.get::<_, String>(1)?,
132                    row.get::<_, Vec<u8>>(2)?,
133                    row.get::<_, String>(3)?,
134                ))
135            },
136        )
137        .optional()
138        .map_err(DbError::from)?
139    else {
140        return Ok(None);
141    };
142    let uploaded = circle_operation_uploaded_steps_on(conn, &stored_operation_id)?;
143    parse_circle_operation_row(
144        &stored_operation_id,
145        &circle_id,
146        &prepared,
147        &phase,
148        uploaded,
149    )
150    .map(Some)
151}
152
153/// Every operation whose phase is one this caller acts on, oldest first.
154///
155/// The phase is its own column, so the filter runs before any operation is
156/// parsed — a caller looking for the one discarding operation does not pay for
157/// the prepared payload of every other.
158pub(crate) fn circle_operation_ids_in_phase_on(
159    conn: &Connection,
160    accept: impl Fn(&CircleOperationProgress) -> bool,
161) -> Result<Vec<String>, DbError> {
162    let rows = crate::query_mapped_rows(
163        conn,
164        "SELECT operation_id, phase FROM circle_operations ORDER BY rowid",
165        [],
166        |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
167    )
168    .map_err(DbError::from)?;
169    let mut matching = Vec::new();
170    for (operation_id, phase) in rows {
171        let progress: CircleOperationProgress = serde_json::from_str(&phase)
172            .map_err(|error| DbError::context("parse circle operation phase", error))?;
173        if accept(&progress) {
174            matching.push(operation_id);
175        }
176    }
177    Ok(matching)
178}
coven_database/circle_operation_records.rs

coven_database/
circle_operation_records.rs
