diff --git a/crates/cascade-core/src/account.rs b/crates/cascade-core/src/account.rs index 6497203..2bdea37 100644 --- a/crates/cascade-core/src/account.rs +++ b/crates/cascade-core/src/account.rs @@ -5,9 +5,17 @@ //! request may start, what the user is told, and what success, failure and a //! 401 do. A shell only carries the requests it is handed (each one generic //! request, settled with one generic response) and stores the blob. +//! +//! The [`Account`] answers each account intent and each of its responses +//! itself, and tells the reducer only what follows for listening and storage, +//! as one [`AccountChange`]. use serde::{Deserialize, Serialize}; +use crate::effect::Effect; +use crate::server::{parse_verified, Request, RequestIds, StatusClass}; +use crate::snapshot::AccountSnapshot; + /// Storage-schema version for the persisted account blob. pub const ACCOUNT_VERSION: u32 = 1; @@ -22,14 +30,14 @@ pub struct Session { /// The one account request the shell is carrying, if any, and the id its /// response will carry. Only one runs at a time. #[derive(Debug, Clone, PartialEq, Eq)] -pub struct Pending { - pub id: u64, - pub request: PendingRequest, +struct Pending { + id: u64, + request: PendingRequest, } /// What a pending account request is for, and what its success needs. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum PendingRequest { +enum PendingRequest { SignInLink { email: String, }, @@ -46,7 +54,7 @@ pub enum PendingRequest { /// What the user was last told about their account. [`Self::label`] holds the /// copy, so every shell shows the same words. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum AccountStatus { +enum AccountStatus { EnterEmail, LinkSent { email: String }, LinkFailed, @@ -62,7 +70,7 @@ pub enum AccountStatus { } impl AccountStatus { - pub fn label(&self) -> String { + fn label(&self) -> String { match self { Self::EnterEmail => "Enter your email address.".into(), Self::LinkSent { email } => format!("Check {email} for a sign-in link."), @@ -80,33 +88,57 @@ impl AccountStatus { } } -/// The account as the core holds it. Held inside [`crate::State`]. +/// What an account transition means for listening and storage. The reducer +/// carries it through to the listening ledger and the persisted blobs; an +/// outcome that only changes what the user is told is not a change. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AccountChange { + /// A session was verified. `replaced` when it signed in over another + /// account, whose cross-device total must not show. + SignedIn { replaced: bool }, + /// The session is gone: signed out, or a 401. + SignedOut, + /// The server deleted this device's listening data: rotate the slot. + ListeningDeleted { new_device_id: String }, + /// The server deleted the account: rotate the slot, and the session is + /// gone. + AccountDeleted { new_device_id: String }, +} + +/// The account as the core holds it: the session, the one pending request, +/// and what the user was last told. Held inside [`crate::State`]; every +/// account rule lives here. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct Account { - pub session: Option, - pub pending: Option, - pub status: Option, + session: Option, + pending: Option, + status: Option, } impl Account { - /// An account request is out; every account command but sign-out waits. - pub fn busy(&self) -> bool { - self.pending.is_some() - } - - /// Take the pending request if response `id` settles it. A response for - /// anything else (nothing pending, or a dropped request) changes nothing. - pub fn settle(&mut self, id: u64) -> Option { - if self.pending.as_ref()?.id != id { - return None; - } - self.pending.take().map(|p| p.request) + /// Take the session from a stored blob: the versioned shape, or the + /// version-less one web and Apple stored before the core held the + /// account. Empty, garbage, an unknown version or an empty token mean no + /// account. + pub fn restore(&mut self, json: &str) { + self.session = session_from_json(json); } pub fn session_token(&self) -> Option { self.session.as_ref().map(|s| s.session_token.clone()) } + /// The account view for the UI. Never carries the session token. + pub fn snapshot(&self) -> AccountSnapshot { + let email = self.session.as_ref().map(|s| s.email.clone()); + AccountSnapshot { + signed_in_label: email.as_ref().map(|e| format!("Syncing · {e}")), + email, + status_label: self.status.as_ref().map(AccountStatus::label), + busy: self.busy(), + } + } + /// The blob for [`crate::Effect::PersistAccount`]: empty when signed out, /// meaning "delete the stored account". pub fn to_json(&self) -> String { @@ -123,19 +155,205 @@ impl Account { .unwrap_or_default() } - /// Read a stored blob: the versioned shape, or the version-less one web - /// and Apple stored before the core held the account. Empty, garbage, an - /// unknown version or an empty token mean no account. - pub fn session_from_json(json: &str) -> Option { - let stored: PersistedAccount = serde_json::from_str(json).ok()?; - if stored.version.is_some_and(|v| v != ACCOUNT_VERSION) || stored.session_token.is_empty() { - return None; + /// Ask the server to email a sign-in link to `email`, trimmed. A blank + /// email asks for one instead. + pub fn request_sign_in_link( + &mut self, + email: &str, + platform: Option<&str>, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + let email = email.trim().to_string(); + if email.is_empty() { + self.status = Some(AccountStatus::EnterEmail); + return; + } + let request = Request::sign_in_link(&email, platform); + self.begin( + PendingRequest::SignInLink { email }, + None, + request, + ids, + effects, + ); + } + + /// Redeem the sign-in token found in what the user pasted. + pub fn submit_sign_in_link( + &mut self, + input: &str, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + match parse_sign_in_token(input) { + Some(token) => self.begin( + PendingRequest::Verify, + Some(AccountStatus::SigningIn), + Request::verify(&token), + ids, + effects, + ), + None => self.status = Some(AccountStatus::PasteFullLink), + } + } + + /// Ask the server to delete this account's listening data. The slot + /// rotates to `new_device_id` only once the server confirms. + pub fn delete_listening( + &mut self, + new_device_id: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + if let Some(session_token) = self.session_token() { + self.begin( + PendingRequest::DeleteListening { new_device_id }, + None, + Request::delete_listening(session_token), + ids, + effects, + ); + } + } + + /// Ask the server to delete the account. The slot rotates to + /// `new_device_id` only once the server confirms. + pub fn delete_account( + &mut self, + new_device_id: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + if let Some(session_token) = self.session_token() { + self.begin( + PendingRequest::DeleteAccount { new_device_id }, + None, + Request::delete_account(session_token), + ids, + effects, + ); } - Some(Session { - session_token: stored.session_token, - email: stored.email, - }) } + + /// Sign out, cutting in on any pending request. Revokes the session if + /// there is one; nothing waits for the revoke's response. + pub fn sign_out( + &mut self, + ids: &mut RequestIds, + effects: &mut Vec, + ) -> Option { + self.pending = None; + self.status = None; + let session = self.session.take()?; + ids.send(Request::revoke(session.session_token), effects); + Some(AccountChange::SignedOut) + } + + /// Whether the pending request is request `id`. + pub fn awaits(&self, id: u64) -> bool { + self.pending.as_ref().is_some_and(|p| p.id == id) + } + + /// Settle the pending request with its response. The caller has already + /// matched the response's id with [`Self::awaits`]. A 401 while signed in + /// expires the session; any other failure only changes what the user is + /// told. + pub fn settle(&mut self, class: StatusClass, body: &str) -> Option { + let request = self.pending.take()?.request; + if class == StatusClass::Unauthorized && self.session.is_some() { + return Some(self.expire()); + } + let succeeded = class == StatusClass::Success; + let (status, change) = match request { + PendingRequest::SignInLink { email } if succeeded => { + (AccountStatus::LinkSent { email }, None) + } + PendingRequest::Verify if succeeded => match parse_verified(body) { + Some(session) => { + let status = AccountStatus::SignedIn { + email: session.email.clone(), + }; + let replaced = self.session.replace(session).is_some(); + (status, Some(AccountChange::SignedIn { replaced })) + } + None => (AccountStatus::LinkInvalid, None), + }, + PendingRequest::DeleteListening { new_device_id } if succeeded => ( + AccountStatus::ListeningDeleted, + Some(AccountChange::ListeningDeleted { new_device_id }), + ), + PendingRequest::DeleteAccount { new_device_id } if succeeded => { + self.session = None; + ( + AccountStatus::AccountDeleted, + Some(AccountChange::AccountDeleted { new_device_id }), + ) + } + PendingRequest::SignInLink { .. } => (AccountStatus::LinkFailed, None), + PendingRequest::Verify => (AccountStatus::LinkInvalid, None), + PendingRequest::DeleteListening { .. } => (AccountStatus::ListeningDeleteFailed, None), + PendingRequest::DeleteAccount { .. } => (AccountStatus::AccountDeleteFailed, None), + }; + self.status = Some(status); + change + } + + /// The server rejected the session (a 401 on any request, including a + /// listening push): sign out and tell the user to sign in again. No + /// revoke, since the session is already gone server-side. + pub fn expire(&mut self) -> AccountChange { + self.pending = None; + self.status = Some(AccountStatus::SessionExpired); + self.session = None; + AccountChange::SignedOut + } + + /// An account request is out; every account intent but sign-out waits. + fn busy(&self) -> bool { + self.pending.is_some() + } + + /// Send the one account request, remembering its `purpose`. + fn begin( + &mut self, + purpose: PendingRequest, + status: Option, + request: Request, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + let id = ids.send(request, effects); + self.pending = Some(Pending { + id, + request: purpose, + }); + self.status = status; + } +} + +/// Read a stored account blob into a session; see [`Account::restore`]. +fn session_from_json(json: &str) -> Option { + let stored: PersistedAccount = serde_json::from_str(json).ok()?; + if stored.version.is_some_and(|v| v != ACCOUNT_VERSION) || stored.session_token.is_empty() { + return None; + } + Some(Session { + session_token: stored.session_token, + email: stored.email, + }) } /// The persisted account blob. `version` is optional on read only, so the @@ -151,7 +369,7 @@ pub struct PersistedAccount { /// Find the sign-in token in what the user pasted: a link carrying a query /// parameter named exactly `token`, or the bare token itself. -pub fn parse_sign_in_token(input: &str) -> Option { +fn parse_sign_in_token(input: &str) -> Option { let input = input.trim(); if let Some(value) = token_query_value(input) { let token = percent_decode(value); diff --git a/crates/cascade-core/src/listening.rs b/crates/cascade-core/src/listening.rs index 43d3a12..e37e207 100644 --- a/crates/cascade-core/src/listening.rs +++ b/crates/cascade-core/src/listening.rs @@ -17,6 +17,9 @@ use serde::{Deserialize, Serialize}; +use crate::effect::Effect; +use crate::server::{parse_server_total, Request, RequestIds, StatusClass}; + /// Storage-schema version for the persisted listening blob. Separate from /// `SETTINGS_VERSION` because listening data lives in its own blob /// (`cascade.listening.v1`), distinct from user settings. @@ -104,6 +107,20 @@ impl InFlight { } } +/// How the PUT in flight settled. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Settled { + /// The server accepted it: the ledger moved forward and needs persisting. + Synced, + /// Not sent, no response, or any failure but a 401. + Failed, + /// The server rejected the session. + Unauthorized, + /// It was superseded (or nothing was in flight), so its response was + /// dropped. + Dropped, +} + impl Default for ListeningLedger { fn default() -> Self { Self { @@ -159,79 +176,95 @@ impl ListeningLedger { self.device_total_ms = self.device_total_ms.saturating_add(delta); } - /// Decide whether to sync for a shell's `reason`, and if so what to send: - /// `(device_id, device_total_ms)`. Marks that total as in flight as - /// request `id`, so a second call returns `None` until - /// [`Self::sync_succeeded`] or [`Self::sync_failed`] settles it — the one - /// re-entrancy guard every shell shares. - pub fn begin_sync(&mut self, reason: SyncReason, id: u64) -> Option<(String, u64)> { + /// Sync for a shell's `reason` if there is anything to say: push this + /// slot's total under `session_token`, with a fresh id from `ids`, and + /// mark it in flight, so nothing more is pushed until [`Self::settle`] + /// settles it — the one re-entrancy guard every shell shares. + pub fn begin_sync( + &mut self, + reason: SyncReason, + session_token: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { let worth_sending = match reason { SyncReason::Flush => self.unsynced_ms() > 0, SyncReason::Refresh => true, }; - self.begin_sync_if(worth_sending, id) + self.begin_sync_if(worth_sending, session_token, ids, effects); } /// The core's own routine sync, checked on every tick that counts - /// listening: send only once at least [`LISTENING_SYNC_THRESHOLD_MS`] is - /// unsynced, counted past any failed PUT. Same result and in-flight guard + /// listening: push only once at least [`LISTENING_SYNC_THRESHOLD_MS`] is + /// unsynced, counted past any failed PUT. Same push and in-flight guard /// as [`Self::begin_sync`]. - pub fn begin_threshold_sync(&mut self, id: u64) -> Option<(String, u64)> { + pub fn begin_threshold_sync( + &mut self, + session_token: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { let worth_sending = self.unsent_ms() >= LISTENING_SYNC_THRESHOLD_MS; - self.begin_sync_if(worth_sending, id) - } - - fn begin_sync_if(&mut self, worth_sending: bool, id: u64) -> Option<(String, u64)> { - if self.in_flight.is_some() { - return None; - } - let device_id = self.device_id.clone()?; - if !worth_sending { - return None; + self.begin_sync_if(worth_sending, session_token, ids, effects); + } + + fn begin_sync_if( + &mut self, + worth_sending: bool, + session_token: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.in_flight.is_some() || !worth_sending { + return; } + let Some(device_id) = self.device_id.as_deref() else { + return; + }; + let request = Request::push_listening(device_id, self.device_total_ms, session_token); + let id = ids.send(request, effects); self.in_flight = Some(InFlight::Sent { id, device_total_ms: self.device_total_ms, }); - Some((device_id, self.device_total_ms)) - } - - /// The in-flight PUT succeeded. The server has accepted exactly what was - /// sent — not whatever accrued while the request was out — and reports - /// `server_total_ms` as the cross-device aggregate. This only moves the - /// display baseline forward; it never touches `device_total_ms`, and is - /// monotonic so an out-of-order ack can't rewind the high-water mark. - /// Returns `false` (and changes nothing else) if no sync for the current - /// slot was in flight. The caller has already matched the response's id - /// to [`Self::in_flight`]. - pub fn sync_succeeded(&mut self, server_total_ms: u64) -> bool { + } + + /// Settle the PUT in flight with its response. The caller has already + /// matched the response's id with [`Self::awaits`]. + /// + /// On success the server has accepted exactly what was sent — not + /// whatever accrued while the request was out — and reports the + /// cross-device aggregate. That only moves the display baseline forward; + /// it never touches `device_total_ms`, and is monotonic so an + /// out-of-order ack can't rewind the high-water mark. On failure nothing + /// was acknowledged: a flush or refresh may try again at once, and a + /// threshold sync once another threshold has accrued past what was sent. + /// A 2xx whose body does not parse is a failure. A superseded PUT only + /// frees the slot: its outcome refers to a slot or session already + /// replaced. + pub fn settle(&mut self, class: StatusClass, body: &str) -> Settled { let Some(InFlight::Sent { device_total_ms: sent, .. }) = self.in_flight.take() else { - return false; + return Settled::Dropped; + }; + let server_total_ms = match class { + StatusClass::Success => parse_server_total(body), + StatusClass::Unauthorized | StatusClass::Failed => None, + }; + let Some(server_total_ms) = server_total_ms else { + self.failed_through_ms = Some(sent); + return match class { + StatusClass::Unauthorized => Settled::Unauthorized, + StatusClass::Success | StatusClass::Failed => Settled::Failed, + }; }; self.synced_through_ms = self.synced_through_ms.max(sent); self.server_total_ms = Some(server_total_ms); self.failed_through_ms = None; - true - } - - /// The in-flight PUT failed. Nothing was acknowledged; a flush or refresh - /// may try again at once, and a threshold sync once another threshold has - /// accrued past what was sent. Returns `false` if no sync for the current - /// slot was in flight, so a superseded request's failure means nothing. - pub fn sync_failed(&mut self) -> bool { - let Some(InFlight::Sent { - device_total_ms: sent, - .. - }) = self.in_flight.take() - else { - return false; - }; - self.failed_through_ms = Some(sent); - true + Settled::Synced } /// Take `fallback` as this device's id if it doesn't have one yet. @@ -351,12 +384,20 @@ pub fn format_listening_total(total_ms: u64) -> String { mod tests { use super::*; + /// Start a sync for `reason`, returning the effects it pushed. + fn begin(l: &mut ListeningLedger, reason: SyncReason, ids: &mut RequestIds) -> Vec { + let mut effects = Vec::new(); + l.begin_sync(reason, "token".into(), ids, &mut effects); + effects + } + /// Send whatever is unsynced and have the server accept it. fn sync(l: &mut ListeningLedger, server_total_ms: u64) { l.device_id.get_or_insert_with(|| "device".into()); - l.begin_sync(SyncReason::Refresh, 1) - .expect("nothing in flight"); - assert!(l.sync_succeeded(server_total_ms)); + let pushed = begin(l, SyncReason::Refresh, &mut RequestIds::default()); + assert_eq!(pushed.len(), 1, "nothing in flight"); + let body = format!(r#"{{"serverTotalMs":{server_total_ms}}}"#); + assert_eq!(l.settle(StatusClass::Success, &body), Settled::Synced); } #[test] @@ -424,16 +465,42 @@ mod tests { l.accrue(4_000); sync(&mut l, 9_000); l.accrue(1_000); - l.begin_sync(SyncReason::Flush, 2) - .expect("nothing in flight"); + let pushed = begin(&mut l, SyncReason::Flush, &mut RequestIds::default()); + assert_eq!(pushed.len(), 1, "nothing in flight"); l.forget_server(); assert_eq!(l.server_total_ms, None); - assert_eq!(l.in_flight, Some(InFlight::Superseded { id: 2 })); + assert_eq!(l.in_flight, Some(InFlight::Superseded { id: 1 })); assert_eq!(l.device_total_ms, 5_000); assert_eq!(l.synced_through_ms, 4_000); assert_eq!(l.displayed_total_ms(), 5_000); } + #[test] + fn settle_names_each_outcome() { + let mut ids = RequestIds::default(); + let mut l = ListeningLedger { + device_id: Some("device".into()), + ..ListeningLedger::default() + }; + l.accrue(1_000); + for (class, body, outcome) in [ + (StatusClass::Failed, "", Settled::Failed), + (StatusClass::Success, "not json", Settled::Failed), + (StatusClass::Unauthorized, "", Settled::Unauthorized), + ] { + begin(&mut l, SyncReason::Refresh, &mut ids); + assert_eq!(l.settle(class, body), outcome, "{class:?} {body}"); + assert_eq!(l.failed_through_ms, Some(1_000)); + assert_eq!(l.in_flight, None); + } + begin(&mut l, SyncReason::Refresh, &mut ids); + l.forget_server(); + let ok = r#"{"serverTotalMs":9000}"#; + assert_eq!(l.settle(StatusClass::Success, ok), Settled::Dropped); + assert_eq!(l.server_total_ms, None); + assert_eq!(l.in_flight, None); + } + #[test] fn restore_never_regresses_the_lifetime_total() { let mut live = ListeningLedger::default(); diff --git a/crates/cascade-core/src/server.rs b/crates/cascade-core/src/server.rs index 01e31ff..8fad101 100644 --- a/crates/cascade-core/src/server.rs +++ b/crates/cascade-core/src/server.rs @@ -109,7 +109,7 @@ impl Request { } /// The effect that asks the shell to carry this request as `id`. - pub fn into_effect(self, id: u64) -> Effect { + fn into_effect(self, id: u64) -> Effect { Effect::ServerRequest { id, method: self.method, @@ -120,6 +120,31 @@ impl Request { } } +/// The one source of request ids, held in [`crate::State`] and shared by every +/// module that sends. Starts at 1, is never persisted (a restart forgets every +/// request in flight along with the shell process that was carrying it), and +/// never reuses an id. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RequestIds { + next: u64, +} + +impl Default for RequestIds { + fn default() -> Self { + Self { next: 1 } + } +} + +impl RequestIds { + /// Hand the shell `request` under a fresh id, and return that id. + pub fn send(&mut self, request: Request, effects: &mut Vec) -> u64 { + let id = self.next; + self.next += 1; + effects.push(request.into_effect(id)); + id + } +} + fn to_json(body: &impl Serialize) -> Option { serde_json::to_string(body).ok() } diff --git a/crates/cascade-core/src/snapshot.rs b/crates/cascade-core/src/snapshot.rs index af8c8bf..82959d9 100644 --- a/crates/cascade-core/src/snapshot.rs +++ b/crates/cascade-core/src/snapshot.rs @@ -114,19 +114,6 @@ pub struct AccountSnapshot { pub busy: bool, } -impl AccountSnapshot { - fn from_state(state: &State) -> Self { - let account = &state.account; - let email = account.session.as_ref().map(|s| s.email.clone()); - Self { - signed_in_label: email.as_ref().map(|e| format!("Syncing · {e}")), - email, - status_label: account.status.as_ref().map(|s| s.label()), - busy: account.busy(), - } - } -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Snapshot { @@ -238,7 +225,7 @@ impl Snapshot { unsynced_ms: state.listening.unsynced_ms(), total_label: format_listening_total(state.listening.displayed_total_ms()), }, - account: AccountSnapshot::from_state(state), + account: state.account.snapshot(), tick_interval_ms: tick_interval_ms(state), } } diff --git a/crates/cascade-core/src/state.rs b/crates/cascade-core/src/state.rs index 762e944..04ec37a 100644 --- a/crates/cascade-core/src/state.rs +++ b/crates/cascade-core/src/state.rs @@ -6,13 +6,11 @@ use serde::{Deserialize, Serialize}; -use crate::account::{ - parse_sign_in_token, Account, AccountStatus, Pending, PendingRequest, Session, -}; +use crate::account::{Account, AccountChange}; use crate::command::Command; use crate::effect::Effect; -use crate::listening::{ListeningLedger, PersistedListening, SyncReason}; -use crate::server::{parse_server_total, parse_verified, Request, StatusClass}; +use crate::listening::{ListeningLedger, PersistedListening, Settled, SyncReason}; +use crate::server::{RequestIds, StatusClass}; use crate::settings::{PersistedSettings, SETTINGS_VERSION}; use crate::timer::{clamp_minutes, ActiveTimer, TimerKind, DEFAULT_TIMER_MINUTES}; @@ -66,10 +64,8 @@ pub struct State { /// The optional sign-in. Restored from its own blob via /// [`Command::Restore`]'s `account_json`. pub account: Account, - /// The id the next [`Effect::ServerRequest`] gets. Starts at 1 and only - /// grows; never persisted, since a restart forgets every request in - /// flight along with the shell process that was carrying it. - next_request_id: u64, + /// Where every [`Effect::ServerRequest`] gets its id. + request_ids: RequestIds, /// Whether [`Command::Restore`] has already run. Only the first restore /// takes effect, so a replayed stale blob can't undo a later reset. restored: bool, @@ -89,7 +85,7 @@ impl Default for State { audio_confirmed_playing: false, listening: ListeningLedger::default(), account: Account::default(), - next_request_id: 1, + request_ids: RequestIds::default(), restored: false, } } @@ -159,59 +155,60 @@ fn push_persist_account(state: &State, effects: &mut Vec) { }); } -/// Hand the shell `request` under a fresh id, and return that id. -fn send(state: &mut State, request: Request, effects: &mut Vec) -> u64 { - let id = state.next_request_id; - state.next_request_id += 1; - effects.push(request.into_effect(id)); - id +/// Carry an account transition through to listening and storage. The one +/// place the order of those writes is decided: within one update the +/// listening blob is written before the account blob, so a crash between the +/// two never leaves a deleted account's slot with no session to answer for it +/// (ADR 0001, decision 6). +fn apply_account_change(state: &mut State, change: AccountChange, effects: &mut Vec) { + match change { + AccountChange::SignedIn { replaced } => { + // Signing in over another account: nothing of its total may show. + if replaced { + state.listening.forget_server(); + push_persist_listening(state, effects); + } + push_persist_account(state, effects); + begin_listening_sync(state, effects, |l, token, ids, effects| { + l.begin_sync(SyncReason::Refresh, token, ids, effects) + }); + } + AccountChange::SignedOut => { + state.listening.forget_server(); + push_persist_listening(state, effects); + push_persist_account(state, effects); + } + AccountChange::ListeningDeleted { new_device_id } => { + state.listening.reset(new_device_id); + push_persist_listening(state, effects); + } + AccountChange::AccountDeleted { new_device_id } => { + // The reset forgets the server total and supersedes any in-flight + // sync too, so the slot and the sign-out share one listening write. + state.listening.reset(new_device_id); + push_persist_listening(state, effects); + push_persist_account(state, effects); + } + } } -/// Start a listening sync if `begin` finds anything to send and the account -/// allows it, answering with one push. `begin` is handed the id the push -/// will carry, so the ledger can remember it. +/// Offer a listening sync if signed in: `begin` is handed the session token +/// and the id source, and the ledger decides whether to push. fn begin_listening_sync( state: &mut State, effects: &mut Vec, - begin: impl FnOnce(&mut ListeningLedger, u64) -> Option<(String, u64)>, + begin: impl FnOnce(&mut ListeningLedger, String, &mut RequestIds, &mut Vec), ) { - let Some(session_token) = state.account.session_token() else { - return; - }; - let id = state.next_request_id; - if let Some((device_id, device_total_ms)) = begin(&mut state.listening, id) { - let request = Request::push_listening(&device_id, device_total_ms, session_token); - let sent = send(state, request, effects); - debug_assert_eq!(sent, id, "the push must carry the id the ledger remembers"); + if let Some(session_token) = state.account.session_token() { + begin( + &mut state.listening, + session_token, + &mut state.request_ids, + effects, + ); } } -/// Start the one account request, answering with `request`. The caller has -/// already checked that none is pending. -fn begin_account_request( - state: &mut State, - pending: PendingRequest, - status: Option, - request: Request, - effects: &mut Vec, -) { - let id = send(state, request, effects); - state.account.pending = Some(Pending { - id, - request: pending, - }); - state.account.status = status; -} - -/// Drop the session and everything learned through it: the stored blob, the -/// cross-device total, and any sync still in flight for it. -fn forget_session(state: &mut State, effects: &mut Vec) { - state.account.session = None; - push_persist_account(state, effects); - state.listening.forget_server(); - push_persist_listening(state, effects); -} - /// Whether a tick should count as listening: tracking on, audio *confirmed* /// playing (not merely intended), and not muted. fn is_accruing(state: &State) -> bool { @@ -385,10 +382,11 @@ pub fn reduce(state: &mut State, command: Command, effects: &mut Vec) { // Persist a newly adopted id right away, so the next launch sees // the same slot rather than another fallback. let mut persist_listening = state.listening.adopt_device_id(fallback_device_id); - state.account.session = Account::session_from_json(&account_json); + state.account.restore(&account_json); // A cross-device total with no account behind it is left over // from a sign-out before the core held the account. - if state.account.session.is_none() && state.listening.server_total_ms.is_some() { + if state.account.session_token().is_none() && state.listening.server_total_ms.is_some() + { state.listening.forget_server(); persist_listening = true; } @@ -397,177 +395,61 @@ pub fn reduce(state: &mut State, command: Command, effects: &mut Vec) { } } Command::BeginListeningSync { reason } => { - begin_listening_sync(state, effects, |l, id| l.begin_sync(reason, id)) + begin_listening_sync(state, effects, |l, token, ids, effects| { + l.begin_sync(reason, token, ids, effects) + }) } Command::ResetListeningData { new_device_id } => { state.listening.reset(new_device_id); push_persist_listening(state, effects); } - // Every account request waits for the pending one; only sign-out - // cuts in. - Command::RequestSignInLink { .. } - | Command::SubmitSignInLink { .. } - | Command::DeleteListeningData { .. } - | Command::DeleteAccount { .. } - if state.account.busy() => {} - Command::RequestSignInLink { email, platform } => { - let email = email.trim().to_string(); - if email.is_empty() { - state.account.status = Some(AccountStatus::EnterEmail); - return; - } - let request = Request::sign_in_link(&email, platform.as_deref()); - begin_account_request( - state, - PendingRequest::SignInLink { email }, - None, - request, - effects, - ); + Command::RequestSignInLink { email, platform } => state.account.request_sign_in_link( + &email, + platform.as_deref(), + &mut state.request_ids, + effects, + ), + Command::SubmitSignInLink { input } => { + state + .account + .submit_sign_in_link(&input, &mut state.request_ids, effects) } - Command::SubmitSignInLink { input } => match parse_sign_in_token(&input) { - Some(token) => begin_account_request( - state, - PendingRequest::Verify, - Some(AccountStatus::SigningIn), - Request::verify(&token), - effects, - ), - None => state.account.status = Some(AccountStatus::PasteFullLink), - }, Command::DeleteListeningData { new_device_id } => { - if let Some(session_token) = state.account.session_token() { - begin_account_request( - state, - PendingRequest::DeleteListening { new_device_id }, - None, - Request::delete_listening(session_token), - effects, - ); - } + state + .account + .delete_listening(new_device_id, &mut state.request_ids, effects) } Command::DeleteAccount { new_device_id } => { - if let Some(session_token) = state.account.session_token() { - begin_account_request( - state, - PendingRequest::DeleteAccount { new_device_id }, - None, - Request::delete_account(session_token), - effects, - ); - } + state + .account + .delete_account(new_device_id, &mut state.request_ids, effects) } Command::SignOut => { - state.account.pending = None; - state.account.status = None; - if let Some(session_token) = state.account.session_token() { - // Remembered nowhere, so its response is ignored. - send(state, Request::revoke(session_token), effects); - forget_session(state, effects); + if let Some(change) = state.account.sign_out(&mut state.request_ids, effects) { + apply_account_change(state, change, effects); } } Command::ServerResponse { id, status, body } => { let class = StatusClass::of(status); - if let Some(request) = state.account.settle(id) { - account_response(state, request, class, &body, effects); + if state.account.awaits(id) { + if let Some(change) = state.account.settle(class, &body) { + apply_account_change(state, change, effects); + } } else if state.listening.awaits(id) { - sync_response(state, class, &body, effects); + match state.listening.settle(class, &body) { + Settled::Synced => push_persist_listening(state, effects), + Settled::Unauthorized => { + let change = state.account.expire(); + apply_account_change(state, change, effects); + } + Settled::Failed | Settled::Dropped => {} + } } } } } -/// Settle the in-flight listening push. A superseded push only frees the -/// slot: its success or 401 refers to a slot or session already replaced. -fn sync_response(state: &mut State, class: StatusClass, body: &str, effects: &mut Vec) { - let server_total_ms = match class { - StatusClass::Success => parse_server_total(body), - StatusClass::Unauthorized | StatusClass::Failed => None, - }; - if let Some(server_total_ms) = server_total_ms { - if state.listening.sync_succeeded(server_total_ms) { - push_persist_listening(state, effects); - } - } else if state.listening.sync_failed() && class == StatusClass::Unauthorized { - session_expired(state, effects); - } -} - -/// Settle the pending account `request`, already taken off the account. -fn account_response( - state: &mut State, - request: PendingRequest, - class: StatusClass, - body: &str, - effects: &mut Vec, -) { - if class == StatusClass::Unauthorized && state.account.session.is_some() { - session_expired(state, effects); - return; - } - let succeeded = class == StatusClass::Success; - match request { - PendingRequest::SignInLink { email } if succeeded => { - state.account.status = Some(AccountStatus::LinkSent { email }); - } - PendingRequest::Verify if succeeded => match parse_verified(body) { - Some(session) => signed_in(state, session, effects), - None => state.account.status = Some(AccountStatus::LinkInvalid), - }, - PendingRequest::DeleteListening { new_device_id } if succeeded => { - state.account.status = Some(AccountStatus::ListeningDeleted); - state.listening.reset(new_device_id); - push_persist_listening(state, effects); - } - PendingRequest::DeleteAccount { new_device_id } if succeeded => { - state.account.status = Some(AccountStatus::AccountDeleted); - // The reset forgets the server total and supersedes any in-flight - // sync too, so the slot and the sign-out share one listening write. - state.listening.reset(new_device_id); - push_persist_listening(state, effects); - state.account.session = None; - push_persist_account(state, effects); - } - PendingRequest::SignInLink { .. } => { - state.account.status = Some(AccountStatus::LinkFailed); - } - PendingRequest::Verify => state.account.status = Some(AccountStatus::LinkInvalid), - PendingRequest::DeleteListening { .. } => { - state.account.status = Some(AccountStatus::ListeningDeleteFailed); - } - PendingRequest::DeleteAccount { .. } => { - state.account.status = Some(AccountStatus::AccountDeleteFailed); - } - } -} - -/// The sign-in token was redeemed for a session: persist the account and -/// start a refresh sync for the cross-device total. -fn signed_in(state: &mut State, session: Session, effects: &mut Vec) { - // Signing in over another account: nothing of its total may show. - if state.account.session.is_some() { - state.listening.forget_server(); - push_persist_listening(state, effects); - } - state.account.status = Some(AccountStatus::SignedIn { - email: session.email.clone(), - }); - state.account.session = Some(session); - push_persist_account(state, effects); - begin_listening_sync(state, effects, |l, id| { - l.begin_sync(SyncReason::Refresh, id) - }); -} - -/// The server rejected the session (a 401 on any request): sign out and tell -/// the user to sign in again. Listening carries on locally. -fn session_expired(state: &mut State, effects: &mut Vec) { - state.account.pending = None; - state.account.status = Some(AccountStatus::SessionExpired); - forget_session(state, effects); -} - fn start_playback(state: &mut State, effects: &mut Vec) { if state.intent.is_playing() { return; diff --git a/crates/cascade-core/tests/account.rs b/crates/cascade-core/tests/account.rs index 73570a9..abd84f0 100644 --- a/crates/cascade-core/tests/account.rs +++ b/crates/cascade-core/tests/account.rs @@ -1008,6 +1008,47 @@ fn a_401_while_signed_out_is_an_ordinary_failure() { ); } +// ---- write order ----------------------------------------------------------- +// +// Within one update the listening blob is written before the account blob, so +// a crash between the two never leaves a deleted account's slot with no +// session to answer for it (ADR 0001, decision 6). + +/// The persist effects in `effects`, in order, by blob. +fn persist_order(effects: &[Effect]) -> Vec<&'static str> { + effects + .iter() + .filter_map(|e| match e { + Effect::PersistListening { .. } => Some("listening"), + Effect::PersistAccount { .. } => Some("account"), + _ => None, + }) + .collect() +} + +#[test] +fn an_account_delete_writes_listening_before_the_account() { + let mut core = signed_in(); + listen(&mut core, 2_000); + let id = start(&mut core, delete_account()); + let effects = core.dispatch(bare_response(id, 204)).effects; + assert_eq!(persist_order(&effects), ["listening", "account"]); +} + +#[test] +fn a_401_writes_listening_before_the_account() { + let mut core = signed_in(); + listen(&mut core, 2_000); + let id = request_id(&begin(&mut core, SyncReason::Refresh)); + let effects = core.dispatch(bare_response(id, 401)).effects; + assert_eq!(persist_order(&effects), ["listening", "account"], "sync"); + + let mut core = signed_in(); + let id = start(&mut core, delete_listening()); + let effects = core.dispatch(bare_response(id, 401)).effects; + assert_eq!(persist_order(&effects), ["listening", "account"], "request"); +} + // ---- wire shape ------------------------------------------------------------ // // Every shell hand-writes these JSON shapes. Lock them. diff --git a/docs/adr/0001-listening-time-crdt.md b/docs/adr/0001-listening-time-crdt.md index b7b088e..bf77599 100644 --- a/docs/adr/0001-listening-time-crdt.md +++ b/docs/adr/0001-listening-time-crdt.md @@ -69,7 +69,7 @@ The table and its routing are tested through `Core::dispatch` in `crates/cascade Auth is email magic-link (no passwords) with **opaque server-side session tokens** (not JWT), so logout / delete-account revoke instantly with one `DELETE`. Tokens are stored only as SHA-256 hashes. "Delete my data" rotates the client's `device_id`, closing the one loophole inherent to grow-only counters: a forgotten offline device can't later resurrect a deleted total by pushing a stale higher counter — it lands in a fresh slot. -*Amended 2026-09-27 (issue #18).* The core owns `device_id`, inside the persisted listening blob. `ResetListeningData { newDeviceId }` zeroes the slot and rotates the id in the same state change, so both reach disk in one `PersistListening` write: a crash can no longer leave a fresh id holding the old total (which the next sync would write into a new server slot — exactly the resurrection this decision exists to prevent). The server cannot enforce this — deletes are by `user_id` only — so the client-side atomicity is the whole guarantee. The core has no randomness, so shells supply ids: a fresh UUID on reset, and `restore`'s `fallbackDeviceId` that the core adopts only when the listening blob has none (shells pass the id they used to store themselves; the core takes it only on the first launch after upgrading, so existing server slots carry over). +*Amended 2026-09-27 (issue #18).* The core owns `device_id`, inside the persisted listening blob. `ResetListeningData { newDeviceId }` zeroes the slot and rotates the id in the same state change, so both reach disk in one `PersistListening` write: a crash can no longer leave a fresh id holding the old total (which the next sync would write into a new server slot — exactly the resurrection this decision exists to prevent). The server cannot enforce this — deletes are by `user_id` only — so the client-side atomicity is the whole guarantee. Within one update, the listening blob is written before the account blob, so a crash between the two writes never leaves a deleted account's slot with no session to answer for it. The core has no randomness, so shells supply ids: a fresh UUID on reset, and `restore`'s `fallbackDeviceId` that the core adopts only when the listening blob has none (shells pass the id they used to store themselves; the core takes it only on the first launch after upgrading, so existing server slots carry over). ## Consequences