///|
/// Wire formats supported by the production telemetry exporter.
pub(all) enum ProductionExportFormat {
  ExportJsonLines
  ExportCsv
  ExportPrometheus
  ExportMarkdown
  ExportSummaryJson
}

///|
pub fn production_export_format_name(format : ProductionExportFormat) -> String {
  match format {
    ExportJsonLines => "jsonl"
    ExportCsv => "csv"
    ExportPrometheus => "prometheus"
    ExportMarkdown => "markdown"
    ExportSummaryJson => "summary-json"
  }
}

///|
/// Fields that can be selected in a telemetry export.
pub(all) enum ProductionExportField {
  ExportTimestamp
  ExportMetric
  ExportValue
  ExportBaseline
  ExportScore
  ExportHealth
  ExportState
  ExportSource
}

///|
pub fn production_export_field_name(field : ProductionExportField) -> String {
  match field {
    ExportTimestamp => "timestamp"
    ExportMetric => "metric"
    ExportValue => "value"
    ExportBaseline => "baseline"
    ExportScore => "score"
    ExportHealth => "health"
    ExportState => "state"
    ExportSource => "source"
  }
}

///|
/// Export settings shared by all serializers.
pub struct ProductionExportOptions {
  format : ProductionExportFormat
  delimiter : String
  include_header : Bool
  include_metadata : Bool
  pretty : Bool
  max_records : Int
  metric_prefix : String
  metric_namespace : String
  line_ending : String
}

///|
pub fn ProductionExportOptions::new(
  format? : ProductionExportFormat = ExportJsonLines,
  delimiter? : String = ",",
  include_header? : Bool = true,
  include_metadata? : Bool = false,
  pretty? : Bool = false,
  max_records? : Int = 10000,
  metric_prefix? : String = "",
  metric_namespace? : String = "moon_change_point",
  line_ending? : String = "\n",
) -> ProductionExportOptions {
  {
    format,
    delimiter: if delimiter.length() == 0 {
      ","
    } else {
      delimiter
    },
    include_header,
    include_metadata,
    pretty,
    max_records: if max_records < 1 {
      1
    } else {
      max_records
    },
    metric_prefix,
    metric_namespace,
    line_ending: if line_ending.length() == 0 {
      "\n"
    } else {
      line_ending
    },
  }
}

///|
pub fn ProductionExportOptions::format(
  self : ProductionExportOptions,
) -> ProductionExportFormat {
  self.format
}

///|
pub fn ProductionExportOptions::delimiter(
  self : ProductionExportOptions,
) -> String {
  self.delimiter
}

///|
pub fn ProductionExportOptions::include_header(
  self : ProductionExportOptions,
) -> Bool {
  self.include_header
}

///|
pub fn ProductionExportOptions::include_metadata(
  self : ProductionExportOptions,
) -> Bool {
  self.include_metadata
}

///|
pub fn ProductionExportOptions::pretty(self : ProductionExportOptions) -> Bool {
  self.pretty
}

///|
pub fn ProductionExportOptions::max_records(
  self : ProductionExportOptions,
) -> Int {
  self.max_records
}

///|
pub fn ProductionExportOptions::metric_prefix(
  self : ProductionExportOptions,
) -> String {
  self.metric_prefix
}

///|
pub fn ProductionExportOptions::metric_namespace(
  self : ProductionExportOptions,
) -> String {
  self.metric_namespace
}

///|
pub fn ProductionExportOptions::line_ending(
  self : ProductionExportOptions,
) -> String {
  self.line_ending
}

///|
pub fn ProductionExportOptions::with_format(
  self : ProductionExportOptions,
  format : ProductionExportFormat,
) -> ProductionExportOptions {
  {
    format,
    delimiter: self.delimiter,
    include_header: self.include_header,
    include_metadata: self.include_metadata,
    pretty: self.pretty,
    max_records: self.max_records,
    metric_prefix: self.metric_prefix,
    metric_namespace: self.metric_namespace,
    line_ending: self.line_ending,
  }
}

///|
/// One normalized telemetry row emitted by a monitor or replay.
pub struct ProductionExportRecord {
  timestamp : Int64
  metric : String
  value : Double
  baseline : Double
  score : Double
  health : Double
  state : String
  source : String
}

///|
pub fn ProductionExportRecord::new(
  timestamp : Int64,
  metric : String,
  value : Double,
  baseline? : Double = 0.0,
  score? : Double = 0.0,
  health? : Double = 1.0,
  state? : String = "unknown",
  source? : String = "runtime",
) -> ProductionExportRecord {
  {
    timestamp,
    metric,
    value,
    baseline,
    score,
    health: clamp_probability(health),
    state,
    source,
  }
}

///|
pub fn ProductionExportRecord::timestamp(
  self : ProductionExportRecord,
) -> Int64 {
  self.timestamp
}

///|
pub fn ProductionExportRecord::metric(self : ProductionExportRecord) -> String {
  self.metric
}

///|
pub fn ProductionExportRecord::value(self : ProductionExportRecord) -> Double {
  self.value
}

///|
pub fn ProductionExportRecord::baseline(
  self : ProductionExportRecord,
) -> Double {
  self.baseline
}

///|
pub fn ProductionExportRecord::score(self : ProductionExportRecord) -> Double {
  self.score
}

///|
pub fn ProductionExportRecord::health(self : ProductionExportRecord) -> Double {
  self.health
}

///|
pub fn ProductionExportRecord::state(self : ProductionExportRecord) -> String {
  self.state
}

///|
pub fn ProductionExportRecord::source(self : ProductionExportRecord) -> String {
  self.source
}

///|
pub fn ProductionExportRecord::is_finite(self : ProductionExportRecord) -> Bool {
  self.value == self.value &&
  self.value > -1.0e308 &&
  self.value < 1.0e308 &&
  self.baseline == self.baseline &&
  self.score == self.score
}

///|
pub fn ProductionExportRecord::with_state(
  self : ProductionExportRecord,
  state : String,
) -> ProductionExportRecord {
  {
    timestamp: self.timestamp,
    metric: self.metric,
    value: self.value,
    baseline: self.baseline,
    score: self.score,
    health: self.health,
    state,
    source: self.source,
  }
}

///|
pub fn ProductionExportRecord::field(
  self : ProductionExportRecord,
  field : ProductionExportField,
) -> String {
  match field {
    ExportTimestamp => self.timestamp.to_string()
    ExportMetric => self.metric
    ExportValue => self.value.to_string()
    ExportBaseline => self.baseline.to_string()
    ExportScore => self.score.to_string()
    ExportHealth => self.health.to_string()
    ExportState => self.state
    ExportSource => self.source
  }
}

///|
pub fn ProductionExportRecord::as_key(self : ProductionExportRecord) -> String {
  self.metric + ":" + self.timestamp.to_string()
}

///|
/// Bounded in-memory batch used to make exports deterministic.
pub struct ProductionExportBatch {
  mut records : Array[ProductionExportRecord]
  capacity : Int
  mut accepted : Int
  mut rejected : Int
  mut duplicate : Int
}

///|
pub fn ProductionExportBatch::new(
  capacity? : Int = 10000,
) -> ProductionExportBatch {
  {
    records: [],
    capacity: if capacity < 1 {
      1
    } else {
      capacity
    },
    accepted: 0,
    rejected: 0,
    duplicate: 0,
  }
}

///|
fn production_export_has_key(
  records : Array[ProductionExportRecord],
  key : String,
) -> Bool {
  for record in records {
    if record.as_key() == key {
      return true
    }
  }
  false
}

///|
pub fn ProductionExportBatch::add(
  self : ProductionExportBatch,
  record : ProductionExportRecord,
) -> Bool {
  if self.records.length() >= self.capacity || !record.is_finite() {
    self.rejected = self.rejected + 1
    false
  } else if production_export_has_key(self.records, record.as_key()) {
    self.duplicate = self.duplicate + 1
    false
  } else {
    self.records.push(record)
    self.accepted = self.accepted + 1
    true
  }
}

///|
pub fn ProductionExportBatch::add_many(
  self : ProductionExportBatch,
  records : Array[ProductionExportRecord],
) -> Int {
  let mut added = 0
  for record in records {
    if self.add(record) {
      added = added + 1
    }
  }
  added
}

///|
pub fn ProductionExportBatch::records(
  self : ProductionExportBatch,
) -> Array[ProductionExportRecord] {
  self.records[:].to_owned()
}

///|
pub fn ProductionExportBatch::capacity(self : ProductionExportBatch) -> Int {
  self.capacity
}

///|
pub fn ProductionExportBatch::count(self : ProductionExportBatch) -> Int {
  self.records.length()
}

///|
pub fn ProductionExportBatch::accepted(self : ProductionExportBatch) -> Int {
  self.accepted
}

///|
pub fn ProductionExportBatch::rejected(self : ProductionExportBatch) -> Int {
  self.rejected
}

///|
pub fn ProductionExportBatch::duplicate(self : ProductionExportBatch) -> Int {
  self.duplicate
}

///|
pub fn ProductionExportBatch::clear(self : ProductionExportBatch) -> Unit {
  self.records = []
  self.accepted = 0
  self.rejected = 0
  self.duplicate = 0
}

///|
/// Return records ordered by event time without mutating the input batch.
pub fn ProductionExportBatch::ordered(
  self : ProductionExportBatch,
) -> Array[ProductionExportRecord] {
  let ordered : Array[ProductionExportRecord] = []
  for record in self.records {
    let mut inserted = false
    for i = 0; i < ordered.length(); i = i + 1 {
      if record.timestamp() < ordered[i].timestamp() {
        ordered.insert(i, record)
        inserted = true
        break
      }
    }
    if !inserted {
      ordered.push(record)
    }
  }
  ordered
}

///|
fn production_export_json_escape(raw : String) -> String {
  let pieces : Array[String] = []
  for character in raw {
    match character.to_int() {
      34 => pieces.push("\\\"")
      92 => pieces.push("\\\\")
      10 => pieces.push("\\n")
      13 => pieces.push("\\r")
      9 => pieces.push("\\t")
      _ => pieces.push(character.to_string())
    }
  }
  pieces.join("")
}

///|
fn production_export_csv_escape(raw : String, delimiter : String) -> String {
  let mut needs_quotes = false
  if raw.contains(delimiter) || raw.contains("\"") || raw.contains("\n") {
    needs_quotes = true
  }
  let pieces : Array[String] = []
  for character in raw {
    if character.to_int() == 34 {
      pieces.push("\"\"")
    } else {
      pieces.push(character.to_string())
    }
  }
  if needs_quotes {
    "\"" + pieces.join("") + "\""
  } else {
    pieces.join("")
  }
}

///|
fn production_export_prometheus_name(
  options : ProductionExportOptions,
  metric : String,
) -> String {
  let pieces : Array[String] = []
  for
    character in options.metric_namespace() +
    "_" +
    options.metric_prefix() +
    metric {
    let code = character.to_int()
    if (code >= 48 && code <= 57) ||
      (code >= 65 && code <= 90) ||
      (code >= 97 && code <= 122) ||
      code == 95 {
      pieces.push(character.to_string())
    } else {
      pieces.push("_")
    }
  }
  pieces.join("")
}

///|
pub fn production_export_record_json(
  record : ProductionExportRecord,
  pretty? : Bool = false,
) -> String {
  let separator = if pretty { ", " } else { "," }
  "{\"timestamp\":" +
  record.timestamp().to_string() +
  separator +
  "\"metric\":\"" +
  production_export_json_escape(record.metric()) +
  "\"" +
  separator +
  "\"value\":" +
  record.value().to_string() +
  separator +
  "\"baseline\":" +
  record.baseline().to_string() +
  separator +
  "\"score\":" +
  record.score().to_string() +
  separator +
  "\"health\":" +
  record.health().to_string() +
  separator +
  "\"state\":\"" +
  production_export_json_escape(record.state()) +
  "\"" +
  separator +
  "\"source\":\"" +
  production_export_json_escape(record.source()) +
  "\"}"
}

///|
pub fn production_export_record_csv(
  record : ProductionExportRecord,
  delimiter? : String = ",",
) -> String {
  let safe_delimiter = if delimiter.length() == 0 { "," } else { delimiter }
  let fields : Array[String] = []
  fields.push(record.timestamp().to_string())
  fields.push(production_export_csv_escape(record.metric(), safe_delimiter))
  fields.push(record.value().to_string())
  fields.push(record.baseline().to_string())
  fields.push(record.score().to_string())
  fields.push(record.health().to_string())
  fields.push(production_export_csv_escape(record.state(), safe_delimiter))
  fields.push(production_export_csv_escape(record.source(), safe_delimiter))
  fields.join(safe_delimiter)
}

///|
pub fn production_export_record_prometheus(
  record : ProductionExportRecord,
  options : ProductionExportOptions,
) -> String {
  let name = production_export_prometheus_name(options, record.metric())
  name +
  "{state=\"" +
  production_export_json_escape(record.state()) +
  "\",source=\"" +
  production_export_json_escape(record.source()) +
  "\"} " +
  record.value().to_string() +
  " " +
  record.timestamp().to_string()
}

///|
pub fn production_export_record_markdown(
  record : ProductionExportRecord,
) -> String {
  "| " +
  record.timestamp().to_string() +
  " | " +
  record.metric() +
  " | " +
  record.value().to_string() +
  " | " +
  record.baseline().to_string() +
  " | " +
  record.score().to_string() +
  " | " +
  record.health().to_string() +
  " | " +
  record.state() +
  " | " +
  record.source() +
  " |"
}

///|

///|
/// Exporter counters are kept separately from the batch so operators can
/// monitor serialization failures across multiple flushes.
pub struct ProductionExportStats {
  mut batches : Int
  mut records : Int
  mut bytes : Int
  failures : Int
  mut truncated : Int
}

///|
pub fn ProductionExportStats::new() -> ProductionExportStats {
  { batches: 0, records: 0, bytes: 0, failures: 0, truncated: 0 }
}

///|
pub fn ProductionExportStats::batches(self : ProductionExportStats) -> Int {
  self.batches
}

///|
pub fn ProductionExportStats::records(self : ProductionExportStats) -> Int {
  self.records
}

///|
pub fn ProductionExportStats::bytes(self : ProductionExportStats) -> Int {
  self.bytes
}

///|
pub fn ProductionExportStats::failures(self : ProductionExportStats) -> Int {
  self.failures
}

///|
pub fn ProductionExportStats::truncated(self : ProductionExportStats) -> Int {
  self.truncated
}

///|
pub fn ProductionExportStats::success_rate(
  self : ProductionExportStats,
) -> Double {
  if self.batches == 0 {
    1.0
  } else {
    (self.batches - self.failures).to_double() / self.batches.to_double()
  }
}

///|
pub fn ProductionExportStats::summary(self : ProductionExportStats) -> String {
  "batches=" +
  self.batches.to_string() +
  " records=" +
  self.records.to_string() +
  " bytes=" +
  self.bytes.to_string() +
  " failures=" +
  self.failures.to_string() +
  " truncated=" +
  self.truncated.to_string()
}

///|
pub struct ProductionTelemetryExporter {
  options : ProductionExportOptions
  mut stats : ProductionExportStats
}

///|
pub fn ProductionTelemetryExporter::new(
  options? : ProductionExportOptions = ProductionExportOptions::new(),
) -> ProductionTelemetryExporter {
  { options, stats: ProductionExportStats::new() }
}

///|
pub fn ProductionTelemetryExporter::options(
  self : ProductionTelemetryExporter,
) -> ProductionExportOptions {
  self.options
}

///|
pub fn ProductionTelemetryExporter::stats(
  self : ProductionTelemetryExporter,
) -> ProductionExportStats {
  self.stats
}

///|
fn ProductionTelemetryExporter::take_records(
  self : ProductionTelemetryExporter,
  batch : ProductionExportBatch,
) -> Array[ProductionExportRecord] {
  let records = batch.ordered()
  let limit = if records.length() > self.options.max_records() {
    self.stats.truncated = self.stats.truncated +
      records.length() -
      self.options.max_records()
    self.options.max_records()
  } else {
    records.length()
  }
  records[:limit].to_owned()
}

///|
pub fn ProductionTelemetryExporter::export_batch(
  self : ProductionTelemetryExporter,
  batch : ProductionExportBatch,
) -> String {
  self.stats.batches = self.stats.batches + 1
  let records = self.take_records(batch)
  self.stats.records = self.stats.records + records.length()
  let result = match self.options.format() {
    ExportJsonLines => self.export_json_lines(records)
    ExportCsv => self.export_csv(records)
    ExportPrometheus => self.export_prometheus(records)
    ExportMarkdown => self.export_markdown(records)
    ExportSummaryJson => self.export_summary(records)
  }
  self.stats.bytes = self.stats.bytes + result.length()
  result
}

///|
pub fn ProductionTelemetryExporter::export_json_lines(
  self : ProductionTelemetryExporter,
  records : Array[ProductionExportRecord],
) -> String {
  let lines : Array[String] = []
  if self.options.include_metadata() {
    lines.push(
      "{\"type\":\"metadata\",\"format\":\"jsonl\",\"namespace\":\"" +
      production_export_json_escape(self.options.metric_namespace()) +
      "\"}",
    )
  }
  for record in records {
    lines.push(
      production_export_record_json(record, pretty=self.options.pretty()),
    )
  }
  lines.join(self.options.line_ending())
}

///|
pub fn ProductionTelemetryExporter::export_csv(
  self : ProductionTelemetryExporter,
  records : Array[ProductionExportRecord],
) -> String {
  let lines : Array[String] = []
  if self.options.include_header() {
    let headers : Array[String] = []
    for
      field in [
        ExportTimestamp,
        ExportMetric,
        ExportValue,
        ExportBaseline,
        ExportScore,
        ExportHealth,
        ExportState,
        ExportSource,
      ] {
      headers.push(production_export_field_name(field))
    }
    lines.push(headers.join(self.options.delimiter()))
  }
  for record in records {
    lines.push(
      production_export_record_csv(record, delimiter=self.options.delimiter()),
    )
  }
  lines.join(self.options.line_ending())
}

///|
pub fn ProductionTelemetryExporter::export_prometheus(
  self : ProductionTelemetryExporter,
  records : Array[ProductionExportRecord],
) -> String {
  let lines : Array[String] = []
  let names : Array[String] = []
  for record in records {
    let name = production_export_prometheus_name(self.options, record.metric())
    if !names.contains(name) {
      names.push(name)
      lines.push("# TYPE " + name + " gauge")
    }
    lines.push(production_export_record_prometheus(record, self.options))
  }
  lines.join(self.options.line_ending())
}

///|
pub fn ProductionTelemetryExporter::export_markdown(
  self : ProductionTelemetryExporter,
  records : Array[ProductionExportRecord],
) -> String {
  let lines : Array[String] = []
  lines.push("# Telemetry export")
  lines.push("")
  lines.push(
    "| timestamp | metric | value | baseline | score | health | state | source |",
  )
  lines.push("| ---: | --- | ---: | ---: | ---: | ---: | --- | --- |")
  for record in records {
    lines.push(production_export_record_markdown(record))
  }
  lines.join(self.options.line_ending())
}

///|
pub fn ProductionTelemetryExporter::export_summary(
  self : ProductionTelemetryExporter,
  records : Array[ProductionExportRecord],
) -> String {
  let _ = self.options.pretty()
  let mut sum = 0.0
  let mut min = 0.0
  let mut max = 0.0
  let mut health = 0.0
  if records.length() > 0 {
    min = records[0].value()
    max = records[0].value()
  }
  for record in records {
    sum = sum + record.value()
    health = health + record.health()
    if record.value() < min {
      min = record.value()
    }
    if record.value() > max {
      max = record.value()
    }
  }
  let mean = if records.length() == 0 {
    0.0
  } else {
    sum / records.length().to_double()
  }
  let average_health = if records.length() == 0 {
    1.0
  } else {
    health / records.length().to_double()
  }
  "{\"count\":" +
  records.length().to_string() +
  ",\"mean\":" +
  mean.to_string() +
  ",\"minimum\":" +
  min.to_string() +
  ",\"maximum\":" +
  max.to_string() +
  ",\"health\":" +
  average_health.to_string() +
  "}"
}

///|
pub fn ProductionTelemetryExporter::reset_stats(
  self : ProductionTelemetryExporter,
) -> Unit {
  self.stats = ProductionExportStats::new()
}

///|
/// Produce a manifest describing the selected export schema.
pub fn production_export_manifest(options : ProductionExportOptions) -> String {
  "format=" +
  production_export_format_name(options.format()) +
  " namespace=" +
  options.metric_namespace() +
  " delimiter=" +
  options.delimiter() +
  " max_records=" +
  options.max_records().to_string() +
  " fields=timestamp|metric|value|baseline|score|health|state|source"
}

///|
pub fn production_export_checksum(payload : String) -> String {
  let mut checksum = 17
  for character in payload {
    checksum = (checksum * 31 + character.to_int()) % 2147483647
  }
  checksum.to_string()
}

///|
pub fn production_export_validate_metric_name(metric : String) -> Bool {
  if metric.length() == 0 {
    false
  } else {
    let mut valid = true
    for character in metric {
      let code = character.to_int()
      if !((code >= 48 && code <= 57) ||
        (code >= 65 && code <= 90) ||
        (code >= 97 && code <= 122) ||
        code == 45 ||
        code == 95 ||
        code == 46) {
        valid = false
      }
    }
    valid
  }
}

///|
pub fn production_export_normalize_metric_name(metric : String) -> String {
  let pieces : Array[String] = []
  for character in metric {
    let code = character.to_int()
    if (code >= 48 && code <= 57) ||
      (code >= 65 && code <= 90) ||
      (code >= 97 && code <= 122) ||
      code == 95 {
      pieces.push(character.to_string())
    } else {
      pieces.push("_")
    }
  }
  let normalized = pieces.join("")
  if normalized.length() == 0 {
    "metric"
  } else {
    normalized
  }
}

///|
pub fn production_export_merge_batches(
  left : ProductionExportBatch,
  right : ProductionExportBatch,
) -> ProductionExportBatch {
  let merged = ProductionExportBatch::new(
    capacity=left.capacity() + right.capacity(),
  )
  let _ = merged.add_many(left.records())
  let _ = merged.add_many(right.records())
  merged
}

///|
pub fn production_export_group_by_metric(
  records : Array[ProductionExportRecord],
) -> Array[String] {
  let names : Array[String] = []
  for record in records {
    if !names.contains(record.metric()) {
      names.push(record.metric())
    }
  }
  names
}

///|
pub fn production_export_metric_count(
  records : Array[ProductionExportRecord],
  metric : String,
) -> Int {
  let mut count = 0
  for record in records {
    if record.metric() == metric {
      count = count + 1
    }
  }
  count
}

///|
pub fn production_export_metric_mean(
  records : Array[ProductionExportRecord],
  metric : String,
) -> Double {
  let mut sum = 0.0
  let mut count = 0
  for record in records {
    if record.metric() == metric {
      sum = sum + record.value()
      count = count + 1
    }
  }
  if count == 0 {
    0.0
  } else {
    sum / count.to_double()
  }
}