///|
/// Durable batch recovery primitives for long-running de-identification jobs.
/// The journal is intentionally serializable so an application can persist it
/// in a file, object store, or database without coupling the core package to a
/// particular storage driver.
pub(all) enum BatchJournalState {
  BatchJournalQueued
  BatchJournalRunning
  BatchJournalSucceeded
  BatchJournalFailed
  BatchJournalSkipped
  BatchJournalCancelled
} derive(Debug, Eq)

///|
pub(all) enum BatchRecoveryMode {
  BatchRecoveryResume
  BatchRecoveryReplayFailed
  BatchRecoveryRetryTransient
  BatchRecoveryAuditOnly
} derive(Debug, Eq)

///|
pub(all) struct RetrySchedule {
  max_attempts : Int
  base_delay_seconds : Int
  max_delay_seconds : Int
  jitter_percent : Int
  retryable_errors : Array[String]
} derive(Debug, Eq)

///|
pub(all) struct BatchJournalEntry {
  entry_id : String
  batch_id : String
  item_id : String
  sequence : Int
  state : BatchJournalState
  attempt : Int
  input_checksum : String
  output_checksum : String
  error_code : String
  error_message : String
  started_at : String
  finished_at : String
} derive(Debug, Eq)

///|
pub(all) struct BatchJournal {
  journal_id : String
  created_at : String
  mut entries : Array[BatchJournalEntry]
  mut checksum : String
} derive(Debug)

///|
pub(all) struct RecoveryCandidate {
  entry : BatchJournalEntry
  mode : BatchRecoveryMode
  next_attempt : Int
  delay_seconds : Int
  reason : String
} derive(Debug)

///|
pub(all) struct RecoveryPlan {
  batch_id : String
  mode : BatchRecoveryMode
  candidates : Array[RecoveryCandidate]
  completed_items : Array[String]
  blocked_items : Array[String]
  checksum : String
} derive(Debug)

///|
pub fn batch_journal_state_name(state : BatchJournalState) -> String {
  match state {
    BatchJournalQueued => "queued"
    BatchJournalRunning => "running"
    BatchJournalSucceeded => "succeeded"
    BatchJournalFailed => "failed"
    BatchJournalSkipped => "skipped"
    BatchJournalCancelled => "cancelled"
  }
}

///|
pub fn batch_recovery_mode_name(mode : BatchRecoveryMode) -> String {
  match mode {
    BatchRecoveryResume => "resume"
    BatchRecoveryReplayFailed => "replay-failed"
    BatchRecoveryRetryTransient => "retry-transient"
    BatchRecoveryAuditOnly => "audit-only"
  }
}

///|
pub fn retry_schedule_default() -> RetrySchedule {
  {
    max_attempts: 4,
    base_delay_seconds: 5,
    max_delay_seconds: 300,
    jitter_percent: 10,
    retryable_errors: ["TIMEOUT", "RATE_LIMIT", "TEMPORARY_IO", "UPSTREAM_BUSY"],
  }
}

///|
pub fn retry_schedule(
  max_attempts : Int,
  base_delay_seconds : Int,
  max_delay_seconds : Int,
) -> RetrySchedule {
  {
    max_attempts: if max_attempts < 1 {
      1
    } else {
      max_attempts
    },
    base_delay_seconds: if base_delay_seconds < 0 {
      0
    } else {
      base_delay_seconds
    },
    max_delay_seconds: if max_delay_seconds < base_delay_seconds {
      base_delay_seconds
    } else {
      max_delay_seconds
    },
    jitter_percent: 0,
    retryable_errors: [],
  }
}

///|
pub fn RetrySchedule::with_errors(
  schedule : RetrySchedule,
  errors : Array[String],
) -> RetrySchedule {
  { ..schedule, retryable_errors: errors }
}

///|
pub fn RetrySchedule::with_jitter(
  schedule : RetrySchedule,
  jitter_percent : Int,
) -> RetrySchedule {
  {
    ..schedule,
    jitter_percent: if jitter_percent < 0 {
      0
    } else if jitter_percent > 100 {
      100
    } else {
      jitter_percent
    },
  }
}

///|
pub fn RetrySchedule::is_retryable(
  self : RetrySchedule,
  error_code : String,
) -> Bool {
  self.retryable_errors.contains(error_code)
}

///|
pub fn RetrySchedule::delay_for(self : RetrySchedule, attempt : Int) -> Int {
  let safe_attempt = if attempt < 1 { 1 } else { attempt }
  let mut delay = self.base_delay_seconds
  for _ in 1.. self.max_delay_seconds {
        delay = self.max_delay_seconds
      }
    }
  }
  delay
}

///|
pub fn RetrySchedule::can_retry(
  self : RetrySchedule,
  attempt : Int,
  error_code : String,
) -> Bool {
  attempt < self.max_attempts && self.is_retryable(error_code)
}

///|
pub fn batch_journal(journal_id : String, created_at : String) -> BatchJournal {
  {
    journal_id,
    created_at,
    entries: [],
    checksum: stable_hash(journal_id + created_at),
  }
}

///|
pub fn batch_entry(
  batch_id : String,
  item_id : String,
  sequence : Int,
  input_checksum : String,
) -> BatchJournalEntry {
  {
    entry_id: stable_hash(batch_id + ":" + item_id + ":" + sequence.to_string()),
    batch_id,
    item_id,
    sequence,
    state: BatchJournalQueued,
    attempt: 0,
    input_checksum,
    output_checksum: "",
    error_code: "",
    error_message: "",
    started_at: "",
    finished_at: "",
  }
}

///|
pub fn BatchJournal::append(
  self : BatchJournal,
  entry : BatchJournalEntry,
) -> String {
  self.entries.push(entry)
  self.checksum = stable_hash(
    self.entries.map(batch_journal_entry_key).join("\n"),
  )
  self.checksum
}

///|
fn batch_journal_entry_key(entry : BatchJournalEntry) -> String {
  entry.entry_id +
  "|" +
  entry.item_id +
  "|" +
  batch_journal_state_name(entry.state) +
  "|" +
  entry.attempt.to_string() +
  "|" +
  entry.output_checksum
}

///|
pub fn BatchJournal::find(
  self : BatchJournal,
  item_id : String,
) -> BatchJournalEntry? {
  let mut result : BatchJournalEntry? = None
  for entry in self.entries {
    if entry.item_id == item_id {
      result = Some(entry)
    }
  }
  result
}

///|
pub fn BatchJournal::latest_for(
  self : BatchJournal,
  batch_id : String,
) -> Array[BatchJournalEntry] {
  self.entries.filter(fn(entry) { entry.batch_id == batch_id })
}

///|
pub fn BatchJournal::count_by_state(self : BatchJournal) -> Map[String, Int] {
  let counts : Map[String, Int] = Map([])
  for entry in self.entries {
    let name = batch_journal_state_name(entry.state)
    counts[name] = counts.get_or_default(name, 0) + 1
  }
  counts
}

///|
pub fn BatchJournal::is_consistent(self : BatchJournal) -> Bool {
  self.entries.all(fn(entry) {
    entry.entry_id.length() > 0 &&
    entry.batch_id.length() > 0 &&
    entry.item_id.length() > 0 &&
    entry.sequence >= 0 &&
    entry.attempt >= 0 &&
    entry.input_checksum.length() > 0 &&
    (entry.state != BatchJournalSucceeded || entry.output_checksum.length() > 0)
  })
}

///|
pub fn batch_mark_running(
  entry : BatchJournalEntry,
  started_at : String,
) -> BatchJournalEntry {
  {
    ..entry,
    state: BatchJournalRunning,
    attempt: entry.attempt + 1,
    started_at,
  }
}

///|
pub fn batch_mark_succeeded(
  entry : BatchJournalEntry,
  output_checksum : String,
  finished_at : String,
) -> BatchJournalEntry {
  {
    ..entry,
    state: BatchJournalSucceeded,
    output_checksum,
    error_code: "",
    error_message: "",
    finished_at,
  }
}

///|
pub fn batch_mark_failed(
  entry : BatchJournalEntry,
  error_code : String,
  error_message : String,
  finished_at : String,
) -> BatchJournalEntry {
  { ..entry, state: BatchJournalFailed, error_code, error_message, finished_at }
}

///|
pub fn batch_mark_skipped(
  entry : BatchJournalEntry,
  reason : String,
  finished_at : String,
) -> BatchJournalEntry {
  {
    ..entry,
    state: BatchJournalSkipped,
    error_code: "SKIPPED",
    error_message: reason,
    finished_at,
  }
}

///|
pub fn recovery_candidate(
  entry : BatchJournalEntry,
  mode : BatchRecoveryMode,
  schedule : RetrySchedule,
) -> RecoveryCandidate {
  let next_attempt = entry.attempt + 1
  {
    entry,
    mode,
    next_attempt,
    delay_seconds: schedule.delay_for(next_attempt),
    reason: batch_recovery_reason(entry, mode),
  }
}

///|
fn batch_recovery_reason(
  entry : BatchJournalEntry,
  mode : BatchRecoveryMode,
) -> String {
  match mode {
    BatchRecoveryResume => "resume unfinished item"
    BatchRecoveryReplayFailed => "replay failed item: " + entry.error_code
    BatchRecoveryRetryTransient => "retry transient error: " + entry.error_code
    BatchRecoveryAuditOnly => "audit without mutation"
  }
}

///|
pub fn build_recovery_plan(
  journal : BatchJournal,
  batch_id : String,
  mode : BatchRecoveryMode,
  schedule : RetrySchedule,
) -> RecoveryPlan {
  let candidates = []
  let completed = []
  let blocked = []
  for entry in journal.latest_for(batch_id) {
    match entry.state {
      BatchJournalSucceeded | BatchJournalSkipped =>
        completed.push(entry.item_id)
      BatchJournalQueued | BatchJournalRunning =>
        candidates.push(recovery_candidate(entry, mode, schedule))
      BatchJournalFailed =>
        if mode == BatchRecoveryReplayFailed {
          candidates.push(recovery_candidate(entry, mode, schedule))
        } else if mode == BatchRecoveryRetryTransient &&
          schedule.can_retry(entry.attempt, entry.error_code) {
          candidates.push(recovery_candidate(entry, mode, schedule))
        } else {
          blocked.push(entry.item_id)
        }
      BatchJournalCancelled => blocked.push(entry.item_id)
    }
  }
  let checksum = stable_hash(
    batch_id +
    ":" +
    batch_recovery_mode_name(mode) +
    ":" +
    candidates.map(fn(item) { item.entry.item_id }).join(","),
  )
  {
    batch_id,
    mode,
    candidates,
    completed_items: completed,
    blocked_items: blocked,
    checksum,
  }
}

///|
pub fn RecoveryPlan::is_empty(self : RecoveryPlan) -> Bool {
  self.candidates.is_empty()
}

///|
pub fn RecoveryPlan::summary(self : RecoveryPlan) -> String {
  [
    "batch_id=" + self.batch_id,
    "mode=" + batch_recovery_mode_name(self.mode),
    "candidates=" + self.candidates.length().to_string(),
    "completed=" + self.completed_items.length().to_string(),
    "blocked=" + self.blocked_items.length().to_string(),
    "checksum=" + self.checksum,
  ].join("\n")
}

///|
pub fn RecoveryPlan::to_json(self : RecoveryPlan) -> String {
  "{" +
  "\"batch_id\":\"" +
  json_escape(self.batch_id) +
  "\"," +
  "\"mode\":\"" +
  batch_recovery_mode_name(self.mode) +
  "\"," +
  "\"candidates\":" +
  self.candidates.length().to_string() +
  "," +
  "\"completed\":" +
  self.completed_items.length().to_string() +
  "," +
  "\"blocked\":" +
  self.blocked_items.length().to_string() +
  "," +
  "\"checksum\":\"" +
  json_escape(self.checksum) +
  "\"}"
}

///|
pub fn recovery_is_idempotent(
  first : RecoveryPlan,
  second : RecoveryPlan,
) -> Bool {
  first.batch_id == second.batch_id &&
  first.mode == second.mode &&
  first.checksum == second.checksum
}

///|
pub fn journal_export_lines(journal : BatchJournal) -> Array[String] {
  journal.entries.map(fn(entry) {
    [
      entry.entry_id,
      entry.batch_id,
      entry.item_id,
      entry.sequence.to_string(),
      batch_journal_state_name(entry.state),
      entry.attempt.to_string(),
      entry.input_checksum,
      entry.output_checksum,
      entry.error_code,
      entry.error_message,
      entry.started_at,
      entry.finished_at,
    ]
    .map(json_escape)
    .join("\t")
  })
}

///|
pub fn journal_export_checksum(journal : BatchJournal) -> String {
  stable_hash(journal_export_lines(journal).join("\n"))
}

///|
pub fn recovery_capacity(plan : RecoveryPlan, workers : Int) -> Int {
  if workers <= 0 {
    0
  } else if plan.candidates.length() < workers {
    plan.candidates.length()
  } else {
    workers
  }
}

///|
pub fn recovery_batches(
  plan : RecoveryPlan,
  workers : Int,
) -> Array[Array[RecoveryCandidate]] {
  let width = recovery_capacity(plan, workers)
  if width <= 0 {
    []
  } else {
    let result : Array[Array[RecoveryCandidate]] = []
    let mut current : Array[RecoveryCandidate] = []
    for candidate in plan.candidates {
      current.push(candidate)
      if current.length() >= width {
        result.push(current)
        current = []
      }
    }
    if !current.is_empty() {
      result.push(current)
    }
    result
  }
}

///|
pub fn batch_entry_is_terminal(entry : BatchJournalEntry) -> Bool {
  match entry.state {
    BatchJournalSucceeded | BatchJournalSkipped | BatchJournalCancelled => true
    _ => false
  }
}

///|
pub fn batch_entry_requires_review(entry : BatchJournalEntry) -> Bool {
  entry.state == BatchJournalFailed &&
  (entry.error_code == "DATA_INTEGRITY" || entry.error_code == "POLICY_DENIED")
}

///|
pub fn journal_terminal_count(journal : BatchJournal) -> Int {
  journal.entries.filter(batch_entry_is_terminal).length()
}

///|
pub fn journal_review_count(journal : BatchJournal) -> Int {
  journal.entries.filter(batch_entry_requires_review).length()
}

///|
pub fn journal_health_score(journal : BatchJournal) -> Int {
  if journal.entries.is_empty() {
    100
  } else {
    let terminal = journal_terminal_count(journal)
    let failed = journal.entries
      .filter(fn(entry) { entry.state == BatchJournalFailed })
      .length()
    let success = terminal * 100 / journal.entries.length()
    if success - failed * 5 < 0 {
      0
    } else {
      success - failed * 5
    }
  }
}

///|
pub fn BatchJournal::summary(self : BatchJournal) -> String {
  let counts = self.count_by_state()
  [
    "journal_id=" + self.journal_id,
    "entries=" + self.entries.length().to_string(),
    "queued=" + counts.get_or_default("queued", 0).to_string(),
    "running=" + counts.get_or_default("running", 0).to_string(),
    "succeeded=" + counts.get_or_default("succeeded", 0).to_string(),
    "failed=" + counts.get_or_default("failed", 0).to_string(),
    "health=" + journal_health_score(self).to_string(),
    "checksum=" + self.checksum,
  ].join("\n")
}