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

use super::*;

#[tokio::test]
async fn occupancy_initialization_preserves_existing_rows_and_uses_filtered_epoch() {
    let temp = tempfile::tempdir().unwrap();
    let db = crate::db_connection::init_turso_local_with_retry(&temp.path().join("schema.db"))
        .await
        .unwrap();
    db.create_table("events")
        .column(schema_column("event_seq", DataType::Int, false))
        .primary_key("event_seq")
        .column(schema_column("event_type", DataType::Text, false))
        .execute(&*db)
        .await
        .unwrap();
    db.create_table("projection_checkpoints")
        .column(schema_column("projection_name", DataType::Text, false))
        .primary_key("projection_name")
        .column(schema_column("last_event_seq", DataType::Int, false))
        .column(schema_column("projection_version", DataType::Int, false))
        .column(schema_column("updated_at_ms", DataType::Int, false))
        .execute(&*db)
        .await
        .unwrap();
    for (seq, kind) in [
        (2, "model_changed"),
        (8, "context_compacted"),
        (13, "other"),
    ] {
        db.insert("events")
            .value("event_seq", seq)
            .value("event_type", kind)
            .execute(&*db)
            .await
            .unwrap();
    }
    InitializeOccupancy.execute(&*db).await.unwrap();
    let row = db
        .select("context_occupancy_projection")
        .execute_first(&*db)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(row.get("context_epoch"), Some(DatabaseValue::Int64(8)));
    let checkpoint = db
        .select("projection_checkpoints")
        .execute_first(&*db)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        checkpoint.get("last_event_seq"),
        Some(DatabaseValue::Int64(13))
    );
    db.update("context_occupancy_projection")
        .value("schema_version", 99)
        .execute(&*db)
        .await
        .unwrap();
    InitializeOccupancy.execute(&*db).await.unwrap();
    let row = db
        .select("context_occupancy_projection")
        .execute_first(&*db)
        .await
        .unwrap()
        .unwrap();
    assert_eq!(row.get("schema_version"), Some(DatabaseValue::Int64(99)));
    db.close().await.unwrap();
}