use std::sync::Arc;
use sim_codec_bridge::{
BridgeBook, BridgePacket, BridgePatchPayload, BridgeVotePayload, stamp_packet_cid,
};
use sim_kernel::{Cx, DefaultFactory, EagerPolicy, Error, Result, Symbol};
use crate::rx_check;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum MergePolicy {
Single,
Quorum {
min_votes: u32,
},
SynthesisThenVote {
synthesizer: String,
min_votes: u32,
},
}
pub fn merge_bridge_replies(
base: &BridgePacket,
replies: &[BridgePacket],
policy: &MergePolicy,
) -> Result<BridgePacket> {
let base_cid = base.header.cid.as_deref().ok_or_else(|| {
Error::Eval("BRIDGE collaboration merge requires a stamped base packet".to_owned())
})?;
let patches = patch_candidates(base_cid, replies)?;
if patches.is_empty() {
return Err(Error::Eval(
"BRIDGE collaboration merge found no patch for the base packet".to_owned(),
));
}
require_one_target(&patches)?;
let selected = select_patch(base_cid, replies, &patches, policy)?;
let merged = stamp_packet_cid(&selected.packet.canonicalized())?;
check_merged_reply(base, &merged)?;
Ok(merged)
}
#[derive(Clone)]
struct PatchCandidate<'a> {
packet: &'a BridgePacket,
patch: BridgePatchPayload,
}
fn patch_candidates<'a>(
base_cid: &str,
replies: &'a [BridgePacket],
) -> Result<Vec<PatchCandidate<'a>>> {
let mut patches = Vec::new();
for packet in replies {
if !packet
.header
.parents
.iter()
.any(|parent| parent == base_cid)
{
continue;
}
for part in &packet.body {
if part.kind != Symbol::qualified("bridge", "Patch") {
continue;
}
let patch = BridgePatchPayload::from_expr(&part.payload)?;
if patch.parent_cid == base_cid {
patches.push(PatchCandidate { packet, patch });
}
}
}
Ok(patches)
}
fn require_one_target(patches: &[PatchCandidate<'_>]) -> Result<()> {
let target = patches[0].patch.target.as_str();
if patches
.iter()
.all(|candidate| candidate.patch.target == target)
{
return Ok(());
}
Err(Error::Eval(
"BRIDGE collaboration merge patches must share one exact target path".to_owned(),
))
}
fn select_patch<'a>(
base_cid: &str,
replies: &'a [BridgePacket],
patches: &[PatchCandidate<'a>],
policy: &MergePolicy,
) -> Result<PatchCandidate<'a>> {
match policy {
MergePolicy::Single => {
if patches.len() == 1 {
Ok(patches[0].clone())
} else {
Err(Error::Eval(
"BRIDGE single merge requires exactly one patch".to_owned(),
))
}
}
MergePolicy::Quorum { min_votes } => {
select_quorum_patch(base_cid, replies, patches, *min_votes)
}
MergePolicy::SynthesisThenVote {
synthesizer,
min_votes,
} => {
if synthesizer.is_empty() {
return Err(Error::Eval(
"BRIDGE synthesis merge requires a synthesizer seat".to_owned(),
));
}
let synthesized = patches
.iter()
.filter(|candidate| candidate.packet.header.from == *synthesizer)
.cloned()
.collect::<Vec<_>>();
select_quorum_patch(base_cid, replies, &synthesized, *min_votes)
}
}
}
fn select_quorum_patch<'a>(
base_cid: &str,
replies: &[BridgePacket],
patches: &[PatchCandidate<'a>],
min_votes: u32,
) -> Result<PatchCandidate<'a>> {
if min_votes == 0 {
return Err(Error::Eval(
"BRIDGE quorum merge requires at least one vote".to_owned(),
));
}
let eligible = patches
.iter()
.filter(|candidate| {
votes_for_target(base_cid, replies, &candidate.patch.target)
.map(|votes| votes >= min_votes)
.unwrap_or(false)
})
.cloned()
.collect::<Vec<_>>();
if eligible.len() == 1 {
Ok(eligible[0].clone())
} else if eligible.is_empty() {
Err(Error::Eval(
"BRIDGE quorum merge found no patch with enough votes".to_owned(),
))
} else {
Err(Error::Eval(
"BRIDGE quorum merge selected more than one patch".to_owned(),
))
}
}
fn votes_for_target(base_cid: &str, replies: &[BridgePacket], target: &str) -> Result<u32> {
let mut votes = 0u32;
for packet in replies {
if !packet
.header
.parents
.iter()
.any(|parent| parent == base_cid)
{
continue;
}
for part in &packet.body {
if part.kind != Symbol::qualified("bridge", "Vote") {
continue;
}
let vote = BridgeVotePayload::from_expr(&part.payload)?;
if vote.target == target {
votes += 1;
}
}
}
Ok(votes)
}
fn check_merged_reply(base: &BridgePacket, merged: &BridgePacket) -> Result<()> {
let book = BridgeBook::standard();
let mut cx = Cx::new(Arc::new(EagerPolicy), Arc::new(DefaultFactory));
let report = rx_check(&mut cx, &book, merged, Some(base))?;
if report.accepted() {
return Ok(());
}
Err(Error::Eval(format!(
"BRIDGE merged reply failed receive check: {:?}",
report.obligations
)))
}