every.channel/crates/ec-eth/src/lib.rs

//! Ethereum-compatible representations and commitments for every.channel core types.

use alloy_primitives::{keccak256, B256};
use alloy_sol_types::{eip712_domain, sol, Eip712Domain, SolStruct, SolValue};
use ec_core::{
    BroadcastId, ChainCommitment, ChunkId, Manifest, ManifestBody, ManifestSignature,
    ManifestVariant, SourceId, StreamControlAnnouncement, StreamDescriptor, StreamKey,
    StreamMetadata, StreamTransportDescriptor,
};
use std::fmt;

pub const ETHEREUM_CHAIN: &str = "ethereum";
pub const SCHEME_STREAM_ID_KECCAK: &str = "stream-id-keccak256-v1";
pub const SCHEME_STREAM_DESCRIPTOR_ABI: &str = "stream-descriptor-abi-keccak256-v1";
pub const SCHEME_CONTROL_ANNOUNCEMENT_ABI: &str = "control-announcement-abi-keccak256-v1";
pub const SCHEME_MANIFEST_DATA_ROOT: &str = "manifest-data-merkle-keccak256-v1";
pub const SCHEME_MANIFEST_BODY_ABI: &str = "manifest-body-abi-keccak256-v1";
pub const SCHEME_MANIFEST_ENVELOPE_ABI: &str = "manifest-envelope-abi-keccak256-v1";
pub const ETH_MANIFEST_SIG_ALG: &str = "secp256k1-eip712-manifest-body-v1";
pub const ZERO_B256_HEX: &str =
    "0x0000000000000000000000000000000000000000000000000000000000000000";

sol! {
    struct EthStreamMetadata {
        string key;
        string value;
    }

    struct EthBroadcastId {
        string standard;
        bool hasTransportStreamId;
        uint16 transportStreamId;
        bool hasProgramNumber;
        uint16 programNumber;
        string virtualChannel;
        string callsign;
        string region;
        string frequency;
    }

    struct EthSourceId {
        string kind;
        string deviceId;
        string channel;
    }

    struct EthStreamKey {
        uint16 version;
        bool hasBroadcast;
        EthBroadcastId broadcast;
        bool hasSource;
        EthSourceId source;
        string profile;
        string variant;
    }

    struct EthStreamDescriptor {
        string id;
        string title;
        string number;
        string source;
        EthStreamMetadata[] metadata;
    }

    struct EthStreamTransportDescriptor {
        uint8 kind;
        string url;
        string endpoint;
        string broadcastName;
        string trackName;
    }

    struct EthStreamControlAnnouncement {
        EthStreamDescriptor stream;
        EthStreamTransportDescriptor[] transports;
        uint64 updatedUnixMs;
        uint64 ttlMs;
    }

    struct EthChunkId {
        string streamId;
        string epochId;
        uint64 chunkIndex;
        bytes32 chunkHash;
    }

    struct EthManifestVariant {
        string variantId;
        string streamId;
        uint64 chunkStartIndex;
        uint64 totalChunks;
        bytes32 merkleRoot;
        bytes32[] chunkHashes;
        EthStreamMetadata[] metadata;
    }

    struct EthManifestBody {
        string streamId;
        string epochId;
        uint64 chunkDurationMs;
        uint64 totalChunks;
        uint64 chunkStartIndex;
        string encoderProfileId;
        bytes32 merkleRoot;
        uint64 createdUnixMs;
        EthStreamMetadata[] metadata;
        bytes32[] chunkHashes;
        EthManifestVariant[] variants;
    }

    struct EthManifestSignature {
        string signerId;
        string alg;
        bytes signature;
    }

    struct EthManifest {
        EthManifestBody body;
        bytes32 manifestId;
        EthManifestSignature[] signatures;
    }

    struct EthObservationHeader {
        bytes32 streamHash;
        bytes32 epochHash;
        bytes32 parentObservationHash;
        bytes32 dataRoot;
        bytes32 locatorHash;
        uint64 observedUnixMs;
        uint64 sequence;
    }
}

#[derive(Debug, Clone)]
pub enum EthCommitmentError {
    Empty,
    InvalidHex(String),
}

impl fmt::Display for EthCommitmentError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            EthCommitmentError::Empty => write!(f, "no hashes supplied"),
            EthCommitmentError::InvalidHex(value) => {
                write!(f, "invalid 32-byte hex value: {value}")
            }
        }
    }
}

impl std::error::Error for EthCommitmentError {}

fn commitment(scheme: &str, digest: B256) -> ChainCommitment {
    ChainCommitment {
        chain: ETHEREUM_CHAIN.to_string(),
        scheme: scheme.to_string(),
        digest: b256_hex(digest),
    }
}

fn abi_commitment<T: SolValue>(scheme: &str, value: &T) -> ChainCommitment {
    commitment(scheme, keccak256(value.abi_encode()))
}

pub fn b256_hex(value: B256) -> String {
    format!("0x{}", hex::encode(value))
}

pub fn keccak256_bytes_hex(value: &[u8]) -> String {
    b256_hex(keccak256(value))
}

pub fn parse_b256(value: &str) -> Result<B256, EthCommitmentError> {
    let trimmed = value.trim().strip_prefix("0x").unwrap_or(value.trim());
    let bytes =
        hex::decode(trimmed).map_err(|_| EthCommitmentError::InvalidHex(value.to_string()))?;
    if bytes.len() != 32 {
        return Err(EthCommitmentError::InvalidHex(value.to_string()));
    }
    let mut out = [0u8; 32];
    out.copy_from_slice(&bytes);
    Ok(B256::from(out))
}

fn parse_bytes(value: &str) -> Vec<u8> {
    let trimmed = value.trim().strip_prefix("0x").unwrap_or(value.trim());
    hex::decode(trimmed).unwrap_or_default()
}

pub fn manifest_eip712_domain() -> Eip712Domain {
    eip712_domain! {
        name: "every.channel",
        version: "1",
    }
}

fn eth_metadata(items: &[StreamMetadata]) -> Vec<EthStreamMetadata> {
    items
        .iter()
        .map(|item| EthStreamMetadata {
            key: item.key.clone(),
            value: item.value.clone(),
        })
        .collect()
}

pub fn eth_broadcast_id(value: &BroadcastId) -> EthBroadcastId {
    EthBroadcastId {
        standard: value.standard.clone(),
        hasTransportStreamId: value.transport_stream_id.is_some(),
        transportStreamId: value.transport_stream_id.unwrap_or_default(),
        hasProgramNumber: value.program_number.is_some(),
        programNumber: value.program_number.unwrap_or_default(),
        virtualChannel: value.virtual_channel.clone().unwrap_or_default(),
        callsign: value.callsign.clone().unwrap_or_default(),
        region: value.region.clone().unwrap_or_default(),
        frequency: value.frequency.clone().unwrap_or_default(),
    }
}

pub fn eth_source_id(value: &SourceId) -> EthSourceId {
    EthSourceId {
        kind: value.kind.clone(),
        deviceId: value.device_id.clone().unwrap_or_default(),
        channel: value.channel.clone().unwrap_or_default(),
    }
}

pub fn eth_stream_key(value: &StreamKey) -> EthStreamKey {
    EthStreamKey {
        version: value.version,
        hasBroadcast: value.broadcast.is_some(),
        broadcast: value
            .broadcast
            .as_ref()
            .map(eth_broadcast_id)
            .unwrap_or_else(|| EthBroadcastId {
                standard: String::new(),
                hasTransportStreamId: false,
                transportStreamId: 0,
                hasProgramNumber: false,
                programNumber: 0,
                virtualChannel: String::new(),
                callsign: String::new(),
                region: String::new(),
                frequency: String::new(),
            }),
        hasSource: value.source.is_some(),
        source: value
            .source
            .as_ref()
            .map(eth_source_id)
            .unwrap_or_else(|| EthSourceId {
                kind: String::new(),
                deviceId: String::new(),
                channel: String::new(),
            }),
        profile: value.profile.clone().unwrap_or_default(),
        variant: value.variant.clone().unwrap_or_default(),
    }
}

pub fn eth_stream_descriptor(value: &StreamDescriptor) -> EthStreamDescriptor {
    EthStreamDescriptor {
        id: value.id.0.clone(),
        title: value.title.clone(),
        number: value.number.clone().unwrap_or_default(),
        source: value.source.clone(),
        metadata: eth_metadata(&value.metadata),
    }
}

pub fn eth_stream_transport_descriptor(
    value: &StreamTransportDescriptor,
) -> EthStreamTransportDescriptor {
    match value {
        StreamTransportDescriptor::RelayMoq {
            url,
            broadcast_name,
            track_name,
        } => EthStreamTransportDescriptor {
            kind: 0,
            url: url.clone(),
            endpoint: String::new(),
            broadcastName: broadcast_name.clone(),
            trackName: track_name.clone(),
        },
        StreamTransportDescriptor::IrohDirect {
            endpoint,
            broadcast_name,
            track_name,
        } => EthStreamTransportDescriptor {
            kind: 1,
            url: String::new(),
            endpoint: endpoint.clone(),
            broadcastName: broadcast_name.clone(),
            trackName: track_name.clone(),
        },
    }
}

pub fn eth_stream_control_announcement(
    value: &StreamControlAnnouncement,
) -> EthStreamControlAnnouncement {
    EthStreamControlAnnouncement {
        stream: eth_stream_descriptor(&value.stream),
        transports: value
            .transports
            .iter()
            .map(eth_stream_transport_descriptor)
            .collect(),
        updatedUnixMs: value.updated_unix_ms,
        ttlMs: value.ttl_ms,
    }
}

pub fn eth_chunk_id(value: &ChunkId) -> Result<EthChunkId, EthCommitmentError> {
    Ok(EthChunkId {
        streamId: value.stream_id.0.clone(),
        epochId: value.epoch_id.clone(),
        chunkIndex: value.chunk_index,
        chunkHash: parse_b256(&value.chunk_hash)?,
    })
}

pub fn eth_manifest_variant(
    value: &ManifestVariant,
) -> Result<EthManifestVariant, EthCommitmentError> {
    Ok(EthManifestVariant {
        variantId: value.variant_id.clone(),
        streamId: value.stream_id.0.clone(),
        chunkStartIndex: value.chunk_start_index,
        totalChunks: value.total_chunks,
        merkleRoot: parse_b256(&value.merkle_root)?,
        chunkHashes: value
            .chunk_hashes
            .iter()
            .map(|hash| parse_b256(hash))
            .collect::<Result<Vec<_>, _>>()?,
        metadata: eth_metadata(&value.metadata),
    })
}

pub fn eth_manifest_body(value: &ManifestBody) -> Result<EthManifestBody, EthCommitmentError> {
    let variants = value
        .variants
        .as_ref()
        .map(|variants| {
            variants
                .iter()
                .map(eth_manifest_variant)
                .collect::<Result<Vec<_>, _>>()
        })
        .transpose()?
        .unwrap_or_default();

    Ok(EthManifestBody {
        streamId: value.stream_id.0.clone(),
        epochId: value.epoch_id.clone(),
        chunkDurationMs: value.chunk_duration_ms,
        totalChunks: value.total_chunks,
        chunkStartIndex: value.chunk_start_index,
        encoderProfileId: value.encoder_profile_id.clone(),
        merkleRoot: parse_b256(&value.merkle_root)?,
        createdUnixMs: value.created_unix_ms,
        metadata: eth_metadata(&value.metadata),
        chunkHashes: value
            .chunk_hashes
            .iter()
            .map(|hash| parse_b256(hash))
            .collect::<Result<Vec<_>, _>>()?,
        variants,
    })
}

pub fn manifest_body_eip712_signing_hash(value: &ManifestBody) -> Result<B256, EthCommitmentError> {
    Ok(eth_manifest_body(value)?.eip712_signing_hash(&manifest_eip712_domain()))
}

pub fn eth_manifest_signature(value: &ManifestSignature) -> EthManifestSignature {
    EthManifestSignature {
        signerId: value.signer_id.clone(),
        alg: value.alg.clone(),
        signature: parse_bytes(&value.signature).into(),
    }
}

pub fn eth_manifest(value: &Manifest) -> Result<EthManifest, EthCommitmentError> {
    Ok(EthManifest {
        body: eth_manifest_body(&value.body)?,
        manifestId: parse_b256(&value.manifest_id)?,
        signatures: value
            .signatures
            .iter()
            .map(eth_manifest_signature)
            .collect(),
    })
}

fn keccak_merkle_root(leaves: &[B256]) -> Result<B256, EthCommitmentError> {
    if leaves.is_empty() {
        return Err(EthCommitmentError::Empty);
    }
    let mut nodes = leaves.to_vec();
    while nodes.len() > 1 {
        if nodes.len() % 2 == 1 {
            if let Some(last) = nodes.last().copied() {
                nodes.push(last);
            }
        }
        let mut parents = Vec::with_capacity(nodes.len() / 2);
        for pair in nodes.chunks(2) {
            let mut merged = [0u8; 64];
            merged[..32].copy_from_slice(pair[0].as_slice());
            merged[32..].copy_from_slice(pair[1].as_slice());
            parents.push(keccak256(merged));
        }
        nodes = parents;
    }
    Ok(nodes[0])
}

pub fn ethereum_merkle_root_from_hashes(hashes: &[String]) -> Result<String, EthCommitmentError> {
    let leaves = hashes
        .iter()
        .map(|hash| parse_b256(hash))
        .collect::<Result<Vec<_>, _>>()?;
    Ok(format!("0x{}", hex::encode(keccak_merkle_root(&leaves)?)))
}

pub fn ethereum_merkle_proof_for_index(
    hashes: &[String],
    index: usize,
) -> Result<Vec<String>, EthCommitmentError> {
    if hashes.is_empty() {
        return Err(EthCommitmentError::Empty);
    }
    if index >= hashes.len() {
        return Err(EthCommitmentError::InvalidHex(format!(
            "index {index} out of bounds"
        )));
    }

    let mut nodes = hashes
        .iter()
        .map(|hash| parse_b256(hash))
        .collect::<Result<Vec<_>, _>>()?;
    let mut proof = Vec::new();
    let mut pos = index;
    while nodes.len() > 1 {
        if nodes.len() % 2 == 1 {
            if let Some(last) = nodes.last().copied() {
                nodes.push(last);
            }
        }
        let sibling_index = if pos % 2 == 0 { pos + 1 } else { pos - 1 };
        let sibling = nodes[sibling_index];
        proof.push(format!("0x{}", hex::encode(sibling)));

        let mut parents = Vec::with_capacity(nodes.len() / 2);
        for pair in nodes.chunks(2) {
            let mut merged = [0u8; 64];
            merged[..32].copy_from_slice(pair[0].as_slice());
            merged[32..].copy_from_slice(pair[1].as_slice());
            parents.push(keccak256(merged));
        }
        nodes = parents;
        pos /= 2;
    }

    Ok(proof)
}

pub fn verify_ethereum_merkle_proof(
    leaf_hash: &str,
    mut index: usize,
    branch: &[String],
    expected_root: &str,
) -> bool {
    let Ok(mut acc) = parse_b256(leaf_hash) else {
        return false;
    };
    for sibling_hex in branch {
        let Ok(sibling) = parse_b256(sibling_hex) else {
            return false;
        };
        let (left, right) = if index % 2 == 0 {
            (acc, sibling)
        } else {
            (sibling, acc)
        };
        let mut merged = [0u8; 64];
        merged[..32].copy_from_slice(left.as_slice());
        merged[32..].copy_from_slice(right.as_slice());
        acc = keccak256(merged);
        index /= 2;
    }
    match parse_b256(expected_root) {
        Ok(root) => acc == root,
        Err(_) => false,
    }
}

pub fn stream_id_commitment(stream_id: &str) -> ChainCommitment {
    commitment(SCHEME_STREAM_ID_KECCAK, keccak256(stream_id.as_bytes()))
}

pub fn broadcast_id_commitment(value: &BroadcastId) -> ChainCommitment {
    abi_commitment("broadcast-id-abi-keccak256-v1", &eth_broadcast_id(value))
}

pub fn stream_key_commitment(value: &StreamKey) -> ChainCommitment {
    abi_commitment("stream-key-abi-keccak256-v1", &eth_stream_key(value))
}

pub fn stream_descriptor_commitments(value: &StreamDescriptor) -> Vec<ChainCommitment> {
    vec![
        stream_id_commitment(&value.id.0),
        abi_commitment(SCHEME_STREAM_DESCRIPTOR_ABI, &eth_stream_descriptor(value)),
    ]
}

pub fn control_announcement_commitments(value: &StreamControlAnnouncement) -> Vec<ChainCommitment> {
    vec![abi_commitment(
        SCHEME_CONTROL_ANNOUNCEMENT_ABI,
        &eth_stream_control_announcement(value),
    )]
}

fn manifest_data_root_commitment(
    body: &ManifestBody,
) -> Result<ChainCommitment, EthCommitmentError> {
    let root = if let Some(variants) = body
        .variants
        .as_ref()
        .filter(|variants| !variants.is_empty())
    {
        let roots = variants
            .iter()
            .map(|variant| variant.merkle_root.clone())
            .collect::<Vec<_>>();
        ethereum_merkle_root_from_hashes(&roots)?
    } else {
        ethereum_merkle_root_from_hashes(&body.chunk_hashes)?
    };
    Ok(ChainCommitment {
        chain: ETHEREUM_CHAIN.to_string(),
        scheme: SCHEME_MANIFEST_DATA_ROOT.to_string(),
        digest: root,
    })
}

pub fn manifest_commitments(value: &Manifest) -> Result<Vec<ChainCommitment>, EthCommitmentError> {
    let eth_body = eth_manifest_body(&value.body)?;
    let eth_manifest = eth_manifest(value)?;
    Ok(vec![
        manifest_data_root_commitment(&value.body)?,
        abi_commitment(SCHEME_MANIFEST_BODY_ABI, &eth_body),
        abi_commitment(SCHEME_MANIFEST_ENVELOPE_ABI, &eth_manifest),
    ])
}

pub fn manifest_commitment_digest(
    value: &Manifest,
    scheme: &str,
) -> Result<Option<String>, EthCommitmentError> {
    Ok(manifest_commitments(value)?
        .into_iter()
        .find(|commitment| commitment.scheme == scheme)
        .map(|commitment| commitment.digest))
}

pub fn manifest_observation_header(
    value: &Manifest,
    parent_observation_hash: Option<&str>,
    locator_hash: &str,
    sequence: u64,
) -> Result<EthObservationHeader, EthCommitmentError> {
    let data_root = manifest_commitment_digest(value, SCHEME_MANIFEST_DATA_ROOT)?
        .ok_or(EthCommitmentError::Empty)?;

    Ok(EthObservationHeader {
        streamHash: keccak256(value.body.stream_id.0.as_bytes()),
        epochHash: keccak256(value.body.epoch_id.as_bytes()),
        parentObservationHash: parse_b256(parent_observation_hash.unwrap_or(ZERO_B256_HEX))?,
        dataRoot: parse_b256(&data_root)?,
        locatorHash: parse_b256(locator_hash)?,
        observedUnixMs: value.body.created_unix_ms,
        sequence,
    })
}

pub fn observation_header_hash(value: &EthObservationHeader) -> String {
    b256_hex(keccak256(value.abi_encode()))
}

pub fn observation_slot_hash(
    stream_hash: &str,
    epoch_hash: &str,
) -> Result<String, EthCommitmentError> {
    let stream_hash = parse_b256(stream_hash)?;
    let epoch_hash = parse_b256(epoch_hash)?;
    Ok(b256_hex(keccak256((stream_hash, epoch_hash).abi_encode())))
}

pub fn manifest_commitments_match(value: &Manifest) -> Result<bool, EthCommitmentError> {
    let present = value
        .commitments
        .iter()
        .filter(|item| item.chain == ETHEREUM_CHAIN)
        .collect::<Vec<_>>();
    if present.is_empty() {
        return Ok(true);
    }
    let expected = manifest_commitments(value)?;
    Ok(present.into_iter().all(|actual| expected.contains(actual)))
}

#[cfg(test)]
mod tests {
    use super::*;
    use ec_core::{ManifestBody, StreamId};

    fn sample_body() -> ManifestBody {
        let chunk_hashes = vec![
            blake3::hash(b"chunk0").to_hex().to_string(),
            blake3::hash(b"chunk1").to_hex().to_string(),
        ];
        let merkle_root = ec_core::merkle_root_from_hashes(&chunk_hashes).unwrap();
        ManifestBody {
            stream_id: StreamId("ec/stream/v1/broadcast/atsc/tsid-42/program-3".to_string()),
            epoch_id: "epoch-1".to_string(),
            chunk_duration_ms: 2000,
            total_chunks: 2,
            chunk_start_index: 10,
            encoder_profile_id: "deterministic-h264-aac".to_string(),
            merkle_root,
            created_unix_ms: 1234,
            metadata: vec![StreamMetadata {
                key: "callsign".to_string(),
                value: "KCBS".to_string(),
            }],
            chunk_hashes,
            variants: None,
        }
    }

    #[test]
    fn keccak_merkle_root_and_proof_roundtrip() {
        let body = sample_body();
        let root = ethereum_merkle_root_from_hashes(&body.chunk_hashes).unwrap();
        let proof = ethereum_merkle_proof_for_index(&body.chunk_hashes, 1).unwrap();
        assert!(verify_ethereum_merkle_proof(
            &body.chunk_hashes[1],
            1,
            &proof,
            &root
        ));
    }

    #[test]
    fn manifest_commitments_are_stable_and_match_present_values() {
        let body = sample_body();
        let manifest_id = body.manifest_id().unwrap();
        let mut manifest = Manifest {
            body,
            manifest_id,
            signatures: Vec::new(),
            commitments: Vec::new(),
        };
        let commitments = manifest_commitments(&manifest).unwrap();
        assert_eq!(commitments.len(), 3);
        manifest.commitments = commitments.clone();
        assert!(manifest_commitments_match(&manifest).unwrap());
        manifest.commitments[0].digest = "0xdeadbeef".to_string();
        assert!(!manifest_commitments_match(&manifest).unwrap());
    }

    #[test]
    fn manifest_body_eip712_hash_is_stable() {
        let body = sample_body();
        let h1 = manifest_body_eip712_signing_hash(&body).unwrap();
        let h2 = manifest_body_eip712_signing_hash(&body).unwrap();
        assert_eq!(h1, h2);
    }

    #[test]
    fn manifest_observation_header_uses_manifest_data_root() {
        let body = sample_body();
        let manifest_id = body.manifest_id().unwrap();
        let mut manifest = Manifest {
            body,
            manifest_id,
            signatures: Vec::new(),
            commitments: Vec::new(),
        };
        manifest.commitments = manifest_commitments(&manifest).unwrap();

        let locator_hash = keccak256_bytes_hex(b"locator");
        let header = manifest_observation_header(&manifest, None, &locator_hash, 7).unwrap();
        assert_eq!(header.sequence, 7);
        assert_eq!(header.observedUnixMs, manifest.body.created_unix_ms);
        assert_eq!(
            b256_hex(header.dataRoot),
            manifest_commitment_digest(&manifest, SCHEME_MANIFEST_DATA_ROOT)
                .unwrap()
                .unwrap()
        );
        assert_eq!(
            observation_slot_hash(&b256_hex(header.streamHash), &b256_hex(header.epochHash))
                .unwrap(),
            b256_hex(keccak256(
                (header.streamHash, header.epochHash).abi_encode()
            ))
        );
    }

    #[test]
    fn stream_descriptor_commitments_include_stream_id_and_descriptor_hashes() {
        let descriptor = StreamDescriptor {
            id: StreamId("ec/stream/v1/source/test/device-a/channel-b".to_string()),
            title: "Test".to_string(),
            number: Some("2.1".to_string()),
            source: "control".to_string(),
            metadata: vec![StreamMetadata {
                key: "broadcast".to_string(),
                value: "la-nbc".to_string(),
            }],
            commitments: Vec::new(),
        };
        let commitments = stream_descriptor_commitments(&descriptor);
        assert_eq!(commitments.len(), 2);
        assert_eq!(commitments[0].scheme, SCHEME_STREAM_ID_KECCAK);
        assert_eq!(commitments[1].scheme, SCHEME_STREAM_DESCRIPTOR_ABI);
    }
}