ethl 0.1.42

use std::{collections::VecDeque, sync::Arc};

use crate::storage::{
    codec::{SolEventCodec, decoder::DecodedEventWithHeader},
    store::EventStore,
};
use alloy::json_abi::Event;
use anyhow::Result;
use futures_util::{Stream, StreamExt, stream};
use object_store::{ObjectStore, path::Path};

pub struct EventStoreReader {
    store: EventStore,
    codec: SolEventCodec,
}

impl EventStoreReader {
    pub fn new(store: Arc<dyn ObjectStore>, path: Path, event: &Event) -> Result<Self> {
        let codec = SolEventCodec::new(event)?;
        let store = EventStore::new(store, path, &codec)?;
        Ok(EventStoreReader { store, codec })
    }

    /// The returned stream is `Send` and may be spawned onto an executor (e.g. `tokio::spawn`).
    pub async fn stream(
        &self,
        from_block: Option<u64>,
        to_block: Option<u64>,
    ) -> Result<impl Stream<Item = Vec<Result<DecodedEventWithHeader>>> + Send> {
        let raw = self.store.read_range(from_block, to_block).await?;
        Ok(raw.map(|batch_result| match batch_result {
            Ok(batch) => self
                .codec
                .decoder
                .decode_batch_iter(&batch)
                .collect::<Vec<_>>(),
            Err(e) => vec![Err(anyhow::Error::from(e))],
        }))
    }

    /// The returned stream is `Send` and may be spawned onto an executor (e.g. `tokio::spawn`).
    pub async fn stream_deq(
        &self,
        from_block: Option<u64>,
        to_block: Option<u64>,
    ) -> Result<impl Stream<Item = Result<VecDeque<DecodedEventWithHeader>>> + Send> {
        let raw = self.store.read_range(from_block, to_block).await?;
        Ok(raw.map(|batch_result| {
            let batch = batch_result.map_err(anyhow::Error::from)?;
            self.codec.decoder.par_decode_batch_deque(&batch)
        }))
    }

    /// The returned stream is `Send` and may be spawned onto an executor (e.g. `tokio::spawn`).
    pub async fn stream_flat(
        &self,
        from_block: Option<u64>,
        to_block: Option<u64>,
    ) -> Result<impl Stream<Item = Result<DecodedEventWithHeader>> + Send> {
        let raw = self.store.read_range(from_block, to_block).await?;
        Ok(raw.flat_map(|batch_result| match batch_result {
            Ok(batch) => {
                let iter = self
                    .codec
                    .decoder
                    .decode_batch_iter(&batch)
                    .collect::<Vec<_>>();
                stream::iter(iter)
            }
            Err(e) => stream::iter(vec![Err(anyhow::Error::from(e))]),
        }))
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use alloy::json_abi::Event;
    use anyhow::Result;
    use arrow::array::{RecordBatch, StringArray, UInt32Array, UInt64Array};
    use futures_util::TryStreamExt;
    use object_store::ObjectStore;

    use super::*;
    use crate::storage::{codec::SolEventCodec, store::parse_store_uri};

    /// Writes a small archive of its own, under its own tmp path, using real 20-byte hex
    /// addresses -- `store/mod.rs`'s fixture writes placeholder addresses ("0x0", ...) that
    /// fail `Address::parse` in the decode path `stream_deq` exercises.
    async fn write_test_archive(
        path: &str,
    ) -> Result<(Arc<dyn ObjectStore>, object_store::path::Path, Event)> {
        let _ = std::fs::remove_dir_all(path);
        let event = Event::parse("event Hello()")?;
        let codec = SolEventCodec::new(&event)?;

        let (store, base_path) = parse_store_uri(format!("file://{path}"))?;
        let store: Arc<dyn ObjectStore> = Arc::new(store);

        let archive = EventStore::new(store.clone(), base_path.clone(), &codec)?;
        let batch = RecordBatch::try_new(
            codec.schema.clone(),
            vec![
                Arc::new(UInt64Array::from(vec![1, 2])),
                Arc::new(UInt32Array::from(vec![0, 0])),
                Arc::new(StringArray::from(vec![
                    "0x0000000000000000000000000000000000000001",
                    "0x0000000000000000000000000000000000000002",
                ])),
            ],
        )?;
        archive.write_records((1, 2), batch).await?;

        Ok((store, base_path, event))
    }

    /// `stream_deq`'s `Send` bound is what lets `tokio::spawn` accept the future that
    /// constructs and drains an `EventStoreReader` -- an annotation alone cannot satisfy
    /// `tokio::spawn`'s `Future + Send + 'static` bound, which is checked structurally from
    /// the hidden `impl Stream` type. The reader is constructed *inside* the spawned future
    /// (settled decision 8): its returned stream borrows `&self`, so building it outside and
    /// moving only the stream in is not `'static`.
    #[tokio::test]
    async fn test_stream_deq_spawn_guard() -> Result<()> {
        let (store, path, event) =
            write_test_archive("/tmp/test_archive/test_reader_spawn_guard").await?;

        let inline_reader = EventStoreReader::new(store.clone(), path.clone(), &event)?;
        let inline_batches: Vec<_> = inline_reader
            .stream_deq(None, None)
            .await?
            .try_collect()
            .await?;
        let inline_records: Vec<DecodedEventWithHeader> =
            inline_batches.into_iter().flatten().collect();

        let handle = tokio::spawn(async move {
            let reader = EventStoreReader::new(store, path, &event)?;
            let batches: Vec<_> = reader.stream_deq(None, None).await?.try_collect().await?;
            Ok::<_, anyhow::Error>(batches.into_iter().flatten().collect::<Vec<_>>())
        });
        let spawned_records = handle.await??;

        let summarize = |records: &[DecodedEventWithHeader]| {
            records
                .iter()
                .map(|r| (r.log_block, r.log_index, r.log_address))
                .collect::<Vec<_>>()
        };

        assert!(!inline_records.is_empty(), "fixture produced no records");
        assert_eq!(summarize(&inline_records), summarize(&spawned_records));

        Ok(())
    }
}