evento-sql 2.0.0-alpha.25

SQL database implementations for evento event sourcing library.
Documentation 
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

//! `SqlJournal` integration tests on **PostgreSQL** (a live server at the repo's
//! convention URL, `postgres://postgres:postgres@localhost:5432`). Each scenario runs
//! in its own database for isolation. Gated by the `postgres` feature.

use evento_accord::Journal;
use evento_sql::SqlJournal;
use sqlx::migrate::MigrateDatabase;
use sqlx::{PgPool, Postgres};

mod sql_support;
use sql_support::{command, timestamp};

const BASE: &str = "postgres://postgres:postgres@localhost:5432";

/// A freshly-created, migrated journal in its own database `name`.
async fn fresh(name: &str) -> SqlJournal<Postgres> {
    let url = format!("{BASE}/{name}");
    let _ = Postgres::drop_database(&url).await;
    Postgres::create_database(&url).await.unwrap();
    let journal = SqlJournal::new(PgPool::connect(&url).await.unwrap());
    journal.migrate().await.unwrap();
    journal
}

#[tokio::test]
async fn round_trips_commands_metadata_and_acceptors() {
    sql_support::round_trip(&fresh("accord_pg_round_trip").await).await;
}

#[tokio::test]
async fn group_commit_batches_staged_writes() {
    sql_support::group_commit(&fresh("accord_pg_group_commit").await).await;
}

#[tokio::test]
async fn truncate_drops_below_the_watermark() {
    sql_support::truncate(&fresh("accord_pg_truncate").await).await;
}

#[tokio::test]
async fn survives_reconnect() {
    let name = "accord_pg_reconnect";
    let b = command(200, 2);
    {
        let journal = fresh(name).await;
        journal.record(&command(100, 1)).await.unwrap();
        journal.record(&b).await.unwrap();
        journal.truncate(timestamp(150, 0)).await.unwrap(); // drops the 100-record
        journal.close().await;
    }
    // Reconnect a brand-new pool to the same database (tables already migrated).
    let journal = SqlJournal::new(PgPool::connect(&format!("{BASE}/{name}")).await.unwrap());
    let all = journal.load_all().await.unwrap();
    assert_eq!(
        all.len(),
        1,
        "consensus state is durable across a reconnect"
    );
    assert_eq!(all[0].txn, b.txn);
    assert_eq!(
        journal.load_watermark().await.unwrap(),
        Some(timestamp(150, 0))
    );
}