chore(kona-derive): Online Pipeline Cleanup (#241)

* chore: clean up online pipeline constructor

* chore: re-export input types for the online pipeline constructor

* fix: re-export pipeline trait

crates/derive/src/lib.rs

@@ -21,5 +21,3 @@ pub mod types;
 
 #[cfg(feature = "online")]
 pub mod online;
-#[cfg(feature = "online")]
-pub use online::new_online_stack;

crates/derive/src/online/mod.rs

@@ -1,52 +1,19 @@
 //! Contains "online" implementations for providers.
 
-use crate::{
-    stages::{
-        AttributesQueue, BatchQueue, ChannelBank, ChannelReader, FrameQueue, L1Retrieval,
-        L1Traversal, StatefulAttributesBuilder,
-    },
-    traits::DataAvailabilityProvider,
+// Re-export types for the online pipeline construction.
+pub use crate::{
+    pipeline::{DerivationPipeline, PipelineBuilder},
+    sources::EthereumDataSource,
+    stages::StatefulAttributesBuilder,
+    traits::Pipeline,
     types::RollupConfig,
 };
-use alloc::sync::Arc;
-use core::fmt::Debug;
 
-/// An `online` payload attributes builder for the `AttributesQueue` stage of the derivation
-/// pipeline.
-pub type OnlineAttributesBuilder =
-    StatefulAttributesBuilder<AlloyChainProvider, AlloyL2ChainProvider>;
-
-/// An `online` attributes queue for the derivation pipeline.
-pub type OnlineAttributesQueue<DAP> = AttributesQueue<
-    BatchQueue<
-        ChannelReader<ChannelBank<FrameQueue<L1Retrieval<DAP, L1Traversal<AlloyChainProvider>>>>>,
-        AlloyL2ChainProvider,
-    >,
-    OnlineAttributesBuilder,
->;
-
-/// Creates a new online stack.
-pub fn new_online_stack<DAP>(
-    rollup_config: Arc<RollupConfig>,
-    chain_provider: AlloyChainProvider,
-    dap_source: DAP,
-    fetcher: AlloyL2ChainProvider,
-    builder: OnlineAttributesBuilder,
-) -> OnlineAttributesQueue<DAP>
-where
-    DAP: DataAvailabilityProvider + Debug + Send,
-{
-    let l1_traversal = L1Traversal::new(chain_provider, rollup_config.clone());
-    let l1_retrieval = L1Retrieval::new(l1_traversal, dap_source);
-    let frame_queue = FrameQueue::new(l1_retrieval);
-    let channel_bank = ChannelBank::new(rollup_config.clone(), frame_queue);
-    let channel_reader = ChannelReader::new(channel_bank, rollup_config.clone());
-    let batch_queue = BatchQueue::new(rollup_config.clone(), channel_reader, fetcher);
-    AttributesQueue::new(*rollup_config, batch_queue, builder)
-}
-
-#[cfg(any(test, feature = "test-utils"))]
-pub mod test_utils;
+mod pipeline;
+pub use pipeline::{
+    new_online_pipeline, OnlineAttributesBuilder, OnlineAttributesQueue, OnlineDataProvider,
+    OnlinePipeline,
+};
 
 mod validation;
 pub use validation::{OnlineValidator, Validator};
@@ -59,3 +26,6 @@ pub use alloy_providers::{AlloyChainProvider, AlloyL2ChainProvider};
 
 mod blob_provider;
 pub use blob_provider::{OnlineBlobProvider, SimpleSlotDerivation};
+
+#[cfg(any(test, feature = "test-utils"))]
+pub mod test_utils;

crates/derive/src/online/pipeline.rs

@@ -0,0 +1,58 @@
+//! Contains online pipeline types.
+
+use crate::{
+    online::{
+        AlloyChainProvider, AlloyL2ChainProvider, OnlineBeaconClient, OnlineBlobProvider,
+        SimpleSlotDerivation,
+    },
+    pipeline::{DerivationPipeline, PipelineBuilder},
+    sources::EthereumDataSource,
+    stages::{
+        AttributesQueue, BatchQueue, ChannelBank, ChannelReader, FrameQueue, L1Retrieval,
+        L1Traversal, StatefulAttributesBuilder,
+    },
+    types::RollupConfig,
+};
+use alloc::sync::Arc;
+
+/// An online derivation pipeline.
+pub type OnlinePipeline =
+    DerivationPipeline<OnlineAttributesQueue<OnlineDataProvider>, AlloyL2ChainProvider>;
+
+/// An `online` Ethereum data source.
+pub type OnlineDataProvider = EthereumDataSource<
+    AlloyChainProvider,
+    OnlineBlobProvider<OnlineBeaconClient, SimpleSlotDerivation>,
+>;
+
+/// An `online` payload attributes builder for the `AttributesQueue` stage of the derivation
+/// pipeline.
+pub type OnlineAttributesBuilder =
+    StatefulAttributesBuilder<AlloyChainProvider, AlloyL2ChainProvider>;
+
+/// An `online` attributes queue for the derivation pipeline.
+pub type OnlineAttributesQueue<DAP> = AttributesQueue<
+    BatchQueue<
+        ChannelReader<ChannelBank<FrameQueue<L1Retrieval<DAP, L1Traversal<AlloyChainProvider>>>>>,
+        AlloyL2ChainProvider,
+    >,
+    OnlineAttributesBuilder,
+>;
+
+/// Creates a new online [DerivationPipeline] from the given inputs.
+/// Internally, this uses the [PipelineBuilder] to construct the pipeline.
+pub fn new_online_pipeline(
+    rollup_config: Arc<RollupConfig>,
+    chain_provider: AlloyChainProvider,
+    dap_source: OnlineDataProvider,
+    l2_chain_provider: AlloyL2ChainProvider,
+    builder: OnlineAttributesBuilder,
+) -> OnlinePipeline {
+    PipelineBuilder::new()
+        .rollup_config(rollup_config)
+        .dap_source(dap_source)
+        .l2_chain_provider(l2_chain_provider)
+        .chain_provider(chain_provider)
+        .builder(builder)
+        .build()
+}