bcode_session 0.0.1-alpha.0

Session lifecycle and attachment management for bcode
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

//! Current canonical event-row reads and writes.

use crate::db::{SessionDbError, SessionDbResult};
use crate::persisted::{decode_session_event, encode_session_event};
pub use bcode_session_models::persisted_session_event_kind_name as event_kind_name;
use bcode_session_models::{SessionEvent, SessionId};
use switchy::database::{Database, DatabaseValue, query::SortDirection};

pub const fn event_created_at_ms(event: &SessionEvent) -> u64 {
    event.timestamp_ms
}

pub fn seq_to_value(sequence: u64) -> DatabaseValue {
    DatabaseValue::Int64(i64::try_from(sequence).unwrap_or(i64::MAX))
}

pub async fn insert_event(
    db: &dyn Database,
    event: &SessionEvent,
    activity_timestamp_ms: Option<u64>,
) -> SessionDbResult<()> {
    insert_event_payload(
        db,
        event,
        activity_timestamp_ms,
        encode_session_event(event)?,
    )
    .await
}

pub async fn insert_event_payload(
    db: &dyn Database,
    event: &SessionEvent,
    activity_timestamp_ms: Option<u64>,
    payload: String,
) -> SessionDbResult<()> {
    db.insert("events")
        .value("event_seq", seq_to_value(event.sequence))
        .value("event_type", event_kind_name(&event.kind))
        .value(
            "schema_version",
            DatabaseValue::Int32(i32::from(event.schema_version)),
        )
        .value(
            "created_at_ms",
            seq_to_value(activity_timestamp_ms.unwrap_or_else(|| event_created_at_ms(event))),
        )
        .value("payload", payload)
        .execute(db)
        .await?;
    Ok(())
}

pub async fn strict_events(db: &dyn Database) -> SessionDbResult<Vec<SessionEvent>> {
    let rows = db
        .select("events")
        .columns(&["payload"])
        .sort("event_seq", SortDirection::Asc)
        .execute(db)
        .await?;
    rows.into_iter()
        .map(|row| decode_session_event(&required_string(&row, "payload")?).map_err(Into::into))
        .collect()
}

pub fn strict_event_from_row(
    row: &switchy::database::Row,
    session_id: SessionId,
) -> SessionDbResult<SessionEvent> {
    let event_seq = required_non_negative_u64(row, "event_seq")?;
    let payload = required_string(row, "payload")?;
    let event = decode_session_event(&payload)?;
    if event.sequence != event_seq {
        return Err(SessionDbError::InvalidCanonicalSequence {
            expected: event_seq,
            actual: event.sequence,
        });
    }
    if event.session_id != session_id {
        return Err(SessionDbError::InvalidRow {
            column: "events.session_id".to_owned(),
        });
    }
    Ok(event)
}

pub async fn last_sequence(db: &dyn Database) -> SessionDbResult<Option<u64>> {
    let row = db
        .select("events")
        .columns(&["event_seq"])
        .sort("event_seq", SortDirection::Desc)
        .limit(1)
        .execute_first(db)
        .await?;
    let value = row
        .as_ref()
        .and_then(|row| row.get("event_seq"))
        .and_then(|value| value.as_i64());
    match value {
        Some(value) if value.is_negative() => Err(SessionDbError::InvalidRow {
            column: "event_seq".to_owned(),
        }),
        Some(value) => Ok(Some(value.cast_unsigned())),
        None => Ok(None),
    }
}

fn required_string(row: &switchy::database::Row, column: &str) -> SessionDbResult<String> {
    row.get(column)
        .and_then(|value| value.as_str().map(ToOwned::to_owned))
        .ok_or_else(|| SessionDbError::InvalidRow {
            column: column.to_owned(),
        })
}

fn required_non_negative_u64(row: &switchy::database::Row, column: &str) -> SessionDbResult<u64> {
    let value = row
        .get(column)
        .and_then(|value| value.as_i64())
        .ok_or_else(|| SessionDbError::InvalidRow {
            column: column.to_owned(),
        })?;
    if value.is_negative() {
        return Err(SessionDbError::InvalidRow {
            column: column.to_owned(),
        });
    }
    Ok(value.cast_unsigned())
}