///|
priv struct EngineOwner {
  mut closing : Bool
  mut planning : Int32
  idle : @async.CondVar
}

///|
pub struct StaticEngine {
  config : @core.Config
  priv mime_registry : @core.MimeRegistry
  priv root : @native.File
  priv runtime : @native.Runtime
  priv owner : EngineOwner
  priv responses : Map[Int32, BodyState]
  priv mut next_id : Int32
}

///|
pub(all) enum HandleResult {
  Handled(Response)
  Next
  Error(ServerError)
}

///|
pub suberror ServerError {
  InvalidRequest(String)
  Forbidden(String)
  Io(String)
  FileChanged(String)
  Closed
  Busy
  Overloaded
}

///|
pub fn ServerError::invalid_request(message : String) -> ServerError {
  InvalidRequest(message)
}

///|
pub fn ServerError::forbidden(message : String) -> ServerError {
  Forbidden(message)
}

///|
pub fn ServerError::io(message : String) -> ServerError {
  Io(message)
}

///|
priv enum PlanBody {
  Empty
  Bytes(Bytes)
  FileRegion(@native.File, Int64, Int64)
  Listing(ListingStream)
}

///|
priv struct ListingStream {
  stream : @plan.ListingStream
  metadata_charge : Int32
}

///|
fn ListingStream::read(self : ListingStream, limit : Int32) -> Bytes {
  self.stream.read(limit)
}

///|
priv struct ResponsePlan {
  status : Int32
  headers : Map[String, String]
  body : PlanBody
}

///|
priv enum PlanResult {
  Handled(ResponsePlan)
  Next
}

///|
priv struct BodyState {
  body : PlanBody
  mut cursor : Int64
  mut closed : Bool
  mut reading : Bool
  idle : @async.CondVar
  runtime : @native.Runtime
  mut body_charge : Int32
  mut released : Bool
}

///|
pub struct Response {
  status : Int32
  headers : Map[String, String]
  priv state : BodyState
  priv owner : EngineOwner
  priv registry : Map[Int32, BodyState]
  priv id : Int32
  priv block_size : Int32
}

///|
/// Open and validate a Native static engine. Pair with async close, or prefer with_engine.
pub async fn StaticEngine::open(config : @core.Config) -> StaticEngine {
  config.validate()
  let runtime = @native.Runtime::new(
    workers=config.limits.file_workers,
    capacity=config.limits.queued_operations,
    body_bytes=config.limits.body_buffer_bytes,
    directory_bytes=config.limits.directory_metadata_bytes,
    connections=config.limits.connections,
    backend=config.io_backend,
  )
  errdefer runtime.close()
  let root = runtime.open_root(config.root)
  let mime_registry = @core.MimeRegistry::new()
  mime_registry.set_all(config.mime_types)
  {
    config,
    mime_registry,
    root,
    runtime,
    owner: { closing: false, planning: 0, idle: @async.CondVar::Cond(), },
    responses: Map([]),
    next_id: 0,
  }
}

///|
pub async fn[T] with_engine(
  config : @core.Config,
  handler : async (StaticEngine) -> T,
) -> T {
  let engine = StaticEngine::open(config)
  defer engine.close()
  handler(engine)
}

///|
async fn BodyState::close(self : BodyState) -> Unit {
  self.closed = true
  @async.protect_from_cancel(() => {
    match self.body {
      FileRegion(file, _, _) => file.close()
      _ => ()
    }
    while self.reading {
      self.idle.wait()
    }
    match self.body {
      Listing(stream) => {
        stream.stream.close()
        if !self.released {
          self.runtime.release(directory=true, stream.metadata_charge)
        }
      }
      _ => ()
    }
    if !self.released {
      self.runtime.release(directory=false, self.body_charge)
      self.released = true
    }
  })
}

///|
pub async fn StaticEngine::close(self : StaticEngine) -> Unit {
  self.owner.closing = true
  self.runtime.stop()
  @async.protect_from_cancel(() => {
    while self.owner.planning > 0 {
      self.owner.idle.wait()
    }
    for state in self.responses.values().to_array() {
      state.close()
    }
    self.responses.clear()
    self.root.close()
    self.runtime.close()
  })
}

///|
fn map_native_error(e : Error) -> ServerError {
  match e {
    ServerError::InvalidRequest(s) => InvalidRequest(s)
    ServerError::Forbidden(s) => Forbidden(s)
    ServerError::Io(s) => Io(s)
    ServerError::FileChanged(s) => FileChanged(s)
    ServerError::Closed => Closed
    ServerError::Busy => Busy
    ServerError::Overloaded => Overloaded
    @native.NativeError::FileChanged =>
      FileChanged("opened representation changed")
    @plan.Changed =>
      FileChanged("chosen gzip representation changed before opening")
    @native.NativeError::Forbidden =>
      Forbidden("path is outside the root or inaccessible")
    @native.NativeError::Overloaded => Overloaded
    @native.NativeError::Closed => Closed
    @native.NativeError::Busy => Busy
    _ => Io("native file operation failed")
  }
}

///|
pub async fn StaticEngine::handle(
  self : StaticEngine,
  request : @core.Request,
) -> HandleResult {
  if self.owner.closing {
    return Error(Closed)
  }
  self.owner.planning += 1
  defer {
    self.owner.planning -= 1
    self.owner.idle.broadcast()
  }
  let result = self.plan_request(request) catch {
    e => return Error(map_native_error(e))
  }
  match result {
    Next => Next
    Handled(plan) => {
      let state : BodyState = {
        body: plan.body,
        cursor: 0L,
        closed: false,
        reading: false,
        idle: @async.CondVar::Cond(),
        runtime: self.runtime,
        body_charge: 0,
        released: false,
      }
      if self.owner.closing {
        state.close()
        return Error(Closed)
      }
      let charge = match plan.body {
        Empty => 0
        Bytes(b) => b.length() * 2
        _ => self.config.limits.body_block_bytes * 2
      }
      self.runtime.reserve(directory=false, charge) catch {
        e => {
          state.close()
          return Error(map_native_error(e))
        }
      }
      state.body_charge = charge
      let id = self.next_id
      self.next_id += 1
      self.responses[id] = state
      Handled({
        status: plan.status,
        headers: plan.headers,
        state,
        owner: self.owner,
        registry: self.responses,
        id,
        block_size: self.config.limits.body_block_bytes,
      })
    }
  }
}

///|
pub fn Response::body_length(self : Response) -> Int64 {
  match self.state.body {
    Empty => 0L
    Bytes(b) => b.length().to_int64()
    FileRegion(_, _, n) => n
    Listing(_) => -1L
  }
}

///|
/// None denotes a streaming body whose length is not precomputed.
pub fn Response::content_length(self : Response) -> Int64? {
  let n = self.body_length()
  if n < 0L {
    None
  } else {
    Some(n)
  }
}

///|
/// Revalidate the representation immediately before committing response headers.
pub async fn Response::check(self : Response) -> Unit {
  if self.state.closed || self.owner.closing {
    raise Closed
  }
  if self.state.reading {
    raise Busy
  }
  self.state.reading = true
  defer {
    self.state.reading = false
    self.state.idle.broadcast()
  }
  errdefer {
    self.state.reading = false
    self.state.idle.broadcast()
    self.close()
  }
  match self.state.body {
    FileRegion(file, _, _) =>
      file.check() catch {
        e => raise map_native_error(e)
      }
    _ => ()
  }
}

///|
/// Read the next bounded chunk. Only successful EOF returns empty bytes.
pub async fn Response::read(self : Response, max_len~ : Int32 = 65536) -> Bytes {
  if self.state.closed || self.owner.closing {
    raise Closed
  }
  if self.state.reading {
    raise Busy
  }
  if max_len <= 0 {
    raise InvalidRequest("max_len must be positive")
  }
  self.state.reading = true
  defer {
    self.state.reading = false
    self.state.idle.broadcast()
  }
  errdefer {
    self.state.reading = false
    self.state.idle.broadcast()
    self.close()
  }
  let remaining = if self.state.body is Listing(_) {
    max_len.to_int64()
  } else {
    self.body_length() - self.state.cursor
  }
  let count = if remaining < max_len.to_int64() {
    remaining.to_int()
  } else {
    max_len
  }
  let count = if count > self.block_size { self.block_size } else { count }
  let chunk = match self.state.body {
    Empty => b""
    Bytes(b) =>
      b[self.state.cursor.to_int():self.state.cursor.to_int() + count].to_owned()
    FileRegion(file, offset, _) => {
      file.check() catch {
        e => raise map_native_error(e)
      }
      file.read(offset + self.state.cursor, count) catch {
        e => raise map_native_error(e)
      }
    }
    Listing(stream) => stream.read(count)
  }
  if self.state.closed || self.owner.closing {
    raise Closed
  }
  self.state.cursor += chunk.length().to_int64()
  chunk
}

///|
/// Explicitly bounded convenience collection; normal server paths stream instead.
pub async fn Response::to_bytes(self : Response, max_bytes~ : Int32) -> Bytes {
  errdefer self.close()
  if max_bytes < 0 ||
    self.body_length() - self.state.cursor > max_bytes.to_int64() {
    raise Overloaded
  }
  let buf : Array[Byte] = []
  let mut reserved = 0
  defer self.state.runtime.release(directory=false, reserved)
  for ;; {
    let chunk = self.read()
    if chunk.length() == 0 {
      break
    }
    if buf.length() > max_bytes - chunk.length() {
      raise Overloaded
    }
    // Array growth and the final contiguous copy coexist until return. Returned
    // bytes belong to the caller; engine budgets account for collection in flight.
    if chunk.length() > (2147483647 - reserved) / 4 {
      raise Overloaded
    }
    let charge = chunk.length() * 4
    self.state.runtime.reserve(directory=false, charge) catch {
      e => raise map_native_error(e)
    }
    reserved += charge
    for b in chunk {
      buf.push(b)
    }
  }
  Bytes::from_array(buf)
}

///|
pub async fn Response::close(self : Response) -> Unit {
  self.state.close()
  self.registry.remove(self.id)
}

///|
/// Write a body through a bounded writer; chunked is HTTP/1.1 framing only.
pub async fn Response::write_to(
  self : Response,
  writer : &@io.Writer,
  chunked~ : Bool = false,
) -> Unit {
  for ;; {
    let chunk = self.read()
    if chunk.length() == 0 {
      break
    }
    if chunked {
      writer.write(chunk.length().to_string(radix=16) + "\r\n")
    }
    writer.write(chunk)
    if chunked {
      writer.write("\r\n")
    }
  }
  if chunked {
    writer.write("0\r\n\r\n")
  }
}

///|
/// Send the remaining file through the socket's kernel transfer path, falling back
/// from the exact current offset when the platform reports unsupported operation.
pub async fn Response::send_to(self : Response, tcp : @socket.Tcp) -> Unit {
  match self.state.body {
    FileRegion(file, offset, length) => {
      if self.state.closed || self.owner.closing {
        raise Closed
      }
      if self.state.reading {
        raise Busy
      }
      self.state.reading = true
      defer {
        self.state.reading = false
        self.state.idle.broadcast()
      }
      errdefer {
        self.state.reading = false
        self.state.idle.broadcast()
        self.close()
      }
      let mut kernel = true
      while self.state.cursor < length {
        if self.state.closed || self.owner.closing {
          raise Closed
        }
        let left = length - self.state.cursor
        let count = if left < self.block_size.to_int64() {
          left.to_int()
        } else {
          self.block_size
        }
        if kernel {
          let sent = file.send(tcp.fd(), offset + self.state.cursor, count) catch {
            e => raise map_native_error(e)
          }
          match sent {
            Some(n) => {
              self.state.cursor += n
              if n == 0L {
                @async.sleep(1)
              }
            }
            None => kernel = false
          }
        } else {
          let chunk = file.read(offset + self.state.cursor, count) catch {
            e => raise map_native_error(e)
          }
          tcp.write(chunk)
          self.state.cursor += chunk.length().to_int64()
        }
        @async.pause()
      }
      file.check() catch {
        e => raise map_native_error(e)
      }
    }
    _ =>
      for ;; {
        let chunk = self.read()
        if chunk.length() == 0 {
          break
        }
        tcp.write(chunk)
      }
  }
}

///|
fn StaticEngine::relative(self : StaticEngine, path : String) -> String raise {
  let root = normalize_root_join(self.config.root, "")
  if path == root || path == self.config.root {
    return ""
  }
  let prefix = if root.has_suffix("/") { root } else { root + "/" }
  if !path.has_prefix(prefix) {
    raise Forbidden("candidate outside root")
  }
  path[prefix.length():].to_owned()
}

///|
async fn StaticEngine::open_file(
  self : StaticEngine,
  path : String,
) -> @native.File {
  self.root.open(self.relative(path))
}