///|
priv struct TransactionBuffer {
  id : UInt
  mutations : Array[Mutation]
}

///|
fn mutation_result(
  valid : Bool,
  message : String,
  mutation? : Mutation,
) -> MutationDecode {
  { valid, mutation, message }
}

///|
/// Encodes a put payload as key length, key bytes, then value bytes.
pub fn encode_put_payload(key : BytesView, value : BytesView) -> Bytes {
  let output = Buffer(size_hint=4 + key.length() + value.length())
  output.write_uint_le(key.length().reinterpret_as_uint())
  output.write_bytes(key)
  output.write_bytes(value)
  output.to_bytes()
}

///|
/// Encodes a delete payload as key length followed by key bytes.
pub fn encode_delete_payload(key : BytesView) -> Bytes {
  let output = Buffer(size_hint=4 + key.length())
  output.write_uint_le(key.length().reinterpret_as_uint())
  output.write_bytes(key)
  output.to_bytes()
}

///|
/// Decodes a put or delete record without allocating before bounds checks pass.
pub fn decode_mutation(
  record : JournalRecord,
  max_key? : Int = 1024 * 1024,
) -> MutationDecode {
  if record.kind != Put && record.kind != Delete {
    return mutation_result(false, "record is not a mutation")
  }
  if max_key < 0 {
    return mutation_result(false, "max key must not be negative")
  }
  if record.payload.length() < 4 {
    return mutation_result(false, "mutation payload has no key length")
  }
  let key_length_u = record.payload.unsafe_read_uint32_le(0)
  if key_length_u > max_key.reinterpret_as_uint() {
    return mutation_result(false, "mutation key exceeds configured limit")
  }
  let key_length = key_length_u.reinterpret_as_int()
  if key_length > record.payload.length() - 4 {
    return mutation_result(false, "mutation key is truncated")
  }
  if record.kind == Delete && key_length != record.payload.length() - 4 {
    return mutation_result(false, "delete payload contains trailing bytes")
  }
  let key = record.payload[4:4 + key_length].to_owned()
  let value = if record.kind == Put {
    record.payload[4 + key_length:].to_owned()
  } else {
    b""
  }
  let kind = if record.kind == Put { PutValue } else { DeleteKey }
  mutation_result(true, "mutation decoded", mutation={
    transaction: record.transaction,
    sequence: record.sequence,
    kind,
    key,
    value,
  })
}

///|
/// Encodes the highest sequence known to be durable in a checkpoint payload.
pub fn encode_checkpoint_payload(sequence : UInt) -> Bytes {
  let output = Buffer(size_hint=4)
  output.write_uint_le(sequence)
  output.to_bytes()
}

///|
fn decode_checkpoint(record : JournalRecord) -> UInt? {
  if record.kind != Checkpoint || record.payload.length() != 4 {
    return None
  }
  Some(record.payload.unsafe_read_uint32_le(0))
}

///|
fn transaction_index(active : Array[TransactionBuffer], id : UInt) -> Int {
  for index, transaction in active {
    if transaction.id == id {
      return index
    }
  }
  -1
}

///|
fn add_issue(
  issues : Array[RecoveryIssue],
  code : String,
  record : JournalRecord,
  message : String,
) -> Unit {
  issues.push({
    code,
    sequence: record.sequence,
    transaction: record.transaction,
    message,
  })
}

///|
/// Builds a replay plan. Only mutations from explicitly committed transactions
/// are returned; aborted and crash-incomplete transactions never leak through.
pub fn build_recovery_plan(
  records : Array[JournalRecord],
  max_key? : Int = 1024 * 1024,
) -> RecoveryPlan {
  let issues : Array[RecoveryIssue] = []
  let mut checkpoint_sequence = 0U
  for record in records {
    if record.kind == Checkpoint {
      match decode_checkpoint(record) {
        Some(sequence) =>
          if record.transaction != 0U {
            add_issue(
              issues, "checkpoint_transaction", record, "checkpoint must not belong to a transaction",
            )
          } else if sequence >= checkpoint_sequence &&
            sequence < record.sequence {
            checkpoint_sequence = sequence
          } else {
            add_issue(
              issues, "checkpoint_range", record, "checkpoint durable sequence is invalid",
            )
          }
        None =>
          add_issue(
            issues, "checkpoint_payload", record, "checkpoint payload must contain one uint32 sequence",
          )
      }
    }
  }
  let active : Array[TransactionBuffer] = []
  let actions : Array[Mutation] = []
  let mut committed_transactions = 0
  let mut aborted_transactions = 0
  for record in records {
    if record.sequence <= checkpoint_sequence || record.kind == Checkpoint {
      continue
    }
    if record.transaction == 0U {
      add_issue(
        issues, "missing_transaction", record, "transactional record has transaction id zero",
      )
      continue
    }
    let index = transaction_index(active, record.transaction)
    match record.kind {
      Begin =>
        if index >= 0 {
          add_issue(
            issues, "duplicate_begin", record, "transaction is already active",
          )
        } else {
          active.push({ id: record.transaction, mutations: [] })
        }
      Put | Delete =>
        if index < 0 {
          add_issue(
            issues, "orphan_mutation", record, "mutation has no active transaction",
          )
        } else {
          let decoded = decode_mutation(record, max_key~)
          match decoded.mutation {
            Some(mutation) => active[index].mutations.push(mutation)
            None =>
              add_issue(issues, "mutation_payload", record, decoded.message)
          }
        }
      Commit =>
        if index < 0 {
          add_issue(
            issues, "orphan_commit", record, "commit has no active transaction",
          )
        } else if !record.payload.is_empty() {
          add_issue(
            issues, "terminal_payload", record, "commit payload must be empty",
          )
        } else {
          for mutation in active[index].mutations {
            actions.push(mutation)
          }
          ignore(active.remove(index))
          committed_transactions += 1
        }
      Abort =>
        if index < 0 {
          add_issue(
            issues, "orphan_abort", record, "abort has no active transaction",
          )
        } else if !record.payload.is_empty() {
          add_issue(
            issues, "terminal_payload", record, "abort payload must be empty",
          )
        } else {
          ignore(active.remove(index))
          aborted_transactions += 1
        }
      Checkpoint => ()
    }
  }
  let incomplete_transactions = active.map(fn(transaction) { transaction.id })
  {
    actions,
    committed_transactions,
    aborted_transactions,
    incomplete_transactions,
    checkpoint_sequence,
    issues,
    recoverable: issues.is_empty(),
  }
}