HTTP Client

// Vapor 4.122.2, abridged
public protocol Client: Sendable {
    var eventLoop: EventLoop { get }
    var byteBufferAllocator: ByteBufferAllocator { get }
    func delegating(to eventLoop: EventLoop) -> Client
    func logging(to logger: Logger) -> Client
    func allocating(to byteBufferAllocator: ByteBufferAllocator) -> Client
    func send(_ request: ClientRequest) -> EventLoopFuture<ClientResponse>
}

import Vapor
import AsyncHTTPClient

struct BookMetadata: Content, Equatable {
    let isbn: String
    let title: String
    let authors: [String]
    let pages: Int
}

struct NewReview: Content {
    let isbn: String
    let stars: Int
    let text: String
}

struct ReviewReceipt: Content, Equatable {
    let id: Int
    let isbn: String
}

/// A typed wrapper around the external catalog service. It owns the URL building, the timeouts and the mapping from
/// upstream failures to the status codes Bookshelf wants to expose.
struct CatalogClient: Sendable {
    let client: any Client
    let baseURL: String

    func metadata(isbn: String, traceContext: HTTPHeaders = [:]) async throws -> BookMetadata {
        // Never interpolate unchecked user input into a URL: allow only ISBN characters.
        guard !isbn.isEmpty, isbn.count <= 17, isbn.allSatisfy({ $0.isNumber || $0 == "-" || $0 == "X" }) else {
            throw Abort(.badRequest, reason: "Malformed ISBN.")
        }
        var headers: HTTPHeaders = ["Accept": "application/json"]
        headers.propagateTraceContext(from: traceContext)

        let response = try await send {
            try await client.get(URI(string: "\(baseURL)/isbn/\(isbn)"), headers: headers) { request in
                request.timeout = .seconds(3)
            }
        }
        switch response.status {
        case .ok:
            return try response.content.decode(BookMetadata.self)
        case .notFound:
            throw Abort(.notFound, reason: "The catalog does not know ISBN \(isbn).")
        default:
            throw Abort(.badGateway, reason: "Catalog answered \(response.status.code).")
        }
    }

    func submit(_ review: NewReview) async throws -> ReviewReceipt {
        let response = try await send {
            try await client.post(URI(string: "\(baseURL)/reviews"), headers: ["X-Api-Key": "replace-me"], content: review)
        }
        guard response.status == .created else {
            throw Abort(.badGateway, reason: "Catalog refused the review with \(response.status.code).")
        }
        return try response.content.decode(ReviewReceipt.self)
    }

    /// Turns transport failures (no response at all) into gateway statuses.
    private func send(_ operation: () async throws -> ClientResponse) async throws -> ClientResponse {
        do {
            return try await operation()
        } catch let error as HTTPClientError where error == .deadlineExceeded || error == .readTimeout {
            throw Abort(.gatewayTimeout, reason: "The catalog did not answer in time.")
        } catch let error as HTTPClientError where error == .connectTimeout {
            throw Abort(.gatewayTimeout, reason: "Could not connect to the catalog.")
        } catch let error as any AbortError {
            throw error
        } catch {
            // Connection refused, reset, TLS failure, DNS failure: no response at all. Do not leak the raw error text.
            throw Abort(.badGateway, reason: "The catalog is unreachable.")
        }
    }
}

extension HTTPHeaders {
    /// Copies the W3C Trace Context headers from an incoming request so the downstream span joins the same trace.
    mutating func propagateTraceContext(from incoming: HTTPHeaders) {
        for name in ["traceparent", "tracestate", "baggage"] {
            if let value = incoming.first(name: name) {
                replaceOrAdd(name: name, value: value)
            }
        }
    }
}

func clientRoutes(_ routes: some RoutesBuilder, catalog: CatalogClient) {
    routes.get("books", ":isbn", "metadata") { req async throws -> BookMetadata in
        let isbn = try req.parameters.require("isbn")
        // req.client logs to req.logger and runs on the request's event loop; build the typed client per request.
        let perRequest = CatalogClient(client: req.client, baseURL: catalog.baseURL)
        return try await withRetries {
            try await perRequest.metadata(isbn: isbn, traceContext: req.headers)
        }
    }

    routes.post("reviews") { req async throws -> Response in
        let review = try req.content.decode(NewReview.self)
        let receipt = try await CatalogClient(client: req.client, baseURL: catalog.baseURL).submit(review)
        return try await receipt.encodeResponse(status: .created, for: req)
    }
}

import Vapor

extension CatalogClient {
    /// GET with a query string and a bearer token, built in the beforeSend closure.
    func search(title: String, limit: Int, token: String) async throws -> [BookMetadata] {
        let response = try await client.get(URI(string: "\(baseURL)/search")) { request in
            try request.query.encode(["q": title, "limit": String(limit)])
            request.headers.bearerAuthorization = BearerAuthorization(token: token)
            request.headers.replaceOrAdd(name: .accept, value: "application/json")
            request.timeout = .seconds(3)
        }
        guard response.status == .ok else {
            throw Abort(.badGateway, reason: "Catalog answered \(response.status.code).")
        }
        return try response.content.decode([BookMetadata].self)
    }
}

struct MissingSecret: Error {
    let variable: String
}

/// Production must provide CATALOG_TOKEN; development and tests use an obvious placeholder.
func catalogToken(for environment: Environment) throws -> String {
    if let token = Environment.get("CATALOG_TOKEN") { return token }
    guard environment != .production else { throw MissingSecret(variable: "CATALOG_TOKEN") }
    return "dev-token"
}

func searchRoutes(_ routes: some RoutesBuilder, catalog: CatalogClient, token: String) {
    routes.get("search") { req async throws -> [BookMetadata] in
        let title = try req.query.get(String.self, at: "title")
        let perRequest = CatalogClient(client: req.client, baseURL: catalog.baseURL)
        return try await perRequest.search(title: title, limit: 5, token: token)
    }
}

import Vapor
import AsyncHTTPClient

func configureHTTPClient(_ app: Application) throws {
    var configuration = HTTPClient.Configuration()
    configuration.timeout = .init(connect: .seconds(5), read: .seconds(15))
    configuration.redirectConfiguration = .follow(max: 5, allowCycles: false)
    configuration.connectionPool.idleTimeout = .seconds(30)
    configuration.connectionPool.concurrentHTTP1ConnectionsPerHostSoftLimit = 16
    configuration.httpVersion = .automatic
    // configuration.proxy = .server(host: "proxy.internal", port: 3128)
    // configuration.decompression = .enabled(limit: .ratio(10))

    // Must run before the first use of app.client; afterwards Vapor only logs a warning and ignores the change.
    app.http.client.configuration = configuration
}

import Vapor

/// Retries an idempotent operation with exponential backoff and jitter, only for failures that may be transient.
/// Never wrap a non-idempotent call (a POST that creates something) unless the upstream supports idempotency keys.
func withRetries<Result: Sendable>(
    attempts: Int = 3,
    initialDelay: Duration = .milliseconds(200),
    _ operation: @Sendable () async throws -> Result
) async throws -> Result {
    var delay = initialDelay
    for attempt in 1...attempts {
        do {
            return try await operation()
        } catch let error as Abort
            where attempt < attempts && [.badGateway, .serviceUnavailable, .gatewayTimeout].contains(error.status) {
            let jitter = Double.random(in: 0.5...1.5)
            try await Task.sleep(for: delay * jitter)
            delay *= 2
        }
    }
    preconditionFailure("unreachable: the last attempt either returns or throws")
}

import Vapor

/// A small cache-aside store with a time to live and request coalescing: concurrent misses for one key share a single
/// upstream call, and failures are never cached.
actor TTLCache<Value: Sendable> {
    private struct Entry {
        let value: Value
        let expires: ContinuousClock.Instant
    }

    private var entries: [String: Entry] = [:]
    private var inFlight: [String: Task<Value, any Error>] = [:]
    private let ttl: Duration
    private let clock = ContinuousClock()

    init(ttl: Duration) {
        self.ttl = ttl
    }

    func value(for key: String, load: @escaping @Sendable () async throws -> Value) async throws -> Value {
        if let entry = entries[key], entry.expires > clock.now {
            return entry.value
        }
        if let running = inFlight[key] {
            return try await running.value
        }
        let task = Task { try await load() }
        inFlight[key] = task
        defer { inFlight[key] = nil }
        let value = try await task.value
        entries[key] = Entry(value: value, expires: clock.now.advanced(by: ttl))
        return value
    }

    func invalidate(_ key: String) {
        entries[key] = nil
    }
}

/// The catalog wrapped with the cache. The client is created from the application, not from a request, because the
/// cached value outlives the request that populated it.
struct CachedCatalog: Sendable {
    let catalog: CatalogClient
    let cache = TTLCache<BookMetadata>(ttl: .seconds(300))

    func metadata(isbn: String) async throws -> BookMetadata {
        let catalog = self.catalog
        return try await cache.value(for: isbn) {
            try await catalog.metadata(isbn: isbn)
        }
    }
}

func cachedRoutes(_ routes: some RoutesBuilder, cached: CachedCatalog) {
    routes.get("books", ":isbn", "cached") { req async throws -> BookMetadata in
        try await cached.metadata(isbn: try req.parameters.require("isbn"))
    }
}

// Package.swift (dependencies and the App target, abridged)
.package(url: "https://github.com/vapor/vapor.git", from: "4.122.2"),
.package(url: "https://github.com/swift-server/async-http-client.git", from: "1.30.0"),
// ...
.executableTarget(name: "App", dependencies: [
    .product(name: "Vapor", package: "vapor"),
    .product(name: "AsyncHTTPClient", package: "async-http-client"),
])

import Vapor
import AsyncHTTPClient
import NIOCore

func streamingRoutes(_ routes: some RoutesBuilder, catalogURL: String) {
    // Stream a large upstream body straight through without buffering it.
    routes.get("covers", ":isbn") { req async throws -> Response in
        let isbn = try req.parameters.require("isbn")
        guard isbn.allSatisfy({ $0.isNumber || $0 == "-" }) else { throw Abort(.badRequest) }
        let http = req.application.http.client.shared

        let upstream = try await http.execute(
            HTTPClientRequest(url: "\(catalogURL)/covers/\(isbn)"),
            timeout: .seconds(20),
            logger: req.logger
        )
        guard upstream.status == .ok else { throw Abort(.badGateway) }

        let response = Response(status: .ok)
        if let type = upstream.headers.first(name: "content-type") {
            response.headers.replaceOrAdd(name: .contentType, value: type)
        }
        response.body = .init(managedAsyncStream: { writer in
            for try await chunk in upstream.body {
                try await writer.writeBuffer(chunk)
            }
        })
        return response
    }

    // Read a small body with a hard limit; collect(upTo:) throws NIOTooManyBytesError past it.
    routes.get("books", ":isbn", "bounded") { req async throws -> BookMetadata in
        let isbn = try req.parameters.require("isbn")
        guard isbn.allSatisfy({ $0.isNumber || $0 == "-" }) else { throw Abort(.badRequest) }
        let upstream = try await req.application.http.client.shared.execute(
            HTTPClientRequest(url: "\(catalogURL)/isbn/\(isbn)"),
            timeout: .seconds(5),
            logger: req.logger
        )
        guard upstream.status == .ok else { throw Abort(.badGateway) }
        do {
            let buffer = try await upstream.body.collect(upTo: 64 * 1024)
            return try JSONDecoder().decode(BookMetadata.self, from: Data(buffer.readableBytesView))
        } catch is NIOTooManyBytesError {
            throw Abort(.badGateway, reason: "Upstream body exceeded 64 KiB.")
        }
    }
}

import Vapor

/// Headers a gateway forwards to the upstream (an allow-list is safer than a block-list).
private let forwardedRequestHeaders: [HTTPHeaders.Name] = [
    .accept, .acceptLanguage, .authorization, .contentType, .ifNoneMatch,
    HTTPHeaders.Name("traceparent"), HTTPHeaders.Name("tracestate"), HTTPHeaders.Name("X-Request-ID"),
]

/// Hop-by-hop headers (RFC 9110, section 7.6.1) must never be copied from the upstream response; `Date` and
/// `Content-Length` are set again by this server.
private let hopByHop: Set<String> = [
    "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding",
    "upgrade", "content-length", "date",
]

func gatewayRoutes(_ routes: some RoutesBuilder, upstream: String) {
    for method in [HTTPMethod.GET, .POST, .PUT, .DELETE] {
        routes.on(method, "gateway", "**", body: .collect(maxSize: "1mb")) { req async throws -> Response in
            // Re-encode every segment and refuse dot segments so the path cannot escape the upstream prefix.
            let segments = req.parameters.getCatchall()
            guard !segments.contains(where: { $0 == ".." || $0 == "." || $0.contains("/") }) else {
                throw Abort(.badRequest)
            }
            let path = segments
                .compactMap { $0.addingPercentEncoding(withAllowedCharacters: .urlPathAllowed) }
                .joined(separator: "/")
            var url = URI(string: "\(upstream)/\(path)")
            url.query = req.url.query

            var headers = HTTPHeaders()
            for name in forwardedRequestHeaders {
                for value in req.headers[name] { headers.add(name: name, value: value) }
            }
            headers.replaceOrAdd(name: "X-Forwarded-Host", value: req.headers.first(name: .host) ?? "")
            headers.replaceOrAdd(name: "X-Forwarded-Proto", value: req.url.scheme ?? "http")
            if let peer = req.peerAddress?.ipAddress {
                headers.replaceOrAdd(name: "X-Forwarded-For", value: peer)
            }
            headers.replaceOrAdd(name: "X-Request-ID", value: req.id)

            var outgoing = ClientRequest(method: req.method, url: url, headers: headers, body: req.body.data)
            outgoing.timeout = .seconds(10)
            let upstreamResponse = try await req.client.send(outgoing)

            var responseHeaders = HTTPHeaders()
            for (name, value) in upstreamResponse.headers where !hopByHop.contains(name.lowercased()) {
                responseHeaders.add(name: name, value: value)
            }
            return Response(
                status: upstreamResponse.status,
                headers: responseHeaders,
                body: upstreamResponse.body.map { Response.Body(buffer: $0) } ?? .empty
            )
        }
    }
}

$ curl -si -H 'traceparent: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01' \
    'http://127.0.0.1:8080/gateway/isbn/9780441013593?x=1'
HTTP/1.1 200 OK
content-type: application/json; charset=utf-8
X-Seen-Trace: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01
content-length: 79

{"title":"Dune","authors":["Frank Herbert"],"pages":412,"isbn":"9780441013593"}

$ curl -si -X POST -H 'content-type: application/json' \
    -d '{"isbn":"9780441013593","stars":5,"text":"Great"}' http://127.0.0.1:8080/gateway/reviews
HTTP/1.1 201 Created
content-type: application/json; charset=utf-8
content-length: 31

{"id":7,"isbn":"9780441013593"}

$ curl -si 'http://127.0.0.1:8080/gateway/a/%2E%2E/b'
HTTP/1.1 400 Bad Request

{"reason":"Bad Request","error":true}

import Vapor
import NIOConcurrencyHelpers

/// A Client that never touches the network: it answers from a closure and records what it was asked.
struct StubClient: Client {
    let eventLoop: any EventLoop
    let recorded: NIOLockedValueBox<[ClientRequest]>
    let handler: @Sendable (ClientRequest) -> ClientResponse

    func delegating(to eventLoop: any EventLoop) -> any Client {
        StubClient(eventLoop: eventLoop, recorded: recorded, handler: handler)
    }

    func send(_ request: ClientRequest) -> EventLoopFuture<ClientResponse> {
        recorded.withLockedValue { $0.append(request) }
        return eventLoop.makeSucceededFuture(handler(request))
    }
}

extension Application {
    /// Replaces the HTTP client for this application, typically from a test, and returns the request log.
    @discardableResult
    func useStubClient(
        handler: @escaping @Sendable (ClientRequest) -> ClientResponse
    ) -> NIOLockedValueBox<[ClientRequest]> {
        let recorded = NIOLockedValueBox<[ClientRequest]>([])
        clients.use { app in
            StubClient(eventLoop: app.eventLoopGroup.next(), recorded: recorded, handler: handler)
        }
        return recorded
    }
}

import VaporTesting
import Testing
@testable import App

@Suite(.serialized) struct ClientTests {
    @Test func mapsUpstreamStatusesAndSendsTraceContext() async throws {
        let app = try await Application.make(.testing)
        do {
            let log = app.useStubClient { request in
                guard request.url.path.hasSuffix("/isbn/9780441013593") else { return ClientResponse(status: .notFound) }
                var response = ClientResponse(status: .ok)
                try? response.content.encode(BookMetadata(isbn: "9780441013593", title: "Dune", authors: ["Frank Herbert"], pages: 412))
                return response
            }
            try await configure(app)

            try await app.testing().test(.GET, "books/9780441013593/metadata", headers: ["traceparent": "00-aaaa-bbbb-01"]) { res async throws in
                #expect(res.status == .ok)
                #expect(try res.content.decode(BookMetadata.self).title == "Dune")
            }
            try await app.testing().test(.GET, "books/0000/metadata") { res async in
                #expect(res.status == .notFound)
            }
            try await app.testing().test(.GET, "books/bad!/metadata") { res async in
                #expect(res.status == .badRequest)
            }
            let first = log.withLockedValue { $0.first }
            #expect(first?.headers.first(name: "traceparent") == "00-aaaa-bbbb-01")
            #expect(first?.timeout == .seconds(3))
            #expect(log.withLockedValue { $0.count } == 2)   // the malformed ISBN never reached the client
        } catch { try await app.asyncShutdown(); throw error }
        try await app.asyncShutdown()
    }

    @Test func cacheCoalescesConcurrentMisses() async throws {
        let app = try await Application.make(.testing)
        do {
            let log = app.useStubClient { _ in
                var response = ClientResponse(status: .ok)
                try? response.content.encode(BookMetadata(isbn: "1", title: "T", authors: [], pages: 1))
                return response
            }
            try await configure(app)
            try await withThrowingTaskGroup(of: Void.self) { group in
                for _ in 0..<5 {
                    group.addTask {
                        try await app.testing().test(.GET, "books/1/cached") { res async in #expect(res.status == .ok) }
                    }
                }
                try await group.waitForAll()
            }
            try await app.testing().test(.GET, "books/1/cached") { res async in #expect(res.status == .ok) }
            #expect(log.withLockedValue { $0.count } == 1)
        } catch { try await app.asyncShutdown(); throw error }
        try await app.asyncShutdown()
    }
}

✔ Test mapsUpstreamStatusesAndSendsTraceContext() passed
✔ Test cacheCoalescesConcurrentMisses() passed
✔ Test run with 2 tests in 1 suite passed

import Vapor
import NIOCore

/// A stand-in for the external catalog so the examples can run against this same process.
func upstreamRoutes(_ routes: some RoutesBuilder) {
    routes.get("isbn", ":isbn") { req async throws -> Response in
        let isbn = try req.parameters.require("isbn")
        guard isbn != "0000" else { throw Abort(.notFound) }
        if isbn == "9999" { try await Task.sleep(for: .seconds(5)) }
        let book = BookMetadata(isbn: isbn, title: "Dune", authors: ["Frank Herbert"], pages: 412)
        let response = try await book.encodeResponse(for: req)
        response.headers.replaceOrAdd(name: "X-Seen-Trace", value: req.headers.first(name: "traceparent") ?? "none")
        return response
    }
    routes.post("reviews") { req async throws -> Response in
        let review = try req.content.decode(NewReview.self)
        return try await ReviewReceipt(id: 7, isbn: review.isbn).encodeResponse(status: .created, for: req)
    }
    routes.get("old") { req -> Response in req.redirect(to: "/upstream/isbn/9780441013593", redirectType: .permanent) }
    routes.get("covers", ":isbn") { req async throws -> Response in
        let response = Response(status: .ok)
        response.headers.contentType = HTTPMediaType(type: "image", subType: "png")
        response.body = .init(managedAsyncStream: { writer in
            for _ in 0..<4 { try await writer.writeBuffer(ByteBuffer(repeating: 0x2A, count: 1024)) }
        })
        return response
    }
    routes.get("search") { req async throws -> [BookMetadata] in
        guard req.headers.bearerAuthorization != nil else { throw Abort(.unauthorized) }
        let q = try req.query.get(String.self, at: "q")
        return [BookMetadata(isbn: "9780441013593", title: q, authors: ["Frank Herbert"], pages: 412)]
    }
    routes.get("echo") { req async throws -> String in
        req.headers.map { "\($0.name.lowercased()): \($0.value)" }.sorted().joined(separator: "\n")
    }
}

import Vapor

func configure(_ app: Application) async throws {
    try configureHTTPClient(app)
    let catalogURL = Environment.get("CATALOG_URL") ?? "http://127.0.0.1:8080/upstream"
    upstreamRoutes(app.grouped("upstream"))
    let catalog = CatalogClient(client: app.client, baseURL: catalogURL)
    clientRoutes(app, catalog: catalog)
    searchRoutes(app, catalog: catalog, token: try catalogToken(for: app.environment))
    cachedRoutes(app, cached: CachedCatalog(catalog: catalog))
    gatewayRoutes(app, upstream: catalogURL)
    streamingRoutes(app, catalogURL: catalogURL)
}

$ swift run App serve --hostname 127.0.0.1 --port 8080

$ curl -si http://127.0.0.1:8080/books/9780441013593/metadata
HTTP/1.1 200 OK
content-type: application/json; charset=utf-8
content-length: 79

{"title":"Dune","pages":412,"authors":["Frank Herbert"],"isbn":"9780441013593"}

$ curl -si http://127.0.0.1:8080/books/0000/metadata
HTTP/1.1 404 Not Found

{"reason":"The catalog does not know ISBN 0000.","error":true}

$ curl -si http://127.0.0.1:8080/books/9999/metadata        # the upstream sleeps 5 s, the per-call timeout is 3 s
HTTP/1.1 504 Gateway Timeout

{"reason":"The catalog did not answer in time.","error":true}

$ curl -s 'http://127.0.0.1:8080/search?title=Dune%20Messiah'
[{"title":"Dune Messiah","isbn":"9780441013593","authors":["Frank Herbert"],"pages":412}]

$ curl -s -o /dev/null -w '%{http_code} %{size_download}\n' http://127.0.0.1:8080/covers/123
200 4096