// Copyright 2026 International Digital Economy Academy
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

///|
type Stream

///|
pub fn Stream::to_handle(self : Stream) -> Handle = "%identity"

///|
pub impl ToHandle for Stream with to_handle(self) {
  self.to_handle()
}

///|
pub impl ToHandle for Stream with of_handle(self) = "%identity"

///|
#owned(stream)
extern "c" fn uv_read_start(
  stream : Stream,
  alloc_cb : (Handle, UInt64, @c.Pointer[Int], @c.Pointer[Int]) -> Bytes,
  read_cb : (Stream, Int64, Bytes?, Int, Int) -> Unit,
) -> Int = "moonbit_uv_read_start"

///|
pub fn Stream::read_start(
  self : Stream,
  alloc_cb : (Handle, Int) -> BytesView,
  read_cb : (Stream, Int, BytesView) -> Unit,
  error_cb : (Stream, Errno) -> Unit,
) -> Unit raise Errno {
  fn uv_alloc_cb(
    handle : Handle,
    suggested_size : UInt64,
    buf_offset : @c.Pointer[Int],
    buf_length : @c.Pointer[Int],
  ) -> Bytes {
    let bytes = alloc_cb(handle, suggested_size.to_int())
    buf_offset[0] = bytes.start_offset()
    buf_length[0] = bytes.length()
    return bytes.data()
  }

  fn uv_read_cb(
    stream : Stream,
    count : Int64,
    buf_data : Bytes?,
    buf_offset : Int,
    buf_length : Int,
  ) -> Unit {
    if count < 0 {
      error_cb(stream, Errno::of_int(count.to_int()))
    } else {
      read_cb(stream, count.to_int(), buf_data.unwrap()[buf_offset:buf_length])
    }
  }

  let result = uv_read_start(self, uv_alloc_cb, uv_read_cb)
  if result < 0 {
    raise Errno::of_int(result)
  }
}

///|
/// Starts reading from a stream with custom allocation and data handling
/// callbacks.
///
/// Parameters:
///
/// * `stream` : The stream to read from.
/// * `alloc_cb` : Callback function to allocate buffer space for incoming data.
/// Takes the stream and suggested buffer size, returns a bytes view.
/// * `read_cb` : Callback function to handle successfully read data. Takes the
/// stream, number of bytes read, and the data as a bytes view.
/// * `error_cb` : Callback function to handle read errors. Takes the stream and
/// the error code.
///
/// Throws an error of type `Errno` if the operation fails to start.
#deprecated("Use Stream::read_start() instead")
pub fn[Stream : ToStream] read_start(
  stream : Stream,
  alloc_cb : (Stream, Int) -> BytesView,
  read_cb : (Stream, Int, BytesView) -> Unit,
  error_cb : (Stream, Errno) -> Unit,
) -> Unit raise Errno {
  stream.read_start(alloc_cb, read_cb, error_cb)
}

///|
#owned(stream)
extern "c" fn uv_read_stop(stream : Stream) -> Int = "moonbit_uv_read_stop"

///|
pub fn Stream::read_stop(self : Stream) -> Unit raise Errno {
  let result = uv_read_stop(self)
  if result < 0 {
    raise Errno::of_int(result)
  }
}

///|
#deprecated("Use Stream::read_stop() instead")
pub fn[Stream : ToStream] read_stop(stream : Stream) -> Unit raise Errno {
  stream.read_stop()
}

///|
type Write

///|
pub impl ToReq for Write with to_req(self : Write) -> Req = "%identity"

///|
extern "c" fn uv_write_make() -> Write = "moonbit_uv_write_make"

///|
#owned(write, handle, bufs_base, bufs_offset, bufs_length)
extern "c" fn uv_write(
  write : Write,
  handle : Stream,
  bufs_base : FixedArray[Bytes],
  bufs_offset : FixedArray[Int],
  bufs_length : FixedArray[Int],
  cb : (Write, Int) -> Unit,
) -> Int = "moonbit_uv_write"

///|
#owned(write, handle, bufs_base, bufs_offset, bufs_length)
extern "c" fn uv_try_write(
  write : Write,
  handle : Stream,
  bufs_base : FixedArray[Bytes],
  bufs_offset : FixedArray[Int],
  bufs_length : FixedArray[Int],
  cb : (Write, Int) -> Unit,
) -> Int = "moonbit_uv_try_write"

///|
pub fn Stream::write(
  self : Stream,
  bufs : Array[BytesView],
  write_cb : () -> Unit,
  error_cb : (Errno) -> Unit,
) -> Write raise Errno {
  fn cb(_ : Write, status : Int) {
    if status < 0 {
      error_cb(Errno::of_int(status))
    } else {
      write_cb()
    }
  }

  let req = uv_write_make()
  let bufs_size = bufs.length()
  let bufs_base : FixedArray[Bytes] = FixedArray::make(bufs_size, [])
  let bufs_offset = FixedArray::make(bufs_size, 0)
  let bufs_length = FixedArray::make(bufs_size, 0)
  for i in 0.. Unit,
  error_cb : (Errno) -> Unit,
) -> Write raise Errno {
  fn cb(_ : Write, status : Int) {
    if status < 0 {
      error_cb(Errno::of_int(status))
    } else {
      write_cb()
    }
  }

  let req = uv_write_make()
  let bufs_size = bufs.length()
  let bufs_base : FixedArray[Bytes] = FixedArray::make(bufs_size, [])
  let bufs_offset = FixedArray::make(bufs_size, 0)
  let bufs_length = FixedArray::make(bufs_size, 0)
  for i in 0.. Unit,
  error_cb : (Errno) -> Unit,
) -> Write raise Errno {
  stream.write(bufs, write_cb, error_cb)
}

///|
#owned(write, handle, bufs_base, bufs_offset, bufs_length, send_handle)
extern "c" fn uv_write2(
  write : Write,
  handle : Stream,
  bufs_base : FixedArray[Bytes],
  bufs_offset : FixedArray[Int],
  bufs_length : FixedArray[Int],
  send_handle : Stream,
  cb : (Write, Int) -> Unit,
) -> Int = "moonbit_uv_write2"

///|
#owned(write, handle, bufs_base, bufs_offset, bufs_length, send_handle)
extern "c" fn uv_try_write2(
  write : Write,
  handle : Stream,
  bufs_base : FixedArray[Bytes],
  bufs_offset : FixedArray[Int],
  bufs_length : FixedArray[Int],
  send_handle : Stream,
  cb : (Write, Int) -> Unit,
) -> Int = "moonbit_uv_try_write2"

///|
pub fn Stream::write2(
  self : Stream,
  bufs : Array[BytesView],
  send_handle : Stream,
  write_cb : () -> Unit,
  error_cb : (Errno) -> Unit,
) -> Write raise Errno {
  fn cb(_ : Write, status : Int) {
    if status < 0 {
      error_cb(Errno::of_int(status))
    } else {
      write_cb()
    }
  }

  let req = uv_write_make()
  let bufs_size = bufs.length()
  let bufs_base : FixedArray[Bytes] = FixedArray::make(bufs_size, [])
  let bufs_offset = FixedArray::make(bufs_size, 0)
  let bufs_length = FixedArray::make(bufs_size, 0)
  for i in 0.. Unit,
  error_cb : (Errno) -> Unit,
) -> Write raise Errno {
  fn cb(_ : Write, status : Int) {
    if status < 0 {
      error_cb(Errno::of_int(status))
    } else {
      write_cb()
    }
  }

  let req = uv_write_make()
  let bufs_size = bufs.length()
  let bufs_base : FixedArray[Bytes] = FixedArray::make(bufs_size, [])
  let bufs_offset = FixedArray::make(bufs_size, 0)
  let bufs_length = FixedArray::make(bufs_size, 0)
  for i in 0.. Stream
  of_stream(Stream) -> Self
  read_start(
    Self,
    (Self, Int) -> BytesView,
    (Self, Int, BytesView) -> Unit,
    (Self, Errno) -> Unit,
  ) -> Unit raise Errno = _
  read_stop(Self) -> Unit raise Errno = _
  write(Self, Array[BytesView], () -> Unit, (Errno) -> Unit) -> Write raise Errno = _
  try_write(Self, Array[BytesView], () -> Unit, (Errno) -> Unit) -> Write raise Errno = _
  listen(Self, Int, (Self) -> Unit, (Self, Errno) -> Unit) -> Unit raise Errno = _
  is_readable(Self) -> Bool = _
  is_writable(Self) -> Bool = _
  shutdown(Self, () -> Unit, (Errno) -> Unit) -> Shutdown raise Errno = _
  set_blocking(Self, Bool) -> Unit raise Errno = _
  write_queue_size(Self) -> UInt64 = _
}

///|
pub impl ToStream for Stream with to_stream(self) = "%identity"

///|
pub impl ToStream for Stream with of_stream(self) = "%identity"

///|
impl ToStream with read_start(self, alloc_cb, read_cb, error_cb) {
  // We are wrapping callbacks here, for two reasons:
  // 1. We need to downcast the `Handle`/`Stream` to concrete types.
  // 2. (Mutually) recursive indirect call will create a reference loop,
  //    and RC will fail to collect it. Therefore, instead of directly
  //    passing `alloc_cb`, `read_cb` and `error_cb` to `read_start()`,
  //    we wrap them into local functions, so that compiler can perform
  //    capture analysis and will not create closures that refer to each
  //    other.
  fn handle_alloc_cb(handle : Handle, suggested_size : Int) {
    alloc_cb(ToHandle::of_handle(handle), suggested_size)
  }

  fn stream_read_cb(stream : Stream, count : Int, bytes : BytesView) {
    read_cb(ToStream::of_stream(stream), count, bytes)
  }

  fn stream_error_cb(stream : Stream, errno : Errno) {
    error_cb(ToStream::of_stream(stream), errno)
  }

  self.to_stream().read_start(handle_alloc_cb, stream_read_cb, stream_error_cb)
}

///|
impl ToStream with read_stop(self) {
  self.to_stream().read_stop()
}

///|
impl ToStream with write(self, bufs, write_cb, error_cb) {
  self.to_stream().write(bufs, write_cb, error_cb)
}

///|
impl ToStream with try_write(self, bufs, write_cb, error_cb) {
  self.to_stream().try_write(bufs, write_cb, error_cb)
}

///|
impl ToStream with set_blocking(self, blocking) {
  let status = uv_stream_set_blocking(self.to_stream(), blocking)
  if status < 0 {
    raise Errno::of_int(status)
  }
}

///|
#borrow(stream)
extern "c" fn uv_is_readable(stream : Stream) -> Bool = "moonbit_uv_is_readable"

///|
#borrow(stream)
extern "c" fn uv_is_writable(stream : Stream) -> Bool = "moonbit_uv_is_writable"

///|
pub fn Stream::is_readable(self : Stream) -> Bool {
  uv_is_readable(self)
}

///|
pub fn Stream::is_writable(self : Stream) -> Bool {
  uv_is_writable(self)
}

///|
#deprecated("Use Stream::is_readable() instead")
pub fn[Stream : ToStream] is_readable(stream : Stream) -> Bool {
  stream.is_readable()
}

///|
#deprecated("Use Stream::is_writable() instead")
pub fn[Stream : ToStream] is_writable(stream : Stream) -> Bool {
  stream.is_writable()
}

///|
impl ToStream with is_readable(self) {
  self.to_stream().is_readable()
}

///|
impl ToStream with is_writable(self) {
  self.to_stream().is_writable()
}

///|
type Shutdown

///|
pub impl ToReq for Shutdown with to_req(self : Shutdown) -> Req = "%identity"

///|
extern "c" fn uv_shutdown_make() -> Shutdown = "moonbit_uv_shutdown_make"

///|
#owned(shutdown, handle)
extern "c" fn uv_shutdown(
  shutdown : Shutdown,
  handle : Stream,
  cb : (Shutdown, Int) -> Unit,
) -> Int = "moonbit_uv_shutdown"

///|
pub fn Stream::shutdown(
  self : Stream,
  shutdown_cb : () -> Unit,
  error_cb : (Errno) -> Unit,
) -> Shutdown raise Errno {
  fn cb(_ : Shutdown, status : Int) {
    if status < 0 {
      error_cb(Errno::of_int(status))
    } else {
      shutdown_cb()
    }
  }

  let req = uv_shutdown_make()
  let result = uv_shutdown(req, self, cb)
  if result < 0 {
    raise Errno::of_int(result)
  }
  return req
}

///|
#deprecated("Use Stream::shutdown() instead")
pub fn[Stream : ToStream] shutdown(
  stream : Stream,
  shutdown_cb : () -> Unit,
  error_cb : (Errno) -> Unit,
) -> Shutdown raise Errno {
  stream.shutdown(shutdown_cb, error_cb)
}

///|
impl ToStream with shutdown(self, shutdown_cb, error_cb) {
  self.to_stream().shutdown(shutdown_cb, error_cb)
}

///|
#owned(stream)
extern "c" fn uv_listen(
  stream : Stream,
  backlog : Int,
  cb : (Stream, Int) -> Unit,
) -> Int = "moonbit_uv_listen"

///|
pub fn[Stream : ToStream] listen(
  stream : Stream,
  backlog : Int,
  connection_cb : (Stream) -> Unit,
  error_cb : (Stream, Errno) -> Unit,
) -> Unit raise Errno {
  stream.listen(backlog, connection_cb, error_cb)
}

///|
impl ToStream with listen(
  self : Self,
  backlog : Int,
  connection_cb : (Self) -> Unit,
  error_cb : (Self, Errno) -> Unit,
) {
  fn uv_cb(self : Stream, status : Int) {
    if status < 0 {
      error_cb(ToStream::of_stream(self), Errno::of_int(status))
    } else {
      connection_cb(ToStream::of_stream(self))
    }
  }

  let status = uv_listen(self.to_stream(), backlog, uv_cb)
  if status < 0 {
    raise Errno::of_int(status)
  }
}

///|
#owned(server, client)
extern "c" fn uv_accept(server : Stream, client : Stream) -> Int = "moonbit_uv_accept"

///|
pub fn[Server : ToStream, Client : ToStream] accept(
  server : Server,
  client : Client,
) -> Unit raise Errno {
  let status = uv_accept(server.to_stream(), client.to_stream())
  if status < 0 {
    raise Errno::of_int(status)
  }
}

///|
struct Connect(Bytes)

///|
pub impl ToReq for Connect with to_req(self : Connect) -> Req = "%identity"

///|
extern "c" fn uv_connect_make() -> Connect = "moonbit_uv_connect_make"

///|
#owned(stream)
extern "c" fn uv_stream_set_blocking(stream : Stream, blocking : Bool) -> Int = "moonbit_uv_stream_set_blocking"

///|
#owned(stream)
extern "c" fn uv_stream_get_write_queue_size(stream : Stream) -> UInt64 = "moonbit_uv_stream_get_write_queue_size"

///|
impl ToStream with write_queue_size(self) {
  uv_stream_get_write_queue_size(self.to_stream())
}