coven_core/sync/store/device_join/
journal.rs

1use super::*;
2
3#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
4#[serde(rename_all = "snake_case", deny_unknown_fields)]
5pub enum DeviceJoinCleanupProgress {
6    AwaitingBoth,
7    AwaitingAdministrator {
8        joiner: JoinerJoinTerminal,
9    },
10    AwaitingJoiner {
11        administrator: ProviderAdminJoinTerminal,
12    },
13    Ready {
14        administrator: ProviderAdminJoinTerminal,
15        joiner: JoinerJoinTerminal,
16    },
17}
18
19#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
20#[serde(rename_all = "snake_case", deny_unknown_fields)]
21pub enum DeviceJoinStatus {
22    OperationInProgress {
23        attempt_id: DeviceJoinAttemptId,
24    },
25    AwaitingAccessRequest {
26        offer: DeviceJoinOffer,
27    },
28    AwaitingProviderAdmission {
29        request: DeviceProviderAccessRequest,
30    },
31    AwaitingRegistrationRequest {
32        approval: DeviceProviderAdmissionApproval,
33    },
34    AwaitingBootstrap {
35        request: DeviceRegistrationRequest,
36    },
37    AwaitingChallengePublication {
38        bootstrap: ProvisionalDeviceBootstrap,
39    },
40    AwaitingReadiness {
41        bootstrap: ProviderReadyDeviceBootstrap,
42    },
43    AwaitingProviderCompletion {
44        readiness: DeviceJoinReadiness,
45    },
46    AwaitingActivation {
47        completion: DeviceProviderAdmissionCompletion,
48    },
49    AwaitingCompletion {
50        activation: DeviceJoinActivation,
51    },
52    Activated {
53        store: JoinedStore,
54    },
55    Abandoned {
56        abandonment: DeviceJoinAbandonment,
57    },
58    ProviderAccessGrantCreatePending {
59        request: DeviceProviderAccessRequest,
60        grant: StoreMemberProviderAccessGrant,
61    },
62    AbandonmentCreatePending {
63        abandonment: DeviceJoinAbandonmentRef,
64    },
65    CancellationCreatePending {
66        cancellation: DeviceJoinOutcomeRef,
67    },
68    ProviderClosurePending {
69        cancellation: DeviceJoinCancellation,
70        producer: DeviceJoinProducer,
71    },
72    ProviderClosed {
73        terminal: ProviderAdminJoinTerminal,
74    },
75    JoinerClosed {
76        terminal: JoinerJoinTerminal,
77    },
78    CleanupReceiptCreatePending {
79        cancellation: DeviceJoinCancellation,
80        receipt: DeviceJoinCleanupReceiptRef,
81    },
82    Cancelled {
83        cancellation: DeviceJoinCancellation,
84    },
85    CleanupPending {
86        cancellation: DeviceJoinCancellation,
87        progress: DeviceJoinCleanupProgress,
88    },
89    AwaitingCleanupActivation {
90        receipt: DeviceJoinCleanupReceipt,
91    },
92    CleanupActivated {
93        activation: DeviceJoinCleanupActivation,
94    },
95}
96
97#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
98#[serde(rename_all = "snake_case", deny_unknown_fields)]
99pub enum DeviceJoinAction {
100    TransferOffer(DeviceJoinOffer),
101    TransferProviderAccessRequest(DeviceProviderAccessRequest),
102    TransferProviderAdmissionApproval(DeviceProviderAdmissionApproval),
103    TransferRegistrationRequest(DeviceRegistrationRequest),
104    TransferProvisionalBootstrap(ProvisionalDeviceBootstrap),
105    TransferProviderReadyBootstrap(ProviderReadyDeviceBootstrap),
106    TransferReadiness(DeviceJoinReadiness),
107    TransferProviderAdmissionCompletion(DeviceProviderAdmissionCompletion),
108    TransferActivation(DeviceJoinActivation),
109    TransferAbandonment(DeviceJoinAbandonment),
110    TransferCancellation(DeviceJoinCancellation),
111    TransferProviderAdminTerminal(ProviderAdminJoinTerminal),
112    TransferJoinerTerminal(JoinerJoinTerminal),
113    TransferCleanupReceipt(DeviceJoinCleanupReceipt),
114    TransferCleanupActivation(DeviceJoinCleanupActivation),
115    CompleteJoin(DeviceJoinActivation),
116    CompleteCleanup(DeviceJoinCleanupActivation),
117    ResumeOperation {
118        attempt_id: DeviceJoinAttemptId,
119        role: DeviceJoinRole,
120    },
121}
122
123#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
124#[serde(rename_all = "snake_case", deny_unknown_fields)]
125pub enum OwnerJoinProgress {
126    Offered(DeviceJoinOffer),
127    RegistrationRequested(DeviceRegistrationRequest),
128    AbandonmentCreateIntent {
129        offer: DeviceJoinOffer,
130        abandonment: DeviceJoinAbandonmentRef,
131        prepared: PreparedDeviceJoinObject,
132    },
133    AttemptActivated(ProvisionalDeviceBootstrap),
134    ActivationCreateIntent {
135        bootstrap: ProvisionalDeviceBootstrap,
136        completion: DeviceProviderAdmissionCompletion,
137        outcome: DeviceJoinOutcomeRef,
138        prepared: PreparedDeviceJoinObject,
139    },
140    CancellationCreateIntent {
141        attempt: DeviceJoinAttemptRef,
142        cancellation: DeviceJoinOutcomeRef,
143        prepared: PreparedDeviceJoinObject,
144    },
145    ActivationPrepared {
146        completion: DeviceProviderAdmissionCompletion,
147        activation: DeviceJoinActivation,
148    },
149    Abandoned(DeviceJoinAbandonment),
150    Cancelled(DeviceJoinCancellation),
151    CleanupReceiptCreateIntent {
152        cancellation: DeviceJoinCancellation,
153        receipt: DeviceJoinCleanupReceiptRef,
154        receipt_bytes: Vec<u8>,
155        prepared: PreparedDeviceJoinObject,
156    },
157    CleanupReceipt(DeviceJoinCleanupReceipt),
158    CleanupActivated(DeviceJoinCleanupActivation),
159    CancelledComplete(DeviceJoinCleanupActivation),
160}
161
162#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
163#[serde(rename_all = "snake_case", deny_unknown_fields)]
164pub enum ProviderAdminJoinProgress {
165    AccessRequested(DeviceProviderAccessRequest),
166    AccessGrantPrepared {
167        request: DeviceProviderAccessRequest,
168        grant: StoreMemberProviderAccessGrant,
169        prepared: PreparedDeviceJoinObject,
170    },
171    ApprovalPrepared(DeviceProviderAdmissionApproval),
172    AttemptObserved(ProvisionalDeviceBootstrap),
173    ChallengeCreateIntent(ProvisionalDeviceBootstrap),
174    ProviderReady(ProviderReadyDeviceBootstrap),
175    ResponseObserved(DeviceJoinReadiness),
176    CleanupIntent {
177        cancellation: DeviceJoinCancellation,
178        challenge: ProviderChallengeDisposition,
179        prior_state_hash: ObjectHash,
180    },
181    Completed(DeviceProviderAdmissionCompletion),
182    Cancelled(ProviderAdminJoinClosure),
183    WriteRevoked(DeviceJoinProducerWriteRevocation),
184}
185
186#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
187#[serde(deny_unknown_fields)]
188pub struct PreparedDeviceJoinObject {
189    pub object: ExactObjectRef,
190    pub stored_bytes: Vec<u8>,
191}
192
193impl PreparedDeviceJoinObject {
194    pub(super) fn from_prepared(prepared: &crate::sync::storage::PreparedExactObject) -> Self {
195        Self {
196            object: prepared.reference().clone(),
197            stored_bytes: prepared.stored_bytes().to_vec(),
198        }
199    }
200}
201
202#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
203#[serde(rename_all = "snake_case", deny_unknown_fields)]
204pub enum JoinerJoinProgress {
205    OfferReceived(DeviceJoinOffer),
206    AccessRequested(DeviceProviderAccessRequest),
207    ApprovalReceived(DeviceProviderAdmissionApproval),
208    RegistrationPrepared(DeviceRegistrationRequest),
209    ProviderReady(ProviderReadyDeviceBootstrap),
210    RegistrationCreateIntent(ProviderReadyDeviceBootstrap),
211    RegistrationCreated(StoreDeviceRegistrationRef),
212    AckCreateIntent(StoreDeviceRegistrationRef),
213    AckCreated(crate::sync::store_commit::StoreAckRef),
214    ResponseCreateIntent(DeviceJoinReadiness),
215    Ready(DeviceJoinReadiness),
216    ActivationObserved {
217        readiness: DeviceJoinReadiness,
218        activation: DeviceJoinActivation,
219    },
220    Activated(JoinedStore),
221    Abandoned(DeviceJoinAbandonment),
222    CleanupIntent {
223        cancellation: DeviceJoinCancellation,
224        registration: SlotDisposition,
225        initial_ack: SlotDisposition,
226        response: JoinerResponseDisposition,
227        prior_state_hash: ObjectHash,
228    },
229    Cancelled(JoinerJoinClosure),
230    WriteRevoked(DeviceJoinProducerWriteRevocation),
231    CleanupActivated(DeviceJoinCleanupActivation),
232    CancelledComplete(DeviceJoinCleanupActivation),
233}
234
235#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
236#[serde(rename_all = "snake_case", deny_unknown_fields)]
237pub enum DeviceJoinRoleProgress {
238    Owner(OwnerJoinProgress),
239    ProviderAdministrator(ProviderAdminJoinProgress),
240    Joiner(JoinerJoinProgress),
241}
242
243impl DeviceJoinRoleProgress {
244    fn role(&self) -> DeviceJoinRole {
245        match self {
246            Self::Owner(_) => DeviceJoinRole::Owner,
247            Self::ProviderAdministrator(_) => DeviceJoinRole::ProviderAdministrator,
248            Self::Joiner(_) => DeviceJoinRole::Joiner,
249        }
250    }
251
252    fn role_name(&self) -> &'static str {
253        self.role().as_str()
254    }
255
256    fn validate_transition(&self, next: &Self) -> Result<(), DeviceJoinError> {
257        let adjacent = match (self, next) {
258            (Self::Owner(previous), Self::Owner(next)) => owner_adjacent(previous, next),
259            (Self::ProviderAdministrator(previous), Self::ProviderAdministrator(next)) => {
260                provider_admin_adjacent(previous, next)
261            }
262            (Self::Joiner(previous), Self::Joiner(next)) => joiner_adjacent(previous, next),
263            _ => false,
264        };
265        if adjacent {
266            Ok(())
267        } else {
268            Err(DeviceJoinError::NonAdjacentJournalTransition)
269        }
270    }
271}
272
273#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
274#[serde(deny_unknown_fields)]
275pub struct DeviceJoinJournalRecord {
276    pub attempt_id: DeviceJoinAttemptId,
277    pub progress: Box<DeviceJoinRoleProgress>,
278}
279
280/// Durable role journal. Each row stores a closed progress value; SQLite's
281/// compare-and-swap update rejects stale or skipped transitions.
282#[derive(Clone, Debug)]
283pub struct DeviceJoinJournalDatabase {
284    path: PathBuf,
285}
286
287impl DeviceJoinJournalDatabase {
288    pub fn open(path: impl AsRef<Path>) -> Result<Self, DeviceJoinError> {
289        let path = path.as_ref().to_path_buf();
290        let connection = Connection::open(&path)?;
291        connection.execute_batch(
292            "PRAGMA foreign_keys = ON;
293             CREATE TABLE IF NOT EXISTS device_join_journals (
294                 attempt_id TEXT NOT NULL,
295                 role TEXT NOT NULL,
296                 payload TEXT NOT NULL,
297                 PRIMARY KEY (attempt_id, role)
298             ) STRICT, WITHOUT ROWID;
299             CREATE TABLE IF NOT EXISTS pending_join_transfers (
300                 attempt_id TEXT PRIMARY KEY,
301                 payload_hash TEXT NOT NULL,
302                 payload TEXT NOT NULL
303             ) STRICT, WITHOUT ROWID;",
304        )?;
305        Ok(Self { path })
306    }
307
308    pub(crate) fn path(&self) -> &Path {
309        &self.path
310    }
311
312    pub fn begin(
313        &self,
314        record: DeviceJoinJournalRecord,
315    ) -> Result<DeviceJoinJournalRecord, DeviceJoinError> {
316        validate_initial_progress(&record.progress)?;
317        let connection = Connection::open(&self.path)?;
318        let tx = connection.unchecked_transaction()?;
319        let attempt_id = attempt_key(record.attempt_id);
320        let role = record.progress.role_name();
321        let payload = serde_json::to_string(&record)?;
322        tx.execute(
323            "INSERT OR IGNORE INTO device_join_journals (attempt_id, role, payload)
324             VALUES (?1, ?2, ?3)",
325            (&attempt_id, role, &payload),
326        )?;
327        let actual: String = tx.query_row(
328            "SELECT payload FROM device_join_journals WHERE attempt_id = ?1 AND role = ?2",
329            (&attempt_id, role),
330            |row| row.get(0),
331        )?;
332        tx.commit()?;
333        let actual = serde_json::from_str::<DeviceJoinJournalRecord>(&actual)?;
334        if actual != record {
335            return Err(DeviceJoinError::JournalConflict);
336        }
337        Ok(actual)
338    }
339
340    pub fn load(
341        &self,
342        attempt_id: DeviceJoinAttemptId,
343        role: DeviceJoinRole,
344    ) -> Result<Option<DeviceJoinJournalRecord>, DeviceJoinError> {
345        let connection = Connection::open(&self.path)?;
346        let raw = connection
347            .query_row(
348                "SELECT payload FROM device_join_journals WHERE attempt_id = ?1 AND role = ?2",
349                (attempt_key(attempt_id), role.as_str()),
350                |row| row.get::<_, String>(0),
351            )
352            .optional()?;
353        let record = raw
354            .map(|value| serde_json::from_str::<DeviceJoinJournalRecord>(&value))
355            .transpose()?;
356        if record
357            .as_ref()
358            .is_some_and(|record| record.attempt_id != attempt_id || record.progress.role() != role)
359        {
360            return Err(DeviceJoinError::JournalConflict);
361        }
362        Ok(record)
363    }
364
365    pub fn records(&self) -> Result<Vec<DeviceJoinJournalRecord>, DeviceJoinError> {
366        let connection = Connection::open(&self.path)?;
367        let mut statement = connection.prepare(
368            "SELECT attempt_id, role, payload FROM device_join_journals
369             ORDER BY attempt_id, role",
370        )?;
371        let rows = statement.query_map([], |row| {
372            Ok((
373                row.get::<_, String>(0)?,
374                row.get::<_, String>(1)?,
375                row.get::<_, String>(2)?,
376            ))
377        })?;
378        let mut records = Vec::new();
379        for row in rows {
380            let (attempt_id, role, payload) = row?;
381            let record: DeviceJoinJournalRecord = serde_json::from_str(&payload)?;
382            if attempt_key(record.attempt_id) != attempt_id || record.progress.role_name() != role {
383                return Err(DeviceJoinError::JournalConflict);
384            }
385            records.push(record);
386        }
387        records.sort_by_key(|record| (record.attempt_id, record.progress.role()));
388        Ok(records)
389    }
390
391    pub fn actions(&self) -> Result<Vec<DeviceJoinAction>, DeviceJoinError> {
392        Ok(self
393            .records()?
394            .iter()
395            .filter_map(device_join_action)
396            .collect())
397    }
398
399    pub fn advance(
400        &self,
401        previous: &DeviceJoinJournalRecord,
402        next: DeviceJoinJournalRecord,
403    ) -> Result<(), DeviceJoinError> {
404        if previous.attempt_id != next.attempt_id {
405            return Err(DeviceJoinError::JournalConflict);
406        }
407        previous.progress.validate_transition(&next.progress)?;
408        let previous_payload = serde_json::to_string(previous)?;
409        let next_payload = serde_json::to_string(&next)?;
410        let connection = Connection::open(&self.path)?;
411        let changed = connection.execute(
412            "UPDATE device_join_journals SET payload = ?1
413             WHERE attempt_id = ?2 AND role = ?3 AND payload = ?4",
414            (
415                &next_payload,
416                attempt_key(previous.attempt_id),
417                previous.progress.role_name(),
418                &previous_payload,
419            ),
420        )?;
421        if changed != 1 {
422            return Err(DeviceJoinError::JournalConflict);
423        }
424        Ok(())
425    }
426
427    pub fn stage_pending_transfer(
428        &self,
429        record: &DeviceJoinJournalRecord,
430    ) -> Result<ObjectHash, DeviceJoinError> {
431        if !matches!(*record.progress, DeviceJoinRoleProgress::Joiner(_)) {
432            return Err(DeviceJoinError::JournalConflict);
433        }
434        let payload = serde_json::to_string(record)?;
435        let payload_hash = ObjectHash::digest(payload.as_bytes());
436        let connection = Connection::open(&self.path)?;
437        connection.execute(
438            "INSERT OR IGNORE INTO pending_join_transfers (attempt_id, payload_hash, payload)
439             VALUES (?1, ?2, ?3)",
440            (
441                attempt_key(record.attempt_id),
442                payload_hash.to_string(),
443                &payload,
444            ),
445        )?;
446        let actual: (String, String) = connection.query_row(
447            "SELECT payload_hash, payload FROM pending_join_transfers WHERE attempt_id = ?1",
448            [attempt_key(record.attempt_id)],
449            |row| Ok((row.get(0)?, row.get(1)?)),
450        )?;
451        if actual != (payload_hash.to_string(), payload) {
452            return Err(DeviceJoinError::JournalConflict);
453        }
454        Ok(payload_hash)
455    }
456
457    /// Move one pending joiner payload into the installed Store journal in one
458    /// SQLite transaction spanning the source and destination databases.
459    pub fn transfer_pending_to(
460        &self,
461        destination: &DeviceJoinJournalDatabase,
462        attempt_id: DeviceJoinAttemptId,
463    ) -> Result<DeviceJoinJournalRecord, DeviceJoinError> {
464        let destination_connection = Connection::open(&destination.path)?;
465        destination_connection.execute(
466            "ATTACH DATABASE ?1 AS pending_join_source",
467            [self.path.to_string_lossy().as_ref()],
468        )?;
469        let tx = destination_connection.unchecked_transaction()?;
470        let attempt = attempt_key(attempt_id);
471        let (payload_hash, payload): (String, String) = tx.query_row(
472            "SELECT payload_hash, payload FROM pending_join_source.pending_join_transfers
473             WHERE attempt_id = ?1",
474            [&attempt],
475            |row| Ok((row.get(0)?, row.get(1)?)),
476        )?;
477        let calculated = ObjectHash::digest(payload.as_bytes()).to_string();
478        if calculated != payload_hash {
479            return Err(DeviceJoinError::PendingTransferHashMismatch);
480        }
481        let record: DeviceJoinJournalRecord = serde_json::from_str(&payload)?;
482        if record.attempt_id != attempt_id
483            || !matches!(*record.progress, DeviceJoinRoleProgress::Joiner(_))
484        {
485            return Err(DeviceJoinError::JournalConflict);
486        }
487        tx.execute(
488            "INSERT OR IGNORE INTO device_join_journals (attempt_id, role, payload)
489             VALUES (?1, 'joiner', ?2)",
490            (&attempt, &payload),
491        )?;
492        let installed: String = tx.query_row(
493            "SELECT payload FROM device_join_journals
494             WHERE attempt_id = ?1 AND role = 'joiner'",
495            [&attempt],
496            |row| row.get(0),
497        )?;
498        if installed != payload {
499            return Err(DeviceJoinError::JournalConflict);
500        }
501        tx.execute(
502            "DELETE FROM pending_join_source.pending_join_transfers WHERE attempt_id = ?1",
503            [&attempt],
504        )?;
505        tx.commit()?;
506        destination_connection.execute_batch("DETACH DATABASE pending_join_source")?;
507        Ok(record)
508    }
509}
510
511#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
512#[serde(rename_all = "snake_case", deny_unknown_fields)]
513pub enum DeviceJoinRole {
514    Owner,
515    ProviderAdministrator,
516    Joiner,
517}
518
519impl DeviceJoinRole {
520    pub(super) fn as_str(self) -> &'static str {
521        match self {
522            Self::Owner => "owner",
523            Self::ProviderAdministrator => "provider_administrator",
524            Self::Joiner => "joiner",
525        }
526    }
527}
528
529impl StoreDatabase {
530    pub async fn device_join_status(
531        &self,
532        attempt_id: DeviceJoinAttemptId,
533        role: DeviceJoinRole,
534    ) -> Result<Option<DeviceJoinStatus>, DeviceJoinError> {
535        load_store_device_join_status(self.sqlite(), attempt_id, role).await
536    }
537
538    pub async fn device_join_actions(&self) -> Result<Vec<DeviceJoinAction>, DeviceJoinError> {
539        load_store_device_join_actions(self.sqlite()).await
540    }
541}
542
543pub(crate) fn device_join_status(record: &DeviceJoinJournalRecord) -> DeviceJoinStatus {
544    match &*record.progress {
545        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Offered(offer))
546        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::OfferReceived(offer)) => {
547            DeviceJoinStatus::AwaitingAccessRequest {
548                offer: offer.clone(),
549            }
550        }
551        DeviceJoinRoleProgress::ProviderAdministrator(
552            ProviderAdminJoinProgress::AccessRequested(request),
553        )
554        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::AccessRequested(request)) => {
555            DeviceJoinStatus::AwaitingProviderAdmission {
556                request: request.clone(),
557            }
558        }
559        DeviceJoinRoleProgress::ProviderAdministrator(
560            ProviderAdminJoinProgress::ApprovalPrepared(approval),
561        )
562        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::ApprovalReceived(approval)) => {
563            DeviceJoinStatus::AwaitingRegistrationRequest {
564                approval: approval.clone(),
565            }
566        }
567        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::RegistrationRequested(request))
568        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::RegistrationPrepared(request)) => {
569            DeviceJoinStatus::AwaitingBootstrap {
570                request: request.clone(),
571            }
572        }
573        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::AttemptActivated(bootstrap))
574        | DeviceJoinRoleProgress::ProviderAdministrator(
575            ProviderAdminJoinProgress::AttemptObserved(bootstrap),
576        )
577        | DeviceJoinRoleProgress::ProviderAdministrator(
578            ProviderAdminJoinProgress::ChallengeCreateIntent(bootstrap),
579        ) => DeviceJoinStatus::AwaitingChallengePublication {
580            bootstrap: bootstrap.clone(),
581        },
582        DeviceJoinRoleProgress::ProviderAdministrator(
583            ProviderAdminJoinProgress::ProviderReady(bootstrap),
584        )
585        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::ProviderReady(bootstrap))
586        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::RegistrationCreateIntent(bootstrap)) => {
587            DeviceJoinStatus::AwaitingReadiness {
588                bootstrap: bootstrap.clone(),
589            }
590        }
591        DeviceJoinRoleProgress::ProviderAdministrator(
592            ProviderAdminJoinProgress::ResponseObserved(readiness),
593        )
594        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Ready(readiness)) => {
595            DeviceJoinStatus::AwaitingProviderCompletion {
596                readiness: readiness.clone(),
597            }
598        }
599        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ActivationCreateIntent {
600            completion,
601            ..
602        })
603        | DeviceJoinRoleProgress::ProviderAdministrator(ProviderAdminJoinProgress::Completed(
604            completion,
605        )) => DeviceJoinStatus::AwaitingActivation {
606            completion: completion.clone(),
607        },
608        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ActivationPrepared {
609            activation, ..
610        })
611        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::ActivationObserved {
612            activation,
613            ..
614        }) => DeviceJoinStatus::AwaitingCompletion {
615            activation: activation.clone(),
616        },
617        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Activated(store)) => {
618            DeviceJoinStatus::Activated {
619                store: store.clone(),
620            }
621        }
622        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Abandoned(abandonment))
623        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Abandoned(abandonment)) => {
624            DeviceJoinStatus::Abandoned {
625                abandonment: abandonment.clone(),
626            }
627        }
628        DeviceJoinRoleProgress::ProviderAdministrator(
629            ProviderAdminJoinProgress::AccessGrantPrepared { request, grant, .. },
630        ) => DeviceJoinStatus::ProviderAccessGrantCreatePending {
631            request: request.clone(),
632            grant: grant.clone(),
633        },
634        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::AbandonmentCreateIntent {
635            abandonment,
636            ..
637        }) => DeviceJoinStatus::AbandonmentCreatePending {
638            abandonment: abandonment.clone(),
639        },
640        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::CancellationCreateIntent {
641            cancellation,
642            ..
643        }) => DeviceJoinStatus::CancellationCreatePending {
644            cancellation: cancellation.clone(),
645        },
646        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Cancelled(cancellation)) => {
647            DeviceJoinStatus::CleanupPending {
648                cancellation: cancellation.clone(),
649                progress: DeviceJoinCleanupProgress::AwaitingBoth,
650            }
651        }
652        DeviceJoinRoleProgress::ProviderAdministrator(
653            ProviderAdminJoinProgress::CleanupIntent { cancellation, .. },
654        ) => DeviceJoinStatus::ProviderClosurePending {
655            cancellation: cancellation.clone(),
656            producer: DeviceJoinProducer::ProviderAdministrator,
657        },
658        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::CleanupIntent {
659            cancellation, ..
660        }) => DeviceJoinStatus::ProviderClosurePending {
661            cancellation: cancellation.clone(),
662            producer: DeviceJoinProducer::Joiner,
663        },
664        DeviceJoinRoleProgress::ProviderAdministrator(ProviderAdminJoinProgress::Cancelled(
665            closure,
666        )) => DeviceJoinStatus::ProviderClosed {
667            terminal: ProviderAdminJoinTerminal::Cancelled(closure.clone()),
668        },
669        DeviceJoinRoleProgress::ProviderAdministrator(ProviderAdminJoinProgress::WriteRevoked(
670            revocation,
671        )) => DeviceJoinStatus::ProviderClosed {
672            terminal: ProviderAdminJoinTerminal::WriteRevoked(revocation.clone()),
673        },
674        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Cancelled(closure)) => {
675            DeviceJoinStatus::JoinerClosed {
676                terminal: JoinerJoinTerminal::Cancelled(closure.clone()),
677            }
678        }
679        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::WriteRevoked(revocation)) => {
680            DeviceJoinStatus::JoinerClosed {
681                terminal: JoinerJoinTerminal::WriteRevoked(revocation.clone()),
682            }
683        }
684        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::CleanupReceiptCreateIntent {
685            cancellation,
686            receipt,
687            ..
688        }) => DeviceJoinStatus::CleanupReceiptCreatePending {
689            cancellation: cancellation.clone(),
690            receipt: receipt.clone(),
691        },
692        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::CleanupReceipt(receipt)) => {
693            DeviceJoinStatus::AwaitingCleanupActivation {
694                receipt: receipt.clone(),
695            }
696        }
697        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::CleanupActivated(activation))
698        | DeviceJoinRoleProgress::Owner(OwnerJoinProgress::CancelledComplete(activation))
699        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::CleanupActivated(activation))
700        | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::CancelledComplete(activation)) => {
701            DeviceJoinStatus::CleanupActivated {
702                activation: activation.clone(),
703            }
704        }
705        _ => DeviceJoinStatus::OperationInProgress {
706            attempt_id: record.attempt_id,
707        },
708    }
709}
710
711fn device_join_action(record: &DeviceJoinJournalRecord) -> Option<DeviceJoinAction> {
712    let resume = || DeviceJoinAction::ResumeOperation {
713        attempt_id: record.attempt_id,
714        role: record.progress.role(),
715    };
716    match &*record.progress {
717        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Offered(offer)) => {
718            Some(DeviceJoinAction::TransferOffer(offer.clone()))
719        }
720        DeviceJoinRoleProgress::Owner(
721            OwnerJoinProgress::RegistrationRequested(_)
722            | OwnerJoinProgress::AbandonmentCreateIntent { .. }
723            | OwnerJoinProgress::ActivationCreateIntent { .. }
724            | OwnerJoinProgress::CancellationCreateIntent { .. }
725            | OwnerJoinProgress::CleanupReceiptCreateIntent { .. },
726        ) => Some(resume()),
727        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::AttemptActivated(bootstrap)) => Some(
728            DeviceJoinAction::TransferProvisionalBootstrap(bootstrap.clone()),
729        ),
730        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::ActivationPrepared {
731            activation, ..
732        }) => Some(DeviceJoinAction::TransferActivation(activation.clone())),
733        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Abandoned(abandonment)) => {
734            Some(DeviceJoinAction::TransferAbandonment(abandonment.clone()))
735        }
736        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Cancelled(cancellation)) => {
737            Some(DeviceJoinAction::TransferCancellation(cancellation.clone()))
738        }
739        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::CleanupReceipt(receipt)) => {
740            Some(DeviceJoinAction::TransferCleanupReceipt(receipt.clone()))
741        }
742        DeviceJoinRoleProgress::Owner(
743            OwnerJoinProgress::CleanupActivated(activation)
744            | OwnerJoinProgress::CancelledComplete(activation),
745        ) => Some(DeviceJoinAction::TransferCleanupActivation(
746            activation.clone(),
747        )),
748
749        DeviceJoinRoleProgress::ProviderAdministrator(
750            ProviderAdminJoinProgress::AccessRequested(_)
751            | ProviderAdminJoinProgress::AccessGrantPrepared { .. }
752            | ProviderAdminJoinProgress::AttemptObserved(_)
753            | ProviderAdminJoinProgress::ChallengeCreateIntent(_)
754            | ProviderAdminJoinProgress::ResponseObserved(_)
755            | ProviderAdminJoinProgress::CleanupIntent { .. },
756        ) => Some(resume()),
757        DeviceJoinRoleProgress::ProviderAdministrator(
758            ProviderAdminJoinProgress::ApprovalPrepared(approval),
759        ) => Some(DeviceJoinAction::TransferProviderAdmissionApproval(
760            approval.clone(),
761        )),
762        DeviceJoinRoleProgress::ProviderAdministrator(
763            ProviderAdminJoinProgress::ProviderReady(bootstrap),
764        ) => Some(DeviceJoinAction::TransferProviderReadyBootstrap(
765            bootstrap.clone(),
766        )),
767        DeviceJoinRoleProgress::ProviderAdministrator(ProviderAdminJoinProgress::Completed(
768            completion,
769        )) => Some(DeviceJoinAction::TransferProviderAdmissionCompletion(
770            completion.clone(),
771        )),
772        DeviceJoinRoleProgress::ProviderAdministrator(ProviderAdminJoinProgress::Cancelled(
773            closure,
774        )) => Some(DeviceJoinAction::TransferProviderAdminTerminal(
775            ProviderAdminJoinTerminal::Cancelled(closure.clone()),
776        )),
777        DeviceJoinRoleProgress::ProviderAdministrator(ProviderAdminJoinProgress::WriteRevoked(
778            revocation,
779        )) => Some(DeviceJoinAction::TransferProviderAdminTerminal(
780            ProviderAdminJoinTerminal::WriteRevoked(revocation.clone()),
781        )),
782
783        DeviceJoinRoleProgress::Joiner(
784            JoinerJoinProgress::OfferReceived(_)
785            | JoinerJoinProgress::ApprovalReceived(_)
786            | JoinerJoinProgress::ProviderReady(_)
787            | JoinerJoinProgress::RegistrationCreateIntent(_)
788            | JoinerJoinProgress::RegistrationCreated(_)
789            | JoinerJoinProgress::AckCreateIntent(_)
790            | JoinerJoinProgress::AckCreated(_)
791            | JoinerJoinProgress::ResponseCreateIntent(_)
792            | JoinerJoinProgress::CleanupIntent { .. },
793        ) => Some(resume()),
794        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::AccessRequested(request)) => Some(
795            DeviceJoinAction::TransferProviderAccessRequest(request.clone()),
796        ),
797        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::RegistrationPrepared(request)) => Some(
798            DeviceJoinAction::TransferRegistrationRequest(request.clone()),
799        ),
800        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Ready(readiness)) => {
801            Some(DeviceJoinAction::TransferReadiness(readiness.clone()))
802        }
803        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::ActivationObserved {
804            activation,
805            ..
806        }) => Some(DeviceJoinAction::CompleteJoin(activation.clone())),
807        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::Cancelled(closure)) => {
808            Some(DeviceJoinAction::TransferJoinerTerminal(
809                JoinerJoinTerminal::Cancelled(closure.clone()),
810            ))
811        }
812        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::WriteRevoked(revocation)) => {
813            Some(DeviceJoinAction::TransferJoinerTerminal(
814                JoinerJoinTerminal::WriteRevoked(revocation.clone()),
815            ))
816        }
817        DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::CleanupActivated(activation)) => {
818            Some(DeviceJoinAction::CompleteCleanup(activation.clone()))
819        }
820        DeviceJoinRoleProgress::Joiner(
821            JoinerJoinProgress::Activated(_)
822            | JoinerJoinProgress::Abandoned(_)
823            | JoinerJoinProgress::CancelledComplete(_),
824        ) => None,
825    }
826}
827
828pub(crate) async fn load_store_device_join_actions(
829    db: &Database,
830) -> Result<Vec<DeviceJoinAction>, DeviceJoinError> {
831    let rows = db
832        .call(|connection| {
833            let mut statement = connection
834                .prepare(
835                    "SELECT key, value FROM protocol_state
836                     WHERE key GLOB 'device_join/*' ORDER BY key",
837                )
838                .map_err(crate::database::DbError::from)?;
839            let rows = statement
840                .query_map([], |row| {
841                    Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
842                })
843                .map_err(crate::database::DbError::from)?;
844            rows.collect::<Result<Vec<_>, _>>()
845                .map_err(crate::database::DbError::from)
846        })
847        .await
848        .map_err(database_error)?;
849    let mut records = Vec::with_capacity(rows.len());
850    for (key, value) in rows {
851        let record: DeviceJoinJournalRecord = serde_json::from_str(&value)?;
852        if store_journal_key(record.attempt_id, record.progress.role_name()) != key {
853            return Err(DeviceJoinError::JournalConflict);
854        }
855        records.push(record);
856    }
857    records.sort_by_key(|record| (record.attempt_id, record.progress.role()));
858    Ok(records.iter().filter_map(device_join_action).collect())
859}
860
861pub fn load_pending_device_join_actions(
862    pending: &DeviceJoinJournalDatabase,
863) -> Result<Vec<DeviceJoinAction>, DeviceJoinError> {
864    pending.actions()
865}
866
867pub(crate) async fn load_store_device_join_status(
868    db: &Database,
869    attempt_id: DeviceJoinAttemptId,
870    role: DeviceJoinRole,
871) -> Result<Option<DeviceJoinStatus>, DeviceJoinError> {
872    load_store_journal(db, attempt_id, role)
873        .await
874        .map(|record| record.as_ref().map(device_join_status))
875}
876
877pub fn load_pending_device_join_status(
878    pending: &DeviceJoinJournalDatabase,
879    attempt_id: DeviceJoinAttemptId,
880) -> Result<Option<DeviceJoinStatus>, DeviceJoinError> {
881    pending
882        .load(attempt_id, DeviceJoinRole::Joiner)
883        .map(|record| record.as_ref().map(device_join_status))
884}
885
886pub(super) async fn begin_store_journal(
887    db: &Database,
888    record: DeviceJoinJournalRecord,
889) -> Result<DeviceJoinJournalRecord, DeviceJoinError> {
890    validate_initial_progress(&record.progress)?;
891    let key = store_journal_key(record.attempt_id, record.progress.role_name());
892    let value = serde_json::to_string(&record)?;
893    db.call(move |connection| {
894        connection
895            .execute(
896                "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
897                (&key, &value),
898            )
899            .map_err(crate::database::DbError::from)?;
900        let actual = connection
901            .query_row(
902                "SELECT value FROM protocol_state WHERE key = ?1",
903                [&key],
904                |row| row.get::<_, String>(0),
905            )
906            .map_err(crate::database::DbError::from)?;
907        serde_json::from_str(&actual)
908            .map_err(|error| crate::database::DbError::Message(error.to_string()))
909    })
910    .await
911    .map_err(database_error)
912}
913
914pub(super) async fn begin_store_replacement_terminal(
915    db: &Database,
916    record: DeviceJoinJournalRecord,
917) -> Result<(), DeviceJoinError> {
918    if !matches!(
919        &*record.progress,
920        DeviceJoinRoleProgress::ProviderAdministrator(ProviderAdminJoinProgress::WriteRevoked(_))
921            | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::WriteRevoked(_))
922    ) {
923        return Err(DeviceJoinError::JournalConflict);
924    }
925    let key = store_journal_key(record.attempt_id, record.progress.role_name());
926    let value = serde_json::to_string(&record)?;
927    let durable = db
928        .call(move |connection| {
929            connection
930                .execute(
931                    "INSERT OR IGNORE INTO protocol_state (key, value) VALUES (?1, ?2)",
932                    (&key, &value),
933                )
934                .map_err(crate::database::DbError::from)?;
935            connection
936                .query_row(
937                    "SELECT value FROM protocol_state WHERE key = ?1",
938                    [&key],
939                    |row| row.get::<_, String>(0),
940                )
941                .map_err(crate::database::DbError::from)
942        })
943        .await
944        .map_err(database_error)?;
945    if durable != serde_json::to_string(&record)? {
946        return Err(DeviceJoinError::JournalConflict);
947    }
948    Ok(())
949}
950
951pub(super) async fn load_store_journal(
952    db: &Database,
953    attempt_id: DeviceJoinAttemptId,
954    role: DeviceJoinRole,
955) -> Result<Option<DeviceJoinJournalRecord>, DeviceJoinError> {
956    let key = store_journal_key(attempt_id, role.as_str());
957    let value = db.get_protocol_state(&key).await.map_err(database_error)?;
958    value
959        .map(|value| serde_json::from_str(&value).map_err(DeviceJoinError::from))
960        .transpose()
961}
962
963pub(super) async fn advance_store_journal(
964    db: &Database,
965    previous: &DeviceJoinJournalRecord,
966    next: DeviceJoinJournalRecord,
967) -> Result<(), DeviceJoinError> {
968    if previous.attempt_id != next.attempt_id {
969        return Err(DeviceJoinError::JournalConflict);
970    }
971    previous.progress.validate_transition(&next.progress)?;
972    let key = store_journal_key(previous.attempt_id, previous.progress.role_name());
973    let previous = serde_json::to_string(previous)?;
974    let next = serde_json::to_string(&next)?;
975    let changed = db
976        .call(move |connection| {
977            connection
978                .execute(
979                    "UPDATE protocol_state SET value = ?1 WHERE key = ?2 AND value = ?3",
980                    (&next, &key, &previous),
981                )
982                .map_err(crate::database::DbError::from)
983        })
984        .await
985        .map_err(database_error)?;
986    if changed == 1 {
987        Ok(())
988    } else {
989        Err(DeviceJoinError::JournalConflict)
990    }
991}
992
993pub(super) fn store_journal_key(attempt_id: DeviceJoinAttemptId, role: &str) -> String {
994    format!("device_join/{}/{role}", attempt_key(attempt_id))
995}
996
997pub(super) fn database_error(error: crate::database::DbError) -> DeviceJoinError {
998    DeviceJoinError::Store(error.into_message())
999}
1000
1001pub(super) fn provider_error(error: impl std::fmt::Display) -> DeviceJoinError {
1002    DeviceJoinError::Provider(error.to_string())
1003}
1004
1005fn owner_adjacent(previous: &OwnerJoinProgress, next: &OwnerJoinProgress) -> bool {
1006    matches!(
1007        (previous, next),
1008        (
1009            OwnerJoinProgress::Offered(_),
1010            OwnerJoinProgress::RegistrationRequested(_)
1011        ) | (
1012            OwnerJoinProgress::Offered(_),
1013            OwnerJoinProgress::AbandonmentCreateIntent { .. }
1014        ) | (
1015            OwnerJoinProgress::RegistrationRequested(_),
1016            OwnerJoinProgress::AttemptActivated(_)
1017        ) | (
1018            OwnerJoinProgress::RegistrationRequested(_),
1019            OwnerJoinProgress::AbandonmentCreateIntent { .. }
1020        ) | (
1021            OwnerJoinProgress::AbandonmentCreateIntent { .. },
1022            OwnerJoinProgress::Abandoned(_)
1023        ) | (
1024            OwnerJoinProgress::AttemptActivated(_),
1025            OwnerJoinProgress::ActivationCreateIntent { .. }
1026        ) | (
1027            OwnerJoinProgress::AttemptActivated(_),
1028            OwnerJoinProgress::CancellationCreateIntent { .. }
1029        ) | (
1030            OwnerJoinProgress::ActivationCreateIntent { .. },
1031            OwnerJoinProgress::ActivationPrepared { .. }
1032        ) | (
1033            OwnerJoinProgress::CancellationCreateIntent { .. },
1034            OwnerJoinProgress::Cancelled(_)
1035        ) | (
1036            OwnerJoinProgress::Cancelled(_),
1037            OwnerJoinProgress::CleanupReceiptCreateIntent { .. }
1038        ) | (
1039            OwnerJoinProgress::CleanupReceiptCreateIntent { .. },
1040            OwnerJoinProgress::CleanupReceipt(_)
1041        ) | (
1042            OwnerJoinProgress::CleanupReceipt(_),
1043            OwnerJoinProgress::CleanupActivated(_)
1044        ) | (
1045            OwnerJoinProgress::CleanupActivated(_),
1046            OwnerJoinProgress::CancelledComplete(_)
1047        )
1048    )
1049}
1050
1051fn provider_admin_adjacent(
1052    previous: &ProviderAdminJoinProgress,
1053    next: &ProviderAdminJoinProgress,
1054) -> bool {
1055    matches!(
1056        (previous, next),
1057        (
1058            ProviderAdminJoinProgress::AccessRequested(_),
1059            ProviderAdminJoinProgress::AccessGrantPrepared { .. }
1060        ) | (
1061            ProviderAdminJoinProgress::AccessGrantPrepared { .. },
1062            ProviderAdminJoinProgress::ApprovalPrepared(_)
1063        ) | (
1064            ProviderAdminJoinProgress::ApprovalPrepared(_),
1065            ProviderAdminJoinProgress::AttemptObserved(_)
1066        ) | (
1067            ProviderAdminJoinProgress::ApprovalPrepared(_),
1068            ProviderAdminJoinProgress::CleanupIntent { .. }
1069        ) | (
1070            ProviderAdminJoinProgress::AttemptObserved(_),
1071            ProviderAdminJoinProgress::ChallengeCreateIntent(_)
1072        ) | (
1073            ProviderAdminJoinProgress::AttemptObserved(_),
1074            ProviderAdminJoinProgress::CleanupIntent { .. }
1075        ) | (
1076            ProviderAdminJoinProgress::ChallengeCreateIntent(_),
1077            ProviderAdminJoinProgress::ProviderReady(_)
1078        ) | (
1079            ProviderAdminJoinProgress::ChallengeCreateIntent(_),
1080            ProviderAdminJoinProgress::CleanupIntent { .. }
1081        ) | (
1082            ProviderAdminJoinProgress::ProviderReady(_),
1083            ProviderAdminJoinProgress::ResponseObserved(_)
1084        ) | (
1085            ProviderAdminJoinProgress::ProviderReady(_),
1086            ProviderAdminJoinProgress::CleanupIntent { .. }
1087        ) | (
1088            ProviderAdminJoinProgress::ResponseObserved(_),
1089            ProviderAdminJoinProgress::Completed(_)
1090        ) | (
1091            ProviderAdminJoinProgress::ResponseObserved(_),
1092            ProviderAdminJoinProgress::CleanupIntent { .. }
1093        ) | (
1094            ProviderAdminJoinProgress::ApprovalPrepared(_),
1095            ProviderAdminJoinProgress::WriteRevoked(_)
1096        ) | (
1097            ProviderAdminJoinProgress::AttemptObserved(_),
1098            ProviderAdminJoinProgress::WriteRevoked(_)
1099        ) | (
1100            ProviderAdminJoinProgress::ChallengeCreateIntent(_),
1101            ProviderAdminJoinProgress::WriteRevoked(_)
1102        ) | (
1103            ProviderAdminJoinProgress::ProviderReady(_),
1104            ProviderAdminJoinProgress::WriteRevoked(_)
1105        ) | (
1106            ProviderAdminJoinProgress::ResponseObserved(_),
1107            ProviderAdminJoinProgress::WriteRevoked(_)
1108        ) | (
1109            ProviderAdminJoinProgress::CleanupIntent { .. },
1110            ProviderAdminJoinProgress::Cancelled(_)
1111        ) | (
1112            ProviderAdminJoinProgress::CleanupIntent { .. },
1113            ProviderAdminJoinProgress::WriteRevoked(_)
1114        )
1115    )
1116}
1117
1118fn joiner_adjacent(previous: &JoinerJoinProgress, next: &JoinerJoinProgress) -> bool {
1119    matches!(
1120        (previous, next),
1121        (
1122            JoinerJoinProgress::OfferReceived(_),
1123            JoinerJoinProgress::AccessRequested(_)
1124        ) | (
1125            JoinerJoinProgress::AccessRequested(_),
1126            JoinerJoinProgress::ApprovalReceived(_)
1127        ) | (
1128            JoinerJoinProgress::AccessRequested(_),
1129            JoinerJoinProgress::Abandoned(_)
1130        ) | (
1131            JoinerJoinProgress::ApprovalReceived(_),
1132            JoinerJoinProgress::RegistrationPrepared(_)
1133        ) | (
1134            JoinerJoinProgress::ApprovalReceived(_),
1135            JoinerJoinProgress::Abandoned(_)
1136        ) | (
1137            JoinerJoinProgress::RegistrationPrepared(_),
1138            JoinerJoinProgress::ProviderReady(_)
1139        ) | (
1140            JoinerJoinProgress::RegistrationPrepared(_),
1141            JoinerJoinProgress::CleanupIntent { .. }
1142        ) | (
1143            JoinerJoinProgress::ProviderReady(_),
1144            JoinerJoinProgress::RegistrationCreateIntent(_)
1145        ) | (
1146            JoinerJoinProgress::ProviderReady(_),
1147            JoinerJoinProgress::CleanupIntent { .. }
1148        ) | (
1149            JoinerJoinProgress::RegistrationCreateIntent(_),
1150            JoinerJoinProgress::RegistrationCreated(_)
1151        ) | (
1152            JoinerJoinProgress::RegistrationCreateIntent(_),
1153            JoinerJoinProgress::CleanupIntent { .. }
1154        ) | (
1155            JoinerJoinProgress::RegistrationCreated(_),
1156            JoinerJoinProgress::AckCreateIntent(_)
1157        ) | (
1158            JoinerJoinProgress::RegistrationCreated(_),
1159            JoinerJoinProgress::CleanupIntent { .. }
1160        ) | (
1161            JoinerJoinProgress::AckCreateIntent(_),
1162            JoinerJoinProgress::AckCreated(_)
1163        ) | (
1164            JoinerJoinProgress::AckCreateIntent(_),
1165            JoinerJoinProgress::CleanupIntent { .. }
1166        ) | (
1167            JoinerJoinProgress::AckCreated(_),
1168            JoinerJoinProgress::ResponseCreateIntent(_)
1169        ) | (
1170            JoinerJoinProgress::AckCreated(_),
1171            JoinerJoinProgress::Ready(_)
1172        ) | (
1173            JoinerJoinProgress::AckCreated(_),
1174            JoinerJoinProgress::CleanupIntent { .. }
1175        ) | (
1176            JoinerJoinProgress::ResponseCreateIntent(_),
1177            JoinerJoinProgress::Ready(_)
1178        ) | (
1179            JoinerJoinProgress::ResponseCreateIntent(_),
1180            JoinerJoinProgress::CleanupIntent { .. }
1181        ) | (
1182            JoinerJoinProgress::Ready(_),
1183            JoinerJoinProgress::ActivationObserved { .. }
1184        ) | (
1185            JoinerJoinProgress::Ready(_),
1186            JoinerJoinProgress::CleanupIntent { .. }
1187        ) | (
1188            JoinerJoinProgress::RegistrationPrepared(_),
1189            JoinerJoinProgress::WriteRevoked(_)
1190        ) | (
1191            JoinerJoinProgress::ProviderReady(_),
1192            JoinerJoinProgress::WriteRevoked(_)
1193        ) | (
1194            JoinerJoinProgress::RegistrationCreateIntent(_),
1195            JoinerJoinProgress::WriteRevoked(_)
1196        ) | (
1197            JoinerJoinProgress::RegistrationCreated(_),
1198            JoinerJoinProgress::WriteRevoked(_)
1199        ) | (
1200            JoinerJoinProgress::AckCreateIntent(_),
1201            JoinerJoinProgress::WriteRevoked(_)
1202        ) | (
1203            JoinerJoinProgress::AckCreated(_),
1204            JoinerJoinProgress::WriteRevoked(_)
1205        ) | (
1206            JoinerJoinProgress::ResponseCreateIntent(_),
1207            JoinerJoinProgress::WriteRevoked(_)
1208        ) | (
1209            JoinerJoinProgress::CleanupIntent { .. },
1210            JoinerJoinProgress::Cancelled(_)
1211        ) | (
1212            JoinerJoinProgress::ActivationObserved { .. },
1213            JoinerJoinProgress::Activated(_)
1214        ) | (
1215            JoinerJoinProgress::Cancelled(_),
1216            JoinerJoinProgress::CleanupActivated(_)
1217        ) | (
1218            JoinerJoinProgress::WriteRevoked(_),
1219            JoinerJoinProgress::CleanupActivated(_)
1220        ) | (
1221            JoinerJoinProgress::CleanupActivated(_),
1222            JoinerJoinProgress::CancelledComplete(_)
1223        )
1224    )
1225}
1226
1227fn validate_initial_progress(progress: &DeviceJoinRoleProgress) -> Result<(), DeviceJoinError> {
1228    if matches!(
1229        progress,
1230        DeviceJoinRoleProgress::Owner(OwnerJoinProgress::Offered(_))
1231            | DeviceJoinRoleProgress::ProviderAdministrator(
1232                ProviderAdminJoinProgress::AccessRequested(_)
1233            )
1234            | DeviceJoinRoleProgress::Joiner(JoinerJoinProgress::OfferReceived(_))
1235    ) {
1236        Ok(())
1237    } else {
1238        Err(DeviceJoinError::NonAdjacentJournalTransition)
1239    }
1240}
1241
1242pub(super) fn require_distinct_slots(slots: &[ObjectSlot]) -> Result<(), DeviceJoinError> {
1243    let unique = slots.iter().collect::<BTreeSet<_>>();
1244    if unique.len() == slots.len() {
1245        Ok(())
1246    } else {
1247        Err(DeviceJoinError::DuplicateReservedSlot)
1248    }
1249}
1250
1251pub(super) fn attempt_key(attempt_id: DeviceJoinAttemptId) -> String {
1252    serde_json::to_value(attempt_id)
1253        .expect("device join attempt id serialization cannot fail")
1254        .as_str()
1255        .expect("device join attempt id serializes as a string")
1256        .to_string()
1257}
coven_core/sync/store/device_join/journal.rs

coven_core/sync/store/device_join/
journal.rs
