///|
/// DeliveryStatus describes how a subscriber handled a message.
pub(all) enum DeliveryStatus {
  Ack
  Rejected(String)
  Skipped(String)
} derive(Eq, @debug.Debug)

///|
/// PublishMode controls whether dispatch continues after a handler failure.
pub(all) enum PublishMode {
  ContinueOnError
  StopOnError
} derive(Eq, @debug.Debug)

///|
/// TopicPattern is the route expression used by subscribers.
pub(all) struct TopicPattern {
  pattern : String
} derive(Eq, @debug.Debug)

///|
/// Message carries a topic, payload, and lightweight metadata.
pub(all) struct Message {
  id : String
  topic : String
  payload : String
  source : String
  trace_id : String
  tags : Array[String]
  headers : Map[String, String]
} derive(Eq, @debug.Debug)

///|
/// Subscriber records routing information and the callback to invoke.
pub(all) struct Subscriber {
  id : String
  pattern : TopicPattern
  priority : Int
  once : Bool
  active : Bool
  handler : (Message) -> DeliveryStatus
} derive(@debug.Debug)

///|
/// DeliveryReport is one subscriber's dispatch result.
pub(all) struct DeliveryReport {
  subscriber_id : String
  topic : String
  status : DeliveryStatus
} derive(Eq, @debug.Debug)

///|
/// PublishReport summarizes one publish operation.
pub(all) struct PublishReport {
  message_id : String
  topic : String
  matched : Int
  delivered : Int
  rejected : Int
  skipped : Int
  reports : Array[DeliveryReport]
} derive(Eq, @debug.Debug)

///|
/// DeadLetter keeps messages that were rejected by at least one subscriber.
pub(all) struct DeadLetter {
  message : Message
  subscriber_id : String
  reason : String
} derive(Eq, @debug.Debug)

///|
/// BusConfig controls dispatch behavior.
pub(all) struct BusConfig {
  mode : PublishMode
  keep_history : Bool
  keep_dead_letters : Bool
} derive(Eq, @debug.Debug)

///|
/// EventBus is an in-memory event hub for tests, tools, and embedded apps.
pub(all) struct EventBus {
  config : BusConfig
  subscribers : Array[Subscriber]
  history : Array[Message]
  dead_letters : Array[DeadLetter]
} derive(@debug.Debug)

///|
pub fn TopicPattern::TopicPattern(pattern : StringView) -> TopicPattern {
  { pattern: pattern.to_owned() }
}

///|
pub fn TopicPattern::matches(self : TopicPattern, topic : StringView) -> Bool {
  topic_matches(self.pattern, topic)
}

///|
pub fn Message::Message(
  id : StringView,
  topic : StringView,
  payload? : StringView = "",
  source? : StringView = "",
  trace_id? : StringView = "",
  tags? : ArrayView[String] = [],
  headers? : Map[String, String] = Map([]),
) -> Message {
  {
    id: id.to_owned(),
    topic: topic.to_owned(),
    payload: payload.to_owned(),
    source: source.to_owned(),
    trace_id: trace_id.to_owned(),
    tags: tags.to_owned(),
    headers,
  }
}

///|
pub fn BusConfig::BusConfig(
  mode? : PublishMode = ContinueOnError,
  keep_history? : Bool = true,
  keep_dead_letters? : Bool = true,
) -> BusConfig {
  { mode, keep_history, keep_dead_letters }
}

///|
pub fn EventBus::EventBus(config? : BusConfig = BusConfig()) -> EventBus {
  { config, subscribers: [], history: [], dead_letters: [] }
}

///|
pub fn new_event_bus(config? : BusConfig = BusConfig()) -> EventBus {
  EventBus(config~)
}

///|
pub fn Subscriber::Subscriber(
  id : StringView,
  pattern : StringView,
  handler : (Message) -> DeliveryStatus,
  priority? : Int = 0,
  once? : Bool = false,
) -> Subscriber {
  {
    id: id.to_owned(),
    pattern: TopicPattern(pattern),
    priority,
    once,
    active: true,
    handler,
  }
}

///|
pub fn EventBus::subscribe(
  self : EventBus,
  id : StringView,
  pattern : StringView,
  handler : (Message) -> DeliveryStatus,
  priority? : Int = 0,
  once? : Bool = false,
) -> Unit {
  self.subscribers.push(Subscriber(id, pattern, handler, priority~, once~))
  self.sort_subscribers()
}

///|
pub fn EventBus::unsubscribe(self : EventBus, id : StringView) -> Bool {
  let mut changed = false
  for i = 0; i < self.subscribers.length(); i = i + 1 {
    if self.subscribers[i].id == id.to_owned() {
      self.subscribers[i] = { ..self.subscribers[i], active: false }
      changed = true
    }
  }
  changed
}

///|
pub fn EventBus::subscriber_count(self : EventBus) -> Int {
  let mut count = 0
  for sub in self.subscribers {
    if sub.active {
      count = count + 1
    }
  }
  count
}

///|
pub fn EventBus::publish(
  self : EventBus,
  topic : StringView,
  payload? : StringView = "",
  source? : StringView = "",
  trace_id? : StringView = "",
  tags? : ArrayView[String] = [],
  headers? : Map[String, String] = Map([]),
) -> PublishReport {
  let id = "msg-\{self.history.length() + 1}"
  let message = Message(
    id,
    topic,
    payload~,
    source~,
    trace_id~,
    tags~,
    headers~,
  )
  self.dispatch(message)
}

///|
pub fn EventBus::dispatch(self : EventBus, message : Message) -> PublishReport {
  if self.config.keep_history {
    self.history.push(message)
  }
  let reports = Array::new()
  let mut matched = 0
  let mut delivered = 0
  let mut rejected = 0
  let mut skipped = 0
  let mut should_stop = false
  for i = 0; i < self.subscribers.length(); i = i + 1 {
    let sub = self.subscribers[i]
    if sub.active && sub.pattern.matches(message.topic) && !should_stop {
      matched = matched + 1
      let status = (sub.handler)(message)
      match status {
        Ack => delivered = delivered + 1
        Rejected(reason) => {
          rejected = rejected + 1
          if self.config.keep_dead_letters {
            self.dead_letters.push({ message, subscriber_id: sub.id, reason })
          }
          if self.config.mode is StopOnError {
            should_stop = true
          }
        }
        Skipped(_) => skipped = skipped + 1
      }
      reports.push({ subscriber_id: sub.id, topic: message.topic, status })
      if sub.once {
        self.subscribers[i] = { ..sub, active: false }
      }
    }
  }
  {
    message_id: message.id,
    topic: message.topic,
    matched,
    delivered,
    rejected,
    skipped,
    reports,
  }
}

///|
pub fn EventBus::clear_history(self : EventBus) -> Unit {
  self.history.clear()
}

///|
pub fn EventBus::clear_dead_letters(self : EventBus) -> Unit {
  self.dead_letters.clear()
}

///|
pub fn EventBus::history_count(self : EventBus) -> Int {
  self.history.length()
}

///|
pub fn EventBus::dead_letter_count(self : EventBus) -> Int {
  self.dead_letters.length()
}

///|
pub fn EventBus::sort_subscribers(self : EventBus) -> Unit {
  for i = 1; i < self.subscribers.length(); i = i + 1 {
    let item = self.subscribers[i]
    let mut j = i - 1
    while j >= 0 && self.subscribers[j].priority < item.priority {
      self.subscribers[j + 1] = self.subscribers[j]
      j = j - 1
    }
    self.subscribers[j + 1] = item
  }
}

///|
pub fn topic_matches(pattern : StringView, topic : StringView) -> Bool {
  if pattern == "**" || pattern == "*" {
    true
  } else if pattern == topic {
    true
  } else {
    let p = split_topic(pattern)
    let t = split_topic(topic)
    match_segments(p, t, 0, 0)
  }
}

///|
fn split_topic(value : StringView) -> Array[String] {
  let parts = Array::new()
  let mut current = StringBuilder()
  for ch in value {
    if ch == '.' {
      parts.push(current.to_string())
      current = StringBuilder()
    } else {
      current.write_char(ch)
    }
  }
  parts.push(current.to_string())
  parts
}

///|
fn match_segments(
  pattern : Array[String],
  topic : Array[String],
  pi : Int,
  ti : Int,
) -> Bool {
  if pi == pattern.length() {
    ti == topic.length()
  } else if pattern[pi] == "**" {
    if pi + 1 == pattern.length() {
      true
    } else {
      let mut cursor = ti
      let mut ok = false
      while cursor <= topic.length() && !ok {
        ok = match_segments(pattern, topic, pi + 1, cursor)
        cursor = cursor + 1
      }
      ok
    }
  } else if ti == topic.length() {
    false
  } else if pattern[pi] == "*" || pattern[pi] == topic[ti] {
    match_segments(pattern, topic, pi + 1, ti + 1)
  } else {
    false
  }
}

///|
pub fn report_to_json(report : PublishReport) -> String {
  let buf = StringBuilder()
  buf.write_string("{")
  buf.write_string("\"message_id\":\"\{escape_json(report.message_id)}\",")
  buf.write_string("\"topic\":\"\{escape_json(report.topic)}\",")
  buf.write_string("\"matched\":\{report.matched},")
  buf.write_string("\"delivered\":\{report.delivered},")
  buf.write_string("\"rejected\":\{report.rejected},")
  buf.write_string("\"skipped\":\{report.skipped},")
  buf.write_string("\"reports\":[")
  for i = 0; i < report.reports.length(); i = i + 1 {
    if i > 0 {
      buf.write_string(",")
    }
    let item = report.reports[i]
    buf.write_string("{")
    buf.write_string(
      "\"subscriber_id\":\"\{escape_json(item.subscriber_id)}\",",
    )
    buf.write_string("\"topic\":\"\{escape_json(item.topic)}\",")
    buf.write_string("\"status\":\"\{status_name(item.status)}\"")
    buf.write_string("}")
  }
  buf.write_string("]}")
  buf.to_string()
}

///|
pub fn status_name(status : DeliveryStatus) -> String {
  match status {
    Ack => "ack"
    Rejected(_) => "rejected"
    Skipped(_) => "skipped"
  }
}

///|
pub fn escape_json(input : StringView) -> String {
  let buf = StringBuilder()
  for ch in input {
    if ch == '"' {
      buf.write_string("\\\"")
    } else if ch == '\\' {
      buf.write_string("\\\\")
    } else if ch == '\n' {
      buf.write_string("\\n")
    } else if ch == '\r' {
      buf.write_string("\\r")
    } else if ch == '\t' {
      buf.write_string("\\t")
    } else {
      buf.write_char(ch)
    }
  }
  buf.to_string()
}