// vtest flaky: true
// vtest retry: 3
module main

import net
import net.html
import net.http
import net.mbedtls
import sync.pool
import time

const concurrent_large_response_requests = 6
const concurrent_large_response_workers = 3
const concurrent_large_response_links = 2_000
const concurrent_large_response_chunk_size = 8_192
const concurrent_large_response_test_cert = '-----BEGIN CERTIFICATE-----\nMIIEOTCCAyECFG64Q2g46jZb3kRbDOJWX/BwjSp6MA0GCSqGSIb3DQEBCwUAMEUx\nCzAJBgNVBAYTAkFVMRMwEQYDVQQIDApTb21lLVN0YXRlMSEwHwYDVQQKDBhJbnRl\ncm5ldCBXaWRnaXRzIFB0eSBMdGQwIBcNMjMwODAyMTcyOTQyWhgPMjA1MDEyMTcx\nNzI5NDJaMGsxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDYWxpZm9ybmlhMRQwEgYD\nVQQHDAtMb3MgQW5nZWxlczEdMBsGA1UECgwUQ2F0YWx5c3QgRGV2ZWxvcG1lbnQx\nEjAQBgNVBAMMCWxvY2FsaG9zdDCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoC\nggIBALqAI4fqUi+QBVWcsXglouLdOML5+w0+1hSR1KdO0Q5XPdQAs/yYWJ+KUkDw\nG++rfy9DUPq7FNRBVurXQkcAtn6gXdllGUSjwUiDo/N4mMOyS/2sufBuaeww7jVi\nrppH+zwP1tUnjRd6khl6bi1Ian9VSzr3Iy9CkXIg1GU4CPXkOydLeoQfepXxWoK1\nOUNwT3VKC/stAfY3j/NIIeiJYkyuRGFCkxn/BUjN+AsXiTugRcYKEFHdIPkOuCXp\nYbhf+lLsczpxCs3rdZG9b/N6mEDCzXTmeHkmsjdPTf+1k5DZZvKzVBBrgdxCgBb7\n5RwjF5v9WmnIc33wWgfJC6FaUzj9NYxYUbPHD+jTz0rJB/jj4u/xJlM/e5NRmXdW\n70pOMKXtWjRSolLOFIPKLY1qs3KMTAZxKKWPDDF7WlMJxMRt7nnnks5yw43Nog4C\njDLk1ZgETnPpLgo3jbmJdIv+OHKTJrBlVvDq7VTyixCoS5G8KoOmyQJhaXG6NwE2\niVhH5JIKgzgCfetfDsnjxqJ/qtrFXPa8FF2TsomD0NK/GZmIcs+9OeVB75Jn5uhF\nfLHScpiTbuu5w3P/LI/MqihLRB6RRNnRzPH8fIg5bYC9b770ta/8GcFRuYE8t+UR\nGtqXJoIKixbDlqV54kal8FQzYzhETf9+NM6Kb/lKEfG/pslvAgMBAAEwDQYJKoZI\nhvcNAQELBQADggEBALI3uNiNO0QE1brA3QYFK+d9ZroB72NrJ0UNkzYHDg2Fc6xg\n4aVVfaxY08+TmKc0JlMOW+pUxeCW/+UBSngdQiR9EE9xm0k0XIrAsy9RXxRvEtPu\nM1VI2h7ayp1Y2BrnQinevTSgtqLRyS1VbOFRl1FiyVvinw2I0KsDdAMNevAPXcOa\nQ8pUgUq6f56DkhocQaj+hxD/uV8HryNxuoSXnPhvfTN3z4YRGzsaWevJ9EYJliOM\n+XugcqfFJ+W7/QCEcAHCL+Bw6OydG5NFORr3p57PXjjcL/uKmxPBrWg2Bz6uT4uR\nMhj0zttiFHLAt9jGfyk6W57UNUja1e1ggftJJhs=\n-----END CERTIFICATE-----\n'
const concurrent_large_response_test_key = '-----BEGIN RSA PRIVATE KEY-----\nMIIJKQIBAAKCAgEAuoAjh+pSL5AFVZyxeCWi4t04wvn7DT7WFJHUp07RDlc91ACz\n/JhYn4pSQPAb76t/L0NQ+rsU1EFW6tdCRwC2fqBd2WUZRKPBSIOj83iYw7JL/ay5\n8G5p7DDuNWKumkf7PA/W1SeNF3qSGXpuLUhqf1VLOvcjL0KRciDUZTgI9eQ7J0t6\nhB96lfFagrU5Q3BPdUoL+y0B9jeP80gh6IliTK5EYUKTGf8FSM34CxeJO6BFxgoQ\nUd0g+Q64JelhuF/6UuxzOnEKzet1kb1v83qYQMLNdOZ4eSayN09N/7WTkNlm8rNU\nEGuB3EKAFvvlHCMXm/1aachzffBaB8kLoVpTOP01jFhRs8cP6NPPSskH+OPi7/Em\nUz97k1GZd1bvSk4wpe1aNFKiUs4Ug8otjWqzcoxMBnEopY8MMXtaUwnExG3ueeeS\nznLDjc2iDgKMMuTVmAROc+kuCjeNuYl0i/44cpMmsGVW8OrtVPKLEKhLkbwqg6bJ\nAmFpcbo3ATaJWEfkkgqDOAJ9618OyePGon+q2sVc9rwUXZOyiYPQ0r8ZmYhyz705\n5UHvkmfm6EV8sdJymJNu67nDc/8sj8yqKEtEHpFE2dHM8fx8iDltgL1vvvS1r/wZ\nwVG5gTy35REa2pcmggqLFsOWpXniRqXwVDNjOERN/340zopv+UoR8b+myW8CAwEA\nAQKCAgEAkcoffF0JOBMOiHlAJhrNtSiX+ZruzNDlCxlgshUjyWEbfQG7sWbqSHUZ\njZflTrqyZqDpyca7Jp2ZM2Vocxa0klIMayfj08trCaOWY3pPeROE4d3HUJMPjEpH\nvEXTFdnVJIOBPgl3+vWfBfm17QIh9j4X3BVbVNNl3WCaiDGAl699Kl+Pe38cFeCh\nD3JZPEWsZ5SlvwjU8sNGbThjAWN8C1NjMuCXG4hGej5Ae3M/nPPR91jgnw4Me4Ut\nIL3K3RVyGqaqAPJjLsu0kWQUArJAGMfvUkXjwVklkaUV5SHtJBs+pdTXjyprTmJR\nvSXWWON5zkAEEJNY7QcZaeKYi96PFLUFI+ciEdnXn74CfSKhgZCBo+OyFZjDWW5R\nNmgAbZTN2RW0z+V54Lg36JfJrmiGs8TN06KwNjFo+iOJCdQnoUSIhTlmMfVbXPah\ntRfQvwqtfqVS9W/jkiGq9yDDqyXx093R/QTM/XqDlWJ2iOJFppOJefGFCWF6Fwll\nVT9povTAGQmXFiAxwFZxWtbFa0i8fP5QG80X6l/gRklSd6ZXAVvcLkaFGqxunDAe\nrYC2jBwHWRpVmbxw880SWRzlAsJXc7M8PQnBTlyX1mFZNnwAJgqplz0BQHQhQh4V\nqNfisUm9smtda+Hr9GBBUxs09ulery3I0lQjsArVxPqPVgUbFPECggEBANqLA5fH\n2LupOBoFH/fK5jixyGdSB8eJvU+XuS8RBBexnzTQApmDHiU7Axa/cKvxAfUgwBpU\n6OIsL6Lq6wowVInBgo7GraACwspGMIP8Z7+A8qDgSWIcpXP21Ny2RW+nukdH8ZnV\nTFtiFxLYU9GRfzSUcqvE0miKfMGP/S9Cqbew00K6CQ2xurLTR2AchfUQZJJIg7eF\nRBoftthXLQ+s1JoiLJX2gqCliFy32RMAUP+pKvKVJmVQh8bxEkoEzTV2eY7eTxsH\nJDH5hD66EZ5bW/nVAMruJ3iKjy3WvjDbnddNAz9IFKrd1RMP9dgSEKuSv/HhqwPe\n1q9Wm6LWZo8BlYcCggEBANp3M14QMcMxRlZE0TiSopi1CaE8OG0C9apToS1dol2s\n4lCsWHVPIC516LMPGU0bmCdtwJey1mgXQEKVxCWHkVhhoCKT/tN53o5qkptrhrXL\npbqmRfoMXI7LwJU+Vqi5fwSPGrSR/IzHwCUL7pHTbYN7wT5rr2rcC84XYSX31TFm\nNfMnbDuUk33ycAo07Vqts5A5FN+xViEUMFSDmfA2XmOAV77awz0l/3n3qOg9lQYe\nU4Av2nT19lGELirLInkB1ndLirWAcLaCBXKOLW4bzpNm9Bt8aiziVzcUzlJlLa+1\nnb/7//xzKi0eM/BhyJfhsmOz5B8AQ6Ca/keDk8M7JtkCggEARl8DDinE6VCpBv/l\ndlX4YgMlQ9fPN3pr4ig58iTpi3Ofj1L3s1TcLSLecMG+Vy9o8PTVxuTWhJWz1SMO\nAh7j6ePM1Yq2N9MLxDRrxOROyASOnCz8lEIjKL8vdc6fdz+sJO3OpzleuAJS6beM\n7euK6XRvpE3hbtZBK9bgsQonOkYPEOp0pds4AgM0dYdZvzrDF7OP7lVUQ5E4wFr5\n4JVHdEZS0wsoru/+g9STaqHscxaXBLvwPCl9Pxs7R2haZ7+5jr6Y/FwFVK5C3ivu\nJm7GpCDpe27KeO8tAZancXYWUlCzHfpo5Ug/Jz85a5UNlyHO+uUuuzVTLeyWew3M\nwnnBGwKCAQEAqGTBP3wUH3TX1p9s9cJxemvxZEra44woeIXF8wX9pV8hgzWVabb4\nA1f3ai31Pq5KdfnvPf8nrUxex/RRIOyCaDG4EW8qOS/zEKutHgef6nly4ZBQ2BC3\nN4pug5ttiNiSw5za5NyyYoGF5ghweA8UlwjJR6gRqri6kL0MsQt7VXyHkUmN787y\ncV5yZiut2PuTMVQOdu5miVDagAqAmdwOnXvMJtzRKU0kw4rWs0zklbbCfkhkh0sf\n9m2AeJPjmoqEGags3wKF3ugR8t8MvZbJgG0XNCiOXtKIj3iGIJTExm+jjNxd0OWk\nWOqy9lMpH4lky91ZtVuqxR0za0RMnWv24QKCAQBe8l0w9AYVNGDLv1jyPcbsncty\nNYI81yqe2mL+TC00sMCeil7C7WCP7kRklY01rH5q5gJ9Q1UV+bOj2fQdXDmQ5Bgo\n41jseh44gkbuXAeWcSDrDkJCrfvlNqFobTmUb8cdb9aQlHYfOJ31367LJspiw2SY\nmCbnLQ5sMnyBiMkcn0GfBV6IAkZVN73DPa8a1m/0Qrrv1GmBJFVbuZd9d/hAWpHa\nekhXPq0Sta+RNDfBR3aI5lAmVA17qRGiubQYJ+Ldq0aRJ40fGE51ctoSU/5RMcmh\n6+Qro+jSC94L46xMFp+1J5atgB1p/jVzTT/Ws7SLyotYUSL8zU7tcLiycQXs\n-----END RSA PRIVATE KEY-----\n'

struct ConcurrentLargeResponseContext {
    expected_body       string
    expected_link_count int
}

struct ConcurrentLargeResponseResult {
    status_code  int
    body_matches bool
    link_count   int
    err          string
}

fn test_https_large_bodies_can_be_processed_concurrently() {
    // Watchdog thread to prevent the test from hanging indefinitely on CI.
    spawn fn () {
        time.sleep(120 * time.second)
        eprintln('mbedtls concurrent large response test: timeout reached, exiting.')
        exit(1)
    }()

    mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0')!
    port := port_listener.addr()!.port()!
    port_listener.close()!

    mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{
        cert:                   concurrent_large_response_test_cert
        cert_key:               concurrent_large_response_test_key
        validate:               false
        in_memory_verification: true
    })!
    server := spawn serve_large_html_responses(mut listener, concurrent_large_response_body(),
        concurrent_large_response_requests)

    mut urls := []string{cap: concurrent_large_response_requests}
    for i in 0 .. concurrent_large_response_requests {
        urls << 'https://127.0.0.1:${port}/?req=${i}'
    }
    expected_body := concurrent_large_response_body()
    mut pp := pool.new_pool_processor(callback: concurrent_large_response_worker)
    pp.set_max_jobs(concurrent_large_response_workers)
    pp.set_shared_context(&ConcurrentLargeResponseContext{
        expected_body:       expected_body
        expected_link_count: concurrent_large_response_links
    })
    pp.work_on_items(urls)
    server.wait()

    for result in pp.get_results_ref[ConcurrentLargeResponseResult]() {
        assert result.err == ''
        assert result.status_code == 200
        assert result.body_matches
        assert result.link_count == concurrent_large_response_links
    }
}

fn test_https_request_timeout_closes_the_connection() {
    mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0')!
    port := port_listener.addr()!.port()!
    port_listener.close()!

    mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{
        cert:                   concurrent_large_response_test_cert
        cert_key:               concurrent_large_response_test_key
        validate:               false
        in_memory_verification: true
    })!
    server := spawn serve_incomplete_https_response(mut listener)

    mut req := http.new_request(.get, 'https://127.0.0.1:${port}', '')
    req.read_timeout = 250 * time.millisecond
    req.validate = false
    req.do() or {
        assert server.wait()
        return
    }
    panic('expected the HTTPS request to time out')
}

fn test_https_chunked_body_is_returned_by_http_get_text() {
    mut port_listener := net.listen_tcp(.ip, '127.0.0.1:0')!
    port := port_listener.addr()!.port()!
    port_listener.close()!

    mut listener := mbedtls.new_ssl_listener('127.0.0.1:${port}', mbedtls.SSLConnectConfig{
        cert:                   concurrent_large_response_test_cert
        cert_key:               concurrent_large_response_test_key
        validate:               false
        in_memory_verification: true
    })!
    expected_body := concurrent_large_response_body()
    server := spawn serve_chunked_html_response(mut listener, expected_body)

    body := http.get_text('https://127.0.0.1:${port}/chunked')
    server.wait()

    assert body == expected_body
}

fn concurrent_large_response_body() string {
    return '<html><body>' +
        '<article><a href="/item">payload</a></article>'.repeat(concurrent_large_response_links) +
        '</body></html>'
}

fn serve_incomplete_https_response(mut listener mbedtls.SSLListener) bool {
    defer {
        listener.shutdown() or {}
    }
    mut conn := listener.accept() or { panic(err) }
    defer {
        conn.shutdown() or {}
    }
    mut request_buf := []u8{len: 2048}
    _ = conn.read(mut request_buf) or { return false }
    conn.write_string('HTTP/1.1 200 OK\r\nContent-Length: 2\r\n') or { return false }
    conn.set_read_timeout(time.second)
    mut drain_buf := []u8{len: 128}
    _ = conn.read(mut drain_buf) or { return err.code() != net.err_timed_out_code }
    return false
}

fn serve_large_html_responses(mut listener mbedtls.SSLListener, body string, request_count int) {
    defer {
        listener.shutdown() or {}
    }
    mut handlers := []thread{cap: request_count}
    for _ in 0 .. request_count {
        mut conn := listener.accept() or { panic(err) }
        handlers << spawn write_large_html_response(mut conn, body)
    }
    handlers.wait()
}

fn serve_chunked_html_response(mut listener mbedtls.SSLListener, body string) {
    defer {
        listener.shutdown() or {}
    }
    mut conn := listener.accept() or { panic(err) }
    write_chunked_html_response(mut conn, body)
}

fn write_large_html_response(mut conn mbedtls.SSLConn, body string) {
    defer {
        conn.shutdown() or {}
    }
    mut request_buf := []u8{len: 2048}
    _ = conn.read(mut request_buf) or { return }
    header := 'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: ${body.len}\r\nConnection: close\r\n\r\n'
    conn.write_string(header) or { return }
    mut start := 0
    for start < body.len {
        end := if start + concurrent_large_response_chunk_size > body.len {
            body.len
        } else {
            start + concurrent_large_response_chunk_size
        }
        conn.write_string(body[start..end]) or { return }
        start = end
        time.sleep(1 * time.millisecond)
    }
}

fn write_chunked_html_response(mut conn mbedtls.SSLConn, body string) {
    defer {
        conn.shutdown() or {}
    }
    mut request_buf := []u8{len: 2048}
    _ = conn.read(mut request_buf) or { return }
    header := 'HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n'
    conn.write_string(header) or { return }
    mut start := 0
    for start < body.len {
        end := if start + concurrent_large_response_chunk_size > body.len {
            body.len
        } else {
            start + concurrent_large_response_chunk_size
        }
        chunk := body[start..end]
        conn.write_string('${chunk.len:x}\r\n') or { return }
        conn.write_string(chunk) or { return }
        conn.write_string('\r\n') or { return }
        start = end
        time.sleep(1 * time.millisecond)
    }
    conn.write_string('0\r\n\r\n') or { return }
}

fn concurrent_large_response_worker(mut pp pool.PoolProcessor, idx int, _wid int) &ConcurrentLargeResponseResult {
    url := pp.get_item[string](idx)
    ctx := unsafe { &ConcurrentLargeResponseContext(pp.get_shared_context()) }
    resp := http.fetch(
        method:   .get
        url:      url
        validate: false
    ) or { return &ConcurrentLargeResponseResult{
        err: '${err}'
    } }
    doc := html.parse(resp.body)
    return &ConcurrentLargeResponseResult{
        status_code:  resp.status_code
        body_matches: resp.body == ctx.expected_body
        link_count:   doc.get_tags(name: 'a').len
    }
}