obzenflow_runtime 0.2.4

// SPDX-License-Identifier: MIT OR Apache-2.0
// SPDX-FileCopyrightText: 2025-2026 ObzenFlow Contributors
// https://obzenflow.dev

use super::types::{
    AdvertisedWriterSeqByEventType, ReaderSelectionPolicy, SelectedDataSeqByEventType,
};
use super::{
    CompositeEntrySpec, ContractTracker, ContractsWiring, DeliveredCount, DeliveryFilter,
    FeedContractChain, FeedIdentity, ReaderSlot, ReaderTiebreakKey, SelectedFeedMetadata, StageKey,
    SubscriptionState, UpstreamSubscription,
};
use crate::contracts::ContractChain;
use crate::control_plane::NoControlPlane;
use crate::messaging::upstream_subscription_policy::build_policy_stack_for_upstream;
use async_trait::async_trait;
use obzenflow_core::event::provenance::FlowContext;
use obzenflow_core::event::types::SeqNo;
use obzenflow_core::event::JournalEvent;
use obzenflow_core::journal::journal_error::JournalError;
use obzenflow_core::journal::reader::JournalReader;
use obzenflow_core::journal::Journal;
use obzenflow_core::{
    DeliveryContract, EventType, JournalRecord, Result, StageId, TransportContract,
};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

/// Fallback reader used when a real journal reader cannot be created.
///
/// This reader behaves as an always-empty journal (EOF), allowing the
/// subscription machinery to continue operating without failing the FSM.
struct EmptyJournalReader<T: JournalEvent> {
    _phantom: std::marker::PhantomData<T>,
}

impl<T: JournalEvent> EmptyJournalReader<T> {
    fn new() -> Self {
        Self {
            _phantom: std::marker::PhantomData,
        }
    }
}

#[async_trait]
impl<T> JournalReader<T> for EmptyJournalReader<T>
where
    T: JournalEvent + Send + Sync + 'static,
{
    async fn next(
        &mut self,
    ) -> std::result::Result<Option<JournalRecord<T::Payload>>, JournalError> {
        Ok(None)
    }

    fn position(&self) -> u64 {
        0
    }

    fn is_at_end(&self) -> bool {
        true
    }
    fn initial_prefix_complete(&self) -> std::result::Result<bool, JournalError> {
        Ok(true)
    }
}

impl<T> UpstreamSubscription<T>
where
    T: JournalEvent + 'static,
{
    /// Create a new subscription from upstream journals
    pub async fn new_with_names(
        owner_label: &str,
        upstream_journals: &[(StageId, String, Arc<dyn Journal<T>>)],
    ) -> Result<Self> {
        // Delegate to the position-aware constructor with all starting
        // positions at 0 (from-beginning semantics).
        let start_positions = vec![0u64; upstream_journals.len()];
        Self::new_with_names_from_positions(owner_label, upstream_journals, &start_positions).await
    }

    /// Create a new subscription from upstream journals, starting each reader
    /// from an explicit position.
    ///
    /// This is used by the metrics aggregator (FLOWIP-059 Phase 6) to
    /// fast-forward readers to the tail while still seeding snapshot metrics
    /// from wide events. Other callers should generally prefer `new_with_names`.
    pub async fn new_with_names_from_positions(
        owner_label: &str,
        upstream_journals: &[(StageId, String, Arc<dyn Journal<T>>)],
        start_positions: &[u64],
    ) -> Result<Self> {
        if upstream_journals.len() != start_positions.len() {
            return Err(format!(
                "start_positions length {} does not match upstream_journals length {}",
                start_positions.len(),
                upstream_journals.len()
            )
            .into());
        }

        let mut readers = Vec::new();

        tracing::debug!(
            "Creating subscription for {} upstream journals",
            upstream_journals.len()
        );

        tracing::debug!(
            target: "flowip-080o",
            owner = owner_label,
            readers = ?upstream_journals
                .iter()
                .map(|(_id, name, journal)| format!("{} ({})", name, journal.id()))
                .collect::<Vec<_>>(),
            "UpstreamSubscription::new_with_names binding readers"
        );

        for ((stage_id, stage_name, journal), position) in
            upstream_journals.iter().zip(start_positions.iter())
        {
            // Get journal ID for debugging
            let journal_id = journal.id();
            tracing::debug!(
                target: "flowip-080o",
                stage_id = ?stage_id,
                stage_name = stage_name,
                journal_id = ?journal_id,
                "Creating reader for upstream journal"
            );
            let reader_result = if *position == 0 {
                journal.reader().await
            } else {
                journal.reader_from(*position).await
            };

            let reader: Box<dyn JournalReader<T>> = match reader_result {
                Ok(reader) => reader,
                Err(e)
                    if crate::runtime_resource_limits::journal_error_is_too_many_open_files(&e) =>
                {
                    return Err(format!(
                        "Too many open files while creating reader for upstream journal (owner={owner_label}, stage_id={stage_id:?}, stage_name={stage_name}, journal_id={journal_id:?}). Increase RLIMIT_NOFILE / `ulimit -n` or reduce pipeline size (for development, disable metrics in obzenflow.toml). Underlying error: {e}"
                    )
                    .into());
                }
                Err(JournalError::Implementation { message, source }) => {
                    // Best-effort: log the failure and use an empty reader so the
                    // FSM can continue operating (upstream treated as having no events).
                    tracing::error!(
                        target: "flowip-080o",
                        stage_id = ?stage_id,
                        stage_name = stage_name,
                        journal_id = ?journal_id,
                        journal_error_message = %message,
                        journal_error_source = %source,
                        "Failed to create reader for upstream journal; using EmptyJournalReader (no events)"
                    );
                    Box::new(EmptyJournalReader::<T>::new()) as Box<dyn JournalReader<T>>
                }
                Err(e) => {
                    return Err(
                        format!("Failed to create reader for stage {stage_id:?}: {e}").into(),
                    );
                }
            };
            readers.push(ReaderSlot {
                stage_id: *stage_id,
                stage_key: StageKey::new(stage_name.clone()),
                reader,
            });
        }

        let state = SubscriptionState::new(readers.len());
        let reader_tiebreak_keys = readers
            .iter()
            .map(|slot| ReaderTiebreakKey {
                stage_key: slot.stage_key.clone(),
                feed_identity: FeedIdentity::unfiltered(),
            })
            .collect();

        Ok(Self {
            delivery_filter: DeliveryFilter::All,
            owner_label: owner_label.to_string(),
            selected_data_seq_by_reader: vec![SeqNo(0); readers.len()],
            selected_data_seq_by_reader_event_type: vec![
                SelectedDataSeqByEventType::default();
                readers.len()
            ],
            advertised_writer_seq_by_reader_event_type: vec![
                AdvertisedWriterSeqByEventType::default(
                );
                readers.len()
            ],
            held_heads: readers.iter().map(|_| None).collect(),
            delivered_count_by_reader: vec![DeliveredCount::default(); readers.len()],
            generation_by_reader: vec![obzenflow_core::ReaderGeneration::default(); readers.len()],
            last_positional_seq: vec![obzenflow_core::AdmissionSeq(0); readers.len()],
            last_delivered_generation: None,
            reader_tiebreak_keys,
            readers,
            archived_stage_ids_by_current: HashMap::new(),
            selected_event_types_by_stage: HashMap::new(),
            selected_feeds_by_stage: HashMap::new(),
            composite_entries_by_stage: HashMap::new(),
            state,
            contract_tracker: None,
            contract_chains: Vec::new(),
            contract_feed_chains: Vec::new(),
            contract_policies: Vec::new(),
            control_plane: Arc::new(NoControlPlane),
            last_eof_outcome: None,
            last_delivered_upstream_stage: None,
            next_stage_input_position: 1,
            last_delivered_stage_input_position: None,
            reader_selection: ReaderSelectionPolicy::default(),
            seq_ordered: false,
            entered_generation: obzenflow_core::ReaderGeneration::default(),
            last_merge_wait: None,
            merge_candidate_index: None,
        })
    }

    /// Choose the reader-selection policy (FLOWIP-095d).
    ///
    /// `CanonicalMerge` requires every reader to start at position zero with
    /// no tail-start baseline: a non-zero starting point would shift
    /// per-reader delivered ordinals between runs and break the
    /// deterministic-order function. The invariant is enforced here, where
    /// the policy is chosen, because construction order (`new_at_tail` then
    /// this) would otherwise admit a silently broken subscription.
    pub fn with_reader_selection(mut self, policy: ReaderSelectionPolicy) -> Self {
        if policy == ReaderSelectionPolicy::CanonicalMerge {
            for (index, slot) in self.readers.iter().enumerate() {
                let position = slot.reader.position();
                assert!(
                    position == 0 && !self.state.baseline_at_tail[index],
                    "FLOWIP-095d: CanonicalMerge requires reader '{}' at position \
                     zero with no tail-start baseline (position {position}); a non-zero \
                     starting point shifts per-reader delivered ordinals between runs and \
                     breaks the deterministic merge",
                    slot.stage_key
                );
            }
        }
        self.reader_selection = policy;
        self
    }

    /// FLOWIP-120n F18: run the seq-ordered merge. Only meaningful under
    /// `CanonicalMerge`; the flow build sets both together on marked
    /// source-fed ordered fan-ins.
    pub fn with_seq_ordered(mut self, seq_ordered: bool) -> Self {
        self.seq_ordered = seq_ordered;
        self
    }

    /// The generation this run entered at (FLOWIP-120n F18): 0 live, archive
    /// max recorded generation + 1 on replay/resume. Gates the seq-mode
    /// quiet-input wait exemption per reader.
    pub fn with_entered_generation(mut self, entered: obzenflow_core::ReaderGeneration) -> Self {
        self.entered_generation = entered;
        self
    }

    /// Install stable replay aliases from current topology stages to the
    /// corresponding archived stage authors.
    pub fn with_archived_stage_ids(
        mut self,
        archived_stage_ids_by_current: HashMap<StageId, StageId>,
    ) -> Self {
        self.archived_stage_ids_by_current = archived_stage_ids_by_current;
        self
    }

    /// Configure this subscription to deliver only transport-relevant events to the caller.
    ///
    /// This is intended for *stage runtime* subscriptions where downstream stages should
    /// not be forced to process upstream observability events (e.g. middleware metrics)
    /// as part of normal draining / shutdown.
    pub fn transport_only(mut self) -> Self {
        self.delivery_filter = DeliveryFilter::TransportOnly;
        self
    }

    /// Configure selected Data event types per upstream reader.
    pub fn with_selected_event_types(
        mut self,
        selected_event_types_by_stage: HashMap<StageId, HashSet<EventType>>,
    ) -> Self {
        self.selected_feeds_by_stage = selected_event_types_by_stage
            .iter()
            .map(|(stage_id, event_types)| {
                let feeds = event_types
                    .iter()
                    .cloned()
                    .map(SelectedFeedMetadata::unscoped)
                    .collect();
                (*stage_id, feeds)
            })
            .collect();
        self.selected_event_types_by_stage = selected_event_types_by_stage;
        self.recompute_tiebreak_keys();
        self
    }

    /// Configure selected logical feeds per upstream reader.
    pub fn with_selected_feeds(
        mut self,
        selected_feeds_by_stage: HashMap<StageId, Vec<SelectedFeedMetadata>>,
    ) -> Self {
        self.selected_event_types_by_stage = selected_feeds_by_stage
            .iter()
            .map(|(stage_id, feeds)| {
                (
                    *stage_id,
                    feeds
                        .iter()
                        .map(|feed| feed.event_type().clone())
                        .collect::<HashSet<_>>(),
                )
            })
            .collect();
        self.selected_feeds_by_stage = selected_feeds_by_stage;
        self.recompute_tiebreak_keys();
        self
    }

    /// Configure exact composite input-boundary stamps per upstream edge.
    pub fn with_composite_entries(
        mut self,
        entries_by_stage: HashMap<StageId, Vec<CompositeEntrySpec>>,
    ) -> Self {
        self.composite_entries_by_stage = entries_by_stage;
        self
    }

    /// Recompute per-reader tiebreak keys (FLOWIP-095d).
    ///
    /// The key is (stage key, feed identity), both stable across runs;
    /// per-run `StageId` ULIDs never participate. Two readers consuming
    /// different selected feeds of the same upstream stage share a stage key,
    /// so the feed identity keeps the key total, and its role qualifier keeps
    /// it total even when two readers share a stage key and event type but
    /// differ by feed role.
    fn recompute_tiebreak_keys(&mut self) {
        self.reader_tiebreak_keys = self
            .readers
            .iter()
            .map(|slot| ReaderTiebreakKey {
                stage_key: slot.stage_key.clone(),
                feed_identity: self
                    .selected_feeds_by_stage
                    .get(&slot.stage_id)
                    .map(|feeds| FeedIdentity::from_feeds(feeds))
                    .unwrap_or_default(),
            })
            .collect();
    }

    /// Create a subscription starting from explicit tail positions.
    ///
    /// Readers are treated as logically at EOF for historical data
    /// (baseline_at_tail = true) but will still observe any new
    /// events appended after subscription creation. This is used by
    /// tail-first observers like the metrics aggregator which seed
    /// from tail snapshots and do not need to re-observe historical
    /// EOF control events.
    pub async fn new_at_tail(
        owner_label: &str,
        upstream_journals: &[(StageId, String, Arc<dyn Journal<T>>)],
        tail_positions: &[u64],
    ) -> Result<Self> {
        let mut sub =
            Self::new_with_names_from_positions(owner_label, upstream_journals, tail_positions)
                .await?;

        // FLOWIP-095d: tail-start subscriptions are incompatible with the
        // canonical merge; `with_reader_selection` enforces the invariant.

        // Only mark a reader as baseline-at-tail if we're actually skipping historical
        // events (i.e., the computed tail position is non-zero). For fresh/empty
        // journals, setting this baseline would incorrectly allow "logical EOF"
        // termination before new events are observed.
        let mut baseline_count = 0usize;
        for (idx, tail_position) in tail_positions.iter().take(sub.readers.len()).enumerate() {
            if *tail_position > 0 {
                sub.state.mark_reader_baseline_at_tail(idx);
                baseline_count += 1;
            }
        }

        tracing::info!(
            target: "flowip-059d",
            owner = owner_label,
            reader_count = sub.readers.len(),
            baseline_count = baseline_count,
            "Created tail-start upstream subscription"
        );

        Ok(sub)
    }

    /// Backwards-compatible constructor using stage IDs as names
    pub async fn new(upstream_journals: &[(StageId, Arc<dyn Journal<T>>)]) -> Result<Self> {
        let with_names: Vec<(StageId, String, Arc<dyn Journal<T>>)> = upstream_journals
            .iter()
            .map(|(id, journal)| (*id, format!("{id:?}"), journal.clone()))
            .collect();
        Self::new_with_names("unknown_owner", &with_names).await
    }

    /// Attach the consuming stage's context before a framework subscription is polled.
    /// Contracts must already be enabled; public construction keeps its existing default.
    pub(crate) fn with_contract_flow_context(mut self, owner: FlowContext) -> Self {
        self.contract_tracker
            .as_mut()
            .expect("contract flow context requires contract wiring")
            .flow_context = Some(owner);
        self
    }

    /// Enable contract emission for at-least-once delivery guarantees
    pub fn with_contracts(mut self, wiring: ContractsWiring) -> Self {
        let ContractsWiring {
            writer_id,
            contract_journal,
            config,
            system_journal,
            reader_stage,
            control_plane,
            include_delivery_contract,
            cycle_guard_config,
        } = wiring;

        self.control_plane = control_plane.clone();

        self.contract_tracker = Some(ContractTracker {
            config,
            writer_id,
            journal: contract_journal,
            system_journal,
            reader_stage,
            receipt_aware_progress: include_delivery_contract,
            flow_context: None,
            output_events_written: SeqNo(0),
        });

        // Initialize per-reader contract chains using the new Contract framework.
        // For 090c v1, we attach a TransportContract to each upstream edge.
        if !self.readers.is_empty() {
            self.contract_chains = self
                .readers
                .iter()
                .map(|slot| {
                    let mut chain = ContractChain::new()
                        .with_contract(TransportContract::new())
                        .with_contract(obzenflow_core::SourceContract::new());
                    if include_delivery_contract {
                        chain = chain.with_contract(DeliveryContract::default());
                    }

                    // Divergence detection is attached only for SCC-internal upstreams
                    // when cycle metadata is available (FLOWIP-080r).
                    if let Some(cycle_cfg) = &cycle_guard_config {
                        if cycle_cfg.internal_upstreams.contains(&slot.stage_id) {
                            let thresholds = obzenflow_core::DivergenceThresholds {
                                max_cycle_depth: cycle_cfg.max_iterations.as_u16(),
                                ..Default::default()
                            };
                            chain = chain.with_contract(
                                obzenflow_core::DivergenceContract::with_thresholds(
                                    cycle_cfg.scc_id,
                                    thresholds,
                                ),
                            );
                        }
                    }
                    Some(chain)
                })
                .collect();

            // Selected-feed chains intentionally start with the transport
            // contract only. Delivery receipts and divergence predicates still
            // use the physical edge chain until those contracts gain explicit
            // selected-feed semantics.
            self.contract_feed_chains = self
                .readers
                .iter()
                .map(|slot| {
                    let Some(feeds) = self.selected_feeds_by_stage.get(&slot.stage_id) else {
                        return Vec::new();
                    };
                    if feeds.len() <= 1 {
                        return Vec::new();
                    }

                    feeds
                        .iter()
                        .cloned()
                        .map(|feed| {
                            FeedContractChain::new(
                                feed,
                                ContractChain::new().with_contract(TransportContract::new()),
                            )
                        })
                        .collect()
                })
                .collect();

            // Initialize per-reader policy stacks using the upstream stage IDs.
            self.contract_policies = self
                .readers
                .iter()
                .map(|slot| {
                    let stack = build_policy_stack_for_upstream(slot.stage_id);
                    Some(stack)
                })
                .collect();
        }

        self
    }
}