module http

// Tests for the connection-pooling Transport (transport.v): keep-alive reuse
// over plain TCP and TLS, no-reuse on `Connection: close` / truncated reads,
// transparent retry on stale pooled connections, idle eviction, and the
// `disable_connection_reuse` opt-out. Each test runs its own loopback server
// with an accept counter: the accept count is the proof of (non-)reuse.
import net
import net.mbedtls
import sync
import time

const tls_test_cert_path = @VEXEROOT +
	'/vlib/net/websocket/tests/autobahn/fuzzing_server_wss/config/server.crt'
const tls_test_key_path = @VEXEROOT +
	'/vlib/net/websocket/tests/autobahn/fuzzing_server_wss/config/server.key'

@[heap]
struct KaSrv {
mut:
	mu      &sync.Mutex = sync.new_mutex()
	accepts int
	// connection_close: respond with `Connection: close` and close afterwards.
	connection_close bool
	// close_after_each: close after each response WITHOUT advertising it — the
	// classic stale-pooled-connection scenario.
	close_after_each bool
	// split_connection_close: emit the close token in a SECOND, repeated
	// `Connection` header (`keep-alive` first, then `close`) while keeping the
	// socket open — a server that says "do not reuse" via a split field. The
	// client must honor it and not pool, even though the connection stays usable.
	split_connection_close bool
	// drop_on_post: after reading a POST request head, close the connection
	// without responding — a reused-connection failure after the bytes were
	// written, for a non-idempotent method.
	drop_on_post bool
	posts        int // POST request heads actually read by the server
	body         string = 'hello'
}

fn (mut s KaSrv) bump_accepts() {
	s.mu.lock()
	s.accepts++
	s.mu.unlock()
}

fn (mut s KaSrv) accept_count() int {
	s.mu.lock()
	defer {
		s.mu.unlock()
	}
	return s.accepts
}

fn (mut s KaSrv) post_count() int {
	s.mu.lock()
	defer {
		s.mu.unlock()
	}
	return s.posts
}

// ka_read_request_head reads from `conn` until a full request head (terminated
// by a blank line) has arrived, returning it. The test requests carry no bodies.
fn ka_read_request_head(mut conn net.TcpConn) !string {
	mut buf := []u8{len: 4096}
	mut sofar := []u8{}
	for {
		n := conn.read(mut buf)!
		if n <= 0 {
			return error('closed')
		}
		sofar << buf[..n]
		if sofar.bytestr().contains('\r\n\r\n') {
			return sofar.bytestr()
		}
	}
	return error('closed')
}

fn ka_srv_serve_conn(mut s KaSrv, mut conn net.TcpConn) {
	defer {
		conn.close() or {}
	}
	conn.set_read_timeout(10 * time.second)
	for {
		head := ka_read_request_head(mut conn) or { return }
		if s.drop_on_post && head.starts_with('POST ') {
			s.mu.lock()
			s.posts++
			s.mu.unlock()
			// Drop the connection without responding: the request bytes were
			// written, so this is not a stale-write.
			return
		}
		mut resp := 'HTTP/1.1 200 OK\r\nContent-Length: ${s.body.len}\r\n'
		if s.connection_close {
			resp += 'Connection: close\r\n'
		}
		if s.split_connection_close {
			resp += 'Connection: keep-alive\r\nConnection: close\r\n'
		}
		resp += '\r\n' + s.body
		conn.write(resp.bytes()) or { return }
		if s.connection_close || s.close_after_each {
			return
		}
	}
}

fn ka_srv_loop(mut s KaSrv, mut listener net.TcpListener) {
	for {
		mut conn := listener.accept() or { return }
		s.bump_accepts()
		ka_srv_serve_conn(mut s, mut conn)
	}
}

// start_ka_srv starts a keep-alive capable loopback HTTP server, returning its
// port, listener and server thread.
fn start_ka_srv(mut s KaSrv) !(int, &net.TcpListener, thread) {
	mut listener := net.listen_tcp(.ip, '127.0.0.1:0')!
	port := listener.addr()!.port()!
	th := spawn ka_srv_loop(mut s, mut listener)
	return port, listener, th
}

// stop_ka_srv tears a test server down fully before the test returns: closing
// the idle pool unblocks a server thread reading a kept-alive connection,
// closing the listener aborts its accept, and the join guarantees the thread
// is gone before the next test creates sockets (otherwise the OS can recycle
// this listener's handle for the next test's listener while the old thread is
// still calling accept on it, stealing its connections).
fn stop_ka_srv(mut listener net.TcpListener, th thread) {
	close_idle_connections()
	listener.close() or {}
	th.wait()
}

// The pool key must isolate distinct TLS configurations even when a field value
// contains the '|' separator, or a request could reuse a connection dialed with
// the wrong cert/CA. cert='a|b',cert_key='c' and cert='a',cert_key='b|c' must
// not collide.
fn test_transport_pool_key_no_delimiter_collision() {
	a := Request{
		cert:     'a|b'
		cert_key: 'c'
	}
	b := Request{
		cert:     'a'
		cert_key: 'b|c'
	}
	assert transport_pool_key(a, 'https', 'h', 443) != transport_pool_key(b, 'https', 'h', 443)
	// Identical configs still share a key.
	a2 := Request{
		cert:     'a|b'
		cert_key: 'c'
	}
	assert transport_pool_key(a, 'https', 'h', 443) == transport_pool_key(a2, 'https', 'h', 443)
	// A host containing '|' must not collide with a different host/port split.
	assert transport_pool_key(Request{}, 'https', 'h|x', 443) != transport_pool_key(Request{},
		'https', 'h', 443)
}

fn test_h1_plain_reuse() {
	mut srv := &KaSrv{}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	for i in 0 .. 3 {
		resp := fetch(url: 'http://127.0.0.1:${port}/r${i}') or {
			assert false, 'fetch ${i}: ${err}'
			return
		}
		assert resp.status_code == 200
		assert resp.body == 'hello'
	}
	// All three requests must have shared one connection.
	assert srv.accept_count() == 1
	stop_ka_srv(mut listener, th)
}

fn test_h1_no_reuse_on_connection_close_response() {
	mut srv := &KaSrv{
		connection_close: true
	}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	for i in 0 .. 2 {
		resp := fetch(url: 'http://127.0.0.1:${port}/r${i}') or {
			assert false, 'fetch ${i}: ${err}'
			return
		}
		assert resp.status_code == 200
	}
	// `Connection: close` responses must not be pooled.
	assert srv.accept_count() == 2
	stop_ka_srv(mut listener, th)
}

// A close token carried in a repeated `Connection` header (after a keep-alive
// one) must still be honored: parse_headers stores repeats separately and get()
// returns only the first, so response_allows_reuse joins all values. The server
// keeps the socket open, so a wrongly-pooled connection would be reused (1
// accept); honoring the close dials fresh each time (2 accepts).
fn test_h1_no_reuse_on_split_connection_close() {
	mut srv := &KaSrv{
		split_connection_close: true
	}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	for i in 0 .. 2 {
		resp := fetch(url: 'http://127.0.0.1:${port}/r${i}') or {
			assert false, 'fetch ${i}: ${err}'
			return
		}
		assert resp.status_code == 200
	}
	assert srv.accept_count() == 2
	stop_ka_srv(mut listener, th)
}

// The total idle pool is bounded by max_idle_conns across all pool keys, not
// just per host: checking in more distinct-keyed connections than the cap
// evicts the least-recently-used ones (here k0, k1) and keeps the newest.
fn test_transport_global_idle_cap() {
	mut t := new_transport()
	t.max_idle_conns = 3
	t.max_idle_conns_per_host = 10 // keep the per-host cap out of the way
	for i in 0 .. 5 {
		mut c := &H1PooledConn{
			key: 'k${i}'
		}
		t.checkin(mut c)
	}
	mut total := 0
	for _, list in t.h1_idle {
		total += list.len
	}
	assert total == 3, 'global idle cap not enforced: ${total}'
	assert 'k0' !in t.h1_idle
	assert 'k1' !in t.h1_idle
	assert 'k2' in t.h1_idle
	assert 'k3' in t.h1_idle
	assert 'k4' in t.h1_idle
}

fn test_h1_stale_pooled_connection_is_retried() {
	mut srv := &KaSrv{
		close_after_each: true
	}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	// First request succeeds and the connection is pooled (the response did not
	// advertise the close). The server then closes it. The second request picks
	// up the stale connection, fails, and must transparently retry on a fresh
	// one.
	for i in 0 .. 2 {
		resp := fetch(url: 'http://127.0.0.1:${port}/r${i}') or {
			assert false, 'fetch ${i}: ${err}'
			return
		}
		assert resp.status_code == 200
		assert resp.body == 'hello'
	}
	assert srv.accept_count() == 2
	stop_ka_srv(mut listener, th)
}

// A non-idempotent request that fails on a reused keep-alive connection after
// its bytes were written must NOT be replayed: doing so could duplicate side
// effects. The server reads the POST then drops the connection; the client must
// surface the error without retrying on a fresh connection.
fn test_h1_unsafe_pooled_post_is_not_retried() {
	mut srv := &KaSrv{
		drop_on_post: true
	}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	// First, a GET to establish and pool a keep-alive connection.
	resp := fetch(url: 'http://127.0.0.1:${port}/warmup') or {
		assert false, 'warmup fetch: ${err}'
		return
	}
	assert resp.status_code == 200
	// The POST reuses that connection; the server reads it and drops the
	// connection. The request must fail rather than be re-sent.
	fetch(method: .post, url: 'http://127.0.0.1:${port}/submit', data: 'payload') or {
		// expected: the POST failed and was not retried.
		assert srv.post_count() == 1, 'POST was replayed ${srv.post_count()} times'
		assert srv.accept_count() == 1, 'a fresh connection was opened to retry the POST'
		stop_ka_srv(mut listener, th)
		return
	}
	assert false, 'the POST unexpectedly succeeded'
	stop_ka_srv(mut listener, th)
}

fn test_h1_opt_out_disables_reuse() {
	mut srv := &KaSrv{}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	for i in 0 .. 2 {
		resp := fetch(
			url:                      'http://127.0.0.1:${port}/r${i}'
			disable_connection_reuse: true
		) or {
			assert false, 'fetch ${i}: ${err}'
			return
		}
		assert resp.status_code == 200
	}
	// Opt-out requests open one connection each.
	assert srv.accept_count() == 2
	stop_ka_srv(mut listener, th)
}

fn test_h1_truncated_read_poisons_connection() {
	mut srv := &KaSrv{
		// Larger than the 64KB read buffer, so the body needs several reads and
		// the stop limit actually interrupts the transfer mid-stream.
		body: 'x'.repeat(200 * 1024)
	}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	// A stop_receiving_limit read leaves unread response bytes on the wire, so
	// that connection must not be reused.
	resp1 := fetch(url: 'http://127.0.0.1:${port}/big', stop_receiving_limit: 1000) or {
		assert false, 'fetch 1: ${err}'
		return
	}
	assert resp1.status_code == 200
	resp2 := fetch(url: 'http://127.0.0.1:${port}/after') or {
		assert false, 'fetch 2: ${err}'
		return
	}
	assert resp2.status_code == 200
	assert srv.accept_count() == 2
	stop_ka_srv(mut listener, th)
}

fn test_h1_idle_eviction() {
	mut srv := &KaSrv{}
	port, mut listener, th := start_ka_srv(mut srv) or {
		assert false, 'server: ${err}'
		return
	}
	mut t := new_transport()
	t.idle_timeout = 50 * time.millisecond
	req := prepare(url: 'http://127.0.0.1:${port}/') or {
		assert false, 'prepare: ${err}'
		return
	}
	r1 := t.round_trip(req, .get, 'http', '127.0.0.1', port, '/', '', req.header) or {
		assert false, 'round_trip 1: ${err}'
		return
	}
	assert r1.status_code == 200
	time.sleep(150 * time.millisecond)
	// The pooled connection has sat idle past the timeout: it must be evicted
	// and a fresh one dialled.
	r2 := t.round_trip(req, .get, 'http', '127.0.0.1', port, '/', '', req.header) or {
		assert false, 'round_trip 2: ${err}'
		return
	}
	assert r2.status_code == 200
	assert srv.accept_count() == 2
	// This test pools in its own private Transport, so flush that one before
	// the joint teardown.
	t.close_idle()
	stop_ka_srv(mut listener, th)
}

// --- TLS (h1 over mbedtls) reuse ---

@[heap]
struct TlsKaSrv {
mut:
	mu      &sync.Mutex = sync.new_mutex()
	accepts int
}

fn (mut s TlsKaSrv) bump_accepts() {
	s.mu.lock()
	s.accepts++
	s.mu.unlock()
}

fn (mut s TlsKaSrv) accept_count() int {
	s.mu.lock()
	defer {
		s.mu.unlock()
	}
	return s.accepts
}

fn tls_ka_srv_serve_conn(mut conn mbedtls.SSLConn) {
	defer {
		conn.shutdown() or {}
	}
	body := 'tls hello'
	for {
		mut buf := []u8{len: 4096}
		mut sofar := []u8{}
		for !sofar.bytestr().contains('\r\n\r\n') {
			n := conn.read(mut buf) or { return }
			if n <= 0 {
				return
			}
			sofar << buf[..n]
		}
		conn.write_string('HTTP/1.1 200 OK\r\nContent-Length: ${body.len}\r\n\r\n${body}') or {
			return
		}
	}
}

fn tls_ka_srv_loop(mut s TlsKaSrv, mut listener mbedtls.SSLListener) {
	for {
		mut conn := listener.accept() or { return }
		s.bump_accepts()
		// Serve each connection on its own thread: a pooled idle connection
		// keeps its serve loop parked in read, which must not block accepting
		// the next connection. (The serve threads exit when the test closes
		// the pooled connections via close_idle_connections.)
		spawn tls_ka_srv_serve_conn(mut conn)
	}
}

fn test_h1_tls_reuse() {
	$if windows && !no_vschannel ? {
		// The default Windows TLS backend (SChannel) keeps its one-shot path
		// until SChannel pooling lands.
		eprintln('skipping: SChannel connection pooling is not implemented yet')
		return
	}
	mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0') or {
		assert false, 'port: ${err}'
		return
	}
	port := port_listener.addr() or {
		assert false, 'addr: ${err}'
		return
	}.port() or {
		assert false, 'port: ${err}'
		return
	}
	port_listener.close() or {}
	mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{
		cert:     tls_test_cert_path
		cert_key: tls_test_key_path
		validate: false
	}) or {
		assert false, 'listener: ${err}'
		return
	}
	mut srv := &TlsKaSrv{}
	th := spawn tls_ka_srv_loop(mut srv, mut listener)
	for i in 0 .. 2 {
		resp := fetch(url: 'https://127.0.0.1:${port}/r${i}', validate: false) or {
			assert false, 'fetch ${i}: ${err}'
			return
		}
		assert resp.status_code == 200
		assert resp.body == 'tls hello'
	}
	// Both https requests must have shared one TLS connection.
	assert srv.accept_count() == 1
	// Free the pooled TLS connection (unblocking the server read), then abort
	// the accept and join the server thread before the test returns.
	close_idle_connections()
	listener.shutdown() or {}
	th.wait()
}

// A TLS connection dialled with HTTP/2 disabled (no ALPN) must not satisfy an
// HTTP/2-enabled request to the same origin — the ALPN preference is part of
// the pool key. Forced-h1 requests still share among themselves.
fn test_h1_tls_no_reuse_across_alpn_preference() {
	$if windows && !no_vschannel ? {
		eprintln('skipping: SChannel connection pooling is not implemented yet')
		return
	}
	mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0') or {
		assert false, 'port: ${err}'
		return
	}
	port := port_listener.addr() or {
		assert false, 'addr: ${err}'
		return
	}.port() or {
		assert false, 'port: ${err}'
		return
	}
	port_listener.close() or {}
	mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{
		cert:     tls_test_cert_path
		cert_key: tls_test_key_path
		validate: false
	}) or {
		assert false, 'listener: ${err}'
		return
	}
	mut srv := &TlsKaSrv{}
	th := spawn tls_ka_srv_loop(mut srv, mut listener)
	// 1. Forced-HTTP/1.1 request: dialled without ALPN, pooled under its key.
	r1 := fetch(url: 'https://127.0.0.1:${port}/h1', validate: false, enable_http2: false) or {
		assert false, 'fetch 1: ${err}'
		return
	}
	assert r1.status_code == 200
	// 2. Default (HTTP/2-enabled) request: must NOT reuse the no-ALPN
	// connection; it dials fresh and advertises h2.
	r2 := fetch(url: 'https://127.0.0.1:${port}/h2pref', validate: false) or {
		assert false, 'fetch 2: ${err}'
		return
	}
	assert r2.status_code == 200
	assert srv.accept_count() == 2
	// 3. Another forced-HTTP/1.1 request: reuses connection 1.
	r3 := fetch(url: 'https://127.0.0.1:${port}/h1again', validate: false, enable_http2: false) or {
		assert false, 'fetch 3: ${err}'
		return
	}
	assert r3.status_code == 200
	assert srv.accept_count() == 2
	close_idle_connections()
	listener.shutdown() or {}
	th.wait()
}