///|
/// Write-Ahead Log (WAL) implementation
///
/// Format v1 (wal_version=0x01):
///   Header (12 bytes): Magic (8 bytes "VCDBMBT\0") + Type (1) + Version (1) + Reserved (2)
///   Records: Type (1) + Reserved (1) + ID (8) + AttrsLen (4) + VectorLen (4) + Attrs + Vector
///   Footer (8 bytes, optional): Magic (4 bytes) + CRC32 (4 bytes)
///
/// Format v2 (wal_version=0x02):
///   Header: same as v1 (version byte = 0x02)
///   Records: Type (1) + Flags (1) + Timestamp (8) + ID (8) + AttrsLen (4) + VectorLen (4) + Attrs + Vector
///   Footer: same as v1
///
/// Format v3 (wal_version=0x03, legacy):
///   Header: same as v1 (version byte = 0x03)
///   Records: Type (1) + Flags (1) + Timestamp (8) + IDLen (1) + ID (IDLen bytes) + AttrsLen (4) + VectorLen (4) + Attrs + Vector
///   Footer: same as v1
///
/// Format v4 (wal_version=0x04, current):
///   Header: same as v1 (version byte = 0x04)
///   Records: Type (1) + Flags (1) + Timestamp (8) + IDWireLen (4) + ID (wire bytes) + AttrsLen (4) + VectorLen (4) + Attrs + Vector
///   Footer: same as v1
///   ID uses VectorId::to_wire_bytes() / from_wire_bytes().
///
/// Flags byte:
///   bit 0 (0x01) = has_timestamp  — always set in v2/v3/v4
///   bit 1 (0x02) = has_128bit_id  — v3 only (deprecated in v4)
///
/// v4 reader can read v1/v2/v3 records (version-aware decode).
/// v3 reader cannot read v4 records (IDWireLen is u32, not u8).

///|
/// WAL Footer magic: "WCRC" in little-endian
pub let wal_footer_magic : UInt = 0x43524357U

///|
/// Current WAL format version
pub let wal_version : Byte = b'\x04'

///|
/// V3 WAL version (for backward compatibility reads)
let wal_version_v3 : Byte = b'\x03'

///|
/// V2 WAL version (for backward compatibility reads)
let wal_version_v2 : Byte = b'\x02'

///|
/// V1 WAL version (for backward compatibility)
let wal_version_v1 : Byte = b'\x01'

///|
/// Flags
let flag_has_timestamp : Byte = b'\x01'

///|
/// Flag indicating the ID field is 128-bit (Bytes16Id). Absent = 8-byte Int64Id.
let flag_has_128bit_id : Byte = b'\x02'

///|
/// WAL record types
pub enum WalRecordType {
  Upsert // 1 - Insert or update vector with attrs
  Remove // 2 - Delete vector
  SetAttrs // 3 - Update attrs only
} derive(Eq, Debug)

///|
pub impl Show for WalRecordType with fn output(self, logger) {
  match self {
    Upsert => logger.write_string("Upsert")
    Remove => logger.write_string("Remove")
    SetAttrs => logger.write_string("SetAttrs")
  }
}

///|
pub fn WalRecordType::to_byte(self : WalRecordType) -> Byte {
  match self {
    Upsert => b'\x01'
    Remove => b'\x02'
    SetAttrs => b'\x03'
  }
}

///|
pub fn WalRecordType::from_byte(b : Byte) -> WalRecordType? {
  match b {
    b'\x01' => Some(Upsert)
    b'\x02' => Some(Remove)
    b'\x03' => Some(SetAttrs)
    _ => None
  }
}

///|
/// WAL record with timestamp
pub struct WalRecord {
  record_type : WalRecordType
  id : @types.VectorId
  attrs : @types.Attrs?
  vector : Array[Double]?
  /// Nanoseconds since Unix epoch. 0 = no timestamp (v1 compatibility).
  timestamp : Int64
} derive(Debug)

///|
pub impl Show for WalRecord with fn output(self, logger) {
  logger.write_string("{ record_type: ")
  Show::output(self.record_type, logger)
  logger.write_string(", id: ")
  Show::output(self.id, logger)
  logger.write_string(", attrs: ")
  Debug::to_repr(self.attrs).output(logger)
  logger.write_string(", vector: ")
  Debug::to_repr(self.vector).output(logger)
  logger.write_string(", timestamp: ")
  Show::output(self.timestamp, logger)
  logger.write_string(" }")
}

///|
/// Create an upsert record with timestamp
pub fn WalRecord::upsert(
  id : @types.VectorId,
  vector : Array[Double],
  attrs : @types.Attrs,
  timestamp? : Int64 = 0L,
) -> WalRecord {
  {
    record_type: Upsert,
    id,
    attrs: Some(attrs),
    vector: Some(vector),
    timestamp,
  }
}

///|
/// Create a remove record with timestamp
pub fn WalRecord::remove(
  id : @types.VectorId,
  timestamp? : Int64 = 0L,
) -> WalRecord {
  { record_type: Remove, id, attrs: None, vector: None, timestamp }
}

///|
/// Create a set_attrs record with timestamp
pub fn WalRecord::set_attrs(
  id : @types.VectorId,
  attrs : @types.Attrs,
  timestamp? : Int64 = 0L,
) -> WalRecord {
  { record_type: SetAttrs, id, attrs: Some(attrs), vector: None, timestamp }
}

///|
fn encode_attrs_json(attrs : @types.Attrs?) -> Bytes {
  @codec.encode_attrs_json(attrs)
}

///|
fn decode_attrs_json(bytes : Bytes) -> @types.Attrs? {
  @codec.decode_attrs_json(bytes)
}

///|
/// Encode a WAL record to bytes (v2 format with timestamp)
pub fn encode_wal_record(record : WalRecord) -> Bytes {
  let w = @binary.BinaryWriter::new()
  encode_wal_record_to(w, record)
  w.concat()
}

///|
/// Encode a WAL record to an existing writer (v3 format)
fn encode_wal_record_to(w : @binary.BinaryWriter, record : WalRecord) -> Unit {
  // Type (1 byte)
  w.push_byte(record.record_type.to_byte())
  // Flags (1 byte) — v4: has_timestamp only (ID type encoded in wire bytes)
  w.push_byte(flag_has_timestamp)
  // Timestamp (8 bytes, Int64 nanoseconds)
  w.push_i64(record.timestamp)
  // ID: length-prefixed wire bytes from VectorId::to_wire_bytes
  let id_wire = record.id.to_wire_bytes()
  w.push_u32(id_wire.length().reinterpret_as_uint())
  w.push_bytes(id_wire)
  // Attrs
  let attrs_bytes = encode_attrs_json(record.attrs)
  let vector_len = match record.vector {
    None => 0
    Some(v) => v.length() * 4
  }
  w.push_u32(attrs_bytes.length().reinterpret_as_uint())
  w.push_u32(vector_len.reinterpret_as_uint())
  w.push_bytes(attrs_bytes)
  match record.vector {
    None => ()
    Some(v) =>
      for val in v {
        w.push_f32(Float::from_double(val))
      }
  }
}

///|
/// Decode a WAL record from reader (public entry point — assumes v3 format).
/// For reading v1/v2 WAL files, use decode_wal_records which handles versioning.
pub fn decode_wal_record(r : @binary.BinaryReader) -> WalRecord? {
  decode_wal_record_versioned(r, wal_version)
}

///|
/// Version-aware WAL record decoder (v1, v2, v3).
///
/// v1: Type(1) + Reserved(1) + ID(8) + AttrsLen(4) + VectorLen(4) + ...
/// v2: Type(1) + Flags(1) + Timestamp(8) + ID(8) + AttrsLen(4) + VectorLen(4) + ...
/// v3: Type(1) + Flags(1) + Timestamp(8) + IDLen(1) + ID(IDLen) + AttrsLen(4) + VectorLen(4) + ...
fn decode_wal_record_versioned(
  r : @binary.BinaryReader,
  version : Byte,
) -> WalRecord? {
  if r.remaining() < 2 {
    return None
  }
  // Type
  let type_byte = r.read_byte()
  let record_type = match WalRecordType::from_byte(type_byte) {
    None => return None
    Some(t) => t
  }
  // Flags/Reserved
  let flags = r.read_byte()
  let has_timestamp = flags.land(flag_has_timestamp) != b'\x00'
  let has_128bit_id = flags.land(flag_has_128bit_id) != b'\x00'
  // Timestamp (v2/v3) or absent (v1)
  let timestamp : Int64 = if has_timestamp {
    if r.remaining() < 8 {
      return None
    }
    r.read_i64()
  } else {
    0L
  }
  // ID field layout differs by version:
  //   v1: fixed 8 bytes (no IDLen byte, no timestamp)
  //   v2: fixed 8 bytes (no IDLen byte, has timestamp)
  //   v3: IDLen(1) + payload(IDLen bytes), flag_has_128bit_id for Bytes16Id
  //   v4: WireLen(u32=4) + wire bytes consumed by VectorId::from_wire_bytes
  let id : @types.VectorId = if version == wal_version {
    // v4: length-prefixed wire bytes
    if r.remaining() < 4 {
      return None
    }
    let wire_len = r.read_u32().reinterpret_as_int()
    if r.remaining() < wire_len {
      return None
    }
    match @types.VectorId::from_wire_bytes(r.read_bytes(wire_len)) {
      Some(v) => v
      None => return None
    }
  } else if has_128bit_id {
    // v3 only: 128-bit ID
    if r.remaining() < 1 {
      return None
    }
    let id_len = r.read_byte().to_int()
    if id_len != 16 || r.remaining() < 16 {
      return None
    }
    let wire = r.read_bytes(16)
    match @types.VectorId::from_wire_bytes(wire) {
      Some(v) => v
      None => return None
    }
  } else if version == wal_version_v3 {
    // v3 with Int64Id: IDLen(1) + 8 bytes
    if r.remaining() < 1 {
      return None
    }
    let id_len = r.read_byte().to_int()
    if id_len != 8 || r.remaining() < 8 {
      return None
    }
    @types.Int64Id(r.read_u64().reinterpret_as_int64())
  } else {
    // v1 or v2: fixed 8-byte ID (no IDLen byte)
    if r.remaining() < 8 {
      return None
    }
    @types.Int64Id(r.read_u64().reinterpret_as_int64())
  }
  // Lengths
  let attrs_len = r.read_u32().reinterpret_as_int()
  let vector_len = r.read_u32().reinterpret_as_int()
  if attrs_len < 0 || vector_len < 0 || r.remaining() < attrs_len + vector_len {
    return None
  }
  // Attrs
  let attrs = if attrs_len > 0 {
    let attrs_bytes = r.read_bytes(attrs_len)
    decode_attrs_json(attrs_bytes)
  } else {
    None
  }
  // Vector
  let vector = if vector_len > 0 {
    if vector_len % 4 != 0 {
      return None
    }
    let vec : Array[Double] = []
    let count = vector_len / 4
    for _ in 0.. Bytes {
  let w = @binary.BinaryWriter::new()
  @codec.encode_header(w, Wal, wal_version)
  w.concat()
}

///|
/// Encode WAL footer with CRC32
pub fn encode_wal_footer(body : Bytes) -> Bytes {
  let w = @binary.BinaryWriter::new()
  w.push_u32(wal_footer_magic)
  w.push_u32(@binary.crc32_fast(body))
  w.concat()
}

///|
/// Encode multiple records to a complete WAL segment
pub fn encode_wal_segment(records : Array[WalRecord]) -> Bytes {
  let w = @binary.BinaryWriter::new()
  @codec.encode_header(w, Wal, wal_version)
  for record in records {
    encode_wal_record_to(w, record)
  }
  let body = w.concat()
  w.push_u32(wal_footer_magic)
  w.push_u32(@binary.crc32_fast(body))
  w.concat()
}

///|
/// Verify WAL header (accepts v1, v2, and v3). Returns the version byte on success.
pub fn verify_wal_header(r : @binary.BinaryReader) -> Bool {
  match @codec.read_header(r) {
    None => false
    Some((file_type, version)) =>
      file_type == Wal &&
      (
        version == wal_version ||
        version == wal_version_v3 ||
        version == wal_version_v2 ||
        version == wal_version_v1
      )
  }
}

///|
/// Read the WAL version from a header, returning None if the header is invalid.
fn read_wal_version(data : Bytes) -> Byte? {
  if data.length() < @codec.header_size {
    return None
  }
  let r = @binary.BinaryReader::new(data)
  match @codec.read_header(r) {
    None => None
    Some((file_type, version)) =>
      if file_type == Wal &&
        (
          version == wal_version ||
          version == wal_version_v3 ||
          version == wal_version_v2 ||
          version == wal_version_v1
        ) {
        Some(version)
      } else {
        None
      }
  }
}

///|
fn wal_footer_start(data : Bytes) -> Int? {
  if data.length() < @codec.header_size + 8 {
    return None
  }
  let footer_start = data.length() - 8
  let r = @binary.BinaryReader::new(data)
  r.skip(footer_start)
  let footer_magic = r.read_u32()
  if footer_magic == wal_footer_magic {
    Some(footer_start)
  } else {
    None
  }
}

///|
fn wal_body_length(data : Bytes) -> Int {
  match wal_footer_start(data) {
    Some(footer_start) => footer_start
    None => data.length()
  }
}

///|
/// Check if WAL has valid footer and verify CRC32
pub fn verify_wal_checksum(data : Bytes) -> Bool {
  if data.length() < @codec.header_size {
    return false
  }
  if data.length() < @codec.header_size + 8 {
    return true
  }
  let footer_start = match wal_footer_start(data) {
    None => return true
    Some(offset) => offset
  }
  let footer_reader = @binary.BinaryReader::new(data)
  footer_reader.skip(footer_start + 4)
  let stored_crc = footer_reader.read_u32()
  let body_reader = @binary.BinaryReader::new(data)
  let body = body_reader.read_bytes(footer_start)
  let computed_crc = @binary.crc32_fast(body)
  stored_crc == computed_crc
}

///|
/// Decode all records from WAL data (supports v1, v2, and v3)
pub fn decode_wal_records(data : Bytes) -> Array[WalRecord] {
  let records : Array[WalRecord] = []
  let version = match read_wal_version(data) {
    None => return records
    Some(v) => v
  }
  let r = @binary.BinaryReader::new(data)
  // Skip header (already validated by read_wal_version)
  r.skip(@codec.header_size)
  let body_len = wal_body_length(data)
  while r.position() < body_len {
    match decode_wal_record_versioned(r, version) {
      None => break
      Some(record) => records.push(record)
    }
  }
  records
}

///|
/// Merge an existing WAL with a new WAL segment.
pub fn merge_wal(existing : Bytes, new_segment : Bytes) -> Bytes {
  let w = @binary.BinaryWriter::new()
  let mut use_existing = false

  // Always write v4 header, then re-encode existing records in v4 format
  w.push_bytes(encode_wal_header())

  if existing.length() >= @codec.header_size {
    if verify_wal_checksum(existing) {
      let existing_body_len = wal_body_length(existing)
      match read_wal_version(existing) {
        None => ()
        Some(existing_version) => {
          let er = @binary.BinaryReader::new(existing)
          er.skip(@codec.header_size)
          while er.position() < existing_body_len {
            match decode_wal_record_versioned(er, existing_version) {
              None => break
              Some(record) => encode_wal_record_to(w, record)
            }
          }
          use_existing = true
        }
      }
    }
  }

  let _ = use_existing

  let new_body_len = wal_body_length(new_segment)
  match read_wal_version(new_segment) {
    None => ()
    Some(new_version) => {
      let r = @binary.BinaryReader::new(new_segment)
      r.skip(@codec.header_size)
      while r.position() < new_body_len {
        match decode_wal_record_versioned(r, new_version) {
          None => break
          // Re-encode in current (v4) format
          Some(record) => encode_wal_record_to(w, record)
        }
      }
    }
  }

  let body = w.concat()
  let footer = encode_wal_footer(body)
  let final_w = @binary.BinaryWriter::new()
  final_w.push_bytes(body)
  final_w.push_bytes(footer)
  final_w.concat()
}

///|
/// Decode replayable WAL records after validating header and checksum.
pub fn wal_records_for_replay(
  data : Bytes,
  cutoff_ts? : Int64 = 0L,
) -> Array[WalRecord] {
  if data.length() < @codec.header_size {
    return []
  }
  let header_reader = @binary.BinaryReader::new(data)
  if !verify_wal_header(header_reader) {
    return []
  }
  if !verify_wal_checksum(data) {
    return []
  }
  if cutoff_ts > 0L {
    filter_wal_by_timestamp(data, cutoff_ts)
  } else {
    decode_wal_records(data)
  }
}

///|
/// Replay WAL binary data into a CoreStore (no Storage dependency).
/// Used by:
/// - AsyncWalRuntime::replay_into (reads from AsyncStorage, then delegates here)
/// - JS persistent exports (passes WAL bytes directly)
///
/// Returns the number of records applied. Returns 0 for empty/invalid data.
pub fn replay_wal_data(
  data : Bytes,
  store : @store.CoreStore,
  cutoff_ts? : Int64 = 0L,
) -> Int {
  let records = wal_records_for_replay(data, cutoff_ts~)
  let mut applied = 0
  for record in records {
    match record.record_type {
      Upsert =>
        match record.vector {
          None => ()
          Some(vector) => {
            let attrs = match record.attrs {
              None => @types.empty_attrs()
              Some(a) => a
            }
            let _ = store.add_or_update(record.id, vector, attrs, upsert=true)
            applied = applied + 1
          }
        }
      Remove => {
        let _ = store.remove_by_id(record.id)
        applied = applied + 1
      }
      SetAttrs =>
        match record.attrs {
          None => ()
          Some(attrs) => {
            let _ = store.update_attrs(record.id, attrs)
            applied = applied + 1
          }
        }
    }
  }
  applied
}

///|
/// Filter WAL records by timestamp — returns only records with timestamp <= cutoff.
/// Used for read-at-timestamp (consistent cross-shard reads).
pub fn filter_wal_by_timestamp(
  data : Bytes,
  cutoff_ts : Int64,
) -> Array[WalRecord] {
  let all = decode_wal_records(data)
  let filtered : Array[WalRecord] = []
  for record in all {
    // timestamp=0 means v1 record (no timestamp) — always include
    if record.timestamp == 0L || record.timestamp <= cutoff_ts {
      filtered.push(record)
    }
  }
  filtered
}

///|
/// Get the maximum timestamp across all records in a WAL.
/// Returns 0 if no timestamped records.
pub fn max_wal_timestamp(data : Bytes) -> Int64 {
  let records = decode_wal_records(data)
  let mut max_ts = 0L
  for record in records {
    if record.timestamp > max_ts {
      max_ts = record.timestamp
    }
  }
  max_ts
}