///|
/// Stateful recorder. Failed additions are atomic and finishing freezes the builder.
pub struct LogBuilder {
  log : Log
  limits : Limits
  mut closed : Bool
  mut next_schema : Int32
  mut next_channel : Int32
  sequences : Map[Int32, UInt32]
}

///|
pub fn new_builder(
  profile~ : String = "",
  limits~ : Limits = default_limits(),
) -> LogBuilder raise McapError {
  let log = empty_log(profile~)
  validate_log(log, limits~)
  {
    log,
    limits,
    closed: false,
    next_schema: 1,
    next_channel: 1,
    sequences: {},
  }
}

///|
fn LogBuilder::require_open(self : LogBuilder) -> Unit raise McapError {
  if self.closed {
    raise Invalid("recorder already finished")
  }
}

///|
pub fn LogBuilder::register_schema(
  self : LogBuilder,
  name : String,
  encoding : String,
  data : Bytes,
) -> Int32 raise McapError {
  self.require_open()
  for schema in self.log.schemas {
    if schema.name == name && schema.encoding == encoding && schema.data == data {
      return schema.id
    }
  }
  if self.next_schema > 65535 {
    raise Invalid("schema identifier space exhausted")
  }
  let schema : Schema = {
    id: self.next_schema,
    name,
    encoding,
    data: copy_payload(data),
  }
  let candidate = copy_log(self.log)
  candidate.schemas.push(schema)
  validate_log(candidate, limits=self.limits)
  self.log.schemas.push(schema)
  self.next_schema += 1
  schema.id
}

///|
pub fn LogBuilder::register_channel(
  self : LogBuilder,
  topic : String,
  message_encoding : String,
  schema_id~ : Int32 = 0,
  metadata~ : Array[(String, String)] = [],
) -> Int32 raise McapError {
  self.require_open()
  for channel in self.log.channels {
    if channel.topic == topic &&
      channel.schema_id == schema_id &&
      channel.message_encoding == message_encoding &&
      channel.metadata == metadata {
      return channel.id
    }
  }
  if self.next_channel > 65535 {
    raise Invalid("channel identifier space exhausted")
  }
  let channel : Channel = {
    id: self.next_channel,
    schema_id,
    topic,
    message_encoding,
    metadata: metadata.copy(),
  }
  let candidate = copy_log(self.log)
  candidate.channels.push(channel)
  validate_log(candidate, limits=self.limits)
  self.log.channels.push(channel)
  self.sequences[channel.id] = 0U
  self.next_channel += 1
  channel.id
}

///|
pub fn LogBuilder::append(
  self : LogBuilder,
  channel_id : Int32,
  log_time : UInt64,
  data : Bytes,
  publish_time~ : UInt64 = log_time,
) -> Unit raise McapError {
  self.require_open()
  let sequence = match self.sequences.get(channel_id) {
    None => raise Invalid("recorder channel is not registered")
    Some(value) => value
  }
  let message : Message = {
    channel_id,
    sequence,
    log_time,
    publish_time,
    data: copy_payload(data),
  }
  let candidate = copy_log(self.log)
  candidate.messages.push(message)
  validate_log(candidate, limits=self.limits)
  self.log.messages.push(message)
  self.sequences[channel_id] = if sequence == 4294967295U {
    0U
  } else {
    sequence + 1U
  }
}

///|
pub fn LogBuilder::add_metadata(
  self : LogBuilder,
  value : Metadata,
) -> Unit raise McapError {
  self.require_open()
  let copied = { ..value, entries: value.entries.copy(), }
  let candidate = copy_log(self.log)
  candidate.metadata.push(copied)
  validate_log(candidate, limits=self.limits)
  self.log.metadata.push(copied)
}

///|
pub fn LogBuilder::add_attachment(
  self : LogBuilder,
  value : Attachment,
) -> Unit raise McapError {
  self.require_open()
  let copied = { ..value, data: copy_payload(value.data), }
  let candidate = copy_log(self.log)
  candidate.attachments.push(copied)
  validate_log(candidate, limits=self.limits)
  self.log.attachments.push(copied)
}

///|
pub fn LogBuilder::snapshot(self : LogBuilder) -> Log {
  copy_log(self.log)
}

///|
pub fn LogBuilder::finish(
  self : LogBuilder,
  indexed~ : Bool = true,
  messages_per_chunk~ : Int32 = 256,
  compression~ : Compression = Uncompressed,
) -> Bytes raise McapError {
  self.require_open()
  let bytes = if indexed {
    write_indexed(
      self.log,
      messages_per_chunk~,
      compression~,
      limits=self.limits,
    )
  } else if compression != Uncompressed {
    write_chunked(
      self.log,
      messages_per_chunk~,
      compression~,
      limits=self.limits,
    )
  } else {
    write_log(self.log, limits=self.limits)
  }
  self.closed = true
  bytes
}

///|
pub fn LogBuilder::message_count(self : LogBuilder) -> Int32 {
  self.log.messages.length()
}