///|
pub fn open_disk_sidecar(
  source : ReadSource,
  index : ReadSource,
  verification : IndexVerification,
  limits~ : Limits = default_limits(),
  declaration_bytes~ : Int32 = 8 * 1024 * 1024,
  cache_pages~ : Int32 = 4,
  chunk_cache_bytes~ : Int32 = 8 * 1024 * 1024,
  index_bytes~ : UInt64 = 1099511627776UL,
) -> DiskSidecarReader raise McapError {
  open_disk_sidecar_internal(
    source,
    index,
    verification,
    limits~,
    declaration_bytes~,
    cache_pages~,
    chunk_cache_bytes~,
    index_bytes~,
  )
}

///|
pub struct DiskSidecarAudit {
  reader : DiskSidecarReader
  mut page : Int32
  mut previous : DiskIndexEntry?
  mut failed : Bool
}

///|
/// Headers/source envelopes are read synchronously here; pages are fed by poll.
pub fn start_disk_sidecar_audit(
  source : ReadSource,
  index : ReadSource,
  verification : IndexVerification,
  limits~ : Limits = default_limits(),
  declaration_bytes~ : Int32 = 8 * 1024 * 1024,
  cache_pages~ : Int32 = 4,
  chunk_cache_bytes~ : Int32 = 8 * 1024 * 1024,
  index_bytes~ : UInt64 = 1099511627776UL,
) -> DiskSidecarAudit raise McapError {
  {
    reader: open_disk_sidecar_internal(
      source,
      index,
      verification,
      limits~,
      declaration_bytes~,
      cache_pages~,
      chunk_cache_bytes~,
      index_bytes~,
      audit=false,
    ),
    page: 0,
    previous: None,
    failed: false,
  }
}

///|
pub fn DiskSidecarAudit::poll(
  self : DiskSidecarAudit,
) -> ReadSpan? raise McapError {
  if self.failed {
    raise Invalid("disk index audit failed")
  }
  if self.page.to_uint64() * self.reader.width.to_uint64() >=
    self.reader.count.to_uint64() {
    return None
  }
  let length = self.reader.width * 36 + 4
  Some({
    offset: self.reader.start + self.page.to_uint64() * length.to_uint64(),
    length,
  })
}

///|
pub fn DiskSidecarAudit::feed(
  self : DiskSidecarAudit,
  bytes : Bytes,
) -> Unit raise McapError {
  errdefer {
    self.failed = true
  }

  guard self.poll() is Some(_) else {
    raise Invalid("disk index audit already complete")
  }
  let entries = self.reader.accept_page(self.page, bytes)
  if self.previous is Some(previous) && !disk_entry_before(previous, entries[0]) {
    raise Invalid("disk index cross-page order mismatch")
  }
  self.previous = entries.last()
  self.page += 1
}

///|
pub fn DiskSidecarAudit::finish(
  self : DiskSidecarAudit,
) -> DiskSidecarReader raise McapError {
  if self.poll() is Some(_) {
    raise Invalid("disk index audit is incomplete")
  }
  self.reader.pages.clear()
  self.reader
}

///|
pub fn DiskSidecarReader::declaration_log(self : DiskSidecarReader) -> Log {
  copy_log(self.base.declarations)
}

///|
pub(all) enum DiskSidecarPoll {
  DiskDone
  DiskMessage(Message)
  DiskIndexRead(ReadSpan)
  DiskSourceRead(ReadSpan)
}

///|
pub struct DiskSidecarPollCursor {
  reader : DiskSidecarReader
  original : Query
  allowed : Map[Int32, Bool]
  mut query : Query
  mut low : Int32
  mut high : Int32
  mut searching : Bool
  mut position : Int32
  mut offset : Int32
  mut remaining : Int32
  mut state : FileCursorState
  mut current : (Int32, Array[DiskIndexEntry])?
  mut waiting_page : Int32?
  mut waiting_entry : SidecarEntry?
  mut ready : Message?
}

///|
pub fn DiskSidecarReader::poll_cursor(
  self : DiskSidecarReader,
  query : Query,
) -> DiskSidecarPollCursor raise McapError {
  if query.start_time > query.end_time || query.offset < 0 || query.limit < 0 {
    raise Invalid("invalid disk index query")
  }
  let query = { ..query, topics: query.topics.copy(), }
  let allowed : Map[Int32, Bool] = Map([])
  for channel in self.base.declarations.channels {
    if query.topics.is_empty() || query.topics.contains(channel.topic) {
      allowed[channel.id] = true
    }
  }
  {
    reader: self,
    original: query,
    allowed,
    query,
    low: 0,
    high: self.count,
    searching: true,
    position: 0,
    offset: query.offset,
    remaining: query.limit,
    state: if query.limit == 0 {
      FileExhausted
    } else {
      FileActive
    },
    current: None,
    waiting_page: None,
    waiting_entry: None,
    ready: None,
  }
}

///|
fn DiskSidecarPollCursor::entry(
  self : DiskSidecarPollCursor,
  position : Int32,
) -> SidecarEntry? {
  let page = position / self.reader.width
  if self.current is Some((number, entries)) && number == page {
    return Some(entries[position % self.reader.width].entry)
  }
  for i = 0; i < self.reader.pages.length(); i = i + 1 {
    if self.reader.pages[i].0 == page {
      let found = self.reader.pages.remove(i)
      self.reader.pages.push(found)
      self.reader.cache_hits += 1
      self.current = Some(found)
      return Some(found.1[position % self.reader.width].entry)
    }
  }
  self.current = None
  self.waiting_page = Some(page)
  None
}

///|
pub fn DiskSidecarPollCursor::poll(
  self : DiskSidecarPollCursor,
) -> DiskSidecarPoll raise McapError {
  if self.state == FileFailed {
    raise Invalid("disk index cursor failed")
  }
  if self.state != FileActive {
    return DiskDone
  }
  errdefer {
    self.state = FileFailed
  }

  while self.searching {
    if self.low == self.high {
      self.position = if self.query.descending {
        self.low - 1
      } else {
        self.low
      }
      self.searching = false
      break
    }
    let middle = self.low + (self.high - self.low) / 2
    guard self.entry(middle) is Some(entry) else { return self.index_request() }
    let time = if self.query.descending {
      self.query.end_time
    } else {
      self.query.start_time
    }
    if entry.time < time || (self.query.descending && entry.time == time) {
      self.low = middle + 1
    } else {
      self.high = middle
    }
  }
  while self.remaining > 0 &&
        self.position >= 0 &&
        self.position < self.reader.count {
    guard self.entry(self.position) is Some(entry) else {
      return self.index_request()
    }
    if entry.time < self.query.start_time || entry.time > self.query.end_time {
      break
    }
    if !self.allowed.contains(entry.channel) || self.offset > 0 {
      if self.allowed.contains(entry.channel) {
        self.offset -= 1
      }
      self.position += if self.query.descending { -1 } else { 1 }
      continue
    }
    let message = match self.ready {
      Some(message) => message
      None =>
        if entry.location.inner is Some(_) &&
          self.reader.base.cached_offset == Some(entry.location.offset) &&
          self.reader.base.cached_span == entry.location.span {
          self.reader.base.load(entry)
        } else {
          self.waiting_entry = Some(entry)
          return DiskSourceRead({
            offset: entry.location.offset,
            length: entry.location.span,
          })
        }
    }
    self.ready = None
    self.waiting_entry = None
    self.position += if self.query.descending { -1 } else { 1 }
    self.remaining -= 1
    if self.remaining == 0 {
      self.state = FileExhausted
    }
    return DiskMessage(message)
  }
  self.state = FileExhausted
  DiskDone
}

///|
fn DiskSidecarPollCursor::index_request(
  self : DiskSidecarPollCursor,
) -> DiskSidecarPoll raise McapError {
  guard self.waiting_page is Some(page) else {
    raise Invalid("missing index page request")
  }
  let length = self.reader.width * 36 + 4
  DiskIndexRead({
    offset: self.reader.start + page.to_uint64() * length.to_uint64(),
    length,
  })
}

///|
pub fn DiskSidecarPollCursor::feed_index(
  self : DiskSidecarPollCursor,
  bytes : Bytes,
) -> Unit raise McapError {
  errdefer {
    self.state = FileFailed
  }

  guard self.state == FileActive &&
    self.waiting_page is Some(page) &&
    self.waiting_entry is None else {
    raise Invalid("unexpected index page feed")
  }
  self.current = Some((page, self.reader.accept_page(page, bytes)))
  self.waiting_page = None
}

///|
pub fn DiskSidecarPollCursor::feed_source(
  self : DiskSidecarPollCursor,
  bytes : Bytes,
  expanded? : Bytes,
) -> Unit raise McapError {
  errdefer {
    self.state = FileFailed
    self.ready = None
  }

  guard self.state == FileActive &&
    self.waiting_entry is Some(entry) &&
    self.waiting_page is None &&
    self.ready is None &&
    bytes.length() == entry.location.span else {
    raise Invalid("unexpected indexed source feed")
  }
  let span = entry.location
  let mirror = new_read_source(self.reader.base.source.size, (offset, length) => {
    if offset < span.offset || offset - span.offset > bytes.length().to_uint64() {
      raise Invalid("disk index feed offset mismatch")
    }
    let start = (offset - span.offset).to_int()
    if length > bytes.length() - start {
      raise Invalid("disk index feed length mismatch")
    }
    bytes[start:start + length].to_owned()
  })
  let staged = { ..self.reader.base, source: mirror, }
  self.ready = Some(staged.load(entry, expanded?))
  self.reader.base.cached = staged.cached
  self.reader.base.cached_offset = staged.cached_offset
  self.reader.base.cached_span = staged.cached_span
  self.reader.base.cached_bytes = staged.cached_bytes
  self.waiting_entry = None
}

///|
pub fn DiskSidecarPollCursor::skip_source(
  self : DiskSidecarPollCursor,
) -> Unit raise McapError {
  if self.state != FileActive ||
    self.waiting_entry is None ||
    self.waiting_page is Some(_) ||
    self.ready is Some(_) {
    raise Invalid("disk index cursor has no pending source read")
  }
  self.waiting_entry = None
  self.position += if self.query.descending { -1 } else { 1 }
}

///|
pub fn DiskSidecarPollCursor::pending_bytes(
  self : DiskSidecarPollCursor,
) -> Int32 {
  self.ready.map(fn(m) { m.data.length() + 31 }).unwrap_or(0)
}

///|
pub fn DiskSidecarPollCursor::cancel(self : DiskSidecarPollCursor) -> Unit {
  self.state = FileCancelled
  self.ready = None
  self.waiting_entry = None
  self.waiting_page = None
  self.current = None
}

///|
pub fn DiskSidecarPollCursor::seek(
  self : DiskSidecarPollCursor,
  time : UInt64,
) -> Unit raise McapError {
  if self.state == FileFailed ||
    self.state == FileCancelled ||
    time < self.original.start_time ||
    time > self.original.end_time {
    raise Invalid("invalid disk index seek or state")
  }
  self.query = {
    ..self.original,
    offset: 0,
    start_time: if self.original.descending {
      self.original.start_time
    } else {
      time
    },
    end_time: if self.original.descending {
      time
    } else {
      self.original.end_time
    },
  }
  self.low = 0
  self.high = self.reader.count
  self.searching = true
  self.position = 0
  self.offset = 0
  self.remaining = self.query.limit
  self.ready = None
  self.waiting_entry = None
  self.waiting_page = None
  self.current = None
  self.state = if self.remaining == 0 { FileExhausted } else { FileActive }
}