// Copyright 2026 PaiGack
// Licensed under the Apache License, Version 2.0.
// Ported from jlaffaye/ftp (ISC License), see LICENSE-THIRD-PARTY.

// Layer: IO — uses the `moonbitlang/async` stack from the root moon.pkg.

// The control connection and the request/response framing on top of it.
//
// This file merges `control`, `command` and `response` (that split
// had already lost its files before this change: the block comments below are
// the only trace of it). A control connection is only meaningful together with
// the two operations performed on it (write a command, read a reply), and the
// reply type exists solely as the return of `read_response`. None of the three
// pieces was ever used apart from the others.
//
// The traffic log used to sit at the bottom of this file. The *switch* still
// does -- `Control` owns the `DebugLog?` and installs it -- but the recorder
// and its decorators moved to `debug.mbt`, which knows nothing about `Control`.

///|
/// An FTP control connection: a line buffered reader/writer pair over the
/// underlying socket.
///
/// The reader and the writer are kept as trait objects so the same code drives
/// a plain `Tcp` socket and a `Tls` connection after `AUTH TLS`.
pub struct Control {
  mut reader : &@io.Reader
  mut writer : &@io.Writer
  /// Address of the peer we are talking to; the `PASV` SSRF guard trusts this
  /// one, not the address the server advertises.
  host : String
  /// Deadline applied to every reply read, in milliseconds; `0` disables it.
  ///
  /// A read with no deadline is not merely missing a diagnostic: it is the
  /// shape the async runtime reports as a *deadlock*. `check_dead_lock` fires
  /// as soon as every task is parked, and a suspended socket read counts as
  /// parked — so a server that takes a few milliseconds longer than the client
  /// expects (a real ftpd answers `150` only *after* it accepted the data
  /// connection, which is one scheduling round where the client is already
  /// waiting) aborts the process with SIGABRT instead of waiting. The report
  /// then names whichever unrelated task happened to be parked too, which is
  /// how a plain `STOR` came to be blamed on a `MemoryReader` two files away.
  /// A deadline makes the wait bounded: the loop has a timer to expire, the
  /// deadlock check never runs, and a genuinely silent peer turns into a plain
  /// `FtpError` the caller can read.
  timeout_ms : Int32
  /// Traffic log sink, `None` when logging is off.
  ///
  /// The switch lives on the connection rather than being baked into the
  /// reader/writer at dial time (which is how upstream `debug.go` wires itself
  /// through `wrapConn`) because `Control::upgrade` replaces that pair
  /// wholesale after `AUTH TLS`: a decorator installed once at dial time would
  /// go on logging the plaintext socket and silently stop covering the TLS
  /// session every later command travels over. Keeping the switch here also
  /// means the logging layer never owns a `Reader`, so the `MemoryReader`
  /// ownership rule in AGENTS.md cannot be broken by turning logging on.
  mut debug : DebugLog?
}

///|
/// Wrap an already connected reader/writer pair.
pub fn Control::new(
  reader : &@io.Reader,
  writer : &@io.Writer,
  host : String,
  timeout_ms~ : Int32 = 0,
  debug~ : DebugLog? = None,
) -> Control {
  { reader, writer, host, timeout_ms, debug, }
}

///|
/// Install (or clear) the traffic log on an already built connection.
///
/// Separate from `Control::new` so `dial` can attach the sink the caller asked
/// for without every other call site growing the label.
pub fn Control::set_debug(self : Control, debug : DebugLog?) -> Unit {
  self.debug = debug
}

///|
/// Whether the traffic log is on.
pub fn Control::debug_enabled(self : Control) -> Bool {
  self.debug is Some(_)
}

///|
/// The peer address of the control connection.
pub fn Control::host(self : Control) -> String {
  self.host
}

///|
/// Replace the underlying reader/writer, used when the connection is upgraded
/// to TLS after `AUTH TLS`.
pub fn Control::upgrade(
  self : Control,
  reader : &@io.Reader,
  writer : &@io.Writer,
) -> Unit {
  self.reader = reader
  self.writer = writer
}

///|
/// Write `line` followed by CRLF.
pub async fn Control::send_line(self : Control, line : String) -> Unit {
  match self.debug {
    None => self.writer.write(line + "\r\n")
    Some(log) => {
      // Logging wraps the *same* write, and a sink that raises is swallowed
      // inside the decorator: the log is a diagnostic, so a broken sink must
      // not stop a command upstream would have sent.
      let mirror = MirrorWriter::new(self.writer, log~)
      @io.Writer::write(mirror, line + "\r\n")
    }
  }
}

///|
/// Read one line, without its trailing CRLF. Raises `ReaderClosed` when the
/// connection was closed before a complete line arrived, and
/// `RequestTimeout` when the server stayed silent for `timeout_ms`.
///
/// The deadline is on *the reply*, not on the transfer: only the idle part of
/// a wait is bounded, so a long listing still completes as long as the server
/// keeps the lines coming.
pub async fn Control::read_line(self : Control) -> String {
  // With logging on, the teeing reader *is* the source: the reply is recorded
  // by the same `read_until` that frames it, so a transcript cannot disagree
  // with what the parser saw. The trait object keeps this branch free of any
  // extra buffering.
  let source : &@io.Reader = match self.debug {
    None => self.reader
    Some(log) => {
      let tee = TeeLogReader::new(self.reader, log~)
      tee
    }
  }
  let read = async fn() -> String {
    match source.read_until("\r\n") {
      Some(line) => line
      None => raise @io.ReaderClosed
    }
  }
  // The deadline wraps *only* the socket read. Logging rides inside it, but
  // `TeeLogReader` swallows every sink error, so it can neither raise nor
  // stall: a server that stayed silent is still reported as silent, and a
  // broken sink still cannot turn a reply into a `RequestTimeout` -- the point
  // of the deadline rule in AGENTS.md.
  if self.timeout_ms > 0 {
    @async.with_timeout(self.timeout_ms, read) catch {
      @async.TimeoutError => raise FtpError::RequestTimeout
      err => raise err
    }
  } else {
    read()
  }
}

///|
/// Send `cmd` (already encoded) and return the reply.
pub async fn cmd(control : Control, command : String) -> Response {
  check_for_command_injection(command)
  control.send_line(command)
  read_response(control)
}

///|
/// Send `cmd` and require one of the `expected` status codes.
///
/// An empty `expected` list accepts any code, which is how upstream spells
/// `cmd(..., -1)` for the `expected == -1` case. A mismatch raises
/// `FtpError::ServerError` carrying the original message.
pub async fn cmd_expect(
  control : Control,
  command : String,
  expected : Array[Int32],
) -> Response {
  let response = cmd(control, command)
  if expected.length() == 0 || expected.contains(response.code()) {
    return response
  }
  raise FtpError::ServerError(code=response.code(), msg=response.message())
}

///|
/// Send `\`format\`` with `args` applied through `{}` substitution and require
/// one of the `expected` codes. Convenience wrapper mirroring upstream
/// `cmd(format, args, expected...)`.
pub async fn cmd_format(
  control : Control,
  format : String,
  args : Array[String],
  expected : Array[Int32],
) -> Response {
  cmd_expect(control, render_command(format, args), expected)
}

///|
/// Apply `{}` substitution to `format` without sending anything, validating
/// every argument first.
///
/// Split out of `cmd_format` because the `PRET` handshake needs the rendered
/// command *before* it is sent: it announces the line that follows, so the two
/// must be built by the same code or they drift.
pub fn render_command(
  format : String,
  args : Array[String],
) -> String raise FtpError {
  let mut rendered = format
  for arg in args {
    check_for_command_injection(arg)
    rendered = replace_first(rendered, "{}", arg)
  }
  rendered
}

///|
/// Refuse command arguments that contain CR or LF.
///
/// Upstream calls this `checkForCommandInjection`; the check must happen
/// before a single byte is written to the socket, otherwise a crafted path
/// could smuggle extra FTP commands into the session.
pub fn check_for_command_injection(arg : String) -> Unit raise FtpError {
  for byte in arg {
    if byte == '\r' || byte == '\n' {
      raise FtpError::InvalidCommand(arg~)
    }
  }
}

///|
/// Replace the first occurrence of `{}` in `format` with `value`.
fn replace_first(format : String, pattern : String, value : String) -> String {
  let index = match format.find(pattern) {
    Some(index) => index
    None => return format
  }
  format[0:index].to_owned() +
  value +
  format[index + pattern.length():].to_owned()
}

///|
/// A parsed FTP reply: the numeric status code and the (possibly multi line)
/// message body.
///
/// The body keeps the text of the framing status lines, matching the semantics
/// of Go's `textproto.ReadResponse`; see `read_response` for why the callers
/// depend on that.
pub struct Response {
  /// The 3 digit status code of the reply.
  code : Int32
  /// The message, with continuation lines joined by `\n`.
  message : String
} derive(@debug.Debug)

///|
/// `Response` promotes its `Debug` methods as regular methods; declare the
/// promotion explicitly so the deprecation warning stays away.
pub extend Response with @debug.Debug::{to_repr}

///|
/// Build a reply value.
pub fn Response::new(code : Int32, message : String) -> Response {
  { code, message, }
}

///|
/// The status code of the reply.
pub fn Response::code(self : Response) -> Int32 {
  self.code
}

///|
/// The message body of the reply.
pub fn Response::message(self : Response) -> String {
  self.message
}

///|
/// Read a complete reply from the control connection, handling both the single
/// line and the multi line (RFC 959 `211-...211 End`) shapes.
///
/// The result reproduces Go's `net/textproto.ReadResponse` byte for byte,
/// because the rest of the port (starting with `FEAT` and `MLST`) is written
/// against that shape. Two consequences are load bearing:
///
/// - the message **includes** the text of the first and of the terminating
///   status line, joined by `\n`. Go's `feat()` filters those back out with a
///   leading space test, and its `GetEntry` relies on `lines[1:lc-1]`; a body
///   that dropped the boundary lines would break `MLST` outright.
/// - a line that is exactly three digits, with no separator, is a
///   `short response` error, not an empty message.
///
/// The multi line terminator is the line that repeats the code *without* a
/// dash. Getting that wrong makes `FEAT` swallow the next reply, so it has a
/// dedicated test.
pub async fn read_response(control : Control) -> Response {
  let first = control.read_line()
  let code = match parse_code(first) {
    Some(code) => code
    None => raise FtpError::ParseError(msg="invalid response line: \{first}")
  }
  guard first.length() > 3 else {
    raise FtpError::ParseError(msg="short response: \{first}")
  }
  let head = first[4:].to_owned()
  if is_continuation(first) {
    let body : Array[String] = [head]
    while true {
      let line = control.read_line()
      let line_code = parse_code(line)
      if line_code == Some(code) && is_continuation(line) == false {
        guard line.length() > 3 else {
          raise FtpError::ParseError(msg="short response: \{line}")
        }
        body.push(line[4:].to_owned())
        break
      }
      body.push(line)
    }
    return { code, message: body.join("\n"), }
  }
  { code, message: head, }
}

///|
/// Parse the leading three digits of a reply line.
pub fn parse_code(line : String) -> Int32? {
  guard line.length() >= 3 else { return None }
  let mut value = 0
  for i = 0; i < 3; i = i + 1 {
    let c = line.unsafe_get(i).to_int()
    guard c >= 48 && c <= 57 else { return None }
    value = value * 10 + (c - 48)
  }
  Some(value)
}

///|
/// Whether the line is a multi line continuation (`211-Features:`), i.e. the
/// fourth byte is `-`.
pub fn is_continuation(line : String) -> Bool {
  line.length() > 3 && line.unsafe_get(3).to_int() == 45
}