mj_controller/database/
client_state.rs

1use super::*;
2
3pub fn client_read_frontier(client_id: &str, workspace_id: &str, session_id: &str) -> Result<u64> {
4    client_read_frontier_at(&database_path(), client_id, workspace_id, session_id)
5}
6
7pub(super) fn client_read_frontier_at(
8    path: &Path,
9    client_id: &str,
10    workspace_id: &str,
11    session_id: &str,
12) -> Result<u64> {
13    let connection = open_reader(path)?;
14    let client: Option<u64> = connection
15        .query_row(
16            "SELECT through_event_ordinal
17               FROM client_read_frontiers
18              WHERE client_id = ?1 AND workspace_id = ?2 AND session_id = ?3",
19            params![client_id, workspace_id, session_id],
20            |row| row.get(0),
21        )
22        .optional()?;
23    if let Some(frontier) = client {
24        return Ok(frontier);
25    }
26    connection
27        .query_row(
28            "SELECT s.viewed_through_event_ordinal
29               FROM sessions s JOIN session_contexts c USING(session_id)
30              WHERE s.session_id = ?1 AND c.workspace_id = ?2",
31            params![session_id, workspace_id],
32            |row| row.get(0),
33        )
34        .with_context(|| format!("find session {session_id:?} in workspace {workspace_id:?}"))
35}
36
37pub fn advance_client_read_frontier(
38    client_id: &str,
39    workspace_id: &str,
40    session_id: &str,
41    through: u64,
42) -> Result<u64> {
43    let client_id = client_id.to_owned();
44    let workspace_id = workspace_id.to_owned();
45    let session_id = session_id.to_owned();
46    submit_database_write("advance_client_read_frontier", move |_| {
47        advance_client_read_frontier_at(
48            &database_path(),
49            &client_id,
50            &workspace_id,
51            &session_id,
52            through,
53        )
54    })
55}
56
57/// What this viewer has stored for this session: an unsent draft and how far
58/// it has read.
59pub fn client_session_state(
60    client_id: &str,
61    workspace_id: &str,
62    session_id: &str,
63) -> Result<ClientSessionState> {
64    let connection = open_reader(&database_path())?;
65    let draft = connection
66        .query_row(
67            "SELECT draft FROM client_session_state
68              WHERE client_id = ?1 AND workspace_id = ?2 AND session_id = ?3",
69            params![client_id, workspace_id, session_id],
70            |row| row.get::<_, String>(0),
71        )
72        .optional()?
73        .unwrap_or_default();
74    let through_event_ordinal = connection
75        .query_row(
76            "SELECT through_event_ordinal FROM client_read_frontiers
77              WHERE client_id = ?1 AND workspace_id = ?2 AND session_id = ?3",
78            params![client_id, workspace_id, session_id],
79            |row| row.get::<_, u64>(0),
80        )
81        .optional()?
82        .unwrap_or_default();
83    Ok(ClientSessionState {
84        draft,
85        through_event_ordinal,
86    })
87}
88
89/// Store one viewer's unsent draft.
90///
91/// An empty draft deletes the row rather than storing emptiness, so a viewer
92/// that cleared its composer stops occupying a row and stops being pruned
93/// later for something it no longer holds.
94pub fn persist_client_draft(
95    client_id: &str,
96    workspace_id: &str,
97    session_id: &str,
98    draft: &str,
99) -> Result<()> {
100    ensure!(!client_id.trim().is_empty(), "client id is empty");
101    let client_id = client_id.to_owned();
102    let workspace_id = workspace_id.to_owned();
103    let session_id = session_id.to_owned();
104    let draft = draft.to_owned();
105    submit_database_write("persist_client_draft", move |connection| {
106        let transaction =
107            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
108        if draft.is_empty() {
109            transaction.execute(
110                "DELETE FROM client_session_state
111                  WHERE client_id = ?1 AND workspace_id = ?2 AND session_id = ?3",
112                params![client_id, workspace_id, session_id],
113            )?;
114            transaction.commit()?;
115            return Ok(());
116        }
117        let now = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
118        let changed = transaction.execute(
119            "INSERT INTO client_session_state(
120                 client_id, workspace_id, session_id, draft, updated_at
121             )
122             SELECT ?1, ?2, ?3, ?4, ?5
123              WHERE EXISTS(
124                  SELECT 1 FROM session_contexts
125                   WHERE session_id = ?3 AND workspace_id = ?2
126              )
127             ON CONFLICT(client_id, workspace_id, session_id) DO UPDATE SET
128                 draft = excluded.draft,
129                 updated_at = excluded.updated_at",
130            params![client_id, workspace_id, session_id, draft, now],
131        )?;
132        ensure!(
133            changed == 1,
134            "session {session_id:?} is not in workspace {workspace_id:?}"
135        );
136        transaction.commit()?;
137        Ok(())
138    })
139}
140
141/// Forget web-viewer state that has passed its retention.
142///
143/// Only rows whose client id names a phone are considered. A terminal client's
144/// read frontier is not the phone's to expire, and deleting one would lose a
145/// person's place in a conversation they are still reading.
146pub fn prune_phone_client_state(older_than: Duration) -> Result<usize> {
147    let cutoff = (Utc::now() - chrono::Duration::from_std(older_than).unwrap_or_default())
148        .to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
149    submit_database_write("prune_phone_client_state", move |connection| {
150        let transaction =
151            connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
152        let drafts = transaction.execute(
153            "DELETE FROM client_session_state
154              WHERE client_id LIKE 'phone:%' AND updated_at < ?1",
155            params![cutoff],
156        )?;
157        let frontiers = transaction.execute(
158            "DELETE FROM client_read_frontiers
159              WHERE client_id LIKE 'phone:%' AND updated_at < ?1",
160            params![cutoff],
161        )?;
162        transaction.commit()?;
163        Ok(drafts + frontiers)
164    })
165}
166
167pub fn persist_read_receipt(
168    client_id: &str,
169    workspace_id: &str,
170    session_id: &str,
171    through: u64,
172) -> Result<u64> {
173    let client_id = client_id.to_owned();
174    let workspace_id = workspace_id.to_owned();
175    let session_id = session_id.to_owned();
176    submit_database_write("persist_read_receipt", move |connection| {
177        persist_read_receipt_with(connection, &client_id, &workspace_id, &session_id, through)
178    })
179}
180
181pub(super) fn persist_read_receipt_with(
182    connection: &mut Connection,
183    client_id: &str,
184    workspace_id: &str,
185    session_id: &str,
186    through: u64,
187) -> Result<u64> {
188    ensure!(!client_id.trim().is_empty(), "client id is empty");
189    let transaction =
190        connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
191    let applied = transaction
192        .query_row(
193            "SELECT applied_event_ordinal FROM materialized_sessions WHERE session_id = ?1",
194            [session_id],
195            |row| row.get::<_, u64>(0),
196        )
197        .optional()?
198        .with_context(|| format!("unknown session {session_id}"))?;
199    ensure!(
200        through <= applied,
201        "cannot acknowledge event ordinal {through} for session {session_id}; projection is at {applied}"
202    );
203    let now = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
204    let changed = transaction.execute(
205        "INSERT INTO client_read_frontiers(
206             client_id, workspace_id, session_id, through_event_ordinal, updated_at
207         )
208         SELECT ?1, ?2, ?3, ?4, ?5
209          WHERE EXISTS(
210              SELECT 1 FROM session_contexts
211               WHERE session_id = ?3 AND workspace_id = ?2
212          )
213         ON CONFLICT(client_id, workspace_id, session_id) DO UPDATE SET
214             through_event_ordinal = max(
215                 client_read_frontiers.through_event_ordinal,
216                 excluded.through_event_ordinal
217             ),
218             updated_at = excluded.updated_at",
219        params![client_id, workspace_id, session_id, through, now],
220    )?;
221    ensure!(
222        changed == 1,
223        "session {session_id:?} is not in workspace {workspace_id:?}"
224    );
225    let changed = transaction.execute(
226        "UPDATE sessions
227         SET viewed_through_event_ordinal = max(viewed_through_event_ordinal, ?2)
228         WHERE session_id = ?1",
229        params![session_id, through],
230    )?;
231    ensure!(changed == 1, "unknown session {session_id}");
232    let frontier = transaction.query_row(
233        "SELECT through_event_ordinal
234           FROM client_read_frontiers
235          WHERE client_id = ?1 AND workspace_id = ?2 AND session_id = ?3",
236        params![client_id, workspace_id, session_id],
237        |row| row.get(0),
238    )?;
239    transaction.commit()?;
240    Ok(frontier)
241}
242
243pub(super) fn advance_client_read_frontier_at(
244    path: &Path,
245    client_id: &str,
246    workspace_id: &str,
247    session_id: &str,
248    through: u64,
249) -> Result<u64> {
250    ensure!(!client_id.trim().is_empty(), "client id is empty");
251    let connection = open(path)?;
252    let now = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
253    let changed = connection.execute(
254        "INSERT INTO client_read_frontiers(
255             client_id, workspace_id, session_id, through_event_ordinal, updated_at
256         )
257         SELECT ?1, ?2, ?3, ?4, ?5
258          WHERE EXISTS(
259              SELECT 1 FROM session_contexts
260               WHERE session_id = ?3 AND workspace_id = ?2
261          )
262         ON CONFLICT(client_id, workspace_id, session_id) DO UPDATE SET
263             through_event_ordinal = max(
264                 client_read_frontiers.through_event_ordinal,
265                 excluded.through_event_ordinal
266             ),
267             updated_at = excluded.updated_at",
268        params![client_id, workspace_id, session_id, through, now],
269    )?;
270    ensure!(
271        changed == 1,
272        "session {session_id:?} is not in workspace {workspace_id:?}"
273    );
274    client_read_frontier_at(path, client_id, workspace_id, session_id)
275}
276
277/// Preserve unsent input for explicit recovery and retire its unchanged legacy
278/// seed together. Empty input still retires the seed: clearing is an edit.
279pub fn save_detached_session_draft(
280    workspace_id: &str,
281    session_id: &str,
282    source: &str,
283    owner_pid: u32,
284    draft: DetachedSessionDraft,
285) -> Result<Option<String>> {
286    let workspace_id = workspace_id.to_owned();
287    let session_id = session_id.to_owned();
288    let source = source.to_owned();
289    submit_database_write("save_detached_session_draft", move |connection| {
290        save_detached_session_draft_in(
291            connection,
292            &workspace_id,
293            &session_id,
294            &source,
295            owner_pid,
296            &draft,
297        )
298    })
299}
300
301pub(super) fn save_detached_session_draft_in(
302    connection: &mut Connection,
303    workspace_id: &str,
304    session_id: &str,
305    source: &str,
306    owner_pid: u32,
307    draft: &DetachedSessionDraft,
308) -> Result<Option<String>> {
309    let transaction =
310        connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
311    if let Some(inherited) = &draft.inherited_input {
312        transaction.execute(
313            "UPDATE sessions SET draft_input = ''
314              WHERE session_id = ?1 AND draft_input = ?2
315                AND EXISTS (SELECT 1 FROM session_contexts
316                             WHERE session_id = ?1 AND workspace_id = ?3)",
317            params![session_id, inherited, workspace_id],
318        )?;
319    }
320    let id = insert_detached_draft(
321        &transaction,
322        workspace_id,
323        Some(session_id),
324        source,
325        Some(owner_pid),
326        &draft.text,
327    )?;
328    transaction.commit()?;
329    Ok(id)
330}
331
332pub fn save_detached_draft(
333    workspace_id: &str,
334    session_id: Option<&str>,
335    source: &str,
336    owner_pid: Option<u32>,
337    text: &str,
338) -> Result<Option<String>> {
339    let workspace_id = workspace_id.to_owned();
340    let session_id = session_id.map(str::to_owned);
341    let source = source.to_owned();
342    let text = text.to_owned();
343    submit_database_write("save_detached_draft", move |_| {
344        save_detached_draft_at(
345            &database_path(),
346            &workspace_id,
347            session_id.as_deref(),
348            &source,
349            owner_pid,
350            &text,
351        )
352    })
353}
354
355pub(super) fn save_detached_draft_at(
356    path: &Path,
357    workspace_id: &str,
358    session_id: Option<&str>,
359    source: &str,
360    owner_pid: Option<u32>,
361    text: &str,
362) -> Result<Option<String>> {
363    if text.is_empty() {
364        return Ok(None);
365    }
366    let connection = open(path)?;
367    insert_detached_draft(
368        &connection,
369        workspace_id,
370        session_id,
371        source,
372        owner_pid,
373        text,
374    )
375}
376
377pub(super) fn insert_detached_draft(
378    connection: &Connection,
379    workspace_id: &str,
380    session_id: Option<&str>,
381    source: &str,
382    owner_pid: Option<u32>,
383    text: &str,
384) -> Result<Option<String>> {
385    if text.is_empty() {
386        return Ok(None);
387    }
388    ensure!(!source.trim().is_empty(), "draft source is empty");
389    let id = new_workspace_id()?;
390    let saved_at = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
391    connection.execute(
392        "INSERT INTO detached_drafts(
393             draft_id, workspace_id, session_id, source, owner_pid, saved_at, text
394         ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
395        params![
396            id,
397            workspace_id,
398            session_id,
399            source,
400            owner_pid,
401            saved_at,
402            text
403        ],
404    )?;
405    Ok(Some(id))
406}
407
408pub fn list_detached_drafts(workspace_id: &str) -> Result<Vec<DetachedDraft>> {
409    list_detached_drafts_at(&database_path(), workspace_id)
410}
411
412pub(super) fn list_detached_drafts_at(
413    path: &Path,
414    workspace_id: &str,
415) -> Result<Vec<DetachedDraft>> {
416    let connection = open_reader(path)?;
417    let mut statement = connection.prepare(
418        "SELECT draft_id, workspace_id, session_id, source, owner_pid, saved_at, text,
419                recovered_at
420           FROM detached_drafts
421          WHERE workspace_id = ?1 AND recovered_at IS NULL
422          ORDER BY saved_at DESC, draft_id DESC",
423    )?;
424    let rows = statement.query_map([workspace_id], |row| {
425        Ok(DetachedDraft {
426            id: row.get(0)?,
427            workspace_id: row.get(1)?,
428            session_id: row.get(2)?,
429            source: row.get(3)?,
430            owner_pid: row.get(4)?,
431            saved_at: row.get(5)?,
432            text: row.get(6)?,
433            recovered_at: row.get(7)?,
434        })
435    })?;
436    rows.collect::<rusqlite::Result<_>>().map_err(Into::into)
437}
438
439pub fn mark_draft_recovered(draft_id: &str) -> Result<()> {
440    let draft_id = draft_id.to_owned();
441    submit_database_write("mark_draft_recovered", move |_| {
442        mark_draft_recovered_at(&database_path(), &draft_id)
443    })
444}
445
446pub(super) fn mark_draft_recovered_at(path: &Path, draft_id: &str) -> Result<()> {
447    let connection = open(path)?;
448    let now = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
449    let changed = connection.execute(
450        "UPDATE detached_drafts SET recovered_at = ?2
451          WHERE draft_id = ?1 AND recovered_at IS NULL",
452        params![draft_id, now],
453    )?;
454    ensure!(
455        changed == 1,
456        "unknown or already recovered draft {draft_id:?}"
457    );
458    Ok(())
459}
460
461/// Explicitly restore a detached draft into its session composer. This is the
462/// only operation that merges client-local draft state back into the legacy
463/// session field, and the transaction marks the source draft recovered at the
464/// same durable boundary.
465pub fn recover_detached_draft(draft_id: &str) -> Result<String> {
466    let draft_id = draft_id.to_owned();
467    submit_database_write("recover_detached_draft", move |_| {
468        recover_detached_draft_at(&database_path(), &draft_id)
469    })
470}
471
472pub(super) fn recover_detached_draft_at(path: &Path, draft_id: &str) -> Result<String> {
473    let mut connection = open(path)?;
474    let tx = connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
475    let (session_id, text): (Option<String>, String) = tx
476        .query_row(
477            "SELECT session_id, text FROM detached_drafts
478              WHERE draft_id = ?1 AND recovered_at IS NULL",
479            [draft_id],
480            |row| Ok((row.get(0)?, row.get(1)?)),
481        )
482        .with_context(|| format!("find recoverable draft {draft_id:?}"))?;
483    let session_id = session_id.context("draft is not associated with a session")?;
484    let changed = tx.execute(
485        "UPDATE sessions SET draft_input = ?2 WHERE session_id = ?1",
486        params![session_id, text],
487    )?;
488    ensure!(changed == 1, "draft session no longer exists");
489    let now = Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
490    tx.execute(
491        "UPDATE detached_drafts SET recovered_at = ?2 WHERE draft_id = ?1",
492        params![draft_id, now],
493    )?;
494    tx.commit()?;
495    Ok(session_id)
496}
mj_controller/database/client_state.rs

mj_controller/database/
client_state.rs
