use std::{
io::{self, Cursor},
net::SocketAddr,
};
use anyhow::{Context, Result, anyhow};
use capnp::{
message::{Builder, ReaderOptions},
serialize::write_message_to_words,
};
use capnp_futures::serialize::read_message;
use chrono::{TimeZone, Utc};
use eventdbx::{
replication_capnp::{
replication_hello, replication_hello_response, replication_request, replication_response,
},
replication_capnp_client::{CapnpReplicationClient, REPLICATION_PROTOCOL_VERSION},
replication_noise::{perform_server_handshake, read_encrypted_frame, write_encrypted_frame},
snowflake::SnowflakeId,
store::{AggregatePositionEntry, EventMetadata, EventRecord},
};
use futures::AsyncWriteExt;
use serde_json::json;
use tokio::net::TcpListener;
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
enum ListPositionsFixture {
Positions(Vec<AggregatePositionEntry>),
EmptyOk,
}
#[tokio::test(flavor = "multi_thread")]
async fn capnp_regression_list_positions_and_pull_events() -> Result<()> {
let listener = match TcpListener::bind("127.0.0.1:0").await {
Ok(listener) => listener,
Err(err) if err.kind() == io::ErrorKind::PermissionDenied => {
eprintln!("capnp regression skipped: {}", err);
return Ok(());
}
Err(err) => {
return Err(err).context("failed to bind mock replication server");
}
};
let addr: SocketAddr = listener
.local_addr()
.context("failed to read mock server address")?;
let token = "eyJhbGciOiJFZERTQSJ9.eyJzdWIiOiJ0ZXN0In0.signature".to_string();
let positions = vec![
AggregatePositionEntry {
aggregate_type: "order".to_string(),
aggregate_id: "order-1".to_string(),
version: 3,
},
AggregatePositionEntry {
aggregate_type: "inventory".to_string(),
aggregate_id: "sku-42".to_string(),
version: 7,
},
];
let event_id = SnowflakeId::from_u64(9_999_999_999);
let created_at = Utc
.timestamp_opt(1_700_000_000, 123_000)
.single()
.expect("valid timestamp");
let sample_event = EventRecord {
aggregate_type: "order".to_string(),
aggregate_id: "order-1".to_string(),
event_type: "OrderCreated".to_string(),
payload: json!({"status": "created"}),
extensions: None,
metadata: EventMetadata {
event_id,
created_at,
issued_by: None,
note: Some("initial import".to_string()),
},
version: 3,
hash: "hash-order-1-v3".to_string(),
merkle_root: "root-order-1-v3".to_string(),
};
let server_handle = tokio::spawn(run_mock_replication_server(
listener,
token.clone(),
ListPositionsFixture::Positions(positions.clone()),
vec![sample_event.clone()],
));
let endpoint = addr.to_string();
let mut client = CapnpReplicationClient::connect(&endpoint, &token)
.await
.context("client failed to connect to mock server")?;
let received_positions = client
.list_positions()
.await
.context("list_positions call failed")?;
assert_eq!(received_positions.len(), positions.len());
for (expected, actual) in positions.iter().zip(received_positions.iter()) {
assert_eq!(expected.aggregate_type, actual.aggregate_type);
assert_eq!(expected.aggregate_id, actual.aggregate_id);
assert_eq!(expected.version, actual.version);
}
let pulled_events = client
.pull_events("order", "order-1", 0, Some(10))
.await
.context("pull_events call failed")?;
assert_eq!(pulled_events.len(), 1);
let event = &pulled_events[0];
assert_eq!(event.aggregate_type, sample_event.aggregate_type);
assert_eq!(event.aggregate_id, sample_event.aggregate_id);
assert_eq!(event.event_type, sample_event.event_type);
assert_eq!(event.version, sample_event.version);
assert_eq!(event.hash, sample_event.hash);
assert_eq!(event.merkle_root, sample_event.merkle_root);
assert_eq!(event.payload, sample_event.payload);
assert_eq!(event.metadata.event_id, sample_event.metadata.event_id);
assert_eq!(event.metadata.created_at, sample_event.metadata.created_at);
assert_eq!(event.metadata.note, sample_event.metadata.note);
assert_eq!(event.metadata.issued_by.is_none(), true);
drop(client);
server_handle
.await
.context("mock server task panicked")?
.context("mock server failed")?;
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn capnp_regression_list_positions_ok_response() -> Result<()> {
let listener = match TcpListener::bind("127.0.0.1:0").await {
Ok(listener) => listener,
Err(err) if err.kind() == io::ErrorKind::PermissionDenied => {
eprintln!("capnp regression skipped: {}", err);
return Ok(());
}
Err(err) => {
return Err(err).context("failed to bind mock replication server");
}
};
let addr: SocketAddr = listener
.local_addr()
.context("failed to read mock server address")?;
let token = "eyJhbGciOiJFZERTQSJ9.eyJzdWIiOiJ0ZXN0In0.signature".to_string();
let server_handle = tokio::spawn(run_mock_replication_server(
listener,
token.clone(),
ListPositionsFixture::EmptyOk,
Vec::new(),
));
let endpoint = addr.to_string();
let mut client = CapnpReplicationClient::connect(&endpoint, &token)
.await
.context("client failed to connect to mock server")?;
let received_positions = client
.list_positions()
.await
.context("list_positions call failed")?;
assert!(received_positions.is_empty());
let pulled_events = client
.pull_events("noop", "noop", 0, Some(10))
.await
.context("pull_events call failed")?;
assert!(pulled_events.is_empty());
drop(client);
server_handle
.await
.context("mock server task panicked")?
.context("mock server failed")?;
Ok(())
}
async fn run_mock_replication_server(
listener: TcpListener,
mut expected_token: String,
list_positions_reply: ListPositionsFixture,
events: Vec<EventRecord>,
) -> Result<()> {
let (stream, _) = listener
.accept()
.await
.context("failed to accept mock replication connection")?;
let (read_half, write_half) = stream.into_split();
let mut reader = read_half.compat();
let mut writer = write_half.compat_write();
expected_token = expected_token.trim().to_string();
let hello_message = read_message(&mut reader, Default::default())
.await
.context("failed to read handshake")?;
{
let hello = hello_message
.get_root::<replication_hello::Reader>()
.context("failed to decode handshake")?;
if hello.get_protocol_version() != REPLICATION_PROTOCOL_VERSION {
return Err(anyhow!(
"unexpected protocol version {}",
hello.get_protocol_version()
));
}
let received_token = hello
.get_token()
.context("failed to read expected token")?
.to_str()
.context("invalid UTF-8 in handshake token")?
.trim()
.to_string();
if received_token != expected_token {
return Err(anyhow!("handshake carried unexpected token"));
}
}
let mut hello_response = Builder::new_default();
{
let mut response = hello_response.init_root::<replication_hello_response::Builder>();
response.set_accepted(true);
response.set_message("");
}
let response_bytes = write_message_to_words(&hello_response);
writer
.write_all(&response_bytes)
.await
.context("failed to send handshake response")?;
writer.flush().await.context("failed to flush handshake")?;
let mut noise = perform_server_handshake(&mut reader, &mut writer, expected_token.as_bytes())
.await
.context("failed to complete Noise handshake")?;
let list_request_bytes = read_encrypted_frame(&mut reader, &mut noise)
.await
.context("failed to read encrypted list_positions request")?
.ok_or_else(|| anyhow!("client closed before list_positions request"))?;
let mut cursor = Cursor::new(&list_request_bytes);
let list_request_message = capnp::serialize::read_message(&mut cursor, ReaderOptions::new())
.context("failed to decode list_positions request")?;
{
let list_request = list_request_message
.get_root::<replication_request::Reader>()
.context("failed to decode list_positions request root")?;
if !matches!(
list_request
.which()
.context("invalid request discriminant")?,
replication_request::Which::ListPositions(())
) {
return Err(anyhow!("unexpected first request"));
}
}
let mut response = Builder::new_default();
{
let mut envelope = response.init_root::<replication_response::Builder>();
match &list_positions_reply {
ListPositionsFixture::Positions(positions) => {
let mut ok = envelope.reborrow().init_list_positions();
let mut list = ok.reborrow().init_positions(positions.len() as u32);
for (idx, entry) in positions.iter().enumerate() {
let mut builder = list.reborrow().get(idx as u32);
builder.set_aggregate_type(&entry.aggregate_type);
builder.set_aggregate_id(&entry.aggregate_id);
builder.set_version(entry.version);
}
}
ListPositionsFixture::EmptyOk => {
envelope.set_ok(());
}
}
}
let bytes = write_message_to_words(&response);
write_encrypted_frame(&mut writer, &mut noise, &bytes)
.await
.context("failed to send list_positions response")?;
let pull_request_bytes = match read_encrypted_frame(&mut reader, &mut noise).await? {
Some(bytes) => bytes,
None => return Ok(()),
};
let mut cursor = Cursor::new(&pull_request_bytes);
let pull_request_message = capnp::serialize::read_message(&mut cursor, ReaderOptions::new())
.context("failed to decode pull_events request")?;
let (requested_type, requested_id, from_version, limit) = {
let pull_request = pull_request_message
.get_root::<replication_request::Reader>()
.context("failed to decode pull_events request root")?;
match pull_request
.which()
.context("invalid request discriminant")?
{
replication_request::Which::PullEvents(Ok(request)) => {
let aggregate_type = request
.get_aggregate_type()
.context("failed to read aggregate type from request")?
.to_str()
.context("invalid UTF-8 in aggregate type")?
.to_string();
let aggregate_id = request
.get_aggregate_id()
.context("failed to read aggregate id from request")?
.to_str()
.context("invalid UTF-8 in aggregate id")?
.to_string();
let from_version = request.get_from_version();
let limit = request.get_limit() as usize;
(aggregate_type, aggregate_id, from_version, limit)
}
_ => return Err(anyhow!("unexpected second request")),
}
};
let mut filtered: Vec<EventRecord> = events
.iter()
.filter(|event| {
event.aggregate_type == requested_type
&& event.aggregate_id == requested_id
&& event.version > from_version
})
.cloned()
.collect();
if limit > 0 && filtered.len() > limit {
filtered.truncate(limit);
}
let mut response = Builder::new_default();
{
let mut envelope = response.init_root::<replication_response::Builder>();
let mut pull = envelope.reborrow().init_pull_events();
let mut list = pull.reborrow().init_events(filtered.len() as u32);
for (idx, record) in filtered.iter().enumerate() {
let mut builder = list.reborrow().get(idx as u32);
builder.set_aggregate_type(&record.aggregate_type);
builder.set_aggregate_id(&record.aggregate_id);
builder.set_event_type(&record.event_type);
builder.set_version(record.version);
builder.set_merkle_root(&record.merkle_root);
builder.set_hash(&record.hash);
let payload = serde_json::to_vec(&record.payload)
.context("failed to encode payload for pull_events response")?;
let metadata = serde_json::to_vec(&record.metadata)
.context("failed to encode metadata for pull_events response")?;
builder.set_payload(&payload);
builder.set_metadata(&metadata);
if let Some(extensions) = &record.extensions {
let bytes = serde_json::to_vec(extensions)
.context("failed to encode extensions for pull_events response")?;
builder.set_extensions(&bytes);
} else {
builder.set_extensions(&[]);
}
}
}
let bytes = write_message_to_words(&response);
write_encrypted_frame(&mut writer, &mut noise, &bytes)
.await
.context("failed to send pull_events response")?;
Ok(())
}