pub trait Sink {
  write(Self, Record) -> Unit
}

pub struct ConsoleSink {
  _dummy : Unit
}

pub fn console_sink() -> ConsoleSink {
  { _dummy: () }
}

pub impl Sink for ConsoleSink with write(self, rec) {
  ignore(self)
  println(format_text(rec))
}

pub struct ContextSink[S] {
  sink : S
  context_fields : Array[Field]
}

pub impl[S : Sink] Sink for ContextSink[S] with write(self, rec) {
  let merged = if self.context_fields.length() == 0 {
    rec.fields
  } else if rec.fields.length() == 0 {
    self.context_fields
  } else {
    self.context_fields + rec.fields
  }
  self.sink.write({ ..rec, fields: merged })
}

pub struct JsonConsoleSink {
  _dummy : Unit
}

pub fn json_console_sink() -> JsonConsoleSink {
  { _dummy: () }
}

pub impl Sink for JsonConsoleSink with write(self, rec) {
  ignore(self)
  println(format_json(rec))
}

pub struct FileSink {
  path : String
  append : Ref[Bool]
  default_append : Bool
  handle : Ref[FileHandle?]
  formatter : RecordFormatter
  auto_flush : Ref[Bool]
  default_auto_flush : Bool
  rotation : Ref[FileRotation?]
  default_rotation : FileRotation?
  open_failures : Ref[Int]
  write_failures : Ref[Int]
  flush_failures : Ref[Int]
  rotation_failures : Ref[Int]
}

pub struct FileRotation {
  max_bytes : Int
  max_backups : Int
}

pub struct FileSinkState {
  path : String
  available : Bool
  append : Bool
  auto_flush : Bool
  rotation : FileRotation?
  open_failures : Int
  write_failures : Int
  flush_failures : Int
  rotation_failures : Int
}

pub struct FileSinkPolicy {
  append : Bool
  auto_flush : Bool
  rotation : FileRotation?
}

pub fn FileSinkPolicy::new(
  append~ : Bool = true,
  auto_flush~ : Bool = true,
  rotation~ : FileRotation? = None,
) -> FileSinkPolicy {
  { append, auto_flush, rotation }
}

pub fn FileSinkState::new(
  path : String,
  available~ : Bool = false,
  append~ : Bool = true,
  auto_flush~ : Bool = true,
  rotation~ : FileRotation? = None,
  open_failures~ : Int = 0,
  write_failures~ : Int = 0,
  flush_failures~ : Int = 0,
  rotation_failures~ : Int = 0,
) -> FileSinkState {
  {
    path,
    available,
    append,
    auto_flush,
    rotation,
    open_failures,
    write_failures,
    flush_failures,
    rotation_failures,
  }
}

pub fn file_rotation(max_bytes : Int, max_backups~ : Int = 1) -> FileRotation {
  {
    max_bytes: if max_bytes <= 0 { 1 } else { max_bytes },
    max_backups: if max_backups <= 0 { 1 } else { max_backups },
  }
}

pub fn native_files_supported() -> Bool {
  native_files_supported_internal()
}

pub fn file_sink(
  path : String,
  append~ : Bool = true,
  auto_flush~ : Bool = true,
  rotation~ : FileRotation? = None,
  formatter~ : RecordFormatter = fn(rec) {
    format_text(rec)
  },
) -> FileSink {
  let handle = open_file_handle_internal(path, append)
  {
    path,
    append: Ref::new(append),
    default_append: append,
    handle: Ref::new(handle),
    formatter,
    auto_flush: Ref::new(auto_flush),
    default_auto_flush: auto_flush,
    rotation: Ref::new(rotation),
    default_rotation: rotation,
    open_failures: Ref::new(if handle is Some(_) { 0 } else { 1 }),
    write_failures: Ref::new(0),
    flush_failures: Ref::new(0),
    rotation_failures: Ref::new(0),
  }
}

pub fn FileSink::is_available(self : FileSink) -> Bool {
  self.handle.val is Some(_)
}

pub fn FileSink::flush(self : FileSink) -> Bool {
  match self.handle.val {
    None => false
    Some(handle) => {
      let ok = flush_file_handle_internal(handle)
      if !ok {
        self.flush_failures.val += 1
      }
      ok
    }
  }
}

pub fn FileSink::append_mode(self : FileSink) -> Bool {
  self.append.val
}

pub fn FileSink::set_append_mode(self : FileSink, append : Bool) -> Unit {
  self.append.val = append
}

pub fn FileSink::path(self : FileSink) -> String {
  self.path
}

pub fn FileSink::auto_flush_enabled(self : FileSink) -> Bool {
  self.auto_flush.val
}

pub fn FileSink::rotation_enabled(self : FileSink) -> Bool {
  self.rotation.val is Some(_)
}

pub fn FileSink::rotation_config(self : FileSink) -> FileRotation? {
  self.rotation.val
}

pub fn FileSink::set_auto_flush(self : FileSink, enabled : Bool) -> Unit {
  self.auto_flush.val = enabled
}

pub fn FileSink::set_policy(self : FileSink, policy : FileSinkPolicy) -> Unit {
  self.append.val = policy.append
  self.auto_flush.val = policy.auto_flush
  self.rotation.val = policy.rotation
}

pub fn FileSink::set_rotation(self : FileSink, rotation : FileRotation?) -> Unit {
  self.rotation.val = rotation
}

pub fn FileSink::clear_rotation(self : FileSink) -> Unit {
  self.rotation.val = None
}

pub fn FileSink::close(self : FileSink) -> Bool {
  match self.handle.val {
    None => false
    Some(handle) => {
      let ok = close_file_handle_internal(handle)
      self.handle.val = None
      ok
    }
  }
}

pub fn FileSink::rotation_failures(self : FileSink) -> Int {
  self.rotation_failures.val
}

pub fn FileSink::open_failures(self : FileSink) -> Int {
  self.open_failures.val
}

pub fn FileSink::write_failures(self : FileSink) -> Int {
  self.write_failures.val
}

pub fn FileSink::flush_failures(self : FileSink) -> Int {
  self.flush_failures.val
}

pub fn FileSink::reset_failure_counters(self : FileSink) -> Unit {
  self.open_failures.val = 0
  self.write_failures.val = 0
  self.flush_failures.val = 0
  self.rotation_failures.val = 0
}

pub fn FileSink::reset_policy(self : FileSink) -> Unit {
  self.append.val = self.default_append
  self.auto_flush.val = self.default_auto_flush
  self.rotation.val = self.default_rotation
}

pub fn FileSink::policy(self : FileSink) -> FileSinkPolicy {
  FileSinkPolicy::new(
    append=self.append.val,
    auto_flush=self.auto_flush.val,
    rotation=self.rotation.val,
  )
}

pub fn FileSink::default_policy(self : FileSink) -> FileSinkPolicy {
  FileSinkPolicy::new(
    append=self.default_append,
    auto_flush=self.default_auto_flush,
    rotation=self.default_rotation,
  )
}

pub fn FileSink::policy_matches_default(self : FileSink) -> Bool {
  let current = self.policy()
  let default = self.default_policy()
  current.append == default.append &&
  current.auto_flush == default.auto_flush &&
  policy_rotation_equals_internal(current.rotation, default.rotation)
}

fn policy_rotation_equals_internal(left : FileRotation?, right : FileRotation?) -> Bool {
  match (left, right) {
    (None, None) => true
    (Some(a), Some(b)) => a.max_bytes == b.max_bytes && a.max_backups == b.max_backups
    _ => false
  }
}

pub fn FileSink::state(self : FileSink) -> FileSinkState {
  {
    path: self.path,
    available: self.is_available(),
    append: self.append.val,
    auto_flush: self.auto_flush.val,
    rotation: self.rotation.val,
    open_failures: self.open_failures.val,
    write_failures: self.write_failures.val,
    flush_failures: self.flush_failures.val,
    rotation_failures: self.rotation_failures.val,
  }
}

pub fn FileSink::reopen(self : FileSink, append~ : Bool? = None) -> Bool {
  let append_mode = append.unwrap_or(self.append.val)
  self.append.val = append_mode
  match self.handle.val {
    None => ()
    Some(handle) => {
      ignore(close_file_handle_internal(handle))
      self.handle.val = None
    }
  }
  let reopened = open_file_handle_internal(self.path, append_mode)
  self.handle.val = reopened
  if reopened is Some(_) {
    true
  } else {
    self.open_failures.val += 1
    false
  }
}

pub fn FileSink::reopen_with_current_policy(self : FileSink) -> Bool {
  self.reopen()
}

pub fn FileSink::reopen_append(self : FileSink) -> Bool {
  self.reopen(append=Some(true))
}

pub fn FileSink::reopen_truncate(self : FileSink) -> Bool {
  self.reopen(append=Some(false))
}

fn rotated_file_path(path : String, index : Int) -> String {
  "\{path}.\{index}"
}

fn rotate_file_sink_internal(sink : FileSink, rotation : FileRotation) -> Bool {
  let closed = match sink.handle.val {
    None => true
    Some(handle) => {
      let ok = close_file_handle_internal(handle)
      sink.handle.val = None
      ok
    }
  }
  if !closed {
    return false
  }
  if rotation.max_backups > 0 {
    ignore(remove_file_internal(rotated_file_path(sink.path, rotation.max_backups)))
    for index = rotation.max_backups - 1; index >= 1; {
      let from_path = rotated_file_path(sink.path, index)
      let to_path = rotated_file_path(sink.path, index + 1)
      ignore(rename_file_internal(from_path, to_path))
      continue index - 1
    }
    ignore(rename_file_internal(sink.path, rotated_file_path(sink.path, 1)))
  } else {
    ignore(remove_file_internal(sink.path))
  }
  sink.handle.val = open_file_handle_internal(sink.path, false)
  sink.handle.val is Some(_)
}

fn rotate_if_needed_internal(sink : FileSink, next_line_bytes : Int) -> Bool {
  match sink.rotation.val {
    None => true
    Some(rotation) => match sink.handle.val {
      None => false
      Some(handle) => {
        let size = file_size_internal(handle)
        if size + next_line_bytes <= rotation.max_bytes {
          true
        } else {
          let rotated = rotate_file_sink_internal(sink, rotation)
          if !rotated {
            sink.rotation_failures.val += 1
          }
          rotated
        }
      }
    }
  }
}

pub impl Sink for FileSink with write(self, rec) {
  match self.handle.val {
    None => {
      self.write_failures.val += 1
    }
    Some(_) => {
      let line = "\{(self.formatter)(rec)}\n"
      let can_write = rotate_if_needed_internal(self, string_byte_length_internal(line))
      if can_write {
        match self.handle.val {
          None => {
            self.write_failures.val += 1
          }
          Some(active) => {
            let wrote = write_file_handle_internal(active, line)
            if wrote {
              if self.auto_flush.val {
                let flushed = flush_file_handle_internal(active)
                if !flushed {
                  self.flush_failures.val += 1
                }
              }
            } else {
              self.write_failures.val += 1
            }
          }
        }
      } else {
        self.write_failures.val += 1
      }
    }
  }
}

pub struct FormattedConsoleSink {
  formatter : RecordFormatter
}

pub fn formatted_console_sink(formatter : RecordFormatter) -> FormattedConsoleSink {
  { formatter, }
}

pub fn text_console_sink(formatter : TextFormatter) -> FormattedConsoleSink {
  formatted_console_sink(fn(rec) {
    format_text(rec, formatter=formatter)
  })
}

pub impl Sink for FormattedConsoleSink with write(self, rec) {
  println((self.formatter)(rec))
}

pub struct FormattedCallbackSink {
  formatter : RecordFormatter
  callback : (String) -> Unit
}

pub fn formatted_callback_sink(
  formatter : RecordFormatter,
  callback : (String) -> Unit,
) -> FormattedCallbackSink {
  { formatter, callback }
}

pub fn text_callback_sink(
  formatter : TextFormatter,
  callback : (String) -> Unit,
) -> FormattedCallbackSink {
  formatted_callback_sink(fn(rec) {
    format_text(rec, formatter=formatter)
  }, callback)
}

pub impl Sink for FormattedCallbackSink with write(self, rec) {
  (self.callback)((self.formatter)(rec))
}

pub struct FanoutSink[A, B] {
  left : A
  right : B
}

pub fn[A, B] fanout_sink(left : A, right : B) -> FanoutSink[A, B] {
  { left, right }
}

pub impl[A : Sink, B : Sink] Sink for FanoutSink[A, B] with write(self, rec) {
  self.left.write(rec)
  self.right.write({ ..rec })
}

pub struct SplitSink[A, B] {
  left : A
  right : B
  predicate : (Record) -> Bool
}

pub fn[A, B] split_sink(left : A, right : B, predicate : (Record) -> Bool) -> SplitSink[A, B] {
  { left, right, predicate }
}

pub fn[A, B] split_by_level(
  left : A,
  right : B,
  min_level~ : Level = Level::Warn,
) -> SplitSink[A, B] {
  split_sink(left, right, fn(rec) {
    rec.level.enabled(min_level)
  })
}

pub impl[A : Sink, B : Sink] Sink for SplitSink[A, B] with write(self, rec) {
  if (self.predicate)(rec) {
    self.left.write(rec)
  } else {
    self.right.write(rec)
  }
}

pub struct CallbackSink {
  callback : (Record) -> Unit
}

pub fn callback_sink(callback : (Record) -> Unit) -> CallbackSink {
  { callback, }
}

pub impl Sink for CallbackSink with write(self, rec) {
  (self.callback)(rec)
}

pub struct BufferedSink[S] {
  sink : S
  buffer : Ref[Array[Record]]
  flush_limit : Int
}

pub fn[S] buffered_sink(sink : S, flush_limit~ : Int = 1) -> BufferedSink[S] {
  let actual_limit = if flush_limit <= 0 { 1 } else { flush_limit }
  { sink, buffer: Ref::new([]), flush_limit: actual_limit }
}

pub fn[S] BufferedSink::pending_count(self : BufferedSink[S]) -> Int {
  self.buffer.val.length()
}

pub fn[S : Sink] BufferedSink::flush(self : BufferedSink[S]) -> Unit {
  if self.buffer.val.length() == 0 {
    ()
  } else {
    let pending = self.buffer.val
    self.buffer.val = []
    for rec in pending {
      self.sink.write(rec)
    }
  }
}

pub impl[S : Sink] Sink for BufferedSink[S] with write(self, rec) {
  self.buffer.val.push(rec)
  if self.buffer.val.length() >= self.flush_limit {
    self.flush()
  }
}

pub(all) enum QueueOverflowPolicy {
  DropNewest
  DropOldest
}

pub struct QueuedSink[S] {
  sink : S
  queue : @queue.Queue[Record]
  max_pending : Int
  overflow : QueueOverflowPolicy
  dropped_count : Ref[Int]
}

pub fn[S] queued_sink(
  sink : S,
  max_pending~ : Int = 0,
  overflow~ : QueueOverflowPolicy = QueueOverflowPolicy::DropNewest,
) -> QueuedSink[S] {
  {
    sink,
    queue: @queue.Queue::new(),
    max_pending,
    overflow,
    dropped_count: Ref::new(0),
  }
}

pub fn[S] QueuedSink::pending_count(self : QueuedSink[S]) -> Int {
  self.queue.length()
}

pub fn[S] QueuedSink::dropped_count(self : QueuedSink[S]) -> Int {
  self.dropped_count.val
}

pub fn[S : Sink] QueuedSink::drain(self : QueuedSink[S], max_items~ : Int = -1) -> Int {
  if max_items == 0 {
    return 0
  }
  let limit = if max_items < 0 { self.pending_count() } else { max_items }
  for drained = 0; drained < limit; {
    match self.queue.pop() {
      None => break drained
      Some(rec) => {
        self.sink.write(rec)
        continue drained + 1
      }
    }
  } nobreak {
    limit
  }
}

pub fn[S : Sink] QueuedSink::flush(self : QueuedSink[S]) -> Int {
  self.drain()
}

pub impl[S] Sink for QueuedSink[S] with write(self, rec) {
  let full = self.max_pending > 0 && self.pending_count() >= self.max_pending
  if !full {
    self.queue.push(rec)
  } else {
    self.dropped_count.val += 1
    match self.overflow {
      QueueOverflowPolicy::DropNewest => ()
      QueueOverflowPolicy::DropOldest => {
        ignore(self.queue.pop())
        self.queue.push(rec)
      }
    }
  }
}

pub struct FilterSink[S] {
  sink : S
  predicate : (Record) -> Bool
}

pub fn[S] filter_sink(sink : S, predicate : (Record) -> Bool) -> FilterSink[S] {
  { sink, predicate }
}

pub impl[S : Sink] Sink for FilterSink[S] with write(self, rec) {
  if (self.predicate)(rec) {
    self.sink.write(rec)
  }
}

pub struct PatchSink[S] {
  sink : S
  patch : RecordPatch
}

pub fn[S] patch_sink(sink : S, patch : RecordPatch) -> PatchSink[S] {
  { sink, patch }
}

pub impl[S : Sink] Sink for PatchSink[S] with write(self, rec) {
  self.sink.write((self.patch)(rec))
}