surreal-sync-runtime 0.6.0

//! Ordered transform pipeline: in-place stages and external boundary.

use crate::pipeline::apply::ApplyEvent;
use crate::pipeline::external::ExternalTransform;
use anyhow::{bail, Result};
use std::sync::Arc;
use surreal_sync_core::InPlaceTransform;
use surreal_sync_core::{Change, Relation, RelationChange, Row};

/// A single pipeline stage.
#[derive(Clone)]
pub enum Stage {
    /// In-process mutate-only transform.
    InPlace(Arc<dyn InPlaceTransform>),
    /// External worker boundary (child-stdio NDJSON).
    External(ExternalTransform),
}

impl std::fmt::Debug for Stage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Stage::InPlace(_) => f.write_str("InPlace(_)"),
            Stage::External(ext) => f.debug_tuple("External").field(ext).finish(),
        }
    }
}

/// Ordered list of transform stages.
///
/// An empty pipeline is **identity**: [`is_identity`](Self::is_identity) is
/// true and apply helpers return immediately without dispatching any stage.
///
/// # Identity vs passthrough
///
/// Pushing a lone [`crate::pipeline::Passthrough`] via [`push_inplace`](Self::push_inplace)
/// does **not** make [`is_identity`](Self::is_identity) return `true` — the
/// pipeline still has a stage and will dispatch into it. TOML config loading
/// ([`crate::pipeline::Pipeline::from_config`] / [`crate::pipeline::parse_transforms_toml`])
/// collapses passthrough-only configs to an empty pipeline so the CLI identity
/// path stays zero-dispatch.
///
/// # Apply path
///
/// Shared apply loop: transform → ordered write → watermark.
/// [`crate::pipeline::ApplyContext`] / [`crate::pipeline::SourceDriver`] gate on
/// [`crate::pipeline::BatchTransformer::is_identity`] (implemented for [`Pipeline`] via
/// [`is_identity`](Self::is_identity)) so the transform [`tokio::task::JoinSet`]
/// can take an async no-op path (zero stage dispatch). That is **not** a
/// write-path bypass: identity batches still go through the ordered sink step
/// (including homogeneous `Update` coalesce to `write_universal_*`). Only an
/// empty stage list is identity — not “stages happen to be no-ops.”
///
/// # Schema-aware / FK transforms
///
/// Construct an [`InPlaceTransform`] with schema (e.g. FK → `Thing` links) and
/// [`push_inplace`](Self::push_inplace) it. Relation edges use the same
/// pipeline via relation transform methods; full join-table→relation conversion
/// may still live in a source crate.
///
/// # External + relations
///
/// External stages exchange **both** row changes and relation changes over
/// NDJSON ([`ExternalTransform::exchange_relation_changes`] /
/// [`ExternalTransform::exchange_relations`]). There is no silent relation
/// pass-through. Mixed change+relation batches may not filter/fan-out (length
/// of each kind must be preserved); use homogeneous batches for length changes.
///
/// When a single External stage sees a **mixed** change+relation batch, the two
/// wire exchanges use distinct `batch_id`s: changes keep the apply `batch_id`,
/// relations use [`crate::pipeline::relation_wire_batch_id`] (high bit set) so workers and
/// outstanding-id tracking never confuse the two sequential exchanges.
#[derive(Debug, Default, Clone)]
pub struct Pipeline {
    stages: Vec<Stage>,
}

impl Pipeline {
    /// Create an empty (identity) pipeline.
    pub fn new() -> Self {
        Self { stages: Vec::new() }
    }

    /// Whether this pipeline has no stages (identity / no stage dispatch).
    ///
    /// Only an empty stage list is identity. A pipeline that contains only
    /// [`crate::pipeline::Passthrough`] still returns `false` here — see type-level docs.
    pub fn is_identity(&self) -> bool {
        self.stages.is_empty()
    }

    /// Number of stages (0 = identity).
    pub fn len(&self) -> usize {
        self.stages.len()
    }

    /// Whether there are no stages.
    pub fn is_empty(&self) -> bool {
        self.stages.is_empty()
    }

    /// Borrow the stage list.
    pub fn stages(&self) -> &[Stage] {
        &self.stages
    }

    /// Append an in-place transform stage (library / embedder API).
    ///
    /// Note: appending [`crate::pipeline::Passthrough`] alone does not yield an identity
    /// pipeline ([`is_identity`](Self::is_identity) stays `false`).
    pub fn push_inplace<T>(&mut self, transform: T)
    where
        T: InPlaceTransform + 'static,
    {
        self.stages.push(Stage::InPlace(Arc::new(transform)));
    }

    /// Append a pre-boxed in-place stage.
    pub fn push_inplace_arc(&mut self, transform: Arc<dyn InPlaceTransform>) {
        self.stages.push(Stage::InPlace(transform));
    }

    /// Append an external (child-stdio) stage.
    pub fn push_external(&mut self, external: ExternalTransform) {
        self.stages.push(Stage::External(external));
    }

    /// Transform owned rows in place (sync path — **in-place stages only**).
    ///
    /// Empty pipeline: no-op with no stage dispatch. External stages are not
    /// supported here; use [`crate::pipeline::BatchTransformer::transform_rows`] (async).
    pub fn transform_rows_inplace(&self, rows: &mut [Row]) -> Result<()> {
        if self.is_identity() {
            return Ok(());
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_rows_inplace(rows)?,
                Stage::External(_) => {
                    bail!(
                        "External transforms require the async BatchTransformer path \
                         (transform_rows); sync inplace apply is in-place-only"
                    )
                }
            }
        }
        Ok(())
    }

    /// Transform owned changes in place (sync path — **in-place stages only**).
    ///
    /// Empty pipeline: no-op with no stage dispatch. External stages are not
    /// supported here; use [`crate::pipeline::BatchTransformer::transform_changes`] (async).
    pub fn transform_changes_inplace(&self, changes: &mut [Change]) -> Result<()> {
        if self.is_identity() {
            return Ok(());
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_changes_inplace(changes)?,
                Stage::External(_) => {
                    bail!(
                        "External transforms require the async BatchTransformer path \
                         (transform_changes); sync inplace apply is in-place-only"
                    )
                }
            }
        }
        Ok(())
    }

    /// Consume an owned row batch, transform in place, and return it.
    ///
    /// Preferred path for **in-place-only** pipelines: empty
    /// pipeline is a pure move with no transform dispatch.
    pub fn apply_rows(&self, mut rows: Vec<Row>) -> Result<Vec<Row>> {
        self.transform_rows_inplace(&mut rows)?;
        Ok(rows)
    }

    /// Consume an owned change batch, transform in place, and return it.
    pub fn apply_changes(&self, mut changes: Vec<Change>) -> Result<Vec<Change>> {
        self.transform_changes_inplace(&mut changes)?;
        Ok(changes)
    }

    /// Async stage walk used by [`crate::pipeline::BatchTransformer`]: in-place mutates,
    /// External exchanges over child-stdio (may change batch length).
    pub(crate) async fn apply_changes_async(
        &self,
        batch_id: u64,
        mut changes: Vec<Change>,
    ) -> Result<Vec<Change>> {
        if self.is_identity() {
            return Ok(changes);
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_changes_inplace(&mut changes)?,
                Stage::External(ext) => {
                    changes = ext.exchange_changes(batch_id, changes).await?;
                }
            }
        }
        Ok(changes)
    }

    /// Async stage walk for rows (see [`Self::apply_changes_async`]).
    pub(crate) async fn apply_rows_async(
        &self,
        batch_id: u64,
        mut rows: Vec<Row>,
    ) -> Result<Vec<Row>> {
        if self.is_identity() {
            return Ok(rows);
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_rows_inplace(&mut rows)?,
                Stage::External(ext) => {
                    rows = ext.exchange_rows(batch_id, rows).await?;
                }
            }
        }
        Ok(rows)
    }

    /// Transform owned relation changes in place (sync — **in-place stages only**).
    pub fn transform_relation_changes_inplace(&self, changes: &mut [RelationChange]) -> Result<()> {
        if self.is_identity() {
            return Ok(());
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_relation_changes_inplace(changes)?,
                Stage::External(_) => {
                    bail!(
                        "External transforms require the async BatchTransformer path \
                         (transform_relation_changes); sync inplace apply is in-place-only"
                    )
                }
            }
        }
        Ok(())
    }

    /// Transform owned relations in place (sync — **in-place stages only**).
    pub fn transform_relations_inplace(&self, relations: &mut [Relation]) -> Result<()> {
        if self.is_identity() {
            return Ok(());
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_relations_inplace(relations)?,
                Stage::External(_) => {
                    bail!(
                        "External transforms require the async BatchTransformer path \
                         (transform_relations); sync inplace apply is in-place-only"
                    )
                }
            }
        }
        Ok(())
    }

    /// Consume owned relation changes, transform in place, return them.
    pub fn apply_relation_changes(
        &self,
        mut changes: Vec<RelationChange>,
    ) -> Result<Vec<RelationChange>> {
        self.transform_relation_changes_inplace(&mut changes)?;
        Ok(changes)
    }

    /// Consume owned relations, transform in place, return them.
    pub fn apply_relations(&self, mut relations: Vec<Relation>) -> Result<Vec<Relation>> {
        self.transform_relations_inplace(&mut relations)?;
        Ok(relations)
    }

    pub(crate) async fn apply_relation_changes_async(
        &self,
        batch_id: u64,
        mut changes: Vec<RelationChange>,
    ) -> Result<Vec<RelationChange>> {
        if self.is_identity() {
            return Ok(changes);
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_relation_changes_inplace(&mut changes)?,
                Stage::External(ext) => {
                    changes = ext.exchange_relation_changes(batch_id, changes).await?;
                }
            }
        }
        Ok(changes)
    }

    pub(crate) async fn apply_relations_async(
        &self,
        batch_id: u64,
        mut relations: Vec<Relation>,
    ) -> Result<Vec<Relation>> {
        if self.is_identity() {
            return Ok(relations);
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => t.transform_relations_inplace(&mut relations)?,
                Stage::External(ext) => {
                    relations = ext.exchange_relations(batch_id, relations).await?;
                }
            }
        }
        Ok(relations)
    }

    /// Mixed event walk: InPlace applies to each kind; External exchanges **both**
    /// row changes and relation changes over NDJSON (no silent pass-through).
    ///
    /// Homogeneous batches may filter/fan-out (length may change). Mixed
    /// change+relation batches must preserve length of each kind.
    ///
    /// Mixed External exchanges use distinct wire `batch_id`s (see
    /// [`crate::pipeline::relation_wire_batch_id`]) so change vs relation requests never
    /// share an id.
    pub(crate) async fn apply_events_async(
        &self,
        batch_id: u64,
        mut events: Vec<ApplyEvent>,
    ) -> Result<Vec<ApplyEvent>> {
        if self.is_identity() {
            return Ok(events);
        }
        for stage in &self.stages {
            match stage {
                Stage::InPlace(t) => {
                    for event in &mut events {
                        match event {
                            ApplyEvent::Change(c) => t.transform_change(c)?,
                            ApplyEvent::RelationChange(r) => t.transform_relation_change(r)?,
                        }
                    }
                }
                Stage::External(ext) => {
                    let all_changes = events.iter().all(|e| e.is_change());
                    let all_rels = events.iter().all(|e| e.is_relation_change());
                    if all_changes {
                        let changes: Vec<Change> = events
                            .into_iter()
                            .map(|e| match e {
                                ApplyEvent::Change(c) => c,
                                ApplyEvent::RelationChange(_) => unreachable!(),
                            })
                            .collect();
                        let transformed = ext.exchange_changes(batch_id, changes).await?;
                        events = transformed.into_iter().map(ApplyEvent::Change).collect();
                    } else if all_rels {
                        let rels: Vec<RelationChange> = events
                            .into_iter()
                            .map(|e| match e {
                                ApplyEvent::RelationChange(r) => *r,
                                ApplyEvent::Change(_) => unreachable!(),
                            })
                            .collect();
                        let transformed = ext.exchange_relation_changes(batch_id, rels).await?;
                        events = transformed
                            .into_iter()
                            .map(ApplyEvent::relation_change)
                            .collect();
                    } else {
                        let mut change_idxs = Vec::new();
                        let mut changes = Vec::new();
                        let mut rel_idxs = Vec::new();
                        let mut rels = Vec::new();
                        for (i, event) in events.iter().enumerate() {
                            match event {
                                ApplyEvent::Change(c) => {
                                    change_idxs.push(i);
                                    changes.push(c.clone());
                                }
                                ApplyEvent::RelationChange(r) => {
                                    rel_idxs.push(i);
                                    rels.push((**r).clone());
                                }
                            }
                        }
                        // Distinct wire ids: reuse of `batch_id` for both kinds
                        // would collide in outstanding tracking / worker scripts.
                        let rel_batch_id = crate::pipeline::relation_wire_batch_id(batch_id);
                        if !changes.is_empty() {
                            let n = changes.len();
                            let transformed = ext.exchange_changes(batch_id, changes).await?;
                            if transformed.len() != n {
                                bail!(
                                    "External stage changed change-count ({n} → {}) while relation \
                                     events were present in the same batch; use homogeneous batches \
                                     for filter/fan-out",
                                    transformed.len()
                                );
                            }
                            for (idx, c) in change_idxs.into_iter().zip(transformed) {
                                events[idx] = ApplyEvent::Change(c);
                            }
                        }
                        if !rels.is_empty() {
                            let n = rels.len();
                            let transformed =
                                ext.exchange_relation_changes(rel_batch_id, rels).await?;
                            if transformed.len() != n {
                                bail!(
                                    "External stage changed relation-count ({n} → {}) while row \
                                     changes were present in the same batch; use homogeneous batches \
                                     for filter/fan-out",
                                    transformed.len()
                                );
                            }
                            for (idx, r) in rel_idxs.into_iter().zip(transformed) {
                                events[idx] = ApplyEvent::relation_change(r);
                            }
                        }
                    }
                }
            }
        }
        Ok(events)
    }
}