service.nct

see ./index.nct

see ./configuration.nct

see ./operational.nct

use std/http.{Handler, IncomingRequest, Limits, OutgoingResponse, Responder, ResponseHead, RouteDispatch, RouteMatch, Router, Server, ServerConnection, SessionCookie, Status}
use std/config.{Configuration, PublishedConfiguration}
use std/fs
use std/log.{Event, Field, Level}
use std/net
use std/net.{SocketAddress, TcpStream}
use std/session.SessionId
use std/string.String
use std/store.Store
use std/sync.Mutex
use std/task
use std/task.{TaskGroup, Timeout}
use std/time.{Duration, SystemTime}
use std/vec.Vec

const CONNECTION_LIMIT: usize = 2

const REQUEST_COUNT: usize = 7

const LARGE_BODY_BYTES: usize = 32768

const TRANSFER_BYTES: usize = 1024

struct ServiceState {
    configuration: PublishedConfiguration
    session_cookie: SessionCookie
    event_records: Mutex<Vec<Event>>
    durable_state: Mutex<Store>
}

construct ServiceState {
    func new(configuration: PublishedConfiguration, durable_state: Store): Self! {
        return ServiceState {
            configuration: move configuration,
            session_cookie: SessionCookie.new("session", "/", false)?,
            event_records: Mutex<Vec<Event>>.new(Vec<Event>.empty())?,
            durable_state: Mutex<Store>.new(move durable_state)?,
        }
    }
}

instance ServiceState {
    async method &self.timeout(): Duration! {
        let view = await self.configuration.current()
        let timeout = configured_timeout(&view as &Configuration)?
        return timeout
    }

    async method &self.response(): String! {
        let view = await self.configuration.current()
        let response = configured_response(&view as &Configuration)?
        return move response
    }

    async method &self.token_display(): String! {
        let view = await self.configuration.current()
        let display = configured_token_display(&view as &Configuration)?
        return move display
    }

    noalloc method &self.cookie(): &SessionCookie from self { return &self.session_cookie }

    noalloc method &self.event_store(): &Mutex<Vec<Event>> from self {
        return &self.event_records
    }

    noalloc method &self.store(): &Mutex<Store> from self { return &self.durable_state }
}

func byte_buffer(length: usize): Vec<u8> {
    var output: Vec<u8> = Vec.with_capacity(length)
    while output.len() < length { output.push(u8.truncate(0)) }
    return move output
}

func append_bytes(output: &+Vec<u8>, bytes: &[u8]): void {
    var offset: usize = 0
    while offset < bytes.len() {
        output.push(bytes[offset])
        offset += 1
    }
    return
}

async func stream_large_response(
    responder: Responder,
    timeout: Duration,
): ServerConnection?! {
    let head = ResponseHead.ok()
    var writer = await responder.begin_chunked_with_timeout(move head, timeout)?
    var chunk = byte_buffer(TRANSFER_BYTES)
    var offset: usize = 0
    while offset < chunk.len() {
        chunk[offset] = u8.truncate(120)
        offset += 1
    }
    var written: usize = 0
    while written < LARGE_BODY_BYTES {
        await writer.write_with_timeout(&chunk, timeout)?
        written += chunk.len()
    }
    return await writer.finish_with_timeout(timeout)?
}

async func read_large_request(
    request: &+IncomingRequest,
    timeout: Duration,
): bool {
    var scratch = byte_buffer(TRANSFER_BYTES)
    var received: usize = 0
    loop {
        let count = await request.read_with_timeout(&+scratch, timeout)
        catch _ { return false }
        if count == 0 { return received == LARGE_BODY_BYTES }
        received += count
    }
}

async func handle_large_route(
    state: &ServiceState,
    request: IncomingRequest,
    _route: RouteMatch,
): ServerConnection?! {
    var request_owner = move request
    let timeout = await state.timeout()?
    if !await read_large_request(&+request_owner, timeout) {
        request_owner.close()
        return error.new("http-service.large_body", "large request body changed")
    }
    let responder = await request_owner.finish_body_with_timeout(timeout)?
    return await stream_large_response(move responder, timeout)?
}

async func handle_close_route(
    state: &ServiceState,
    request: IncomingRequest,
    _route: RouteMatch,
): ServerConnection?! {
    let timeout = await state.timeout()?
    var responder = await request.finish_body_with_timeout(timeout)?
    responder.close_after_response()
    let response = OutgoingResponse.text(Status.ok(), "done")?
    return await responder.respond_with_timeout(move response, timeout)?
}

async func record_event(state: &ServiceState, event: Event): void! {
    let event_output = event.to_json_line()?
    await persist_event(state, &event_output)?
    await retain_event(state, move event)
    return
}

async func persist_event(state: &ServiceState, event_output: &str): void! {
    let store = state.store()
    var durable = await store.lock()
    durable.set("last-event".bytes(), event_output.bytes())?
    let _sequence = await durable.commit()?
    return
}

async func retain_event(state: &ServiceState, event: Event): void {
    let event_store = state.event_store()
    var events = await event_store.lock()
    events.push(move event)
    return
}

func session_for_request(state: &ServiceState, request: &IncomingRequest): SessionId! {
    let session = state.cookie().read_headers(request.headers())?
    return session otherwise { return SessionId.issue()? }
}

async func handle_user_route(
    state: &ServiceState,
    request: IncomingRequest,
    route: RouteMatch,
): ServerConnection?! {
    let timeout = await state.timeout()?
    let identifier = session_for_request(state, &request)?
    let user = route.parameter("id") otherwise {
        let response = OutgoingResponse.text(Status.bad_request(), "missing user")?
        return await request.respond_with_timeout(
            move response,
            timeout,
        )?
    }
    var query = request.target().query_pairs()
    let parameter = query.next() otherwise {
        let response = OutgoingResponse.text(Status.bad_request(), "missing query")?
        return await request.respond_with_timeout(
            move response,
            timeout,
        )?
    }
    if user != "alice" || parameter.name() != "view" || parameter.value() != "full" {
        let response = OutgoingResponse.text(Status.bad_request(), "invalid user query")?
        return await request.respond_with_timeout(
            move response,
            timeout,
        )?
    }
    let timestamp = SystemTime.now()?
    var event = Event.new(timestamp, Level.info, "request.complete")?
    let route_field = Field.text("route", "user")?
    event.add(move route_field)?
    let status_field = Field.unsigned("status", 200)?
    event.add(move status_field)?
    let session_field = Field.redacted("session")?
    event.add(move session_field)?
    let token_display = await state.token_display()?
    let token_field = Field.text("configured_token", &token_display)?
    event.add(move token_field)?
    await record_event(state, move event)?
    let response_text = await state.response()?
    var response = OutgoingResponse.json(Status.ok(), &response_text)?
    let cookie_policy = state.cookie()
    let session_header = cookie_policy.set_header(&identifier)?
    response.append_header(move session_header)
    return await request.respond_with_timeout(
        move response,
        timeout,
    )?
}

async func handle_failure_route(
    _state: &ServiceState,
    _request: IncomingRequest,
    _route: RouteMatch,
): ServerConnection?! {
    return error.new("http-service.handler_failed", "intentional handler failure")
}

func build_router(
    configuration: PublishedConfiguration,
    durable_state: Store,
): Router<ServiceState>! {
    var router = Router.new(ServiceState.new(move configuration, move durable_state)?)
    let large: Handler<ServiceState> = (state, request, route) {
        handle_large_route(state, move request, move route)
    }
    router.post("/large", move large)?
    let close: Handler<ServiceState> = (state, request, route) {
        handle_close_route(state, move request, move route)
    }
    router.get("/close", move close)?
    let user: Handler<ServiceState> = (state, request, route) {
        handle_user_route(state, move request, move route)
    }
    router.get("/users/:id", move user)?
    let failure: Handler<ServiceState> = (state, request, route) {
        handle_failure_route(state, move request, move route)
    }
    router.get("/fail", move failure)?
    return move router
}

async func handle_connection(
    router: &Router<ServiceState>,
    connection: ServerConnection,
): bool {
    let timeout = await router.state().timeout() catch _ { return false }
    var owner = move connection
    loop {
        let request = await owner.read_request_with_timeout(timeout) catch failure {
            return failure.has_code("std.http.invalid_syntax")
            || failure.has_code("std.http.timed_out")
        }
        let dispatched = await router.dispatch(move request) catch failure {
            return failure.has_code("http-service.handler_failed")
        }
        match move dispatched {
            RouteDispatch.handled(next) {
                owner = move next otherwise { return true }
            }
            RouteDispatch.not_found(unmatched) {
                let response = OutgoingResponse.text(Status.not_found(), "not found")
                catch _ { return false }
                let next = await unmatched.respond_with_timeout(move response, timeout)
                catch _ { return false }
                owner = move next otherwise { return true }
            }
            RouteDispatch.method_not_allowed(unmatched) {
                let response = OutgoingResponse.text(
                    Status.method_not_allowed(),
                    "method not allowed",
                ) catch _ { return false }
                let next = await unmatched.respond_with_timeout(move response, timeout)
                catch _ { return false }
                owner = move next otherwise { return true }
            }
        }
    }
}

async func drain_handlers(
    handlers: TaskGroup<bool>,
): (usize, usize)! {
    var owner = move handlers
    var succeeded: usize = 0
    var failed: usize = 0
    while !owner.is_empty() {
        if await observe_handler(&+owner)? {
            succeeded += 1
        } else {
            failed += 1
        }
    }
    return (succeeded, failed)
}

async func drain_handlers_before(
    handlers: TaskGroup<bool>,
    deadline: Duration,
): (usize, usize)! {
    match await task.with_timeout(drain_handlers(move handlers), deadline) {
        Timeout.completed(result) { return move result? }
        Timeout.elapsed {
            return error.new(
                "http-service.shutdown_timed_out",
                "handler drain exceeded the shutdown deadline",
            )
        }
    }
}

async func observe_handler(
    handlers: &+TaskGroup<bool>,
): bool! {
    let completed = await handlers.next() otherwise {
        return error.new("http-service.group", "non-empty handler group produced no result")
    }
    return completed
}

async func serve_bounded(
    server: Server,
    configuration: PublishedConfiguration,
    durable_state: Store,
    state_path: &str,
    shutdown_deadline: Duration,
): (usize, usize, String)! {
    var owner = move server
    let router = build_router(move configuration, move durable_state)?
    var handlers: TaskGroup<bool> = TaskGroup.empty()
    var accepted: usize = 0
    var succeeded: usize = 0
    var failed: usize = 0
    while accepted < REQUEST_COUNT {
        if handlers.len() == CONNECTION_LIMIT {
            if await observe_handler(&+handlers)? {
                succeeded += 1
            } else {
                failed += 1
            }
        }
        let connection = await owner.accept_with_timeout(shutdown_deadline)?
        handlers.add(handle_connection(&router, move connection))
        accepted += 1
    }
    owner.close()
    let drained = await drain_handlers_before(move handlers, shutdown_deadline)?
    let state = router.into_state()
    let event_output = await rendered_event(&state)?
    await close_and_recover_event(&state, &event_output, state_path)?
    return (succeeded + drained.0, failed + drained.1, move event_output)
}

async func rendered_event(state: &ServiceState): String! {
    let event_store = state.event_store()
    var events = await event_store.lock()
    let stored_events: &Vec<Event> = &events as &Vec<Event>
    if stored_events.len() == 0 {
        return error.new("http-service.events", "no structured event was recorded")
    }
    let output = stored_events[stored_events.len() - 1].to_json_line()?
    return move output
}

async func close_and_recover_event(
    state: &ServiceState,
    event_output: &str,
    path: &str,
): void! {
    let committed_sequence = await close_store(state)?

    var restarted = await Store.open(path)?
    let recovered = restarted.get("last-event".bytes()) otherwise {
        return error.new("http-service.state", "committed event was not recovered")
    }
    if recovered != event_output.bytes()
    || restarted.committed_sequence() != committed_sequence {
        return error.new("http-service.state", "restarted state changed")
    }
    await restarted.close()?
    await fs.remove_file(path)?
    return
}

async func close_store(state: &ServiceState): u64! {
    let store = state.store()
    var durable = await store.lock()
    let committed_sequence = await durable.commit()?
    await durable.close()?
    return committed_sequence
}

async func read_until_eof(
    stream: &+TcpStream,
    timeout: Duration,
): usize! {
    var scratch = byte_buffer(TRANSFER_BYTES)
    var received: usize = 0
    loop {
        let count = await stream.read_with_timeout(&+scratch, timeout)?
        if count == 0 { return received }
        received += count
    }
}

async func read_response_text(
    stream: &+TcpStream,
    timeout: Duration,
): String! {
    var scratch = byte_buffer(TRANSFER_BYTES)
    var bytes: Vec<u8> = Vec.empty()
    loop {
        let received = await stream.read_with_timeout(&+scratch, timeout)?
        if received == 0 { return String.from_utf8(&bytes)? }
        var offset: usize = 0
        while offset < received {
            bytes.push(scratch[offset])
            offset += 1
        }
    }
}

async func text_exchange(
    address: SocketAddress,
    timeout: Duration,
    request: &str,
    expected: &str,
): bool {
    var stream = await net.connect_tcp_with_timeout(address, timeout) catch _ { return false }
    await stream.write_with_timeout(request.bytes(), timeout) catch _ { return false }
    let response = await read_response_text(&+stream, timeout) catch _ { return false }
    return (&response as &str).contains(expected)
}

async func large_reused_exchange(address: SocketAddress, timeout: Duration): bool {
    var stream = await net.connect_tcp_with_timeout(address, timeout) catch _ { return false }
    var request: Vec<u8> = Vec.with_capacity(LARGE_BODY_BYTES + 256)
    append_bytes(
        &+request,
        "POST /large HTTP/1.1\r\nHost: localhost\r\nContent-Length: 32768\r\n\r\n".bytes(),
    )
    let body_start = request.len()
    while request.len() - body_start < LARGE_BODY_BYTES {
        request.push(u8.truncate(97))
    }
    append_bytes(
        &+request,
        "GET /close HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n".bytes(),
    )
    await stream.write_with_timeout(&request, timeout) catch _ { return false }
    let received = await read_until_eof(&+stream, timeout) catch _ { return false }
    return received > LARGE_BODY_BYTES
}

async func malformed_exchange(address: SocketAddress, timeout: Duration): bool {
    var stream = await net.connect_tcp_with_timeout(address, timeout) catch _ { return false }
    await stream.write_with_timeout(
        "GET /bad HTTP/1.1\r\nHost: localhost\r\nConnection: close,\r\n\r\n".bytes(),
        timeout,
    ) catch _ { return false }
    let received = await read_until_eof(&+stream, timeout) catch _ { return false }
    return received == 0
}

async func timed_out_exchange(address: SocketAddress, timeout: Duration): bool {
    var stream = await net.connect_tcp_with_timeout(address, timeout) catch _ { return false }
    let received = await read_until_eof(&+stream, timeout) catch _ { return false }
    return received == 0
}

async func routed_parameter_exchange(
    address: SocketAddress,
    timeout: Duration,
    expected_response: &str,
): bool {
    var stream = await net.connect_tcp_with_timeout(address, timeout) catch _ { return false }
    await stream.write_with_timeout(
        "GET /users/alice?view=full HTTP/1.1\r\nHost: localhost\r\nCookie: session=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\r\nConnection: close\r\n\r\n".bytes(),
        timeout,
    ) catch _ { return false }
    let response = await read_response_text(&+stream, timeout) catch _ { return false }
    return (&response as &str).contains(expected_response)
    && (&response as &str).contains("set-cookie: session=")
    && (&response as &str).contains("HttpOnly; SameSite=Lax")
}

async func missing_route_exchange(address: SocketAddress, timeout: Duration): bool {
    return await text_exchange(
        address,
        timeout,
        "GET /missing HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n",
        "404",
    )
}

async func method_mismatch_exchange(address: SocketAddress, timeout: Duration): bool {
    return await text_exchange(
        address,
        timeout,
        "POST /users/alice HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
        "405",
    )
}

async func handler_failure_exchange(address: SocketAddress, timeout: Duration): bool {
    var stream = await net.connect_tcp_with_timeout(address, timeout) catch _ { return false }
    await stream.write_with_timeout(
        "GET /fail HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n".bytes(),
        timeout,
    ) catch _ { return false }
    let received = await read_until_eof(&+stream, timeout) catch _ { return false }
    return received == 0
}

async func run_clients(
    address: SocketAddress,
    timeout: Duration,
    expected_response: &str,
): usize! {
    var requests: TaskGroup<bool> = TaskGroup.empty()
    requests.add(large_reused_exchange(address, timeout))
    requests.add(routed_parameter_exchange(address, timeout, expected_response))
    requests.add(missing_route_exchange(address, timeout))
    requests.add(method_mismatch_exchange(address, timeout))
    requests.add(malformed_exchange(address, timeout))
    requests.add(timed_out_exchange(address, timeout))
    requests.add(handler_failure_exchange(address, timeout))
    var passed: usize = 0
    while !requests.is_empty() {
        let result = await requests.next() otherwise {
            return error.new("http-service.group", "non-empty client group produced no result")
        }
        if result { passed += 1 }
    }
    return passed
}

async func run_service_example(): i32! {
    let arguments = parse_service_arguments()?
    let configuration = await build_service_configuration(&arguments)?
    if operational_mode(&arguments) {
        return await run_operational_service(&arguments, move configuration)
    }
    let listen_text = configured_listen(&configuration)?
    let listen = SocketAddress.parse(&listen_text) otherwise {
        return error.new("http-service.configuration", "listen is not a numeric socket address")
    }
    let expected_response = configured_response(&configuration)?
    let state_path = configured_state_path(&configuration)?
    let published = PublishedConfiguration.new(move configuration)?
    let limits = Limits.new(1024, 8192, 32, 4, 1024, 128, 65536)?
    let operation_timeout = Duration.from_seconds(3)
    let durable_state = await Store.open(&state_path)?
    let server = await Server.bind_with_limits(listen, limits)?
    let address = server.local_address()?
    let completed = await task.join(
        serve_bounded(
            move server,
            published.share(),
            move durable_state,
            &state_path,
            operation_timeout,
        ),
        run_clients(address, operation_timeout, &expected_response),
    )
    let service = move completed.0?
    let clients = move completed.1?
    if service.0 != REQUEST_COUNT {
        return error.new("http-service.handlers", "successful handler count changed")
    }
    if service.1 != 0 {
        return error.new("http-service.handlers", "failed handler count changed")
    }
    let event_output: &str = &service.2
    if !event_output.contains("\"event\":\"request.complete\"")
    || !event_output.contains("\"session\":\"[redacted]\"")
    || !event_output.contains("\"configured_token\":\"[redacted]\"")
    || event_output.contains("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
    || event_output.contains("environment-secret") {
        return error.new("http-service.events", "structured operational output changed")
    }
    if clients != REQUEST_COUNT {
        return error.new("http-service.clients", "successful client count changed")
    }
    return 0
}