use std::error::Error;
use liminal::durability::MessageEnvelope as DurableEnvelope;
use liminal::durability::bridge::block_on;
use super::{R3Fixture, self_stream_publication_envelope, wire_config};
use crate::ServerError;
use crate::config::types::ChannelDef;
use crate::server::connection::services::{ConnectionServices, Reach};
use crate::server::self_streams::events::PublishOrigin;
#[test]
fn embedded_publish_refuses_reserved_and_console_namespaces_without_store_work()
-> Result<(), Box<dyn Error>> {
let fixture = R3Fixture::new()?;
let envelope = self_stream_publication_envelope();
let before = fixture.services.store_counts().snapshot();
for channel in [
"liminal.events",
"liminal.cost",
"liminal.unconfigured",
"console.topology",
"console.unconfigured",
] {
let connection = fixture
.services
.publish(7, channel, &envelope, Some("forged"))
.err()
.ok_or("connection bypassed its namespace guard")?;
let embedded = fixture
.services
.publish_embedded("cambium", channel, &envelope, Some("forged"))
.err()
.ok_or("embedded host bypassed its namespace guard")?;
assert_eq!(
std::mem::discriminant(&embedded),
std::mem::discriminant(&connection)
);
assert_eq!(embedded.to_string(), connection.to_string());
let direct = fixture
.services
.publish_with_origin(
channel,
&envelope,
Some("forged"),
PublishOrigin::Host { name: "cambium" },
)
.err()
.ok_or("internal host origin bypassed its namespace guard")?;
assert_eq!(direct.to_string(), connection.to_string());
assert_eq!(fixture.services.store_counts().snapshot(), before);
}
Ok(())
}
#[test]
fn embedded_publish_refuses_blank_hosts_before_admission_or_store_work()
-> Result<(), Box<dyn Error>> {
let fixture = R3Fixture::new()?;
let envelope = self_stream_publication_envelope();
let before = fixture.services.store_counts().snapshot();
let wake = fixture.services.signals().wake();
let arms = wake.arms();
for host in ["", " ", "\t\n"] {
for channel in ["pressure.source", "missing"] {
let error = fixture
.services
.publish_embedded(host, channel, &envelope, Some("blank"))
.err()
.ok_or("blank embedded host was accepted")?;
assert!(matches!(error, ServerError::ConfigValidation { message }
if message == "embedded publication requires a nonempty host name"));
assert_eq!(fixture.services.store_counts().snapshot(), before);
assert_eq!(wake.arms(), arms);
}
}
Ok(())
}
#[test]
fn embedded_publish_preserves_host_identity_in_live_and_durable_keyed_rows()
-> Result<(), Box<dyn Error>> {
let mut config = wire_config()?;
config.channels.push(ChannelDef {
name: "embedded.durable".to_owned(),
schema_ref: None,
durable: true,
loaded_schema: None,
});
let fixture = R3Fixture::with_config(&config)?;
let envelope = self_stream_publication_envelope();
for channel in ["pressure.source", "embedded.durable"] {
let subscription = fixture.services.subscribe_handle_for_test(channel)?;
for key in [None, Some("host-key")] {
let outcome = fixture
.services
.publish_embedded("cambium", channel, &envelope, key)?;
assert!(outcome.delivered);
let live = subscription
.try_next()?
.ok_or("embedded publication was not delivered")?;
assert_eq!(live.publisher_id.as_str(), "liminal.cambium");
assert_eq!(live.payload, envelope.payload);
if let Some(position) = outcome.durable_position {
let row = block_on(fixture.store.read_at(&format!("{channel}:0"), position))??
.ok_or("embedded durable row missing")?;
let stored = DurableEnvelope::deserialize(&row.payload)?;
assert_eq!(stored.publisher_id, "liminal.cambium");
assert_eq!(stored.payload, envelope.payload);
} else {
assert_eq!(channel, "pressure.source");
}
if let Some(key) = key {
let repeated =
fixture
.services
.publish_embedded("cambium", channel, &envelope, Some(key))?;
assert_eq!(repeated.reach, Reach::Suppressed);
assert_eq!(repeated.durable_position, outcome.durable_position);
assert!(subscription.try_next()?.is_none());
}
}
}
Ok(())
}
#[test]
fn embedded_publish_arms_external_cost_work_without_self_row_attribution()
-> Result<(), Box<dyn Error>> {
let fixture = R3Fixture::new()?;
let wake = fixture.services.signals().wake();
let arms = wake.arms();
let outcome = fixture.services.publish_with_origin(
"pressure.source",
&self_stream_publication_envelope(),
None,
PublishOrigin::Host { name: "cambium" },
)?;
assert_eq!(wake.arms() - arms, 1);
assert!(!outcome.self_row_appended);
assert_eq!(outcome.self_row_at, None);
wake.arm(&PublishOrigin::Server { device: "local" });
assert_eq!(wake.arms() - arms, 1);
Ok(())
}
#[test]
fn embedded_publish_names_the_console_tap_without_becoming_server_self_work()
-> Result<(), Box<dyn Error>> {
let fixture = R3Fixture::new()?;
let tap = fixture.services.console().tap();
let subscription = tap.subscribe(crate::tap::TapFilter {
subject: "pressure.source".to_owned(),
})?;
let before = tap.observations();
let server_before = tap.server_observations();
let host = String::from("cambium");
fixture.services.publish_embedded(
&host,
"pressure.source",
&self_stream_publication_envelope(),
None,
)?;
drop(host);
assert_eq!(tap.observations() - before, 1);
assert_eq!(tap.server_observations(), server_before);
let batches = subscription.take_batches()?;
assert_eq!(batches.len(), 1);
let batch = batches.first().ok_or("host tap batch missing")?;
assert_eq!(batch.messages.len(), 1);
let message = batch.messages.first().ok_or("host tap message missing")?;
assert_eq!(message.station.as_deref(), Some("liminal.cambium"));
let wall = crate::console::wall::WallBase::fixed(0, std::time::Instant::now());
let record =
crate::console::records::tap_batch_record("local", "pressure.source", batch, &wall)?;
assert_eq!(
record
.messages
.first()
.ok_or("host console record missing")?
.from,
"liminal.cambium"
);
fixture.services.publish_with_origin(
"pressure.source",
&self_stream_publication_envelope(),
None,
PublishOrigin::Server { device: "local" },
)?;
assert_eq!(tap.server_observations() - server_before, 1);
let server_batches = subscription.take_batches()?;
let server_batch = server_batches.first().ok_or("server tap batch missing")?;
let server_record =
crate::console::records::tap_batch_record("local", "pressure.source", server_batch, &wall)?;
assert_eq!(
server_record
.messages
.first()
.ok_or("server console record missing")?
.from,
"liminal.local"
);
Ok(())
}