Advanced

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.ktor.util.*
import io.ktor.util.pipeline.*

// Plugin that measures request duration and route label for metrics
data class TraceContext(
    val startNanos: Long,                 // => Nanosecond timestamp of request start
    val requestId: String                 // => Unique ID for this request
)

val TraceContextKey = AttributeKey<TraceContext>("TraceContext")

class TracingConfig {
    var emitter: ((route: String, durationMs: Long, statusCode: Int) -> Unit)? = null
    // => Callback for emitting metrics; defaults to no-op
}

val TracingPlugin = createApplicationPlugin("TracingPlugin", ::TracingConfig) {
    val emitter = pluginConfig.emitter

    // Phase 1: Intercept BEFORE any processing (set up context)
    onCall { call ->
        val ctx = TraceContext(
            startNanos = System.nanoTime(),              // => Record start time
            requestId = java.util.UUID.randomUUID().toString()
        )
        call.attributes.put(TraceContextKey, ctx)        // => Store for later phases
    }

    // Phase 2: Intercept AFTER response is built (emit metrics)
    onCallRespond { call, _ ->
        val ctx = call.attributes.getOrNull(TraceContextKey) ?: return@onCallRespond
        val durationMs = (System.nanoTime() - ctx.startNanos) / 1_000_000L
                                                         // => Convert nanos to millis
        val statusCode = call.response.status()?.value ?: 0
        val route = call.request.uri                     // => Use RouteInfo for named routes

        emitter?.invoke(route, durationMs, statusCode)   // => Emit to metrics backend
        call.response.headers.append("X-Trace-ID", ctx.requestId)
        call.response.headers.append("X-Response-Time", "${durationMs}ms")
    }
}

fun Application.configureTracing() {
    install(TracingPlugin) {
        emitter = { route, durationMs, status ->
            // In production: emit to Prometheus, Datadog, CloudWatch
            log.info("METRIC: route=$route status=$status duration=${durationMs}ms")
        }
    }
}

fun Application.configureRouting() {
    routing {
        get("/traced") {
            kotlinx.coroutines.delay(25)  // => Simulate 25ms processing
            call.respondText("Traced response")
        }
    }
}
// => GET /traced
//    Response headers: X-Trace-ID: 550e8400-..., X-Response-Time: 26ms
//    Log: METRIC: route=/traced status=200 duration=26ms

import io.ktor.server.application.*
import io.ktor.server.auth.*
import io.ktor.server.plugins.callid.*
import io.ktor.server.plugins.calllogging.*
import io.ktor.server.plugins.contentnegotiation.*
import io.ktor.server.plugins.statuspages.*
import io.ktor.serialization.kotlinx.json.*

fun Application.configurePluginsInCorrectOrder() {
    // ORDERING MATTERS: install in dependency order

    // 1. Call ID first - generates ID before other plugins log
    install(CallId) {
        generate { java.util.UUID.randomUUID().toString() }
        replyToHeader("X-Request-ID")
    }

    // 2. Call Logging uses Call ID for MDC - must come AFTER CallId
    install(CallLogging) {
        callIdMdc("requestId")             // => Uses Call ID generated in step 1
    }

    // 3. Status Pages - catches exceptions from ALL subsequent plugins and routes
    // Must be installed early so it can handle exceptions from auth, serialization, etc.
    install(StatusPages) {
        exception<Throwable> { call, cause ->
            call.application.environment.log.error("Error", cause)
            call.respondText("Error: ${cause.message}",
                status = io.ktor.http.HttpStatusCode.InternalServerError)
        }
    }

    // 4. Content Negotiation - used by routes, must be before routing
    install(ContentNegotiation) {
        json()
    }

    // 5. Authentication - last plugin, used by protected routes
    install(Authentication) {
        basic("auth-basic") {
            validate { creds ->
                if (creds.name == "user") UserIdPrincipal(creds.name) else null
            }
        }
    }

    // Routes reference plugins installed above
    // configureRouting() called after all plugins installed
}

// Anti-pattern: wrong order causes subtle bugs
fun Application.configurePluginsWrongOrder() {
    install(CallLogging) {
        callIdMdc("requestId")  // => BUG: CallId not installed yet; requestId always null
    }
    install(CallId) {           // => Too late: CallLogging won't see the ID
        generate { java.util.UUID.randomUUID().toString() }
    }
    // => All log lines: requestId=null
    // => Difficult to debug: no error thrown, just missing data
}
// Correct output: requestId=550e8400-e29b-41d4-a716-... in every log line
// Wrong output:   requestId=null in every log line

// build.gradle.kts:
// implementation("io.ktor:ktor-server-metrics-micrometer:3.0.3")
// implementation("io.micrometer:micrometer-registry-prometheus:1.12.0")

import io.ktor.server.application.*
import io.ktor.server.metrics.micrometer.*   // => MicrometerMetrics plugin
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.micrometer.core.instrument.*
import io.micrometer.prometheus.*

fun Application.configureMetrics() {
    // Create Prometheus registry (stores metrics in Prometheus format)
    val prometheusRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
                                              // => Stores counters, timers, gauges

    install(MicrometerMetrics) {              // => Install Micrometer plugin
        registry = prometheusRegistry         // => Use Prometheus as the registry backend
        // => Automatically records per-route metrics:
        // =>   ktor.http.server.requests (timer): count, mean, max, percentiles
        // =>   ktor.http.server.active.requests (gauge): concurrent in-flight requests
    }

    // Custom business metric: counter for user signups
    val signupCounter = prometheusRegistry.counter(
        "myapp.user.signups",                // => Metric name (dot-separated)
        "source", "api"                      // => Tag: source=api (for filtering in dashboards)
    )
    // => Counter is a cumulative value that only increases

    // Custom gauge: current queue size
    var queueSize = 0
    prometheusRegistry.gauge(
        "myapp.queue.size",                  // => Metric name
        emptyList<Tag>(),                    // => No tags
        queueSize                            // => Current value (polled when scraped)
    )

    routing {
        // Expose Prometheus scrape endpoint
        get("/metrics") {
            // prometheusRegistry.scrape() returns metrics in Prometheus text format
            call.respondText(
                prometheusRegistry.scrape(),  // => Text with all metric values
                contentType = io.ktor.http.ContentType.Text.Plain
            )
        }

        // Business endpoint with custom metric tracking
        post("/signup") {
            signupCounter.increment()         // => Increment counter for each signup
            call.respondText("Signed up!")
        }
    }
}
// => GET /metrics (Prometheus format):
// => # HELP ktor_http_server_requests_seconds_count HTTP server requests
// => ktor_http_server_requests_seconds_count{method="GET",route="/metrics",...} 1.0
// => # HELP myapp_user_signups_total User signups
// => myapp_user_signups_total{source="api"} 5.0

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.micrometer.core.instrument.*
import io.micrometer.prometheus.*
import kotlin.time.*
import kotlin.time.Duration.Companion.milliseconds

class MetricsService(private val registry: MeterRegistry) {

    // Timer measures duration with histogram (enables percentile calculation)
    private val dbQueryTimer = Timer.builder("myapp.db.query")
        .description("Database query duration")
        .tag("operation", "select")          // => Tag for filtering in dashboards
        .publishPercentiles(0.5, 0.95, 0.99) // => Publish p50, p95, p99 percentiles
        .register(registry)                   // => Register with registry

    // Distribution summary: records values like response sizes
    private val responseSizeSummary = DistributionSummary.builder("myapp.response.size")
        .baseUnit("bytes")
        .description("HTTP response body size in bytes")
        .register(registry)

    // Record a DB query with timing
    suspend fun executeQuery(query: String): List<String> {
        val timerSample = Timer.start(registry)  // => Start timer
                                                  // => Captures start time

        return try {
            // Simulate DB query
            kotlinx.coroutines.delay(5)          // => 5ms simulated query
            val results = listOf("row1", "row2")

            timerSample.stop(dbQueryTimer)        // => Stop timer; records duration
                                                   // => Emits histogram bucket values
            results
        } catch (e: Exception) {
            // Record failed queries separately (different tag value)
            Timer.builder("myapp.db.query")
                .tag("operation", "select")
                .tag("error", "true")             // => Tag failures for alerting
                .register(registry)
                .record(0.milliseconds.toJavaDuration())
            throw e
        }
    }

    fun recordResponseSize(bytes: Int) {
        responseSizeSummary.record(bytes.toDouble())  // => Record distribution value
    }
}

fun Application.configureRouting() {
    val registry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)

    routing {
        val metricsService = MetricsService(registry)

        get("/data") {
            val results = metricsService.executeQuery("SELECT * FROM items")
            val responseText = results.joinToString(",")
            metricsService.recordResponseSize(responseText.length)
            call.respondText(responseText)
        }

        get("/metrics") {
            call.respondText(registry.scrape(),
                io.ktor.http.ContentType.Text.Plain)
        }
    }
}
// => After 100 GET /data requests:
// => myapp_db_query_seconds{quantile="0.95"} 0.007  (p95 = 7ms)
// => myapp_db_query_seconds{quantile="0.99"} 0.010  (p99 = 10ms)
// => myapp_response_size_bytes_sum 1500.0           (total bytes)

// build.gradle.kts:
// implementation("io.opentelemetry:opentelemetry-api:1.36.0")
// implementation("io.opentelemetry:opentelemetry-sdk:1.36.0")
// implementation("io.opentelemetry.instrumentation:opentelemetry-ktor-3.0:2.3.0-alpha")
// implementation("io.opentelemetry:opentelemetry-exporter-otlp:1.36.0")

import io.opentelemetry.api.GlobalOpenTelemetry
import io.opentelemetry.api.trace.Tracer
import io.opentelemetry.api.trace.SpanKind
import io.opentelemetry.api.trace.StatusCode
import io.opentelemetry.context.Context
import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*

// Initialize OpenTelemetry (typically done at startup)
// In production: configure OTLP exporter pointing to Jaeger/Tempo/Honeycomb
val tracer: Tracer = GlobalOpenTelemetry.getTracer(
    "ktor-app",                          // => Instrumentation library name
    "1.0.0"                              // => Version
)

fun Application.configureRouting() {
    routing {
        get("/orders/{id}") {
            val orderId = call.parameters["id"] ?: "unknown"

            // Create child span for this operation
            val span = tracer.spanBuilder("get-order")  // => Span name
                .setSpanKind(SpanKind.SERVER)            // => Server-side processing span
                .startSpan()                             // => Start timing

            try {
                // Make span active in this scope
                span.makeCurrent().use {
                    // Any spans created here are children of this span
                    span.setAttribute("order.id", orderId)  // => Structured attribute
                    span.setAttribute("http.route", "/orders/{id}")

                    val order = fetchOrderWithTracing(orderId)  // => Child span inside
                    span.setStatus(StatusCode.OK)               // => Mark span successful

                    call.respondText("Order: $order")
                }
            } catch (e: Exception) {
                span.setStatus(StatusCode.ERROR, e.message ?: "Unknown error")
                span.recordException(e)                  // => Attach exception to span
                throw e
            } finally {
                span.end()                               // => Always end span (even on error)
            }
        }
    }
}

// Function that creates its own child span
suspend fun fetchOrderWithTracing(orderId: String): String {
    val span = tracer.spanBuilder("db-fetch-order")
        .setSpanKind(SpanKind.CLIENT)                   // => Client-side DB call
        .startSpan()

    return try {
        kotlinx.coroutines.delay(10)                    // => Simulate DB query
        span.setAttribute("db.statement", "SELECT * FROM orders WHERE id = ?")
        "Order-$orderId"
    } finally {
        span.end()                                       // => End child span
    }
}
// => GET /orders/123 generates trace:
// => Span: get-order (15ms total)
// =>   Span: db-fetch-order (10ms) - child span
// => Trace visible in Jaeger/Tempo UI with parent-child relationship

import io.ktor.server.engine.*
import io.ktor.server.netty.*
import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import java.security.KeyStore

fun main() {
    // Load keystore from classpath (JKS or PKCS12 format)
    // Generate with: keytool -genkeypair -alias myapp -keyalg RSA -keysize 2048
    //               -validity 365 -keystore keystore.jks
    val keyStore = KeyStore.getInstance("JKS")
    keyStore.load(
        ClassLoader.getSystemResourceAsStream("keystore.jks"),
        "keystore-password".toCharArray()  // => Keystore file password
    )

    val environment = applicationEngineEnvironment {
        // HTTPS connector on port 8443
        sslConnector(
            keyStore = keyStore,
            keyAlias = "myapp",                     // => Certificate alias in keystore
            keyStorePassword = { "keystore-password".toCharArray() },
            privateKeyPassword = { "key-password".toCharArray() }
        ) {
            port = 8443                              // => HTTPS port
            host = "0.0.0.0"
            // keyStorePath = File("keystore.jks")  // => Alternative: file path
        }

        // Optional: HTTP connector for redirect to HTTPS
        connector {
            port = 8080                              // => HTTP port (for redirect only)
            host = "0.0.0.0"
        }

        module {
            // Force HTTPS redirect for HTTP requests
            intercept(io.ktor.server.application.ApplicationCallPipeline.Plugins) {
                if (call.request.origin.scheme == "http") {
                    val httpsUrl = call.request.uri
                        .replace("http://", "https://")
                        .let { if (!it.contains(":8443")) it.replace(":8080", ":8443") else it }
                    call.respondRedirect(httpsUrl, permanent = true)
                    return@intercept
                }
            }
            routing {
                get("/secure") {
                    call.respondText("Secure HTTPS response")
                }
            }
        }
    }

    embeddedServer(Netty, environment).start(wait = true)
}
// => HTTP GET http://localhost:8080/secure
//    => 301 Redirect to https://localhost:8443/secure
// => HTTPS GET https://localhost:8443/secure
//    => 200 "Secure HTTPS response"

import io.ktor.server.engine.*
import io.ktor.server.netty.*
import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.ktor.http.*

fun main() {
    // HTTP/2 requires TLS (h2 only; h2c = HTTP/2 cleartext, not widely supported)
    // Netty engine supports HTTP/2 automatically when SSL is configured
    val environment = applicationEngineEnvironment {
        sslConnector(
            keyStore = java.security.KeyStore.getInstance("JKS").also {
                it.load(
                    ClassLoader.getSystemResourceAsStream("keystore.jks"),
                    "password".toCharArray()
                )
            },
            keyAlias = "myapp",
            keyStorePassword = { "password".toCharArray() },
            privateKeyPassword = { "password".toCharArray() }
        ) {
            port = 8443
        }

        module {
            routing {
                get("/api/data") {
                    // HTTP/2 clients automatically multiplex multiple requests
                    // over the same connection without head-of-line blocking
                    call.respondText(
                        """{"data": "HTTP/2 response"}""",
                        ContentType.Application.Json
                    )
                }

                // Server Push: proactively push resources to client
                // (HTTP/2 only; ignored by HTTP/1.1 clients)
                get("/page") {
                    // Push stylesheet before client requests it
                    call.response.link(              // => <Link> header for push hints
                        rel = "preload",
                        href = "/styles.css",
                        type = "text/css"
                    )
                    call.respondText(
                        "<html><link rel='stylesheet' href='/styles.css'><body>Page</body></html>",
                        ContentType.Text.Html
                    )
                    // => Browser uses Link preload header to fetch /styles.css
                    // => With HTTP/2, both page and CSS arrive in same connection
                }
            }
        }
    }

    embeddedServer(Netty, environment) {
        // Netty automatically negotiates HTTP/2 via ALPN when TLS is configured
        // No explicit HTTP/2 enable flag needed with Netty
    }.start(wait = true)
}
// => HTTPS client connecting to :8443
// => Netty negotiates: TLS + ALPN -> h2 (HTTP/2)
// => Multiple requests multiplexed over single TCP connection
// => GET /page => sends Link: preload header, browser fetches /styles.css eagerly

// build.gradle.kts:
// implementation("io.ktor:ktor-client-core:3.0.3")
// implementation("io.ktor:ktor-client-cio:3.0.3")
// implementation("io.ktor:ktor-client-content-negotiation:3.0.3")
// implementation("io.ktor:ktor-client-logging:3.0.3")

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.plugins.contentnegotiation.*
import io.ktor.client.plugins.logging.*
import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.client.call.*
import io.ktor.serialization.kotlinx.json.*
import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import kotlinx.serialization.Serializable

@Serializable
data class GithubUser(
    val login: String,
    val name: String?,
    val publicRepos: Int = 0
)

// Create shared HttpClient (expensive to create; reuse across requests)
// Close client on application shutdown
val httpClient = HttpClient(CIO) {             // => CIO engine: pure Kotlin coroutines
    install(ContentNegotiation) {              // => Auto-deserialize JSON responses
        json()
    }
    install(Logging) {                         // => Log all HTTP client requests
        level = LogLevel.HEADERS               // => Log headers (use BODY for debugging)
    }
    engine {
        requestTimeout = 10_000                // => 10s timeout for full request
        endpoint {
            connectTimeout = 5_000             // => 5s to establish TCP connection
            socketTimeout = 10_000             // => 10s to read response
        }
    }
}

fun Application.configureRouting() {
    routing {
        get("/github/{username}") {
            val username = call.parameters["username"] ?: return@get

            // GET request with automatic JSON deserialization
            val githubUser: GithubUser = httpClient.get(
                "https://api.github.com/users/$username"  // => External API URL
            ) {
                header("Accept", "application/vnd.github.v3+json")  // => GitHub API version
                header("User-Agent", "ktor-example/1.0")            // => Required by GitHub API
            }.body()                           // => Deserializes response body to GithubUser
            // => body() is a suspend function; waits for full response

            call.respond(githubUser)           // => Re-serialize to client as JSON
        }

        get("/post-example") {
            // POST with body
            val response = httpClient.post("https://httpbin.org/post") {
                contentType(io.ktor.http.ContentType.Application.Json)
                setBody(mapOf("key" to "value"))  // => Serialized to JSON body
            }
            call.respondText(response.bodyAsText())  // => Echo response body
        }
    }
}
// => GET /github/jetbrains
//    => Fetches from api.github.com, responds with: {"login":"JetBrains","name":"JetBrains",...}

import io.ktor.client.*
import io.ktor.client.engine.cio.*
import io.ktor.client.plugins.*
import io.ktor.client.plugins.contentnegotiation.*
import io.ktor.client.request.*
import io.ktor.client.statement.*
import io.ktor.serialization.kotlinx.json.*
import io.ktor.http.*
import kotlinx.coroutines.delay

val resilientClient = HttpClient(CIO) {
    install(ContentNegotiation) { json() }

    // Retry plugin: automatically retries failed requests
    install(HttpRequestRetry) {
        retryOnServerErrors(maxRetries = 3)    // => Retry on 5xx responses (not 4xx)
                                                // => 3 retries maximum
        retryOnException(
            maxRetries = 3,
            retryOnTimeout = true              // => Also retry on connection timeout
        )
        exponentialDelay(                      // => Exponential backoff between retries
            base = 2.0,                        // => Base for exponential calculation
            maxDelayMs = 10_000L               // => Cap delay at 10 seconds
        )
        // => Retry 1: 1s delay, Retry 2: 2s delay, Retry 3: 4s delay
        // => Total max wait: 1 + 2 + 4 = 7s before giving up
    }

    // Circuit breaker (manual implementation - Ktor doesn't have built-in)
    // Use Resilience4j for production circuit breaker patterns
}

// Wrapper for robust external API calls with fallback
suspend fun fetchExternalData(url: String): Result<String> {
    return try {
        val response = resilientClient.get(url)

        when (response.status) {
            HttpStatusCode.OK ->
                Result.success(response.bodyAsText())  // => Success with response body
            HttpStatusCode.NotFound ->
                Result.failure(NoSuchElementException("Resource not found at $url"))
                // => 404: don't retry (permanent failure)
            HttpStatusCode.TooManyRequests -> {
                val retryAfter = response.headers["Retry-After"]?.toLongOrNull() ?: 60L
                delay(retryAfter * 1000)              // => Respect server's Retry-After header
                Result.failure(RuntimeException("Rate limited"))
            }
            else ->
                Result.failure(RuntimeException("HTTP ${response.status.value}"))
        }
    } catch (e: Exception) {
        Result.failure(e)                             // => Network errors wrapped in Result
    }
}
// => fetchExternalData("https://api.example.com/data")
// => On 503: retries 3 times with 1s, 2s, 4s backoff
// => On 404: immediately returns Result.failure (not retried)
// => On 429: waits Retry-After seconds, returns Result.failure

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import kotlinx.coroutines.*
import kotlinx.serialization.Serializable

@Serializable
data class UserProfile(
    val userId: String,
    val name: String,
    val recentOrders: List<String>,
    val recommendations: List<String>
)

// Simulate service calls with different latencies
suspend fun fetchUserInfo(userId: String): String {
    delay(50)                              // => 50ms database query
    return "Alice"
}

suspend fun fetchRecentOrders(userId: String): List<String> {
    delay(80)                              // => 80ms order history query
    return listOf("Order-1", "Order-2", "Order-3")
}

suspend fun fetchRecommendations(userId: String): List<String> {
    delay(120)                             // => 120ms ML recommendation API
    return listOf("Product-A", "Product-B")
}

fun Application.configureRouting() {
    routing {
        get("/users/{id}/profile") {
            val userId = call.parameters["id"] ?: return@get

            // Sequential: 50 + 80 + 120 = 250ms total
            // val name = fetchUserInfo(userId)
            // val orders = fetchRecentOrders(userId)
            // val recs = fetchRecommendations(userId)

            // Parallel with coroutineScope: max(50, 80, 120) = 120ms total
            val (name, orders, recommendations) = coroutineScope {
                val nameDeferred = async { fetchUserInfo(userId) }          // => Starts immediately
                val ordersDeferred = async { fetchRecentOrders(userId) }    // => Starts immediately
                val recsDeferred = async { fetchRecommendations(userId) }   // => Starts immediately
                // => All three run concurrently

                Triple(
                    nameDeferred.await(),     // => Waits for result (50ms)
                    ordersDeferred.await(),   // => Waits for result (80ms)
                    recsDeferred.await()      // => Waits for result (120ms)
                )
                // => Total wait = max(50, 80, 120) = 120ms (parallel)
                // => If ANY async throws, coroutineScope cancels ALL others (structured)
            }

            call.respond(UserProfile(userId, name, orders, recommendations))
            // => Response in ~120ms instead of ~250ms
        }
    }
}
// => GET /users/123/profile
//    Concurrent: fetchUserInfo + fetchRecentOrders + fetchRecommendations start simultaneously
//    Response in ~120ms (limited by slowest: fetchRecommendations)
//    Sequential would take ~250ms

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.ktor.http.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.*

// Simulates a database cursor returning rows as a Flow
fun databaseRows(batchSize: Int = 10): Flow<String> = flow {
    var offset = 0
    while (true) {                         // => Infinite scroll; terminate on empty batch
        val batch = (offset until offset + batchSize)
            .map { "row-$it" }             // => Simulate fetched rows
        if (batch.isEmpty()) break
        batch.forEach { emit(it) }         // => Emit rows one at a time
        offset += batchSize
        delay(5)                           // => Simulate DB fetch delay between batches
        if (offset >= 50) break            // => Stop at 50 for this example
    }
}

fun Application.configureRouting() {
    routing {
        // Stream NDJSON (newline-delimited JSON) using Flow
        get("/stream/data") {
            call.response.headers.append(HttpHeaders.CacheControl, "no-cache")

            call.respondTextWriter(
                ContentType.parse("application/x-ndjson"),  // => NDJSON content type
                status = HttpStatusCode.OK
            ) {
                // Collect Flow and write each element as a JSON line
                databaseRows()
                    .filter { !it.endsWith("-3") }          // => Filter operator: skip every N
                    .map { """{"value":"$it"}""" }           // => Transform to JSON
                    .take(20)                                // => Limit to 20 items
                    .collect { jsonLine ->
                        write(jsonLine + "\n")               // => Each line is a JSON object
                        flush()                              // => Send immediately
                    }
                // => Client receives: {"value":"row-0"}\n{"value":"row-1"}\n...
            }
        }

        // Flow with error handling
        get("/stream/safe") {
            call.respondTextWriter(ContentType.Text.Plain) {
                databaseRows()
                    .catch { e ->                            // => Catch errors in Flow
                        emit("ERROR: ${e.message}")         // => Emit error as a value
                        // => Allows stream to complete gracefully on error
                    }
                    .collect { row ->
                        write("$row\n")
                        flush()
                    }
            }
        }
    }
}
// => GET /stream/data
//    Streams: {"value":"row-0"}\n{"value":"row-1"}\n{"value":"row-2"}\n{"value":"row-4"}\n...
//    (row-3 filtered, row-13 filtered, etc.)
//    Client receives 20 items then connection closes

// build.gradle.kts: implementation("io.github.reactivecircus.cache4k:cache4k:0.13.0")

import io.github.reactivecircus.cache4k.Cache
import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import kotlinx.coroutines.*
import kotlinx.serialization.Serializable
import kotlin.time.Duration.Companion.minutes
import kotlin.time.Duration.Companion.seconds

@Serializable
data class ProductCatalog(val products: List<String>, val lastUpdated: Long)

// Application-level cache (singleton pattern)
val productCache = Cache.Builder<String, ProductCatalog>()
    .maximumCacheSize(1000)               // => Max 1000 entries (LRU eviction)
    .expireAfterWrite(5.minutes)          // => Expire 5 minutes after write
    .expireAfterAccess(10.minutes)        // => Reset expiry on access
    .build()

// Simulate slow database query (would be Exposed transaction in real app)
suspend fun fetchProductCatalogFromDb(category: String): ProductCatalog {
    delay(200)                            // => Simulate 200ms database query
    return ProductCatalog(
        products = listOf("$category-A", "$category-B", "$category-C"),
        lastUpdated = System.currentTimeMillis()
    )
}

fun Application.configureRouting() {
    routing {
        get("/catalog/{category}") {
            val category = call.parameters["category"] ?: return@get

            // Cache-aside pattern: check cache first, fall back to DB
            val catalog = productCache.get(category)   // => Returns cached value or null
                ?: fetchProductCatalogFromDb(category)  // => Cache miss: fetch from DB
                    .also { productCache.put(category, it) }
                    // => .also { } stores fetched value in cache before returning
                    // => Subsequent requests for same category hit the cache

            call.respond(catalog)
            // => First request:  200ms (DB query) + cache write
            // => Second request: ~1ms (cache hit) - 200x faster
        }

        // Admin endpoint to invalidate cache (after catalog updates)
        delete("/catalog/{category}/cache") {
            val category = call.parameters["category"] ?: return@delete
            productCache.invalidate(category)           // => Remove specific entry
            call.respondText("Cache invalidated for $category")
        }

        // Refresh all caches (post-deployment hook)
        post("/admin/refresh-cache") {
            productCache.invalidateAll()               // => Clears entire cache
            call.respondText("All caches cleared")
        }
    }
}
// => GET /catalog/electronics (first call) => 200ms (DB hit), stores in cache
// => GET /catalog/electronics (subsequent) => ~1ms (cache hit)
// => DELETE /catalog/electronics/cache     => Clears electronics entry
// => GET /catalog/electronics (after delete) => 200ms (DB hit again)

// build.gradle.kts: implementation("io.lettuce:lettuce-core:6.3.1.RELEASE")

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.ktor.http.*
import io.lettuce.core.RedisClient
import io.lettuce.core.api.async.RedisAsyncCommands

// Sliding window rate limiter using Redis INCR + EXPIRE
class RedisRateLimiter(
    private val redis: RedisAsyncCommands<String, String>,
    private val windowSeconds: Long = 60L,
    private val maxRequests: Int = 100
) {
    suspend fun isAllowed(key: String): RateLimitResult {
        val redisKey = "ratelimit:$key:${System.currentTimeMillis() / (windowSeconds * 1000)}"
                                              // => Bucket key changes each window
                                              // => "ratelimit:ip:203.0.113.42:28516" (minute bucket)

        // INCR atomically increments and returns new value
        val count = redis.incr(redisKey).toCompletableFuture().get().toInt()
                                              // => Atomic increment; count is the new value

        if (count == 1) {
            // First request in this window: set expiry
            redis.expire(redisKey, windowSeconds + 1).toCompletableFuture().get()
            // => Key expires after window; prevents Redis memory leak
        }

        val remaining = (maxRequests - count).coerceAtLeast(0)
        return if (count <= maxRequests) {
            RateLimitResult(allowed = true, remaining = remaining, resetIn = windowSeconds)
        } else {
            RateLimitResult(allowed = false, remaining = 0, resetIn = windowSeconds)
        }
    }
}

data class RateLimitResult(
    val allowed: Boolean,
    val remaining: Int,             // => Requests remaining in current window
    val resetIn: Long               // => Seconds until window resets
)

fun Application.configureRouting() {
    // In production: configure Redis URL from environment variable
    val redisClient = RedisClient.create("redis://localhost:6379")
    val redisConn = redisClient.connect()
    val redis = redisConn.async()
    val rateLimiter = RedisRateLimiter(redis, windowSeconds = 60, maxRequests = 100)

    routing {
        get("/api/data") {
            val clientIp = call.request.origin.remoteHost  // => Client IP as rate limit key

            val result = rateLimiter.isAllowed(clientIp)
            call.response.headers.apply {
                append("X-RateLimit-Limit", "100")
                append("X-RateLimit-Remaining", result.remaining.toString())
                append("X-RateLimit-Reset", result.resetIn.toString())
                // => Standard rate limit headers clients can use for backoff
            }

            if (!result.allowed) {
                call.response.headers.append("Retry-After", result.resetIn.toString())
                call.respond(
                    HttpStatusCode.TooManyRequests,
                    mapOf("error" to "Rate limit exceeded", "resetIn" to result.resetIn)
                )
                return@get
            }

            call.respondText("Data response")
        }
    }
}
// => First 100 GET /api/data from same IP in 60s => 200 X-RateLimit-Remaining: 99..0
// => 101st request => 429 {"error":"Rate limit exceeded","resetIn":45}
// => Works correctly across multiple Ktor instances (Redis is shared)

// --- Build the fat JAR first ---
// build.gradle.kts additions:
// plugins {
//     id("com.github.johnrengelman.shadow") version "8.1.1"  // => Fat JAR plugin
// }
// tasks.shadowJar {
//     manifest { attributes["Main-Class"] = "com.example.ApplicationKt" }
// }

// --- Dockerfile ---
// # Stage 1: Build
// FROM gradle:8.5-jdk21 AS build
// WORKDIR /app
// COPY gradle/ gradle/
// COPY gradlew build.gradle.kts settings.gradle.kts ./
// RUN ./gradlew dependencies --no-daemon  # Cache dependency layer
// COPY src/ src/
// RUN ./gradlew shadowJar --no-daemon    # Build fat JAR
//
// # Stage 2: Runtime (much smaller image)
// FROM eclipse-temurin:21-jre-alpine     # JRE only (not JDK) - smaller
// WORKDIR /app
// RUN addgroup -S appuser && adduser -S appuser -G appuser  # Non-root user
// COPY --from=build /app/build/libs/*-all.jar app.jar
// USER appuser                           # Run as non-root (security requirement)
//
// EXPOSE 8080
// HEALTHCHECK --interval=30s --timeout=10s --start-period=30s --retries=3 \
//   CMD wget -qO- http://localhost:8080/health/live || exit 1
//
// ENTRYPOINT ["java",                    # JVM flags for container environment
//   "-XX:+UseContainerSupport",         # Enable JVM container awareness
//   "-XX:MaxRAMPercentage=75.0",        # Use up to 75% of container RAM for heap
//   "-XX:+ExitOnOutOfMemoryError",      # Exit (not continue) on OOM (trigger restart)
//   "-jar", "app.jar"]

// --- docker-compose.yml for local development ---
// version: '3.8'
// services:
//   app:
//     build: .
//     ports:
//       - "8080:8080"
//     environment:
//       - DATABASE_URL=jdbc:postgresql://db:5432/mydb
//       - JWT_SECRET=${JWT_SECRET}       # From .env file or shell environment
//     depends_on:
//       db:
//         condition: service_healthy    # Wait for DB health check before starting
//   db:
//     image: postgres:16-alpine
//     environment:
//       POSTGRES_DB: mydb
//       POSTGRES_USER: myuser
//       POSTGRES_PASSWORD: ${DB_PASSWORD}
//     healthcheck:
//       test: ["CMD-SHELL", "pg_isready -U myuser -d mydb"]
//       interval: 5s
//       timeout: 5s
//       retries: 5

// Application.kt - Production-ready startup
import io.ktor.server.engine.*
import io.ktor.server.netty.*
import io.ktor.server.application.*

fun main() {
    // PORT env var is standard for container platforms (Heroku, Railway, Render)
    val port = System.getenv("PORT")?.toIntOrNull() ?: 8080

    embeddedServer(
        Netty,
        port = port,
        host = "0.0.0.0"                 // => Bind to all interfaces (required in containers)
                                          // => 127.0.0.1 would only accept localhost connections
    ) {
        // Module functions here
    }.start(wait = true)
}
// => docker build -t myapp .
// => docker run -p 8080:8080 -e JWT_SECRET=prod-secret myapp
// => curl http://localhost:8080/health/live => {"status":"UP"}

import io.ktor.server.application.*

// Configuration data class with validation
data class AppConfig(
    val port: Int,
    val databaseUrl: String,
    val jwtSecret: String,
    val environment: String,
    val allowedOrigins: List<String>
)

// Read and validate configuration at startup
// Throws IllegalStateException with clear message if required config is missing
fun Application.loadAndValidateConfig(): AppConfig {
    val env = System.getenv()            // => All environment variables as Map

    // Helper: require env var or throw with clear message
    fun requireEnv(name: String): String =
        env[name] ?: throw IllegalStateException(
            "Required environment variable '$name' is not set. " +
            "Set it in your .env file or deployment configuration."
        )
        // => Fails fast at startup with actionable error message
        // => Much better than NullPointerException 10 minutes into a request

    val environment = env.getOrDefault("APP_ENV", "development")

    // Validate JWT secret strength in production
    val jwtSecret = requireEnv("JWT_SECRET")
    if (environment == "production" && jwtSecret.length < 32) {
        throw IllegalStateException(
            "JWT_SECRET must be at least 32 characters in production. " +
            "Generate with: openssl rand -base64 32"
        )
    }

    return AppConfig(
        port = env["PORT"]?.toIntOrNull() ?: 8080,
        databaseUrl = requireEnv("DATABASE_URL"),
        jwtSecret = jwtSecret,
        environment = environment,
        allowedOrigins = env["ALLOWED_ORIGINS"]
            ?.split(",")
            ?.map { it.trim() }
            ?: listOf("http://localhost:3000")  // => Dev default
    )
}

fun Application.module() {
    val config = loadAndValidateConfig()  // => Throws at startup if misconfigured
    log.info("Starting in ${config.environment} mode on port ${config.port}")
    // Never log config.jwtSecret or config.databaseUrl (contains passwords)

    // Use config values throughout the application
    // configureCORS(config.allowedOrigins)
    // configureDatabase(config.databaseUrl)
    // configureJWT(config.jwtSecret)
}
// => Missing JWT_SECRET => startup fails immediately:
//    IllegalStateException: Required environment variable 'JWT_SECRET' is not set.
//    Set it in your .env file or deployment configuration.
// => Short JWT_SECRET in production => startup fails:
//    IllegalStateException: JWT_SECRET must be at least 32 characters in production.

import io.ktor.server.application.*
import io.ktor.server.auth.*
import io.ktor.server.auth.jwt.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.ktor.http.*
import com.auth0.jwt.JWT
import com.auth0.jwt.algorithms.Algorithm

// Unified principal regardless of auth method used
data class AuthenticatedUser(
    val userId: String,
    val role: String,
    val authMethod: String              // => "jwt", "api-key", "session"
) : Principal

fun Application.configureMultiAuth() {
    install(Authentication) {
        // Provider 1: JWT for mobile clients and SPAs
        jwt("jwt") {
            realm = "myapp"
            verifier(
                JWT.require(Algorithm.HMAC256("secret-key"))
                    .withIssuer("myapp")
                    .build()
            )
            validate { credential ->
                val userId = credential.payload.getClaim("userId").asString()
                    ?: return@validate null
                JWTPrincipal(credential.payload)
            }
        }

        // Provider 2: API key for programmatic/server-to-server access
        basic("api-key") {
            realm = "myapp-api"
            validate { credentials ->
                // credentials.name = "api-key", credentials.password = actual key
                if (credentials.name == "api-key" && isValidApiKey(credentials.password)) {
                    UserIdPrincipal("service-account")  // => API key -> service account
                } else null
            }
        }

        // Provider 3: Session for web browser clients
        session<UserSession>("session") {  // => Reuse UserSession from Example 30
            challenge {                    // => Called when session is invalid
                call.respondRedirect("/login")
            }
            validate { session ->
                if (session.userId.isNotEmpty()) session else null
                // => Returns Principal (UserSession) if valid, null if invalid
            }
        }
    }
}

// Dummy API key validation (replace with database lookup in production)
fun isValidApiKey(key: String): Boolean = key.startsWith("ak-") && key.length == 40

fun Application.configureRouting() {
    routing {
        // Accepts any of the three auth methods
        authenticate("jwt", "api-key", "session") {
            get("/api/protected") {
                // Different principal types depending on which method was used
                val jwtPrincipal = call.principal<JWTPrincipal>()
                val sessionPrincipal = call.principal<UserSession>()
                val apiKeyPrincipal = call.principal<UserIdPrincipal>()

                val userId = jwtPrincipal?.payload?.getClaim("userId")?.asString()
                    ?: sessionPrincipal?.userId
                    ?: apiKeyPrincipal?.name
                    ?: "unknown"

                call.respondText("Authenticated as: $userId")
            }
        }

        // JWT only (no API key or session)
        authenticate("jwt") {
            get("/mobile/data") {
                call.respondText("Mobile JWT endpoint")
            }
        }
    }
}
// => GET /api/protected (JWT Bearer)   => "Authenticated as: user-123"
// => GET /api/protected (API key Basic auth) => "Authenticated as: service-account"
// => GET /api/protected (session cookie) => "Authenticated as: alice"

import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.ktor.http.*
import org.jetbrains.exposed.sql.*
import org.jetbrains.exposed.sql.transactions.transaction
import kotlinx.serialization.Serializable

// Two tables that must stay consistent
object Orders : Table("orders") {
    val id = integer("id").autoIncrement()
    val userId = integer("user_id")
    val total = decimal("total", 10, 2)
    override val primaryKey = PrimaryKey(id)
}

object Inventory : Table("inventory") {
    val productId = integer("product_id")
    val quantity = integer("quantity")
    override val primaryKey = PrimaryKey(productId)
}

@Serializable
data class PlaceOrderRequest(val userId: Int, val productId: Int, val quantity: Int)

fun Application.configureRouting() {
    routing {
        post("/orders") {
            val request = call.receive<PlaceOrderRequest>()

            try {
                val orderId = transaction {  // => All operations in one transaction
                    // Step 1: Check inventory
                    val currentQty = Inventory
                        .select { Inventory.productId eq request.productId }
                        .singleOrNull()
                        ?.get(Inventory.quantity)
                        ?: throw IllegalStateException("Product not found")
                                             // => Exception rolls back transaction

                    if (currentQty < request.quantity) {
                        rollback()           // => Explicit rollback (alternative to exception)
                        throw IllegalStateException("Insufficient inventory: $currentQty available")
                    }

                    // Step 2: Deduct inventory
                    Inventory.update({ Inventory.productId eq request.productId }) {
                        it[quantity] = currentQty - request.quantity
                        // => If order insert fails below, this update is rolled back
                    }

                    // Step 3: Create order
                    val newOrderId = Orders.insert {
                        it[userId] = request.userId
                        it[total] = java.math.BigDecimal("99.99")
                    } get Orders.id
                    // => If any step threw, ALL previous steps in this transaction are undone

                    newOrderId  // => Return order ID from transaction
                }

                call.respond(HttpStatusCode.Created, mapOf("orderId" to orderId))
            } catch (e: IllegalStateException) {
                call.respond(HttpStatusCode.BadRequest, mapOf("error" to e.message))
            }
        }
    }
}
// => POST /orders {"userId":1,"productId":5,"quantity":3}
//    If inventory >= 3: => 201 {"orderId":42}, inventory decremented, order created atomically
//    If inventory < 3:  => 400 {"error":"Insufficient inventory: 1 available"}
//                          inventory unchanged, no order created

import io.ktor.client.*
import io.ktor.client.engine.mock.*        // => MockEngine for testing
import io.ktor.client.plugins.contentnegotiation.*
import io.ktor.client.request.*
import io.ktor.client.call.*
import io.ktor.serialization.kotlinx.json.*
import io.ktor.http.*
import io.ktor.server.testing.*
import kotlin.test.*
import kotlinx.serialization.Serializable

@Serializable
data class WeatherData(val city: String, val temperature: Double)

// Service class using HttpClient (production code)
class WeatherService(private val client: HttpClient, private val baseUrl: String) {
    suspend fun getWeather(city: String): WeatherData {
        return client.get("$baseUrl/weather?city=$city").body()
    }
}

class WeatherServiceTest {

    @Test
    fun `getWeather returns weather data for valid city`() = runBlocking {
        // Create MockEngine with predefined response
        val mockEngine = MockEngine { request ->
            // Verify the request URL contains expected city
            assertTrue(request.url.parameters["city"] == "London")

            respond(                         // => Return mock response
                content = """{"city":"London","temperature":15.5}""",
                status = HttpStatusCode.OK,
                headers = headersOf(
                    HttpHeaders.ContentType, ContentType.Application.Json.toString()
                )
            )
        }

        val mockClient = HttpClient(mockEngine) {
            install(ContentNegotiation) { json() }
        }

        val service = WeatherService(mockClient, "https://weather-api.example.com")
        val weather = service.getWeather("London")

        assertEquals("London", weather.city)
        assertEquals(15.5, weather.temperature)
        // => No real HTTP call made; MockEngine intercepts and returns test data
    }

    @Test
    fun `getWeather handles 404 from external API`() = runBlocking {
        val mockEngine = MockEngine {
            respond(
                content = """{"error":"City not found"}""",
                status = HttpStatusCode.NotFound
            )
        }

        val mockClient = HttpClient(mockEngine) {
            install(ContentNegotiation) { json() }
        }

        val service = WeatherService(mockClient, "https://weather-api.example.com")

        // Verify that 404 from external API causes appropriate exception
        assertFailsWith<io.ktor.client.plugins.ResponseException> {
            service.getWeather("Atlantis")   // => MockEngine returns 404
        }
    }
}

import com.zaxxer.hikari.HikariConfig
import com.zaxxer.hikari.HikariDataSource
import io.ktor.server.application.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.micrometer.prometheus.PrometheusMeterRegistry
import io.micrometer.prometheus.PrometheusConfig

fun createTunedDataSource(
    jdbcUrl: String,
    registry: PrometheusMeterRegistry  // => Inject metrics registry for pool monitoring
): HikariDataSource {
    val config = HikariConfig().apply {
        this.jdbcUrl = jdbcUrl
        driverClassName = "org.postgresql.Driver"

        // Pool sizing: PostgreSQL recommended formula
        // poolSize = (core_count * 2) + effective_spindle_count
        // For a 4-core server with SSD (1 spindle): 4*2+1 = 9
        maximumPoolSize = 9              // => Max connections to PostgreSQL
                                          // => More connections != more throughput
                                          // => Too many connections increase contention
        minimumIdle = 2                  // => Always keep 2 connections warm
                                          // => New connections take ~50ms to establish
        maxLifetime = 1_800_000L         // => Replace connections after 30 minutes
                                          // => Prevents stale connections from dead TCP sessions
        keepaliveTime = 60_000L          // => Send keepalive after 60s idle
                                          // => Detects connections severed by network equipment

        // Timeouts
        connectionTimeout = 30_000L     // => Wait max 30s for a connection from pool
                                          // => Throw HikariPool.PoolInitializationException on timeout
        idleTimeout = 600_000L           // => Close connections idle for 10 minutes
        validationTimeout = 5_000L       // => Timeout for connection validation query

        // Performance
        isAutoCommit = false             // => Exposed manages transactions
        connectionTestQuery = "SELECT 1" // => Validation query for old drivers
                                          // => Modern drivers: use isReadOnly validation instead
        // Enable HikariCP metrics export to Micrometer/Prometheus
        metricRegistry = registry        // => Exports pool metrics automatically
        // => Metrics: hikaricp.connections.active, .idle, .pending, .timeout.total
    }
    return HikariDataSource(config)
}

fun Application.configureRouting() {
    val registry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)

    routing {
        // Pool health endpoint for operations teams
        get("/admin/db-pool") {
            val source = /* hikariDataSource */ HikariDataSource(HikariConfig())
            call.respondText(
                "Pool stats: active=${source.hikariPoolMXBean?.activeConnections}, " +
                "idle=${source.hikariPoolMXBean?.idleConnections}, " +
                "pending=${source.hikariPoolMXBean?.threadsAwaitingConnection}"
                // => "Pool stats: active=3, idle=6, pending=0"
            )
        }

        get("/metrics") {
            call.respondText(registry.scrape(), io.ktor.http.ContentType.Text.Plain)
        }
    }
}
// => Prometheus metrics include:
// => hikaricp_connections_active{pool="HikariPool-1"} 3.0
// => hikaricp_connections_pending{pool="HikariPool-1"} 0.0
// => hikaricp_connections_timeout_total{pool="HikariPool-1"} 0.0

import io.ktor.server.application.*
import io.ktor.server.plugins.defaultheaders.*
import io.ktor.server.response.*
import io.ktor.server.routing.*
import io.ktor.http.*

fun Application.configureSecurityHeaders() {
    install(DefaultHeaders) {
        // HSTS: Force HTTPS for 1 year (never send over HTTP)
        header(
            HttpHeaders.StrictTransportSecurity,
            "max-age=31536000; includeSubDomains; preload"
            // => max-age=1 year; applies to all subdomains; eligible for preload list
        )

        // CSP: Restrict which sources can load content (prevents XSS)
        header(
            "Content-Security-Policy",
            buildString {
                append("default-src 'self'; ")          // => Only load from same origin by default
                append("script-src 'self'; ")            // => No inline scripts
                append("style-src 'self' 'unsafe-inline'; ") // => Allow inline CSS (relax for styled components)
                append("img-src 'self' data: https:; ") // => Images from same origin, data URIs, any HTTPS
                append("connect-src 'self'; ")           // => XHR/Fetch only to same origin
                append("frame-ancestors 'none'; ")       // => Prevent embedding in iframes (clickjacking)
                append("form-action 'self'")             // => Forms submit only to same origin
            }
        )

        // X-Frame-Options: Prevent clickjacking (legacy, CSP frame-ancestors preferred)
        header("X-Frame-Options", "DENY")
        // => Prevents page from being embedded in iframes

        // X-Content-Type-Options: Prevent MIME sniffing
        header("X-Content-Type-Options", "nosniff")
        // => Browser must use Content-Type as declared, not guess from content

        // Referrer-Policy: Control Referer header in cross-origin requests
        header("Referrer-Policy", "strict-origin-when-cross-origin")
        // => Sends full URL within same origin, only origin for cross-origin requests

        // Permissions-Policy: Disable browser features not needed
        header(
            "Permissions-Policy",
            "camera=(), microphone=(), geolocation=()"
            // => Disable camera, microphone, geolocation access
            // => Limits attack surface if XSS exploit injects malicious code
        )
    }
}
// => Every response includes:
// => Strict-Transport-Security: max-age=31536000; includeSubDomains; preload
// => Content-Security-Policy: default-src 'self'; ...
// => X-Frame-Options: DENY
// => X-Content-Type-Options: nosniff