eventuary-sqlite 0.2.0

use std::num::NonZeroU32;
use std::time::Duration;

use chrono::Utc;
use futures::StreamExt;
use tokio::time::timeout;

use eventuary_core::io::filter::EventFilter;
use eventuary_core::io::{Reader, Writer};
use eventuary_core::partition::{
    EventKeyPartitionKeyResolver, Fnv1a64PartitionHasher, PartitionSelection,
};
use eventuary_core::{
    Event, EventId, Namespace, NamespacePattern, OrganizationId, Partition, Payload, StartFrom,
    StopAt, Topic, TopicPattern,
};
use eventuary_sqlite::database::SqliteDatabase;
use eventuary_sqlite::reader::{SqliteReader, SqliteReaderConfig, SqliteSubscription};
use eventuary_sqlite::writer::{SqlitePartitioningConfig, SqliteWriter, SqliteWriterConfig};

fn writer_config() -> SqliteWriterConfig {
    SqliteWriterConfig {
        partitioning: SqlitePartitioningConfig::inline(
            NonZeroU32::new(4).unwrap(),
            EventKeyPartitionKeyResolver::new(),
            Fnv1a64PartitionHasher,
        ),
        ..SqliteWriterConfig::default()
    }
}

fn ev(org: &str, ns: &str, topic: &str, key: &str) -> Event {
    Event::builder(org, ns, topic, key, Payload::from_string("payload"))
        .unwrap()
        .build()
        .expect("valid event")
}

fn sub_for(org: &str) -> SqliteSubscription {
    SqliteSubscription {
        start: StartFrom::Earliest,
        stop_at: StopAt::Never,
        filter: EventFilter::for_organization(OrganizationId::new(org).unwrap()),
        batch_size: Some(10),
        limit: None,
        ..SqliteSubscription::default()
    }
}

fn fast_config() -> SqliteReaderConfig {
    SqliteReaderConfig {
        poll_interval: Duration::from_millis(20),
        ..SqliteReaderConfig::default()
    }
}

#[tokio::test]
async fn write_read_roundtrip() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    writer
        .write(&ev("acme", "/x", "thing.happened", "k0"))
        .await
        .unwrap();

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(sub_for("acme")).await.unwrap();
    let msg = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(msg.event().key().as_str(), "k0");
}

#[tokio::test]
async fn reader_roundtrips_lineage_fields() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    let parent_id = EventId::new();
    let event = Event::builder(
        "acme",
        "/x",
        "thing.happened",
        "k",
        Payload::from_string("p"),
    )
    .unwrap()
    .parent_id(parent_id)
    .correlation_id("corr")
    .unwrap()
    .causation_id("cause")
    .unwrap()
    .build()
    .unwrap();
    writer.write(&event).await.unwrap();

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(sub_for("acme")).await.unwrap();
    let msg = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    let event = msg.event();
    assert_eq!(event.parent_id(), Some(parent_id));
    assert_eq!(event.correlation_id().map(|i| i.as_str()), Some("corr"));
    assert_eq!(event.causation_id().map(|i| i.as_str()), Some("cause"));
}

#[tokio::test]
async fn sqlite_reader_advances_after_ack() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    writer
        .write(&ev("acme", "/x", "thing.happened", "k0"))
        .await
        .unwrap();
    writer
        .write(&ev("acme", "/x", "thing.happened", "k1"))
        .await
        .unwrap();

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(sub_for("acme")).await.unwrap();

    let first = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(first.event().key().as_str(), "k0");
    first.ack().await.unwrap();

    let second = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(second.event().key().as_str(), "k1");
}

#[tokio::test]
async fn sqlite_reader_redelivers_after_nack() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    writer
        .write(&ev("acme", "/x", "thing.happened", "k0"))
        .await
        .unwrap();

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(sub_for("acme")).await.unwrap();

    let first = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    let first_id = first.event().id();
    first.nack().await.unwrap();

    let second = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(second.event().id(), first_id);
}

#[tokio::test]
async fn start_from_after_cursor_resumes() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    for i in 0..3 {
        writer
            .write(&ev("acme", "/x", "thing.happened", &format!("k{i}")))
            .await
            .unwrap();
    }
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(sub_for("acme")).await.unwrap();
    let msg = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    let cursor = *msg.cursor();
    msg.ack().await.unwrap();
    drop(stream);

    let resume = SqliteSubscription {
        start: StartFrom::After(cursor),
        ..sub_for("acme")
    };
    let reader2 = SqliteReader::new(db.conn(), fast_config());
    let mut stream2 = reader2.read(resume).await.unwrap();
    let msg = timeout(Duration::from_secs(5), stream2.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(msg.event().key().as_str(), "k1");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn start_from_latest_skips_existing_events() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    for i in 0..3 {
        writer
            .write(&ev("acme", "/x", "thing.happened", &format!("old{i}")))
            .await
            .unwrap();
    }
    let subscription = SqliteSubscription {
        start: StartFrom::Latest,
        ..sub_for("acme")
    };
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(subscription).await.unwrap();
    tokio::time::sleep(Duration::from_millis(50)).await;
    writer
        .write(&ev("acme", "/x", "thing.happened", "new"))
        .await
        .unwrap();
    let msg = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(msg.event().key().as_str(), "new");
}

#[tokio::test]
async fn start_from_timestamp_filters_old_events() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    writer
        .write(&ev("acme", "/x", "thing.happened", "before"))
        .await
        .unwrap();
    tokio::time::sleep(Duration::from_millis(50)).await;
    let cutoff = Utc::now();
    tokio::time::sleep(Duration::from_millis(50)).await;
    writer
        .write(&ev("acme", "/x", "thing.happened", "after"))
        .await
        .unwrap();

    let subscription = SqliteSubscription {
        start: StartFrom::Timestamp(cutoff),
        ..sub_for("acme")
    };
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(subscription).await.unwrap();
    let msg = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(msg.event().key().as_str(), "after");
}

#[tokio::test]
async fn topic_filter() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    writer
        .write(&ev("acme", "/x", "task.created", "t1"))
        .await
        .unwrap();
    writer
        .write(&ev("acme", "/x", "task.completed", "t2"))
        .await
        .unwrap();
    writer
        .write(&ev("acme", "/x", "task.created", "t3"))
        .await
        .unwrap();

    let mut filter = EventFilter::for_organization(OrganizationId::new("acme").unwrap());
    filter.topic = Some(TopicPattern::exact(Topic::new("task.created").unwrap()));
    let subscription = SqliteSubscription {
        start: StartFrom::Earliest,
        stop_at: StopAt::Never,
        filter,
        batch_size: Some(10),
        limit: None,
        ..SqliteSubscription::default()
    };
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(subscription).await.unwrap();
    let m1 = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(m1.event().key().as_str(), "t1");
    m1.ack().await.unwrap();
    let m2 = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(m2.event().key().as_str(), "t3");
}

#[tokio::test]
async fn namespace_filter() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    writer
        .write(&ev("acme", "/backend", "thing.happened", "b1"))
        .await
        .unwrap();
    writer
        .write(&ev("acme", "/frontend", "thing.happened", "f1"))
        .await
        .unwrap();
    writer
        .write(&ev("acme", "/backend/auth", "thing.happened", "b2"))
        .await
        .unwrap();

    let mut filter = EventFilter::for_organization(OrganizationId::new("acme").unwrap());
    filter.namespace = Some(NamespacePattern::prefix(
        Namespace::new("/backend").unwrap(),
    ));
    let subscription = SqliteSubscription {
        start: StartFrom::Earliest,
        stop_at: StopAt::Never,
        filter,
        batch_size: Some(10),
        limit: None,
        ..SqliteSubscription::default()
    };
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(subscription).await.unwrap();
    let m1 = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(m1.event().key().as_str(), "b1");
    m1.ack().await.unwrap();
    let m2 = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(m2.event().key().as_str(), "b2");
}

#[tokio::test]
async fn stop_at_current_end_finishes_after_existing_events() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    for i in 0..3 {
        writer
            .write(&ev("acme", "/x", "thing.happened", &format!("k{i}")))
            .await
            .unwrap();
    }

    let subscription = SqliteSubscription {
        stop_at: StopAt::CurrentEnd,
        ..sub_for("acme")
    };
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader.read(subscription).await.unwrap();

    for i in 0..3 {
        let msg = timeout(Duration::from_secs(5), stream.next())
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        assert_eq!(msg.event().key().as_str(), format!("k{i}"));
        msg.ack().await.unwrap();
    }

    let done = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap();
    assert!(done.is_none());
}

#[tokio::test]
async fn stop_at_cursor_finishes_at_inclusive_cursor() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    for i in 0..4 {
        writer
            .write(&ev("acme", "/x", "thing.happened", &format!("k{i}")))
            .await
            .unwrap();
    }

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut probe = reader
        .read(SqliteSubscription {
            stop_at: StopAt::CurrentEnd,
            ..sub_for("acme")
        })
        .await
        .unwrap();
    let first = timeout(Duration::from_secs(5), probe.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    first.ack().await.unwrap();
    let second = timeout(Duration::from_secs(5), probe.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    let stop_cursor = *second.cursor();
    second.ack().await.unwrap();
    drop(probe);

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader
        .read(SqliteSubscription {
            stop_at: StopAt::Cursor(stop_cursor),
            ..sub_for("acme")
        })
        .await
        .unwrap();

    let first = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(first.event().key().as_str(), "k0");
    first.ack().await.unwrap();

    let second = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(second.event().key().as_str(), "k1");
    second.ack().await.unwrap();

    let done = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap();
    assert!(done.is_none());
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn stop_at_never_waits_for_future_events() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &writer_config()).unwrap();
    let writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader
        .read(SqliteSubscription {
            stop_at: StopAt::Never,
            ..sub_for("acme")
        })
        .await
        .unwrap();

    tokio::time::sleep(Duration::from_millis(50)).await;
    writer
        .write(&ev("acme", "/x", "thing.happened", "future"))
        .await
        .unwrap();

    let msg = timeout(Duration::from_secs(5), stream.next())
        .await
        .unwrap()
        .unwrap()
        .unwrap();
    assert_eq!(msg.event().key().as_str(), "future");
}

#[tokio::test]
async fn default_writer_rows_are_readable_by_default_reader() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &SqliteWriterConfig::default()).unwrap();

    let writer = SqliteWriter::new(db.conn());
    writer
        .write(&ev("acme", "/x", "thing.happened", "default-writer-row"))
        .await
        .unwrap();

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader
        .read(SqliteSubscription {
            start: StartFrom::Earliest,
            stop_at: StopAt::CurrentEnd,
            ..SqliteSubscription::default()
        })
        .await
        .unwrap();

    let msg = stream.next().await.unwrap().unwrap();
    assert_eq!(msg.event().key().as_str(), "default-writer-row");
    assert_eq!(msg.cursor().partition().id(), 0);
    assert_eq!(msg.cursor().partition().count(), 1);
    msg.ack().await.unwrap();

    assert!(stream.next().await.is_none());
}

#[tokio::test]
async fn one_partition_reader_ignores_unpartitioned_rows() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &SqliteWriterConfig::default()).unwrap();

    let unpartitioned_writer = SqliteWriter::new(db.conn());
    unpartitioned_writer
        .write(&ev("acme", "/x", "thing.happened", "unpartitioned-row"))
        .await
        .unwrap();

    let partitioned_writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    partitioned_writer
        .write(&ev("acme", "/x", "thing.happened", "partitioned-row"))
        .await
        .unwrap();

    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader
        .read(SqliteSubscription {
            start: StartFrom::Earliest,
            stop_at: StopAt::CurrentEnd,
            partitions: PartitionSelection::One(
                Partition::new(0, NonZeroU32::new(4).unwrap()).unwrap(),
            ),
            ..SqliteSubscription::default()
        })
        .await
        .unwrap();

    while let Some(item) = stream.next().await {
        let msg = item.unwrap();
        assert_ne!(msg.event().key().as_str(), "unpartitioned-row");
        msg.ack().await.unwrap();
    }
}

#[tokio::test]
async fn many_partition_reader_ignores_unpartitioned_rows() {
    let db = SqliteDatabase::open_in_memory().unwrap();
    SqliteWriter::prepare_schema(&db.conn(), &SqliteWriterConfig::default()).unwrap();

    let unpartitioned_writer = SqliteWriter::new(db.conn());
    unpartitioned_writer
        .write(&ev("acme", "/x", "thing.happened", "unpartitioned-row"))
        .await
        .unwrap();

    let partitioned_writer = SqliteWriter::new_with_config(db.conn(), writer_config());
    partitioned_writer
        .write(&ev("acme", "/x", "thing.happened", "partitioned-row"))
        .await
        .unwrap();

    let partition = Partition::new(0, NonZeroU32::new(4).unwrap()).unwrap();
    let reader = SqliteReader::new(db.conn(), fast_config());
    let mut stream = reader
        .read(SqliteSubscription {
            start: StartFrom::Earliest,
            stop_at: StopAt::CurrentEnd,
            partitions: PartitionSelection::Many(
                eventuary_core::partition::PartitionGroup::singleton(partition),
            ),
            ..SqliteSubscription::default()
        })
        .await
        .unwrap();

    while let Some(item) = stream.next().await {
        let msg = item.unwrap();
        assert_ne!(msg.event().key().as_str(), "unpartitioned-row");
        msg.ack().await.unwrap();
    }
}