///|
/// One validated public mutation retained until its transaction is committed.
priv enum TransactionOperation {
  CreateNode(NodeId, Node)
  CreateEdge(Edge)
  AddNodeLabel(NodeId, String)
  RemoveNodeLabel(NodeId, String)
  SetNodeProperty(NodeId, String, PropertyValue)
  RemoveNodeProperty(NodeId, String)
  SetEdgeProperty(EdgeId, String, PropertyValue)
  RemoveEdgeProperty(EdgeId, String)
  DeleteEdge(EdgeId)
  DeleteNode(NodeId)
  DeleteNodeCascade(NodeId)
  CreateNodePropertyIndex(String, String)
  CreateEdgePropertyIndex(String, String)
}

///|
fn wal_append_u64(target : Array[Byte], value : UInt64) -> Unit {
  wal_append_bytes(target, value.to_le_bytes())
}

///|
fn wal_append_string(target : Array[Byte], value : String) -> Unit {
  let bytes = @utf8.encode(value)
  wal_append_u64(target, bytes.length().to_uint64())
  wal_append_bytes(target, bytes)
}

///|
fn wal_append_property_value(
  target : Array[Byte],
  value : PropertyValue,
) -> Unit {
  match value {
    PropertyValue::Bool(value) => {
      target.push(1)
      target.push(if value { 1 } else { 0 })
    }
    PropertyValue::Int64(value) => {
      target.push(2)
      wal_append_u64(target, value.reinterpret_as_uint64())
    }
    PropertyValue::Float64(value) => {
      target.push(3)
      wal_append_u64(
        target,
        value.reinterpret_as_int64().reinterpret_as_uint64(),
      )
    }
    PropertyValue::String(value) => {
      target.push(4)
      wal_append_string(target, value)
    }
    PropertyValue::Bytes(value) => {
      target.push(5)
      wal_append_u64(target, value.length().to_uint64())
      wal_append_bytes(target, value)
    }
  }
}

///|
fn wal_append_properties(target : Array[Byte], properties : Properties) -> Unit {
  let keys : Array[String] = []
  for key, _ in properties {
    keys.push(key)
  }
  keys.sort()
  wal_append_bytes(target, wal_u32_bytes(keys.length().to_uint64()))
  for key in keys {
    wal_append_string(target, key)
    match properties.get(key) {
      Some(value) => wal_append_property_value(target, value)
      None => ()
    }
  }
}

///|
fn wal_encode_transaction_operation(
  operation : TransactionOperation,
) -> WalOperation {
  match operation {
    TransactionOperation::CreateNode(id, node) => {
      let payload : Array[Byte] = []
      match id {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_bytes(payload, wal_u32_bytes(node.labels.length().to_uint64()))
      for label in node.labels {
        wal_append_string(payload, label)
      }
      wal_append_properties(payload, node.properties)
      WalOperation::{ tag: 1, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::CreateEdge(edge) => {
      let payload : Array[Byte] = []
      match edge.id {
        EdgeId(raw) => wal_append_u64(payload, raw)
      }
      match edge.from {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      match edge.to {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_string(payload, edge.edge_type)
      wal_append_properties(payload, edge.properties)
      WalOperation::{ tag: 2, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::AddNodeLabel(id, label) => {
      let payload : Array[Byte] = []
      match id {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_string(payload, label)
      WalOperation::{ tag: 3, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::RemoveNodeLabel(id, label) => {
      let payload : Array[Byte] = []
      match id {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_string(payload, label)
      WalOperation::{ tag: 4, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::SetNodeProperty(id, key, value) => {
      let payload : Array[Byte] = []
      match id {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_string(payload, key)
      wal_append_property_value(payload, value)
      WalOperation::{ tag: 5, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::RemoveNodeProperty(id, key) => {
      let payload : Array[Byte] = []
      match id {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_string(payload, key)
      WalOperation::{ tag: 6, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::SetEdgeProperty(id, key, value) => {
      let payload : Array[Byte] = []
      match id {
        EdgeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_string(payload, key)
      wal_append_property_value(payload, value)
      WalOperation::{ tag: 7, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::RemoveEdgeProperty(id, key) => {
      let payload : Array[Byte] = []
      match id {
        EdgeId(raw) => wal_append_u64(payload, raw)
      }
      wal_append_string(payload, key)
      WalOperation::{ tag: 8, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::DeleteEdge(id) => {
      let payload : Array[Byte] = []
      match id {
        EdgeId(raw) => wal_append_u64(payload, raw)
      }
      WalOperation::{ tag: 9, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::DeleteNode(id) => {
      let payload : Array[Byte] = []
      match id {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      WalOperation::{ tag: 10, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::DeleteNodeCascade(id) => {
      let payload : Array[Byte] = []
      match id {
        NodeId(raw) => wal_append_u64(payload, raw)
      }
      WalOperation::{ tag: 11, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::CreateNodePropertyIndex(label, property) => {
      let payload : Array[Byte] = []
      wal_append_string(payload, label)
      wal_append_string(payload, property)
      WalOperation::{ tag: 12, payload: Bytes::from_array(payload), }
    }
    TransactionOperation::CreateEdgePropertyIndex(edge_type, property) => {
      let payload : Array[Byte] = []
      wal_append_string(payload, edge_type)
      wal_append_string(payload, property)
      WalOperation::{ tag: 13, payload: Bytes::from_array(payload), }
    }
  }
}

///|
fn wal_encode_transaction_operations(
  operations : Array[TransactionOperation],
) -> Array[WalOperation] {
  let encoded : Array[WalOperation] = []
  for operation in operations {
    encoded.push(wal_encode_transaction_operation(operation))
  }
  encoded
}

///|
priv struct WalPayloadReader {
  bytes : Bytes
  mut cursor : Int
  file : String
  record_offset : UInt64
  transaction_id : UInt64
}

///|
fn WalPayloadReader::corrupt(
  self : WalPayloadReader,
  reason : String,
) -> DatabaseError {
  wal_corrupt(
    self.file,
    self.record_offset + self.cursor.to_uint64(),
    Some(self.transaction_id),
    reason,
  )
}

///|
fn WalPayloadReader::read_byte(
  self : WalPayloadReader,
) -> Byte raise DatabaseError {
  if !wal_has_range(self.bytes, self.cursor, 1) {
    raise self.corrupt("operation payload ends before a byte field")
  }
  let byte = self.bytes.at(self.cursor)
  self.cursor += 1
  byte
}

///|
fn WalPayloadReader::read_u32(
  self : WalPayloadReader,
) -> UInt64 raise DatabaseError {
  if !wal_has_range(self.bytes, self.cursor, 4) {
    raise self.corrupt("operation payload ends before a 32-bit field")
  }
  let value = wal_read_u32(self.bytes, self.cursor)
  self.cursor += 4
  value
}

///|
fn WalPayloadReader::read_u64(
  self : WalPayloadReader,
) -> UInt64 raise DatabaseError {
  if !wal_has_range(self.bytes, self.cursor, 8) {
    raise self.corrupt("operation payload ends before a 64-bit field")
  }
  let value = wal_read_u64(self.bytes, self.cursor)
  self.cursor += 8
  value
}

///|
fn WalPayloadReader::read_blob(
  self : WalPayloadReader,
) -> Bytes raise DatabaseError {
  let length = self.read_u64()
  let remaining = self.bytes.length() - self.cursor
  if length > remaining.to_uint64() {
    raise self.corrupt("operation byte string exceeds payload")
  }
  let end = self.cursor + length.to_int()
  let value = self.bytes[self.cursor:end].to_owned()
  self.cursor = end
  value
}

///|
fn WalPayloadReader::read_string(
  self : WalPayloadReader,
) -> String raise DatabaseError {
  let bytes = self.read_blob()
  @utf8.decode(bytes) catch {
    _ => raise self.corrupt("operation string is not valid UTF-8")
  }
}

///|
fn WalPayloadReader::read_properties(
  self : WalPayloadReader,
) -> Properties raise DatabaseError {
  let count = self.read_u32()
  let remaining = self.bytes.length() - self.cursor
  if count > (remaining / 9).to_uint64() {
    raise self.corrupt("property count exceeds operation payload")
  }
  let properties : Properties = Map([])
  for _ in 0.. PropertyValue raise DatabaseError {
  match self.read_byte().to_uint64() {
    1UL => {
      let value = self.read_byte()
      if value == 0 {
        PropertyValue::Bool(false)
      } else if value == 1 {
        PropertyValue::Bool(true)
      } else {
        raise self.corrupt("invalid boolean property encoding")
      }
    }
    2UL => PropertyValue::Int64(self.read_u64().reinterpret_as_int64())
    3UL => {
      let value = self.read_u64().reinterpret_as_int64().reinterpret_as_double()
      match normalize_property_value("", PropertyValue::Float64(value)) {
        PropertyValue::Float64(normalized) => PropertyValue::Float64(normalized)
        _ => raise self.corrupt("invalid float property encoding")
      }
    }
    4UL => PropertyValue::String(self.read_string())
    5UL => PropertyValue::Bytes(self.read_blob())
    _ => raise self.corrupt("unknown property value tag")
  }
}

///|
fn WalPayloadReader::finish(
  self : WalPayloadReader,
) -> Unit raise DatabaseError {
  if self.cursor != self.bytes.length() {
    raise self.corrupt("unexpected trailing bytes in WAL operation")
  }
}

///|
fn GraphState::replay_wal_record(
  self : GraphState,
  record : WalRecord,
  file : String,
  record_offset : UInt64,
) -> Unit raise DatabaseError {
  for operation in record.operations {
    replay_wal_operation(
      self,
      operation,
      file,
      record_offset,
      record.transaction_id,
    ) catch {
      error =>
        match error {
          DatabaseError::CorruptLogRecord(..) => raise error
          _ =>
            raise wal_corrupt(
              file,
              record_offset,
              Some(record.transaction_id),
              "WAL operation violates graph invariants",
            )
        }
    }
  }
}

///|
fn replay_wal_operation(
  graph : GraphState,
  operation : WalOperation,
  file : String,
  record_offset : UInt64,
  transaction_id : UInt64,
) -> Unit raise DatabaseError {
  let reader = WalPayloadReader::{
    bytes: operation.payload,
    cursor: 0,
    file,
    record_offset,
    transaction_id,
  }
  let tag = operation.tag.to_uint64()
  match tag {
    1UL => {
      let expected_id = NodeId(reader.read_u64())
      if expected_id == NodeId(0UL) {
        raise reader.corrupt("node ID zero is invalid")
      }
      let count = reader.read_u32()
      let remaining = reader.bytes.length() - reader.cursor
      if count > (remaining / 8).to_uint64() {
        raise reader.corrupt("node label count exceeds operation payload")
      }
      let labels : Array[String] = []
      for _ in 0.. {
      let expected_id = EdgeId(reader.read_u64())
      let from = NodeId(reader.read_u64())
      let to = NodeId(reader.read_u64())
      let edge_type = reader.read_string()
      let properties = reader.read_properties()
      reader.finish()
      let actual_id = graph.create_edge(from, to, edge_type, properties)
      if actual_id != expected_id {
        raise reader.corrupt("edge ID does not match WAL allocation order")
      }
    }
    3UL | 4UL => {
      let id = NodeId(reader.read_u64())
      let label = reader.read_string()
      reader.finish()
      if tag == 3UL {
        graph.add_label(id, label)
      } else {
        graph.remove_label(id, label)
      }
    }
    5UL | 7UL => {
      let raw_id = reader.read_u64()
      let key = reader.read_string()
      let value = reader.read_property_value()
      reader.finish()
      if tag == 5UL {
        graph.set_node_property(NodeId(raw_id), key, value)
      } else {
        graph.set_edge_property(EdgeId(raw_id), key, value)
      }
    }
    6UL | 8UL => {
      let raw_id = reader.read_u64()
      let key = reader.read_string()
      reader.finish()
      if tag == 6UL {
        graph.remove_node_property(NodeId(raw_id), key)
      } else {
        graph.remove_edge_property(EdgeId(raw_id), key)
      }
    }
    9UL => {
      let id = EdgeId(reader.read_u64())
      reader.finish()
      graph.delete_edge(id)
    }
    10UL | 11UL => {
      let id = NodeId(reader.read_u64())
      reader.finish()
      if tag == 10UL {
        graph.delete_node(id)
      } else {
        graph.delete_node_cascade(id)
      }
    }
    12UL | 13UL => {
      let first = reader.read_string()
      let second = reader.read_string()
      reader.finish()
      if tag == 12UL {
        graph.create_node_property_index(first, second)
      } else {
        graph.create_edge_property_index(first, second)
      }
    }
    _ => raise reader.corrupt("unknown WAL operation tag")
  }
}

///|
fn wal_recover(
  bytes : Bytes,
  file~ : String,
) -> (GraphState, UInt64, Bytes) raise DatabaseError {
  let scan = wal_scan(bytes, file~, offset=0UL, snapshot_transaction_id=0UL)
  let mut graph = GraphState::new()
  let mut record_offset = 0UL
  let clean_log : Array[Byte] = []
  for record in scan.records {
    let candidate = graph.copy_detached()
    candidate.replay_wal_record(record, file, record_offset)
    graph = candidate
    let encoded = wal_encode(record)
    wal_append_bytes(clean_log, encoded)
    record_offset += encoded.length().to_uint64()
  }
  (graph, scan.last_transaction_id, Bytes::from_array(clean_log))
}