///|
let segment_magic : Bytes = b"MOONSEG\x00"

///|
let segment_format_version : UInt = 5U

///|
let previous_segment_format_version : UInt = 4U

///|
let v3_segment_format_version : UInt = 3U

///|
let v2_segment_format_version : UInt = 2U

///|
let legacy_segment_format_version : UInt = 1U

///|
fn encode_schema(buffer : @buffer.Buffer, schema : Schema?) -> Unit {
  match schema {
    None => write_bool(buffer, false)
    Some(schema) => {
      write_bool(buffer, true)
      buffer.write_int_be(schema.field_count())
      for field_index in 0.. name
          None => abort("schema field disappeared during encoding")
        }
        let field_type = match schema.field_type(field_id) {
          Some(field_type) => field_type
          None => abort("schema field type disappeared during encoding")
        }
        write_string(buffer, name)
        buffer.write_byte(
          match field_type {
            @schema.Text => b'\x00'
            @schema.Keyword => b'\x01'
            @schema.I64 => b'\x02'
            @schema.U64 => b'\x03'
            @schema.F64 => b'\x04'
            @schema.Bool => b'\x05'
            @schema.Date => b'\x06'
            @schema.Bytes => b'\x07'
          },
        )
        write_bool(buffer, schema.is_indexed(field_id))
        write_bool(buffer, schema.is_stored(field_id))
        write_bool(buffer, schema.is_fast(field_id))
        write_bool(buffer, schema.is_multi_valued(field_id))
        write_string(
          buffer,
          match schema.tokenizer_name(field_id) {
            Some(name) => name
            None => ""
          },
        )
        write_string(
          buffer,
          match schema.analyzer_version(field_id) {
            Some(version) => version
            None => ""
          },
        )
      }
    }
  }
}

///|
fn encode_field_value(buffer : @buffer.Buffer, value : FieldValue) -> Unit {
  match value {
    @core.Text(value) => {
      buffer.write_byte(b'\x00')
      write_string(buffer, value)
    }
    @core.Keyword(value) => {
      buffer.write_byte(b'\x01')
      write_string(buffer, value)
    }
    @core.I64(value) => {
      buffer.write_byte(b'\x02')
      write_int64(buffer, value)
    }
    @core.U64(value) => {
      buffer.write_byte(b'\x03')
      write_uint64(buffer, value)
    }
    @core.F64(value) => {
      buffer.write_byte(b'\x04')
      write_double(buffer, value)
    }
    @core.Bool(value) => {
      buffer.write_byte(b'\x05')
      write_bool(buffer, value)
    }
    @core.Date(value) => {
      buffer.write_byte(b'\x06')
      write_int64(buffer, value)
    }
    @core.Bytes(value) => {
      buffer.write_byte(b'\x07')
      buffer.write_int_be(value.length())
      buffer.write_bytes(value[:])
    }
  }
}

///|
fn common_text_prefix(left : String, right : String) -> Int {
  let left_chars = left.to_array()
  let right_chars = right.to_array()
  let limit = if left_chars.length() < right_chars.length() {
    left_chars.length()
  } else {
    right_chars.length()
  }
  let mut length = 0
  while length < limit && left_chars[length] == right_chars[length] {
    length += 1
  }
  length
}

///|
fn text_suffix(text : String, prefix_length : Int) -> String {
  let chars = text.to_array()
  let suffix : Array[Char] = []
  for index in prefix_length.. String? {
  let previous_chars = previous.to_array()
  if prefix_length < 0 || prefix_length > previous_chars.length() {
    return None
  }
  let chars : Array[Char] = []
  for index in 0.. Bytes {
  let buffer = @buffer.Buffer()
  encode_schema(buffer, segment.schema)
  buffer.write_int_be(segment.document_count)
  buffer.write_int_be(segment.indexed_terms.length())
  let mut previous_term_field = FieldId::new(-1)
  let mut previous_term_text = ""
  for term_index in 0.. Bytes {
  let buffer = @buffer.Buffer(
    size_hint=segment_header_length + payload.length(),
  )
  buffer.write_bytes(segment_magic[:])
  buffer.write_uint_be(segment_format_version)
  buffer.write_int_be(payload.length())
  buffer.write_uint_be(segment_checksum(payload))
  buffer.write_bytes(payload[:])
  buffer.to_bytes()
}

///|
pub fn encode_segment(segment : Segment) -> Bytes {
  encode_container(encode_payload(segment))
}

///|
fn decode_schema(
  decoder : SegmentDecoder,
  version : UInt,
) -> Schema? raise PersistenceError {
  if !decoder.read_bool("schema marker") {
    return None
  }
  let field_count = decoder.read_count("schema field count")
  let builder = SchemaBuilder::new()
  for _ in 0.. @schema.Text
        b'\x01' => @schema.Keyword
        b'\x02' => @schema.I64
        b'\x03' => @schema.U64
        b'\x04' => @schema.F64
        b'\x05' => @schema.Bool
        b'\x06' => @schema.Date
        b'\x07' => @schema.Bytes
        _ => raise PersistenceError::InvalidFormat("invalid field type")
      }
      let indexed = decoder.read_bool("indexed option")
      let stored = decoder.read_bool("stored option")
      let fast = decoder.read_bool("fast option")
      let multi_valued = decoder.read_bool("multi-valued option")
      let tokenizer_name = decoder.read_string("tokenizer name")
      let analyzer_version = if version == segment_format_version {
        decoder.read_string("analyzer version")
      } else {
        ""
      }
      let field_id = match field_type {
        @schema.Text => {
          guard valid_tokenizer_name(tokenizer_name) else {
            raise PersistenceError::InvalidFormat("invalid tokenizer name")
          }
          builder.add_text_field(
            name,
            TextOptions::new(indexed, stored)
            .with_fast(fast)
            .with_multi_valued(multi_valued)
            .with_tokenizer(tokenizer_name)
            .with_analyzer_version(analyzer_version),
          )
        }
        _ => {
          guard tokenizer_name == "" else {
            raise PersistenceError::InvalidFormat(
              "non-text field cannot have a tokenizer",
            )
          }
          guard analyzer_version == "" else {
            raise PersistenceError::InvalidFormat(
              "non-text field cannot have an analyzer version",
            )
          }
          let options = FieldOptions::new(indexed, stored, fast).with_multi_valued(
            multi_valued,
          )
          match field_type {
            @schema.Keyword => builder.add_keyword_field(name, options)
            @schema.I64 => builder.add_i64_field(name, options)
            @schema.U64 => builder.add_u64_field(name, options)
            @schema.F64 => builder.add_f64_field(name, options)
            @schema.Bool => builder.add_bool_field(name, options)
            @schema.Date => builder.add_date_field(name, options)
            @schema.Bytes => builder.add_bytes_field(name, options)
            @schema.Text => abort("unreachable text field type")
          }
        }
      }
      ignore(field_id)
    } else {
      let indexed = decoder.read_bool("indexed option")
      let stored = decoder.read_bool("stored option")
      let tokenizer_name = if version == legacy_segment_format_version {
        "whitespace"
      } else {
        decoder.read_string("tokenizer name")
      }
      guard valid_tokenizer_name(tokenizer_name) else {
        raise PersistenceError::InvalidFormat("invalid tokenizer name")
      }
      ignore(
        builder.add_text_field(
          name,
          TextOptions::new(indexed, stored).with_tokenizer(tokenizer_name),
        ),
      )
    }
  }
  Some(builder.build())
}

///|
fn valid_field_id(schema : Schema?, field_id : FieldId) -> Bool {
  if field_id.value < 0 {
    return false
  }
  match schema {
    Some(schema) => field_id.value < schema.field_count()
    None => true
  }
}

///|
fn decode_block_postings(
  decoder : SegmentDecoder,
  document_count : Int,
) -> ReadOnlyArray[Posting] raise PersistenceError {
  let block_count = decoder.read_varint("posting block count")
  guard block_count <= decoder.remaining() else {
    raise PersistenceError::InvalidFormat("invalid posting block count")
  }
  let postings : Array[Posting] = []
  let mut previous_doc_id = -1
  let mut previous_block_max = -1
  for _ in 0.. 0 &&
      count <= posting_block_size &&
      block_max_doc > previous_block_max &&
      block_max_doc < document_count else {
      raise PersistenceError::InvalidFormat("invalid posting block metadata")
    }
    for _ in 0..= 0 && doc_id < document_count && doc_id > previous_doc_id else {
        raise PersistenceError::InvalidFormat(
          "invalid posting doc id: \{doc_id}",
        )
      }
      guard term_freq > 0 && position_count == term_freq else {
        raise PersistenceError::InvalidFormat(
          "term frequency and positions disagree",
        )
      }
      let positions : Array[Int] = []
      let position_lengths : Array[Int] = []
      let start_offsets : Array[Int] = []
      let end_offsets : Array[Int] = []
      let value_indexes : Array[Int] = []
      let mut previous_position = -1
      for position_index in 0..= 0 && position >= previous_position else {
          raise PersistenceError::InvalidFormat("invalid term position")
        }
        positions.push(position)
        let position_length = decoder.read_varint("position length")
        let start_offset = decoder.read_varint("start offset")
        let end_offset = decoder.read_varint("end offset")
        let value_index = decoder.read_varint("value index")
        guard position_length > 0 && end_offset >= start_offset else {
          raise PersistenceError::InvalidFormat("invalid posting occurrence")
        }
        position_lengths.push(position_length)
        start_offsets.push(start_offset)
        end_offsets.push(end_offset)
        value_indexes.push(value_index)
        previous_position = position
      }
      postings.push({
        doc_id: DocId::new(doc_id),
        term_freq,
        positions: ReadOnlyArray::from_array(positions),
        position_lengths: ReadOnlyArray::from_array(position_lengths),
        start_offsets: ReadOnlyArray::from_array(start_offsets),
        end_offsets: ReadOnlyArray::from_array(end_offsets),
        value_indexes: ReadOnlyArray::from_array(value_indexes),
      })
      previous_doc_id = doc_id
    }
    guard previous_doc_id == block_max_doc else {
      raise PersistenceError::InvalidFormat("posting block max doc mismatch")
    }
    previous_block_max = block_max_doc
  }
  ReadOnlyArray::from_array(postings)
}

///|
fn decode_postings(
  decoder : SegmentDecoder,
  document_count : Int,
  version : UInt,
) -> ReadOnlyArray[Posting] raise PersistenceError {
  if version == segment_format_version {
    return decode_block_postings(decoder, document_count)
  }
  let uses_varint = version == previous_segment_format_version ||
    version == v3_segment_format_version
  let posting_count = if uses_varint {
    decoder.read_varint("posting count")
  } else {
    decoder.read_count("posting count")
  }
  guard posting_count <= decoder.remaining() else {
    raise PersistenceError::InvalidFormat("invalid posting count")
  }
  let postings : Array[Posting] = []
  let mut previous_doc_id = -1
  for posting_index in 0.. 0 {
      previous_doc_id + encoded_doc_id
    } else {
      encoded_doc_id
    }
    let term_freq = if uses_varint {
      decoder.read_varint("term frequency")
    } else {
      decoder.read_count("term frequency")
    }
    let position_count = if uses_varint {
      decoder.read_varint("position count")
    } else {
      decoder.read_count("position count")
    }
    guard position_count <= decoder.remaining() else {
      raise PersistenceError::InvalidFormat("invalid position count")
    }
    guard doc_id >= 0 && doc_id < document_count && doc_id > previous_doc_id else {
      raise PersistenceError::InvalidFormat("invalid posting doc id: \{doc_id}")
    }
    guard term_freq > 0 && position_count == term_freq else {
      raise PersistenceError::InvalidFormat(
        "term frequency and positions disagree",
      )
    }
    let positions : Array[Int] = []
    let mut previous_position = -1
    for position_index in 0.. 0 {
        previous_position + encoded_position
      } else {
        encoded_position
      }
      guard position >= 0 && position >= previous_position else {
        raise PersistenceError::InvalidFormat("invalid term position")
      }
      positions.push(position)
      previous_position = position
    }
    postings.push({
      doc_id: DocId::new(doc_id),
      term_freq,
      positions: ReadOnlyArray::from_array(positions),
      position_lengths: ReadOnlyArray::from_array(
        Array::make(position_count, 1),
      ),
      start_offsets: ReadOnlyArray::from_array(Array::make(position_count, 0)),
      end_offsets: ReadOnlyArray::from_array(Array::make(position_count, 0)),
      value_indexes: ReadOnlyArray::from_array(Array::make(position_count, 0)),
    })
    previous_doc_id = doc_id
  }
  ReadOnlyArray::from_array(postings)
}

///|
fn decode_stored_document(
  decoder : SegmentDecoder,
  schema : Schema?,
  version : UInt,
) -> StoredDocument raise PersistenceError {
  let value_count = decoder.read_count("stored value count")
  let field_ids : Array[FieldId] = []
  let texts : Array[String] = []
  for _ in 0.. {
          let field_type = match schema.field_type(field_id) {
            Some(field_type) => field_type
            None => raise PersistenceError::InvalidFormat("unknown typed field")
          }
          guard field_value_matches_type(field_type, value) else {
            raise PersistenceError::InvalidFormat("typed field value mismatch")
          }
        }
        None => ()
      }
      value_field_ids.push(field_id)
      values.push(value)
    }
  }
  StoredDocument::from_all_fields(field_ids, texts, value_field_ids, values)
}

///|
fn decode_field_value(
  decoder : SegmentDecoder,
) -> FieldValue raise PersistenceError {
  match decoder.read_byte("field value type") {
    b'\x00' => @core.Text(decoder.read_string("text value"))
    b'\x01' => @core.Keyword(decoder.read_string("keyword value"))
    b'\x02' => @core.I64(decoder.read_int64("i64 value"))
    b'\x03' => @core.U64(decoder.read_uint64("u64 value"))
    b'\x04' => @core.F64(decoder.read_double("f64 value"))
    b'\x05' => @core.Bool(decoder.read_bool("bool value"))
    b'\x06' => @core.Date(decoder.read_int64("date value"))
    b'\x07' => @core.Bytes(decoder.read_bytes("bytes value"))
    _ => raise PersistenceError::InvalidFormat("invalid field value type")
  }
}

///|
fn decode_document_values(
  decoder : SegmentDecoder,
  schema : Schema?,
  document_count : Int,
) -> ReadOnlyArray[ReadOnlyArray[FieldValueEntry]] raise PersistenceError {
  let count = decoder.read_count("typed document count")
  guard count == document_count else {
    raise PersistenceError::InvalidFormat("typed document count mismatch")
  }
  let documents : Array[ReadOnlyArray[FieldValueEntry]] = []
  for _ in 0.. {
          let field_type = match schema.field_type(field_id) {
            Some(field_type) => field_type
            None => raise PersistenceError::InvalidFormat("unknown typed field")
          }
          guard field_value_matches_type(field_type, value) else {
            raise PersistenceError::InvalidFormat("typed field value mismatch")
          }
          guard schema.is_multi_valued(field_id) ||
            entries.search_by(entry => entry.field_id == field_id) is None else {
            raise PersistenceError::InvalidFormat(
              "multiple values for single-valued field",
            )
          }
        }
        None => ()
      }
      entries.push({ field_id, value })
    }
    documents.push(ReadOnlyArray::from_array(entries))
  }
  ReadOnlyArray::from_array(documents)
}

///|
fn decode_field_lengths(
  decoder : SegmentDecoder,
  schema : Schema?,
) -> ReadOnlyArray[FieldLength] raise PersistenceError {
  let field_count = decoder.read_count("document field length count")
  let lengths : Array[FieldLength] = []
  for _ in 0.. entry.field_id == field_id) is None else {
      raise PersistenceError::InvalidFormat("invalid field length entry")
    }
    lengths.push({ field_id, length })
  }
  ReadOnlyArray::from_array(lengths)
}

///|
fn decode_field_stats(
  decoder : SegmentDecoder,
  schema : Schema?,
  document_count : Int,
) -> ReadOnlyArray[FieldStats] raise PersistenceError {
  let field_count = decoder.read_count("field stats count")
  let stats_entries : Array[FieldStats] = []
  for _ in 0.. entry.field_id == field_id) is None else {
      raise PersistenceError::InvalidFormat("invalid field statistics")
    }
    stats_entries.push({
      field_id,
      document_count: field_document_count,
      total_length,
    })
  }
  ReadOnlyArray::from_array(stats_entries)
}

///|
fn decode_payload(
  payload : Bytes,
  version : UInt,
) -> Segment raise PersistenceError {
  let decoder = SegmentDecoder::new(payload)
  let schema = decode_schema(decoder, version)
  let document_count = decoder.read_count("document count")
  let term_count = decoder.read_count("term count")
  let terms : Array[Term] = []
  let posting_lists : Array[ReadOnlyArray[Posting]] = []
  let mut previous_term_field = FieldId::new(-1)
  let mut previous_term_text = ""
  for _ in 0.. text
        None => raise PersistenceError::InvalidFormat("invalid term prefix")
      }
    } else {
      decoder.read_string("term text")
    }
    let term = Term::new(field_id, term_text)
    guard terms.search_by(existing => existing == term) is None else {
      raise PersistenceError::InvalidFormat("duplicate term")
    }
    terms.push(term)
    posting_lists.push(decode_postings(decoder, document_count, version))
    previous_term_field = field_id
    previous_term_text = term_text
  }
  let stored_field_blocks = if version == segment_format_version {
    let block_count = decoder.read_count("stored block count")
    let blocks : Array[StoredFieldBlock] = []
    for _ in 0.. Segment raise PersistenceError {
  guard bytes.length() >= segment_header_length else {
    raise PersistenceError::InvalidFormat("truncated segment header")
  }
  for index in 0..