# -*- encoding: binary -*-
# Copyright (C) 2015-2016 all contributors <yahns-public@yhbt.net>
# License: GPL-3.0+ (https://www.gnu.org/licenses/gpl-3.0.txt)
# frozen_string_literal: true

# loaded by yahns/proxy_pass, this relies on Yahns::HttpResponse for
# constants.
module Yahns::HttpResponse # :nodoc:

  # write everything in buf to our client socket (or wbuf, if it exists)
  # it may return a newly-created wbuf or nil
  def proxy_write(wbuf, buf, alive)
    unless wbuf
      # no write buffer, try to write directly to the client socket
      case rv = String === buf ? kgio_trywrite(buf) : kgio_trywritev(buf)
      when nil then return # done writing buf, likely
      when String, Array # partial write, hope the skb grows
        buf = rv
      when :wait_writable, :wait_readable
        wbuf = Yahns::Wbuf.new(nil, alive, self.class.output_buffer_tmpdir, rv)
        buf = buf.join if Array === buf
        break
      end while true
    end

    wbuf.wbuf_write(self, buf)
    wbuf.busy ? wbuf : nil
  end

  def proxy_err_response(code, req_res, exc, wbuf)
    logger = @hs.env['rack.logger']
    case exc
    when nil
      logger.error('premature upstream EOF')
    when Kcar::ParserError
      logger.error("upstream response error: #{exc.message}")
    else
      Yahns::Log.exception(logger, 'upstream error', exc)
    end
    # try to write something, but don't care if we fail
    Integer === code and
      kgio_trywrite("HTTP/1.1 #{code} #{
                     Rack::Utils::HTTP_STATUS_CODES[code]}\r\n\r\n") rescue nil

    shutdown rescue nil
    @input = @input.close if @input

    # this is safe ONLY because we are in an :ignore state after
    # Fdmap#forget when we got hijacked:
    close

    nil # signal close of req_res from yahns_step in yahns/proxy_pass.rb
  ensure
    wbuf.wbuf_abort if wbuf
  end

  def wait_on_upstream(req_res, alive, wbuf)
    req_res.resbuf = wbuf || Yahns::Wbuf.new(nil, alive,
                                             self.class.output_buffer_tmpdir,
                                             false)
    :wait_readable # self remains in :ignore, wait on upstream
  end

  def proxy_res_headers(res)
    status, headers = res
    code = status.to_i
    msg = Rack::Utils::HTTP_STATUS_CODES[code]
    env = @hs.env
    have_body = !Rack::Utils::STATUS_WITH_NO_ENTITY_BODY.include?(code) &&
                env['REQUEST_METHOD'] != 'HEAD'.freeze
    flags = MSG_DONTWAIT
    alive = @hs.next? && self.class.persistent_connections
    term = false
    response_headers = env['yahns.proxy_pass.response_headers']

    res = "HTTP/1.1 #{msg ? %Q(#{code} #{msg}) : status}\r\n".dup
    headers.each do |key,value| # n.b.: headers is an Array of 2-element Arrays
      case key
      when /\A(?:Connection|Keep-Alive)\z/i
        next # do not let some upstream headers leak through
      when %r{\AContent-Length\z}i
        term = true
        flags |= MSG_MORE if have_body && value.to_i > 0
      when %r{\ATransfer-Encoding\z}i
        term = true if value =~ /\bchunked\b/i
      end

      # response header mapping
      case val = response_headers[key]
      when :ignore
        next
      when String
        value = val
      end

      res << "#{key}: #{value}\r\n"
    end

    # For now, do not add a Date: header, assume upstream already did it
    # but do not care if they did not

    # chunk the response ourselves if the client supports it,
    # but the backend does not terminate properly
    if alive && ! term
      if env['HTTP_VERSION'] == 'HTTP/1.1'.freeze
        res << "Transfer-Encoding: chunked\r\n".freeze
      else # we can't persist HTTP/1.0 and HTTP/0.9 w/o Content-Length
        alive = false
      end
    end
    res << (alive ? "Connection: keep-alive\r\n\r\n".freeze
                  : "Connection: close\r\n\r\n".freeze)

    # send the headers
    case rv = kgio_syssend(res, flags)
    when nil then break # all done, likely
    when String # partial write, highly unlikely
      flags = MSG_DONTWAIT
      res = rv # hope the skb grows
    when :wait_writable, :wait_readable # highly unlikely in real apps
      wbuf = proxy_write(nil, res, alive)
      break # keep buffering as much as possible
    end while true
    [ alive, wbuf, have_body ]
  end

  def proxy_read_body(tip, kcar, req_res, alive, wbuf)
    chunk = ''.dup if kcar.chunked?
    len = kcar.body_bytes_left
    rbuf = Thread.current[:yahns_rbuf]

    case tmp = tip.shift || req_res.kgio_tryread(0x2000, rbuf)
    when String
      if len
        kcar.body_bytes_left -= tmp.size # progress for body_eof? => true
      elsif chunk
        kcar.filter_body(chunk, rbuf = tmp) # progress for body_eof? => true
        next if chunk.empty? # call req_res.kgio_tryread for more
        tmp = chunk_out(chunk)
      elsif alive # HTTP/1.0 upstream, HTTP/1.1 client
        tmp = chunk_out(tmp)
      # else # HTTP/1.0 upstream, HTTP/1.0 client, do nothing
      end
      wbuf = proxy_write(wbuf, tmp, alive)
      chunk.clear if chunk
    when nil # EOF
      # HTTP/1.1 upstream, unexpected premature EOF:
      return proxy_err_response(nil, req_res, nil, wbuf) if len || chunk

      # HTTP/1.0 upstream:
      wbuf = proxy_write(wbuf, "0\r\n\r\n".freeze, true) if alive
      req_res.shutdown
      break
    when :wait_readable
      return wait_on_upstream(req_res, alive, wbuf)
    end until kcar.body_eof?

    if chunk
      # tip is an empty array and becomes trailer storage
      req_res.proxy_trailers = [ rbuf.dup, tip ]
      return proxy_read_trailers(kcar, req_res, alive, wbuf)
    end
    proxy_busy_mod(wbuf, alive)
  end

  def proxy_read_trailers(kcar, req_res, alive, wbuf)
    chunk, tlr = req_res.proxy_trailers
    rbuf = Thread.current[:yahns_rbuf]

    until kcar.trailers(tlr, chunk)
      case rv = req_res.kgio_tryread(0x2000, rbuf)
      when String
        chunk << rv
      when :wait_readable
        return wait_on_upstream(req_res, alive, wbuf)
      when nil # premature EOF
        return proxy_err_response(nil, req_res, nil, wbuf)
      end # no loop here
    end
    wbuf = proxy_write(wbuf, trailer_out(tlr), alive)
    proxy_busy_mod(wbuf, alive)
  end

  # start streaming the response once upstream is done sending headers to us.
  # returns :wait_readable if we need to read more from req_res
  # returns :ignore if we yield control to the client(self)
  # returns nil if completely done
  def proxy_response_start(res, tip, kcar, req_res)
    alive, wbuf, have_body = proxy_res_headers(res)
    tip = tip.empty? ? [] : [ tip ]

    if have_body
      req_res.proxy_trailers = nil # define to avoid uninitialized warnings
      return proxy_read_body(tip, kcar, req_res, alive, wbuf)
    end

    # all done reading response from upstream, req_res will be discarded
    # when we return nil:
    proxy_busy_mod(wbuf, alive)
  rescue => e
    proxy_err_response(502, req_res, e, wbuf)
  end

  def proxy_response_finish(kcar, wbuf, req_res)
    alive = wbuf.wbuf_persist
    req_res.proxy_trailers ? proxy_read_trailers(kcar, req_res, alive, wbuf)
                           : proxy_read_body([], kcar, req_res, alive, wbuf)
  end

  def proxy_wait_next(qflags)
    Thread.current[:yahns_fdmap].remember(self)
    # We must allocate a new, empty request object here to avoid a TOCTTOU
    # in the following timeline
    #
    # original thread:                                 | another thread
    # HttpClient#yahns_step                            |
    # r = k.app.call(env = @hs.env)  # socket hijacked into epoll queue
    # <thread is scheduled away>                       | epoll_wait readiness
    #                                                  | ReqRes#yahns_step
    #                                                  | proxy dispatch ...
    #                                                  | proxy_busy_mod
    # ************************** DANGER BELOW ********************************
    #                                                  | HttpClient#yahns_step
    #                                                  | # clears env
    # sees empty env:                                  |
    # return :ignore if env.include?('rack.hijack_io') |
    #
    # In other words, we cannot touch the original env seen by the
    # original thread since it must see the 'rack.hijack_io' value
    # because both are operating in the same Yahns::HttpClient object.
    # This will happen regardless of GVL existence
    hs = Unicorn::HttpRequest.new
    hs.buf.replace(@hs.buf)
    @hs = hs

    # n.b. we may not touch anything in this object once we call queue_mod,
    # another thread is likely to take it!
    Thread.current[:yahns_queue].queue_mod(self, qflags)
  end

  def proxy_busy_mod(wbuf, alive)
    busy = wbuf.busy if wbuf
    if busy
      # we are completely done reading and buffering the upstream response,
      # but have not completely written the response to the client,
      # yield control to the client socket:
      @state = wbuf
      proxy_wait_next(case busy
        when :wait_readable then Yahns::Queue::QEV_RD
        when :wait_writable then Yahns::Queue::QEV_WR
        else
          raise "BUG: invalid wbuf.busy: #{busy.inspect}"
        end)
      # no touching self after proxy_wait_next, we may be running
      # HttpClient#yahns_step in a different thread at this point
    else
      case http_response_done(alive)
      when :wait_readable then proxy_wait_next(Yahns::Queue::QEV_RD)
      when :wait_writable then proxy_wait_next(Yahns::Queue::QEV_WR)
      when :close then close
      end
    end
    nil # signal close for ReqRes#yahns_step
  end

  # n.b.: we can use String#size for optimized dispatch under YARV instead
  # of String#bytesize because all the IO read methods return a binary
  # string when given a maximum read length
  def chunk_out(buf)
    [ "#{buf.size.to_s(16)}\r\n", buf, "\r\n".freeze ]
  end

  def trailer_out(tlr)
    "0\r\n#{tlr.map! do |k,v| "#{k}: #{v}\r\n" end.join}\r\n"
  end
end