///|
pub(all) enum PipelineAction {
  PipelineValidate(EventSchema)
  PipelineRules(RuleSet)
  PipelineRoute(Bus, DeliveryMode)
  PipelineProject(ProjectionSpec)
  PipelineSnapshot(Bus)
} derive(Eq, @debug.Debug)

///|
pub(all) enum PipelineStageStatus {
  StagePassed
  StageFailed
  StageWarning
} derive(Eq, @debug.Debug)

///|
pub(all) struct PipelineStage {
  index : Int
  action : PipelineAction
  status : PipelineStageStatus
  summary : String
  lines : Array[String]
} derive(Eq, @debug.Debug)

///|
pub(all) struct EventPipeline {
  name : String
  batch : EventBatch
  actions : Array[PipelineAction]
} derive(Eq, @debug.Debug)

///|
pub(all) struct PipelineRun {
  name : String
  stages : Array[PipelineStage]
  passed : Int
  failed : Int
  warnings : Int
} derive(Eq, @debug.Debug)

///|
pub fn event_pipeline(
  name : StringView,
  batch : EventBatch,
  actions? : ArrayView[PipelineAction] = [],
) -> EventPipeline {
  { name: name.to_owned(), batch, actions: actions.to_owned() }
}

///|
pub fn EventPipeline::add(
  self : EventPipeline,
  action : PipelineAction,
) -> EventPipeline {
  let actions = self.actions.copy()
  actions.push(action)
  { ..self, actions, }
}

///|
pub fn EventPipeline::len(self : EventPipeline) -> Int {
  self.actions.length()
}

///|
pub fn EventPipeline::is_empty(self : EventPipeline) -> Bool {
  self.actions.length() == 0
}

///|
pub fn EventPipeline::run(
  self : EventPipeline,
) -> Result[PipelineRun, EventRailError] {
  let stages : Array[PipelineStage] = []
  let mut passed = 0
  let mut failed = 0
  let mut warnings = 0
  for index, action in self.actions {
    let stage = match run_pipeline_action(index, self.batch, action) {
      Err(err) => return Err(err)
      Ok(value) => value
    }
    match stage.status {
      StagePassed => passed += 1
      StageFailed => failed += 1
      StageWarning => warnings += 1
    }
    stages.push(stage)
  }
  Ok({ name: self.name, stages, passed, failed, warnings })
}

///|
pub fn EventPipeline::describe(self : EventPipeline) -> String {
  let lines : Array[String] = []
  lines.push(
    "pipeline=\{escape_wire_text(self.name)} batch=\{escape_wire_text(self.batch.name)} actions=\{self.actions.length()}",
  )
  for index, action in self.actions {
    lines.push("idx=\{index};\{action.to_wire()}")
  }
  lines.join("\n")
}

///|
pub fn PipelineRun::ok(self : PipelineRun) -> Bool {
  self.failed == 0
}

///|
pub fn PipelineRun::summary(self : PipelineRun) -> String {
  "pipeline=\{escape_wire_text(self.name)} stages=\{self.stages.length()} passed=\{self.passed} warnings=\{self.warnings} failed=\{self.failed}"
}

///|
pub fn PipelineRun::manifest_lines(self : PipelineRun) -> Array[String] {
  let lines : Array[String] = []
  lines.push(self.summary())
  for stage in self.stages {
    lines.push(stage.to_wire())
    for line in stage.lines {
      lines.push("  \{line}")
    }
  }
  lines
}

///|
pub fn PipelineRun::manifest(self : PipelineRun) -> String {
  self.manifest_lines().join("\n")
}

///|
pub fn PipelineRun::failed_stages(self : PipelineRun) -> Array[PipelineStage] {
  self.stages.filter(stage => stage.status == StageFailed)
}

///|
pub fn PipelineRun::warning_stages(self : PipelineRun) -> Array[PipelineStage] {
  self.stages.filter(stage => stage.status == StageWarning)
}

///|
pub fn PipelineStage::to_wire(self : PipelineStage) -> String {
  "stage=\{self.index};status=\{self.status.to_wire()};action=\{self.action.to_wire()};summary=\{escape_wire_text(self.summary)}"
}

///|
pub fn PipelineStageStatus::to_wire(self : PipelineStageStatus) -> String {
  match self {
    StagePassed => "passed"
    StageFailed => "failed"
    StageWarning => "warning"
  }
}

///|
pub fn PipelineAction::to_wire(self : PipelineAction) -> String {
  match self {
    PipelineValidate(schema) => "validate:\{escape_wire_text(schema.name)}"
    PipelineRules(rules) => "rules:\{escape_wire_text(rules.name)}"
    PipelineRoute(_, mode) => "route:\{mode.to_wire()}"
    PipelineProject(spec) => "project:\{escape_wire_text(spec.name)}"
    PipelineSnapshot(_) => "snapshot"
  }
}

///|
pub fn DeliveryMode::to_wire(self : DeliveryMode) -> String {
  match self {
    Fanout => "fanout"
    FirstPerGroup => "first-per-group"
  }
}

///|
fn run_pipeline_action(
  index : Int,
  batch : EventBatch,
  action : PipelineAction,
) -> Result[PipelineStage, EventRailError] {
  match action {
    PipelineValidate(schema) =>
      Ok(run_validation_stage(index, batch, action, schema))
    PipelineRules(rules) => Ok(run_rules_stage(index, batch, action, rules))
    PipelineRoute(bus, mode) => run_route_stage(index, batch, action, bus, mode)
    PipelineProject(spec) => run_projection_stage(index, batch, action, spec)
    PipelineSnapshot(bus) => Ok(run_snapshot_stage(index, batch, action, bus))
  }
}

///|
fn run_validation_stage(
  index : Int,
  batch : EventBatch,
  action : PipelineAction,
  schema : EventSchema,
) -> PipelineStage {
  let report = batch.validate(schema)
  let status = if report.invalid == 0 { StagePassed } else { StageFailed }
  {
    index,
    action,
    status,
    summary: report.summary(),
    lines: report.issue_lines(),
  }
}

///|
fn run_rules_stage(
  index : Int,
  batch : EventBatch,
  action : PipelineAction,
  rules : RuleSet,
) -> PipelineStage {
  let report = batch.evaluate_rules(rules)
  let status = if report.rejected == 0 { StagePassed } else { StageWarning }
  {
    index,
    action,
    status,
    summary: report.summary(),
    lines: report.decision_lines(),
  }
}

///|
fn run_route_stage(
  index : Int,
  batch : EventBatch,
  action : PipelineAction,
  bus : Bus,
  mode : DeliveryMode,
) -> Result[PipelineStage, EventRailError] {
  match batch.route_preview(bus, mode~) {
    Err(err) => Err(err)
    Ok(report) => {
      let status = if report.unmatched_events == 0 {
        StagePassed
      } else {
        StageWarning
      }
      Ok({
        index,
        action,
        status,
        summary: report.summary(),
        lines: report.route_lines(),
      })
    }
  }
}

///|
fn run_projection_stage(
  index : Int,
  batch : EventBatch,
  action : PipelineAction,
  spec : ProjectionSpec,
) -> Result[PipelineStage, EventRailError] {
  match batch.project(spec) {
    Err(err) => Err(err)
    Ok(projection) => {
      let status = if projection.issues.length() == 0 {
        StagePassed
      } else {
        StageWarning
      }
      Ok({
        index,
        action,
        status,
        summary: projection.summary(),
        lines: projection.manifest_lines(),
      })
    }
  }
}

///|
fn run_snapshot_stage(
  index : Int,
  batch : EventBatch,
  action : PipelineAction,
  bus : Bus,
) -> PipelineStage {
  let snapshot = batch.snapshot(bus, name="pipeline-snapshot")
  {
    index,
    action,
    status: StagePassed,
    summary: snapshot.summary(),
    lines: snapshot.manifest_lines(),
  }
}