Skip to main content

coven_database/store/store_session/
materialized_commit_index.rs

1use crate::query_mapped_rows;
2use crate::*;
3#[cfg(any(test, feature = "test-utils"))]
4use coven_protocol::store_commit::StoreDeviceRegistration;
5use coven_protocol::store_commit::{
6    ActivatedStoreDeviceRegistration, CommitFrontier, ReferencedStoreDeviceRegistration,
7    ResolvedStoreDeviceState, StoreBatchCommitRef, StoreDeviceProposalAck,
8    StoreDeviceRegistrationRef, StoreDeviceStateRef, StoreHistoryCut,
9};
10use rusqlite::{Connection, OptionalExtension};
11use std::collections::BTreeMap;
12
13use super::*;
14
15impl StoreSession<'_> {
16    fn materialized_frontier(&mut self) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
17        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
18            .materialized_frontier()
19    }
20
21    fn retained_merge_replay_inputs(
22        &mut self,
23        root: coven_protocol::store_commit::StoreRootRef,
24    ) -> Result<Vec<OwnedVerifiedMergeMaterialization>, DbError> {
25        self.verified_store_authority.retained_replay_inputs_on(
26            crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
27            &root,
28        )
29    }
30
31    fn retained_merge_materialization_refs(&mut self) -> Result<Vec<StoreBatchCommitRef>, DbError> {
32        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
33            .retained_merge_materialization_refs()
34    }
35
36    fn retained_merge_materialization(
37        &mut self,
38        root: coven_protocol::store_commit::StoreRootRef,
39        reference: StoreBatchCommitRef,
40    ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
41        self.verified_store_authority
42            .retained_replay_inputs_on(
43                crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
44                &root,
45            )?
46            .into_iter()
47            .find(|materialization| materialization.commit_ref() == &reference)
48            .ok_or_else(|| {
49                DbError::Message(
50                    "retained Merge materialization is absent at its exact coordinate".to_string(),
51                )
52            })
53    }
54
55    fn retained_merge_history_frontier(
56        &mut self,
57        root: coven_protocol::store_commit::StoreRootRef,
58        references: Vec<StoreBatchCommitRef>,
59    ) -> Result<Vec<RetainedMergeHistoryCheckpoint>, DbError> {
60        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
61        let authority = &mut *self.verified_store_authority;
62        let retained = authority.retained_replay_inputs_on(records, &root)?;
63        // Read once from the connection's verified baseline; every reference the
64        // walk finds below the snapshot cut resolves against this same one.
65        let baseline = authority.retained_replay_baseline_on(records)?.clone();
66        let by_reference = retained
67            .iter()
68            .map(|materialization| (materialization.commit_ref().clone(), materialization))
69            .collect::<BTreeMap<_, _>>();
70        let mut pending = references;
71        let mut visited = std::collections::BTreeSet::new();
72        let mut checkpoints = Vec::new();
73        while let Some(reference) = pending.pop() {
74            if !visited.insert(reference.clone()) {
75                continue;
76            }
77            match by_reference.get(&reference) {
78                Some(materialization) => {
79                    pending.extend(
80                        materialization
81                            .commit()
82                            .order
83                            .predecessor_cut()
84                            .map_err(DbError::from)?
85                            .0
86                            .into_values(),
87                    );
88                    checkpoints
89                        .push(authority.retained_history_checkpoint_on(records, &reference)?);
90                }
91                None => checkpoints.push(StoreDatabase::load_retained_merge_history_checkpoint_on(
92                    records, &root, authority, &baseline, &reference,
93                )?),
94            }
95        }
96        Ok(checkpoints)
97    }
98
99    fn exact_materialized_ref(
100        &mut self,
101        stream_id: String,
102        sequence: u64,
103    ) -> Result<Option<StoreBatchCommitRef>, DbError> {
104        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
105            .materialized_commit_ref(&stream_id, sequence)
106    }
107
108    fn snapshot_coverage_frontier(&mut self) -> Result<CommitFrontier, DbError> {
109        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
110            .snapshot_coverage_frontier()
111    }
112
113    fn installed_replay_baseline(&mut self) -> Result<crate::InstalledReplayBaseline, DbError> {
114        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
115        let coverage = self.snapshot_coverage_frontier()?;
116        let covered_states =
117            crate::store::store_device_state::load_covered_store_device_snapshots_on(
118                self.conn, &coverage,
119            )?;
120        // A genesis baseline covers nothing, so there is nothing under it to
121        // summarize and every walk runs to the bottom as it always did.
122        let (summary, snapshot) =
123            match crate::store::retained_replay::load_replay_baseline_metadata_on(records)? {
124                Some(baseline) => match &baseline.authority {
125                    crate::RetainedReplayAuthority::InstalledSnapshot(authority) => {
126                        let snapshot = authority.snapshot.clone();
127                        (
128                            Some(
129                                crate::StoreDatabase::open_installed_baseline_history_summary(
130                                    records, &baseline,
131                                )?,
132                            ),
133                            Some(snapshot),
134                        )
135                    }
136                    crate::RetainedReplayAuthority::Genesis(_) => (None, None),
137                },
138                None => (None, None),
139            };
140        Ok(crate::InstalledReplayBaseline::new(
141            coverage,
142            covered_states,
143            summary,
144            snapshot,
145        ))
146    }
147
148    /// The accepted announcement each stream's installed snapshot restates at
149    /// its covered tip.
150    ///
151    /// The announcement chain is slot-linked, so a walker that starts below a
152    /// position cannot skip to it — it has to read every head in between. This
153    /// is where a walk resumes instead: the head the snapshot's owner signed
154    /// into its history summary, at the coverage this device stands on. A
155    /// device on a genesis baseline has none, and its walks start at the stream
156    /// anchor as they always did.
157    fn snapshot_announcement_frontier(
158        &mut self,
159    ) -> Result<
160        BTreeMap<
161            coven_protocol::causal_grants::AuthorStreamId,
162            coven_protocol::store_commit::RetainedAcceptedStoreAnnouncement,
163        >,
164        DbError,
165    > {
166        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
167        let baseline = self
168            .verified_store_authority
169            .retained_replay_baseline_on(records)?;
170        let crate::RetainedReplayAuthority::InstalledSnapshot(authority) = &baseline.authority
171        else {
172            return Ok(BTreeMap::new());
173        };
174        let summary = &authority.metadata.history_summary;
175        let coverage = &authority.metadata.coverage.0;
176        let frontier = summary.announcement_frontier.clone();
177        for (stream_id, announcement) in &frontier {
178            // The summary carries both, so they can disagree; the coverage is
179            // what every other position question answers from, and an
180            // announcement naming a different commit would resume a walk on the
181            // wrong chain.
182            if coverage.get(stream_id) != Some(&announcement.value.commit) {
183                return Err(DbError::Message(
184                    "snapshot announcement frontier differs from its own coverage".to_string(),
185                ));
186            }
187        }
188        Ok(frontier)
189    }
190
191    fn store_device_state_for_history_cut(
192        &mut self,
193        cut: StoreHistoryCut,
194    ) -> Result<(StoreDeviceStateRef, ResolvedStoreDeviceState), DbError> {
195        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
196            .store_device_state_for_history_cut(&cut)
197    }
198
199    fn resolved_store_device_state(
200        &mut self,
201        reference: StoreDeviceStateRef,
202    ) -> Result<ResolvedStoreDeviceState, DbError> {
203        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
204            .declared_store_device_state(&reference)
205    }
206
207    fn store_device_exclusion_freezes(&mut self) -> Result<Vec<StoreDeviceProposalAck>, DbError> {
208        let root = self
209            .root_authority()?
210            .map(|(reference, _)| reference)
211            .ok_or_else(|| {
212                DbError::Message("Store root is absent while loading exclusion freezes".to_string())
213            })?;
214        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
215            .store_device_exclusion_freezes(&root)
216    }
217
218    fn activated_store_device_registration_records(
219        &mut self,
220    ) -> Result<Vec<ReferencedStoreDeviceRegistration>, DbError> {
221        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
222        let root = self
223            .verified_store_authority
224            .root_authority_on(records)?
225            .map(|(reference, _)| reference)
226            .ok_or_else(|| {
227                DbError::Message("Store root is absent while loading activated devices".to_string())
228            })?;
229        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
230            .activated_registration_references()?
231            .into_iter()
232            .map(|reference| {
233                let device_id = reference.device_id;
234                let registration = self
235                    .verified_store_authority
236                    .activated_registration_on(records, &root, &reference)?;
237                ReferencedStoreDeviceRegistration::verified(reference, registration).map_err(
238                    |error| {
239                        DbError::context(
240                            format!(
241                                "activated Store device registration {device_id} exact reference"
242                            ),
243                            error,
244                        )
245                    },
246                )
247            })
248            .collect::<Result<Vec<_>, DbError>>()
249    }
250
251    fn activated_store_device_registration(
252        &mut self,
253        reference: StoreDeviceRegistrationRef,
254    ) -> Result<ReferencedStoreDeviceRegistration, DbError> {
255        let root = self
256            .root_authority()?
257            .map(|(reference, _)| reference)
258            .ok_or_else(|| {
259                DbError::Message(
260                    "Store root is absent while loading an activated device".to_string(),
261                )
262            })?;
263        let registration = self.verified_store_authority.activated_registration_on(
264            crate::store::store_session::StoreRecords::new(self.conn, self.store_dir),
265            &root,
266            &reference,
267        )?;
268        ReferencedStoreDeviceRegistration::verified(reference, registration).map_err(DbError::from)
269    }
270
271    fn local_activated_registration_ref(
272        &mut self,
273    ) -> Result<Option<StoreDeviceRegistrationRef>, DbError> {
274        crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
275            .local_activated_registration_ref()
276    }
277
278    fn activated_store_device_registration_with_authority(
279        &mut self,
280        root: coven_protocol::store_commit::StoreRootRef,
281        reference: StoreDeviceRegistrationRef,
282    ) -> Result<ActivatedStoreDeviceRegistration, DbError> {
283        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
284        let registration = self
285            .verified_store_authority
286            .activated_registration_on(records, &root, &reference)?;
287        let authority = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
288            .activated_registration_authority(&reference)?;
289        let authority = serde_json::from_str(&authority)
290            .map_err(|error| DbError::context("activated Store registration authority", error))?;
291        let registration = ReferencedStoreDeviceRegistration::verified(reference, registration)
292            .map_err(DbError::from)?;
293        ActivatedStoreDeviceRegistration::verified(registration, authority).map_err(DbError::from)
294    }
295
296    fn activated_store_device_registration_for_device(
297        &mut self,
298        device_id: coven_protocol::store_commit::StoreDeviceId,
299    ) -> Result<Option<ActivatedStoreDeviceRegistration>, DbError> {
300        let records = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir);
301        let root = self
302            .verified_store_authority
303            .root_authority_on(records)?
304            .map(|(reference, _)| reference)
305            .ok_or_else(|| {
306                DbError::Message(
307                    "Store root is absent while loading an activated device".to_string(),
308                )
309            })?;
310        let stored = crate::store::store_session::StoreRecords::new(self.conn, self.store_dir)
311            .activated_registration_row_for_device(device_id)?;
312        let Some((reference, authority)) = stored else {
313            return Ok(None);
314        };
315        let reference: StoreDeviceRegistrationRef = serde_json::from_str(&reference)
316            .map_err(|error| DbError::context("activated Store registration ref", error))?;
317        if reference.device_id != device_id {
318            return Err(DbError::Message(
319                "activated Store registration row names another device".to_string(),
320            ));
321        }
322        let registration = self
323            .verified_store_authority
324            .activated_registration_on(records, &root, &reference)?;
325        let authority = serde_json::from_str(&authority)
326            .map_err(|error| DbError::context("activated Store registration authority", error))?;
327        let registration = ReferencedStoreDeviceRegistration::verified(reference, registration)
328            .map_err(DbError::from)?;
329        ActivatedStoreDeviceRegistration::verified(registration, authority)
330            .map(Some)
331            .map_err(DbError::from)
332    }
333}
334
335impl StoreDatabase {
336    pub async fn materialized_frontier(
337        &self,
338    ) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
339        self.call_store(|session| session.materialized_frontier())
340            .await
341    }
342
343    pub async fn retained_merge_replay_inputs(
344        &self,
345        root: coven_protocol::store_commit::StoreRootRef,
346    ) -> Result<Vec<OwnedVerifiedMergeMaterialization>, DbError> {
347        self.call_store(move |session| session.retained_merge_replay_inputs(root))
348            .await
349    }
350
351    pub async fn retained_merge_materialization_refs(
352        &self,
353    ) -> Result<Vec<StoreBatchCommitRef>, DbError> {
354        self.call_store(|session| session.retained_merge_materialization_refs())
355            .await
356    }
357
358    pub async fn retained_merge_materialization(
359        &self,
360        root: coven_protocol::store_commit::StoreRootRef,
361        reference: StoreBatchCommitRef,
362    ) -> Result<OwnedVerifiedMergeMaterialization, DbError> {
363        self.call_store(move |session| session.retained_merge_materialization(root, reference))
364            .await
365    }
366
367    pub async fn retained_merge_history_frontier(
368        &self,
369        root: coven_protocol::store_commit::StoreRootRef,
370        references: Vec<StoreBatchCommitRef>,
371    ) -> Result<Vec<RetainedMergeHistoryCheckpoint>, DbError> {
372        self.call_store(move |session| session.retained_merge_history_frontier(root, references))
373            .await
374    }
375
376    pub async fn exact_materialized_ref(
377        &self,
378        stream_id: &str,
379        sequence: u64,
380    ) -> Result<Option<StoreBatchCommitRef>, DbError> {
381        let stream_id = stream_id.to_string();
382        self.call_store(move |session| session.exact_materialized_ref(stream_id, sequence))
383            .await
384    }
385
386    pub async fn snapshot_coverage_frontier(&self) -> Result<CommitFrontier, DbError> {
387        self.call_store(|session| session.snapshot_coverage_frontier())
388            .await
389    }
390
391    /// The baseline a history walk stops at, with the device states it keeps
392    /// for the covered positions commits above it still name.
393    pub async fn installed_replay_baseline(
394        &self,
395    ) -> Result<crate::InstalledReplayBaseline, DbError> {
396        self.call_store(|session| session.installed_replay_baseline())
397            .await
398    }
399
400    pub async fn snapshot_announcement_frontier(
401        &self,
402    ) -> Result<
403        BTreeMap<
404            coven_protocol::causal_grants::AuthorStreamId,
405            coven_protocol::store_commit::RetainedAcceptedStoreAnnouncement,
406        >,
407        DbError,
408    > {
409        self.call_store(|session| session.snapshot_announcement_frontier())
410            .await
411    }
412
413    pub async fn store_device_state_for_order(
414        &self,
415        order: &coven_protocol::store_commit::StoreCommitOrder,
416    ) -> Result<(StoreDeviceStateRef, ResolvedStoreDeviceState), DbError> {
417        let cut = order.predecessor_cut().map_err(DbError::from)?;
418        self.call_store(move |session| session.store_device_state_for_history_cut(cut))
419            .await
420    }
421
422    pub async fn store_device_state_for_history_cut(
423        &self,
424        cut: &StoreHistoryCut,
425    ) -> Result<(StoreDeviceStateRef, ResolvedStoreDeviceState), DbError> {
426        let cut = cut.clone();
427        self.call_store(move |session| session.store_device_state_for_history_cut(cut))
428            .await
429    }
430
431    pub async fn resolved_store_device_state(
432        &self,
433        reference: &StoreDeviceStateRef,
434    ) -> Result<ResolvedStoreDeviceState, DbError> {
435        let reference = reference.clone();
436        self.call_store(move |session| session.resolved_store_device_state(reference))
437            .await
438    }
439
440    pub async fn store_device_exclusion_freezes(
441        &self,
442    ) -> Result<Vec<StoreDeviceProposalAck>, DbError> {
443        self.call_store(|session| session.store_device_exclusion_freezes())
444            .await
445    }
446
447    pub async fn activated_store_device_registration_records(
448        &self,
449    ) -> Result<Vec<ReferencedStoreDeviceRegistration>, DbError> {
450        self.call_store(|session| session.activated_store_device_registration_records())
451            .await
452    }
453
454    pub async fn activated_store_device_registration(
455        &self,
456        reference: StoreDeviceRegistrationRef,
457    ) -> Result<ReferencedStoreDeviceRegistration, DbError> {
458        self.call_store(move |session| session.activated_store_device_registration(reference))
459            .await
460    }
461
462    /// The exact registration this device is activated under, or `None` before
463    /// it has one. This is the identity a signed artifact names when it names a
464    /// device, so it is what a role check compares against.
465    pub async fn local_activated_registration_ref(
466        &self,
467    ) -> Result<Option<StoreDeviceRegistrationRef>, DbError> {
468        self.call_store(|session| session.local_activated_registration_ref())
469            .await
470    }
471
472    pub async fn local_blob_write_authority(
473        &self,
474    ) -> Result<ReferencedStoreDeviceRegistration, DbError> {
475        self.call_store(|session| session.local_store_authority())
476            .await
477    }
478
479    pub async fn activated_store_device_registration_with_authority(
480        &self,
481        root: &coven_protocol::store_commit::StoreRootRef,
482        reference: StoreDeviceRegistrationRef,
483    ) -> Result<ActivatedStoreDeviceRegistration, DbError> {
484        let root = root.clone();
485        self.call_store(move |session| {
486            session.activated_store_device_registration_with_authority(root, reference)
487        })
488        .await
489    }
490
491    pub async fn activated_store_device_registration_for_device(
492        &self,
493        device_id: coven_protocol::store_commit::StoreDeviceId,
494    ) -> Result<Option<ActivatedStoreDeviceRegistration>, DbError> {
495        self.call_store(move |session| {
496            session.activated_store_device_registration_for_device(device_id)
497        })
498        .await
499    }
500
501    #[cfg(any(test, feature = "test-utils"))]
502    pub async fn activated_store_device_registrations(
503        &self,
504    ) -> Result<Vec<StoreDeviceRegistration>, DbError> {
505        Ok(self
506            .activated_store_device_registration_records()
507            .await?
508            .into_iter()
509            .map(|registration| registration.value().clone())
510            .collect())
511    }
512}
513
514pub(crate) fn materialized_frontier_on(
515    conn: &Connection,
516    exclude_device: Option<&str>,
517) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
518    let mut frontier = BTreeMap::new();
519    let rows = query_mapped_rows(
520        conn,
521        "SELECT m.device_id, m.seq, m.commit_ref,
522                    m.retained_commit_ref, m.retained_input_hash \
523             FROM materialized_commits m \
524             JOIN (SELECT device_id, MAX(seq) AS seq FROM materialized_commits \
525                   GROUP BY device_id) latest \
526               ON latest.device_id = m.device_id AND latest.seq = m.seq",
527        [],
528        |row| {
529            Ok((
530                row.get::<_, String>(0)?,
531                row.get::<_, i64>(1)?,
532                row.get::<_, String>(2)?,
533                row.get::<_, Option<String>>(3)?,
534                row.get::<_, Option<String>>(4)?,
535            ))
536        },
537    )?;
538    for row in rows {
539        let (device_id, seq, reference, retained_commit_ref, retained_input_hash) = row;
540        if exclude_device == Some(device_id.as_str()) {
541            continue;
542        }
543        let seq = Database::sequence_from_sqlite(&device_id, seq)?;
544        frontier.insert(
545            device_id.clone(),
546            parse_materialized_commit_row_on(
547                &device_id,
548                seq,
549                &reference,
550                retained_commit_ref.as_deref(),
551                retained_input_hash.as_deref(),
552            )?,
553        );
554    }
555
556    for (device_id, reference) in snapshot_coverage_on(conn)? {
557        if exclude_device == Some(device_id.as_str()) {
558            continue;
559        }
560        if frontier
561            .get(&device_id)
562            .is_none_or(|current| current.coord.sequence() < reference.coord.sequence())
563        {
564            frontier.insert(device_id, reference);
565        }
566    }
567    Ok(frontier)
568}
569
570/// The exact commit each stream's installed snapshot image reaches. The image
571/// records one tip per stream and nothing below it, so this is the whole of
572/// what a snapshot says it materialized.
573pub(crate) fn snapshot_coverage_on(
574    conn: &Connection,
575) -> Result<BTreeMap<String, StoreBatchCommitRef>, DbError> {
576    let rows = query_mapped_rows(
577        conn,
578        "SELECT device_id, seq, commit_ref FROM snapshot_coverage",
579        [],
580        |row| {
581            Ok((
582                row.get::<_, String>(0)?,
583                row.get::<_, i64>(1)?,
584                row.get::<_, String>(2)?,
585            ))
586        },
587    )?;
588    let mut coverage = BTreeMap::new();
589    for (device_id, seq, reference) in rows {
590        let seq = Database::sequence_from_sqlite(&device_id, seq)?;
591        let reference = parse_stored_commit_ref(&device_id, seq, &reference)?;
592        coverage.insert(device_id, reference);
593    }
594    Ok(coverage)
595}
596
597pub(crate) fn parse_stored_commit_ref(
598    stream_id: &str,
599    sequence: u64,
600    encoded: &str,
601) -> Result<StoreBatchCommitRef, DbError> {
602    let reference: StoreBatchCommitRef = serde_json::from_str(encoded)
603        .map_err(|error| DbError::context("stored exact Store commit ref", error))?;
604    let coordinate_matches =
605        reference.coord.stream_id.to_string() == stream_id && reference.coord.sequence == sequence;
606    if !coordinate_matches {
607        return Err(DbError::Message(format!(
608            "stored exact Store commit ref differs from {stream_id}/{sequence}"
609        )));
610    }
611    Ok(reference)
612}
613
614fn parse_materialized_commit_row_on(
615    stream_id: &str,
616    sequence: u64,
617    encoded: &str,
618    retained_commit_ref: Option<&str>,
619    retained_input_hash: Option<&str>,
620) -> Result<StoreBatchCommitRef, DbError> {
621    let reference = parse_stored_commit_ref(stream_id, sequence, encoded)?;
622    if retained_commit_ref != Some(encoded) {
623        return Err(DbError::Message(format!(
624            "materialized coordinate {stream_id}/{sequence} does not bind its exact retained commit"
625        )));
626    }
627    let input_hash = retained_input_hash.ok_or_else(|| {
628        DbError::Message(format!(
629            "materialized coordinate {stream_id}/{sequence} has no retained input hash"
630        ))
631    })?;
632    input_hash
633        .parse::<coven_protocol::store_commit::ObjectHash>()
634        .map_err(|error| {
635            DbError::context(
636                format!(
637                    "materialized coordinate {stream_id}/{sequence} retained input hash is invalid"
638                ),
639                error,
640            )
641        })?;
642    Ok(reference)
643}
644
645pub(crate) fn materialized_commit_ref_on(
646    conn: &Connection,
647    stream_id: &str,
648    sequence: u64,
649) -> Result<Option<StoreBatchCommitRef>, DbError> {
650    let seq = Database::sequence_to_sqlite(stream_id, sequence)?;
651    conn.query_row(
652        "SELECT commit_ref, retained_commit_ref, retained_input_hash
653         FROM materialized_commits WHERE device_id = ?1 AND seq = ?2",
654        (stream_id, seq),
655        |row| {
656            Ok((
657                row.get::<_, String>(0)?,
658                row.get::<_, Option<String>>(1)?,
659                row.get::<_, Option<String>>(2)?,
660            ))
661        },
662    )
663    .optional()
664    .map_err(DbError::from)?
665    .map(|(encoded, retained_commit_ref, retained_input_hash)| {
666        parse_materialized_commit_row_on(
667            stream_id,
668            sequence,
669            &encoded,
670            retained_commit_ref.as_deref(),
671            retained_input_hash.as_deref(),
672        )
673    })
674    .transpose()
675}
676
677pub(crate) fn latest_position_for_device_on(
678    conn: &Connection,
679    device_id: &str,
680) -> Result<Option<StoreBatchCommitRef>, DbError> {
681    let materialized = conn
682        .query_row(
683            "SELECT seq, commit_ref, retained_commit_ref, retained_input_hash
684             FROM materialized_commits
685             WHERE device_id = ?1 ORDER BY seq DESC LIMIT 1",
686            [device_id],
687            |row| {
688                Ok((
689                    row.get::<_, i64>(0)?,
690                    row.get::<_, String>(1)?,
691                    row.get::<_, Option<String>>(2)?,
692                    row.get::<_, Option<String>>(3)?,
693                ))
694            },
695        )
696        .optional()
697        .map_err(DbError::from)?;
698    let coverage = conn
699        .query_row(
700            "SELECT seq, commit_ref FROM snapshot_coverage WHERE device_id = ?1",
701            [device_id],
702            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
703        )
704        .optional()
705        .map_err(DbError::from)?;
706    let mut references = Vec::new();
707    if let Some((seq, reference, retained_commit_ref, retained_input_hash)) = materialized {
708        let seq = Database::sequence_from_sqlite(device_id, seq)?;
709        references.push(parse_materialized_commit_row_on(
710            device_id,
711            seq,
712            &reference,
713            retained_commit_ref.as_deref(),
714            retained_input_hash.as_deref(),
715        )?);
716    }
717    if let Some((seq, reference)) = coverage {
718        let seq = Database::sequence_from_sqlite(device_id, seq)?;
719        references.push(parse_stored_commit_ref(device_id, seq, &reference)?);
720    }
721    if references.len() == 2
722        && references[0].coord.sequence() == references[1].coord.sequence()
723        && references[0] != references[1]
724    {
725        return Err(DbError::Message(format!(
726            "materialized ledger and snapshot coverage fork {device_id:?} at sequence {}",
727            references[0].coord.sequence()
728        )));
729    }
730    Ok(references
731        .into_iter()
732        .max_by_key(|reference| reference.coord.sequence()))
733}