///|
pub fn default_stream_filter() -> StreamFilter {
  {
    names: [],
    first_id: None,
    last_id: None,
    minimum_length: 0,
    maximum_length: 0,
    maximum_records: 0,
  }
}

///|
pub fn StreamFilter::named(self : StreamFilter, name : String) -> StreamFilter {
  let names = self.names.copy()
  let normalized = name.to_upper()
  if normalized.length() > 0 && !names.contains(normalized) {
    names.push(normalized)
  }
  { ..self, names, }
}

///|
pub fn StreamFilter::between_ids(
  self : StreamFilter,
  first : Int?,
  last : Int?,
) -> StreamFilter {
  { ..self, first_id: first, last_id: last }
}

///|
pub fn StreamFilter::with_minimum_length(
  self : StreamFilter,
  length : Int,
) -> StreamFilter {
  { ..self, minimum_length: if length < 0 { 0 } else { length } }
}

///|
pub fn StreamFilter::with_maximum_length(
  self : StreamFilter,
  length : Int,
) -> StreamFilter {
  { ..self, maximum_length: if length < 0 { 0 } else { length } }
}

///|
pub fn StreamFilter::limiting(self : StreamFilter, count : Int) -> StreamFilter {
  { ..self, maximum_records: if count < 0 { 0 } else { count } }
}

///|
pub fn StreamFilter::is_unbounded(self : StreamFilter) -> Bool {
  self.maximum_records == 0 &&
  self.maximum_length == 0 &&
  self.names.length() == 0 &&
  self.first_id is None &&
  self.last_id is None
}

///|
fn stream_filter_name_matches(
  filter : StreamFilter,
  record : StepRecord,
) -> Bool {
  filter.names.length() == 0 || filter.names.contains(record.name.to_upper())
}

///|
fn stream_filter_id_matches(filter : StreamFilter, record : StepRecord) -> Bool {
  let first_ok = match filter.first_id {
    Some(value) => record.id >= value
    None => true
  }
  let last_ok = match filter.last_id {
    Some(value) => record.id <= value
    None => true
  }
  first_ok && last_ok
}

///|
fn stream_filter_length_matches(
  filter : StreamFilter,
  record : StepRecord,
) -> Bool {
  let length = record.length()
  length >= filter.minimum_length &&
  (filter.maximum_length == 0 || length <= filter.maximum_length)
}

///|
fn stream_filter_matches(filter : StreamFilter, record : StepRecord) -> Bool {
  stream_filter_name_matches(filter, record) &&
  stream_filter_id_matches(filter, record) &&
  stream_filter_length_matches(filter, record)
}

///|
/// Select raw records without parsing values or constructing a model.
pub fn stream_select(
  source : String,
  filter? : StreamFilter = default_stream_filter(),
) -> StreamSelection {
  let records = scan_step_records(source)
  let selected : Array[StepRecord] = []
  let limit = if filter.maximum_records <= 0 {
    2147483647
  } else {
    filter.maximum_records
  }
  let mut examined = 0
  let mut truncated = false
  for record in records {
    examined += 1
    if selected.length() >= limit {
      truncated = true
      break
    }
    if stream_filter_matches(filter, record) {
      selected.push(record)
    }
  }
  {
    records: selected,
    examined,
    selected: selected.length(),
    truncated,
    first_offset: selected
    .get(0)
    .map(record => record.start_offset)
    .unwrap_or(0),
    last_offset: selected
    .get(selected.length() - 1)
    .map(record => record.end_offset)
    .unwrap_or(0),
  }
}

///|
pub fn stream_select_named(source : String, name : String) -> StreamSelection {
  stream_select(source, filter=default_stream_filter().named(name))
}

///|
pub fn stream_select_id_range(
  source : String,
  first : Int,
  last : Int,
) -> StreamSelection {
  stream_select(
    source,
    filter=default_stream_filter().between_ids(Some(first), Some(last)),
  )
}

///|
pub fn StreamSelection::is_empty(self : StreamSelection) -> Bool {
  self.records.length() == 0
}

///|
pub fn StreamSelection::first(self : StreamSelection) -> StepRecord? {
  self.records.get(0)
}

///|
pub fn StreamSelection::last(self : StreamSelection) -> StepRecord? {
  self.records.get(self.records.length() - 1)
}

///|
pub fn StreamSelection::names(self : StreamSelection) -> Array[String] {
  self.records.map(record => record.name)
}

///|
pub fn StreamSelection::ids(self : StreamSelection) -> Array[Int] {
  self.records.map(record => record.id)
}

///|
pub fn StreamSelection::headline(self : StreamSelection) -> String {
  "selected=" +
  self.selected.to_string() +
  " examined=" +
  self.examined.to_string() +
  " truncated=" +
  self.truncated.to_string() +
  " offsets=" +
  self.first_offset.to_string() +
  ".." +
  self.last_offset.to_string()
}

///|
pub fn StreamSelection::parse_entities(
  self : StreamSelection,
) -> Array[Entity] raise ParseError {
  self.records.fold(init=Array::new(), (result, record) => {
    result.push(parse_step_record(record))
    result
  })
}

///|
fn stream_batch_from_records(
  records : Array[StepRecord],
  sequence : Int,
  complete : Bool,
) -> StreamBatch {
  {
    sequence,
    start_offset: records.get(0).map(record => record.start_offset).unwrap_or(0),
    end_offset: records
    .get(records.length() - 1)
    .map(record => record.end_offset)
    .unwrap_or(0),
    records,
    complete,
  }
}

///|
/// Partition selected raw records into deterministic source-order batches.
pub fn stream_batches(
  source : String,
  batch_size : Int,
  filter? : StreamFilter = default_stream_filter(),
) -> Array[StreamBatch] {
  let selection = stream_select(source, filter~)
  let safe_size = if batch_size <= 0 { 1 } else { batch_size }
  let batches : Array[StreamBatch] = []
  let mut offset = 0
  let mut sequence = 0
  while offset < selection.records.length() {
    let end = if offset + safe_size > selection.records.length() {
      selection.records.length()
    } else {
      offset + safe_size
    }
    let records = selection.records[offset:end].to_owned()
    batches.push(
      stream_batch_from_records(
        records,
        sequence,
        end == selection.records.length() && !selection.truncated,
      ),
    )
    sequence += 1
    offset = end
  }
  if batches.length() == 0 && selection.truncated {
    batches.push(stream_batch_from_records([], 0, false))
  }
  batches
}

///|
pub fn StreamBatch::record_count(self : StreamBatch) -> Int {
  self.records.length()
}

///|
pub fn StreamBatch::is_empty(self : StreamBatch) -> Bool {
  self.records.length() == 0
}

///|
pub fn StreamBatch::first_id(self : StreamBatch) -> Int? {
  self.records.get(0).map(record => record.id)
}

///|
pub fn StreamBatch::last_id(self : StreamBatch) -> Int? {
  self.records.get(self.records.length() - 1).map(record => record.id)
}

///|
pub fn StreamBatch::parse_entities(
  self : StreamBatch,
) -> Array[Entity] raise ParseError {
  self.records.fold(init=Array::new(), (result, record) => {
    result.push(parse_step_record(record))
    result
  })
}

///|
pub fn StreamBatch::headline(self : StreamBatch) -> String {
  "batch=" +
  self.sequence.to_string() +
  " records=" +
  self.records.length().to_string() +
  " offsets=" +
  self.start_offset.to_string() +
  ".." +
  self.end_offset.to_string() +
  " complete=" +
  self.complete.to_string()
}

///|
pub fn stream_batch_summary(batches : Array[StreamBatch]) -> Array[String] {
  batches.map(batch => batch.headline())
}

///|
pub fn stream_batch_count(
  source : String,
  batch_size : Int,
  filter? : StreamFilter = default_stream_filter(),
) -> Int {
  stream_batches(source, batch_size, filter~).length()
}

///|
pub fn stream_records_in_batches(
  source : String,
  batch_size : Int,
  filter? : StreamFilter = default_stream_filter(),
) -> Array[StepRecord] {
  stream_batches(source, batch_size, filter~).fold(init=Array::new(), (
    result,
    batch,
  ) => {
    result.append(batch.records)
    result
  })
}

///|
pub fn stream_filter_names(filter : StreamFilter) -> Array[String] {
  filter.names.copy()
}

///|
pub fn StreamFilter::describe(self : StreamFilter) -> String {
  "names=" +
  self.names.join(",") +
  " ids=" +
  self.first_id.map(value => value.to_string()).unwrap_or("-") +
  ".." +
  self.last_id.map(value => value.to_string()).unwrap_or("-") +
  " length=" +
  self.minimum_length.to_string() +
  ".." +
  self.maximum_length.to_string() +
  " limit=" +
  self.maximum_records.to_string()
}

///|
pub fn StreamFilter::matches(self : StreamFilter, record : StepRecord) -> Bool {
  stream_filter_matches(self, record)
}

///|
pub fn StepRecord::source_window(
  self : StepRecord,
  source : String,
) -> SourceWindow {
  source_window(source, self.start_offset, self.end_offset)
}

///|
pub fn StepRecord::summary(self : StepRecord) -> String {
  "#" +
  self.id.to_string() +
  " " +
  self.name +
  " line=" +
  self.line.to_string() +
  " length=" +
  self.length().to_string()
}

///|
pub fn StepRecord::contains_reference(self : StepRecord, id : Int) -> Bool {
  self.reference_count() > 0 &&
  references(parse_step_value_for_record(self)).contains(id)
}

///|
fn parse_step_value_for_record(record : StepRecord) -> Value {
  let entity = try! parse_step_record(record)
  List(entity.arguments)
}

///|
pub fn stream_record_names(source : String) -> Array[String] {
  scan_step_records(source).map(record => record.name)
}

///|
pub fn stream_record_ids(source : String) -> Array[Int] {
  scan_step_records(source).map(record => record.id)
}

///|
pub fn stream_records_with_references(source : String) -> Array[StepRecord] {
  scan_step_records(source).filter(record => record.reference_count() > 0)
}

///|
pub fn stream_records_longer_than(
  source : String,
  length : Int,
) -> Array[StepRecord] {
  let minimum = if length < 0 { 0 } else { length }
  scan_step_records(source).filter(record => record.length() >= minimum)
}

///|
pub fn stream_group_by_name(source : String) -> Map[String, Int] {
  scan_step_records(source).fold(init=Map([]), (result, record) => {
    result.update_or_default(record.name.to_upper(), 1, count => count + 1)
    result
  })
}

///|
pub fn stream_group_by_line(source : String, line_width : Int) -> Map[Int, Int] {
  let width = if line_width <= 0 { 1 } else { line_width }
  scan_step_records(source).fold(init=Map([]), (result, record) => {
    let key = record.line / width
    result.update_or_default(key, 1, count => count + 1)
    result
  })
}

///|
pub fn stream_reference_histogram(source : String) -> Map[Int, Int] {
  scan_step_records(source).fold(init=Map([]), (result, record) => {
    let entity = try! parse_step_record(record)
    for id in entity_references(entity) {
      result.update_or_default(id, 1, count => count + 1)
    }
    result
  })
}

///|
pub fn stream_stats_for_filter(
  source : String,
  filter? : StreamFilter = default_stream_filter(),
) -> StreamStats {
  let selection = stream_select(source, filter~)
  let records = selection.records
  {
    source_length: source.length(),
    record_count: records.length(),
    reference_count: records.fold(init=0, (result, record) => {
      result + record.reference_count()
    }),
    max_record_length: records.fold(init=0, (result, record) => {
      if record.length() > result {
        record.length()
      } else {
        result
      }
    }),
    comment_count: 0,
    line_count: records.fold(init=0, (result, record) => {
      if record.line > result {
        record.line
      } else {
        result
      }
    }),
    schema: stream_schema(source),
  }
}

///|
pub fn StreamSelection::to_json(self : StreamSelection) -> String {
  "{\"examined\":" +
  self.examined.to_string() +
  ",\"selected\":" +
  self.selected.to_string() +
  ",\"truncated\":" +
  self.truncated.to_string() +
  ",\"first_offset\":" +
  self.first_offset.to_string() +
  ",\"last_offset\":" +
  self.last_offset.to_string() +
  ",\"records\":[" +
  self.records
  .map(record => {
    "{\"id\":" +
    record.id.to_string() +
    ",\"name\":\"" +
    record.name +
    "\",\"line\":" +
    record.line.to_string() +
    ",\"start\":" +
    record.start_offset.to_string() +
    ",\"end\":" +
    record.end_offset.to_string() +
    "}"
  })
  .join(",") +
  "]}"
}

///|
pub fn StreamBatch::to_json(self : StreamBatch) -> String {
  "{\"sequence\":" +
  self.sequence.to_string() +
  ",\"start_offset\":" +
  self.start_offset.to_string() +
  ",\"end_offset\":" +
  self.end_offset.to_string() +
  ",\"complete\":" +
  self.complete.to_string() +
  ",\"records\":[" +
  self.records.map(record => record.id.to_string()).join(",") +
  "]}"
}

///|
pub fn StreamFilter::with_name_set(
  self : StreamFilter,
  names : Array[String],
) -> StreamFilter {
  names.fold(init=self, (result, name) => result.named(name))
}

///|
pub fn stream_select_products(
  source : String,
  filter? : StreamFilter = default_stream_filter(),
) -> Array[Entity] raise ParseError {
  stream_select(source, filter~).parse_entities()
}

///|
pub fn stream_select_relationships(source : String) -> Array[StepRecord] {
  stream_select(source, filter=default_stream_filter().named("IFCREL")).records
}

///|
pub fn stream_select_geometry(source : String) -> Array[StepRecord] {
  let records = scan_step_records(source)
  records.filter(record => {
    let name = record.name.to_upper()
    name.contains("SOLID") ||
    name.contains("BREP") ||
    name.contains("FACESET") ||
    name.contains("MESH")
  })
}

///|
pub fn stream_selection_offsets(selection : StreamSelection) -> (Int, Int) {
  (selection.first_offset, selection.last_offset)
}