///|
pub(all) struct ProjectionSpec {
  name : String
  topic : TopicPattern
  key_path : String
  include_paths : Array[String]
  require_key : Bool
} derive(Eq, @debug.Debug)

///|
pub(all) struct ProjectionIssue {
  event_id : String
  topic : String
  code : String
  message : String
} derive(Eq, @debug.Debug)

///|
pub(all) struct ProjectionRow {
  key : String
  topic : String
  count : Int
  first_timestamp_ms : Int
  last_timestamp_ms : Int
  last_event_id : String
  fields : Array[Pair]
} derive(Eq, @debug.Debug)

///|
pub(all) struct EventProjection {
  spec : ProjectionSpec
  rows : Array[ProjectionRow]
  skipped : Int
  issues : Array[ProjectionIssue]
} derive(Eq, @debug.Debug)

///|
pub fn projection_spec(
  name : StringView,
  key_path : StringView,
  topic? : StringView = "**",
  include_paths? : ArrayView[String] = [],
  require_key? : Bool = true,
) -> Result[ProjectionSpec, EventRailError] {
  match topic_pattern(topic) {
    Err(err) => Err(err)
    Ok(parsed) =>
      Ok({
        name: name.to_owned(),
        topic: parsed,
        key_path: key_path.to_owned(),
        include_paths: include_paths.to_owned(),
        require_key,
      })
  }
}

///|
pub fn event_projection(spec : ProjectionSpec) -> EventProjection {
  { spec, rows: [], skipped: 0, issues: [] }
}

///|
pub fn ProjectionSpec::empty(self : ProjectionSpec) -> EventProjection {
  event_projection(self)
}

///|
pub fn ProjectionSpec::project_events(
  self : ProjectionSpec,
  events : ArrayView[Envelope],
) -> Result[EventProjection, EventRailError] {
  let mut projection = event_projection(self)
  for event in events {
    match projection.apply(event) {
      Err(err) => return Err(err)
      Ok(next) => projection = next
    }
  }
  Ok(projection)
}

///|
pub fn EventTape::project(
  self : EventTape,
  spec : ProjectionSpec,
) -> Result[EventProjection, EventRailError] {
  let mut projection = event_projection(spec)
  for entry in self.entries {
    match projection.apply(entry.event) {
      Err(err) => return Err(err)
      Ok(next) => projection = next
    }
  }
  Ok(projection)
}

///|
pub fn EventBatch::project(
  self : EventBatch,
  spec : ProjectionSpec,
) -> Result[EventProjection, EventRailError] {
  spec.project_events(self.events)
}

///|
pub fn EventProjection::apply(
  self : EventProjection,
  event : Envelope,
) -> Result[EventProjection, EventRailError] {
  match self.spec.topic.matches_topic(event.topic) {
    Err(err) => Err(err)
    Ok(false) => Ok({ ..self, skipped: self.skipped + 1 })
    Ok(true) =>
      match projection_key(event, self.spec.key_path, self.spec.require_key) {
        None => {
          let issues = self.issues.copy()
          issues.push({
            event_id: event.id,
            topic: event.topic,
            code: "key.missing",
            message: "projection key \{self.spec.key_path} is missing",
          })
          Ok({ ..self, skipped: self.skipped + 1, issues })
        }
        Some(key) => {
          let rows = self.rows.copy()
          upsert_projection_row(rows, key, event, self.spec.include_paths)
          Ok({ ..self, rows, })
        }
      }
  }
}

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

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

///|
pub fn EventProjection::find(
  self : EventProjection,
  key : StringView,
) -> ProjectionRow? {
  let wanted = key.to_owned()
  for row in self.rows {
    if row.key == wanted {
      return Some(row)
    }
  }
  None
}

///|
pub fn EventProjection::summary(self : EventProjection) -> String {
  "projection=\{escape_wire_text(self.spec.name)} rows=\{self.rows.length()} skipped=\{self.skipped} issues=\{self.issues.length()}"
}

///|
pub fn EventProjection::manifest_lines(self : EventProjection) -> Array[String] {
  self.rows.map(row => row.to_wire())
}

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

///|
pub fn EventProjection::issue_lines(self : EventProjection) -> Array[String] {
  self.issues.map(issue => issue.to_wire())
}

///|
pub fn ProjectionRow::field(self : ProjectionRow, key : StringView) -> String? {
  let wanted = key.to_owned()
  for item in self.fields {
    if item.key == wanted {
      return Some(item.value)
    }
  }
  None
}

///|
pub fn ProjectionRow::to_wire(self : ProjectionRow) -> String {
  "key=\{escape_wire_text(self.key)};topic=\{escape_wire_text(self.topic)};count=\{self.count};first=\{self.first_timestamp_ms};last=\{self.last_timestamp_ms};event=\{escape_wire_text(self.last_event_id)};fields=\{pairs_to_wire(self.fields)}"
}

///|
pub fn ProjectionIssue::to_wire(self : ProjectionIssue) -> String {
  "event=\{escape_wire_text(self.event_id)};topic=\{escape_wire_text(self.topic)};code=\{escape_wire_text(self.code)};message=\{escape_wire_text(self.message)}"
}

///|
fn upsert_projection_row(
  rows : Array[ProjectionRow],
  key : String,
  event : Envelope,
  include_paths : Array[String],
) -> Unit {
  let fields = projection_fields(event, include_paths)
  for index, row in rows {
    if row.key == key {
      rows[index] = {
        key: row.key,
        topic: event.topic,
        count: row.count + 1,
        first_timestamp_ms: row.first_timestamp_ms,
        last_timestamp_ms: event.timestamp_ms,
        last_event_id: event.id,
        fields,
      }
      return
    }
  }
  rows.push({
    key,
    topic: event.topic,
    count: 1,
    first_timestamp_ms: event.timestamp_ms,
    last_timestamp_ms: event.timestamp_ms,
    last_event_id: event.id,
    fields,
  })
}

///|
fn projection_fields(
  event : Envelope,
  include_paths : Array[String],
) -> Array[Pair] {
  let fields : Array[Pair] = []
  for path in include_paths {
    match projection_value(event, path) {
      Some(value) => fields.push(pair(path, value))
      None => ()
    }
  }
  fields
}

///|
fn projection_key(
  event : Envelope,
  path : String,
  require_key : Bool,
) -> String? {
  match projection_value(event, path) {
    Some(value) => Some(value)
    None => if require_key { None } else { Some(event.id) }
  }
}

///|
fn projection_value(event : Envelope, path : String) -> String? {
  if path == "$id" {
    return Some(event.id)
  }
  if path == "$topic" {
    return Some(event.topic)
  }
  if path == "$timestamp_ms" {
    return Some(event.timestamp_ms.to_string())
  }
  if path == "$attempt" {
    return Some(event.attempt.to_string())
  }
  match event.payload_path(path) {
    Some(value) => Some(projection_value_to_text(value))
    None => None
  }
}

///|
fn projection_value_to_text(value : EventValue) -> String {
  match value {
    VNull => "null"
    VBool(actual) => if actual { "true" } else { "false" }
    VInt(actual) => actual.to_string()
    VText(actual) => actual
    VList(_) | VRecord(_) => value.to_wire()
  }
}