diff --git a/AGENTS.md b/AGENTS.md index fd26fd9..c79fd74 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -13,7 +13,7 @@ The project is structured into distinct, decoupled framework modules. **Do not v - **`cache`**: Two-level caching engine (`Cache`), L1 LRU memory cache (`L1Cache`), L2 Valkey client (`ValkeyWrapper`), CBOR binary serialization (`BinaryCodec`), bucket versioning, and single-flight loader deduplication (`SingleFlight`). - **`exposed`**: Database connection management (`DatabaseWrapper`), Exposed ORM integration, Liquibase migrations, `@Transactional` annotations & proxying (`TransactionalProxy`), and repository base classes (`UpsertRepository`, `EntityRepository`). - **`validator`**: Reflection-based validation engine (`Validator`), constraints (`@RequireAtLeastOneValid`, `@NotNull`, `@NotBlank`, `@NotEmpty`), and `ObjectNotValidException`. -- **`ktor`**: Ktor (server and client) integrations, route annotation binding (`@RestController`, `@GetMapping`, `@PostMapping`, `@PatchMapping`), request parameter resolvers (`@QueryParam`, `@PathParam`, `@RequestBody`), automatic `@Valid` validation integration, `ResponseEntity` wrapper, `MessageDto` error response standardization, OpenAPI metadata (`@Operation`, `@ApiResponses`, `@ApiResponse`), a preconfigured HTTP client (`createHttpClient`), and the built-in `/health` UP endpoint auto-registered by `configureDefaultModules`. +- **`ktor`**: Ktor (server and client) integrations, route annotation binding (`@RestController`, `@GetMapping`, `@PostMapping`, `@PatchMapping`), request parameter resolvers (`@QueryParam`, `@PathParam`, `@RequestBody`), automatic `@Valid` validation integration, `ResponseEntity` wrapper, `MessageDto` error response standardization, OpenAPI metadata (`@Operation`, `@ApiResponses`, `@ApiResponse`), a preconfigured HTTP client (`createHttpClient`) forwarding the current session, a persistent per-session journal (`SessionContext`, `SessionStore`, `/v1/session-traces`), and the built-in `/health` UP endpoint auto-registered by `configureDefaultModules`. - **`koin`**: Dependency carrier publishing the Koin runtime APIs (`koin-bom`, `koin-core`, `koin-annotations`) via `api(...)`; contains no source code. - **`koin-exposed`**: Koin bridge for transactions — `applyTransactionalProxies()` post-processor replacing eligible singleton definitions with `TransactionalProxy`-wrapping factories; depends on `exposed` and the `koin` carrier. - **`ktor-test`**: Dependency carrier publishing the Ktor server test host (`ktor-server-test-host`) and the Ktor client mock engine (`ktor-client-mock`) via `api(...)`; contains no source code. @@ -66,4 +66,6 @@ Before concluding your task, perform a final check: - Is validation logic tested and error responses sanitized? - Have you verified the build and tests with `./gradlew test`? -Consult additional guides for [Cache](guidelines/CACHE.md), [Database](guidelines/DATABASE.md), and [Performance](guidelines/PERFORMANCE.md) rules as needed. +Consult additional guides for [Cache](guidelines/CACHE.md), [Database](guidelines/DATABASE.md), +[Performance](guidelines/PERFORMANCE.md), and [Session Tracking](guidelines/SESSION_TRACKING.md) +rules as needed. diff --git a/cache/src/main/kotlin/SingleFlight.kt b/cache/src/main/kotlin/SingleFlight.kt index 88281ba..5bb95bf 100644 --- a/cache/src/main/kotlin/SingleFlight.kt +++ b/cache/src/main/kotlin/SingleFlight.kt @@ -6,7 +6,7 @@ import java.util.concurrent.ConcurrentHashMap /** * A concurrency control utility that suppresses duplicate in-flight async calls. * - * Ensures that for a given [key], only one concurrent execution of [block] is performed at a time. + * Ensures that for a given `key`, only one concurrent execution of `block` is performed at a time. * Subsequent concurrent callers for the same key await and receive the result (or exception) * produced by the first execution. */ diff --git a/cache/src/main/kotlin/ValkeyWrapperImpl.kt b/cache/src/main/kotlin/ValkeyWrapperImpl.kt index 4d85d30..1f21866 100644 --- a/cache/src/main/kotlin/ValkeyWrapperImpl.kt +++ b/cache/src/main/kotlin/ValkeyWrapperImpl.kt @@ -79,6 +79,7 @@ class ValkeyWrapperImpl(private val client: GlideClient) : ValkeyWrapper { * @param port Network port of the Valkey server (defaults to `6379`). * @return A newly initialized and connected [ValkeyWrapperImpl]. */ + @Suppress("unused") // Public factory consumed by downstream projects. suspend fun connect(host: String = "localhost", port: Int = 6379): ValkeyWrapperImpl { val configuration = GlideClientConfiguration.builder() .address( diff --git a/cache/src/test/kotlin/SingleFlightTest.kt b/cache/src/test/kotlin/SingleFlightTest.kt index bb54589..453a72e 100644 --- a/cache/src/test/kotlin/SingleFlightTest.kt +++ b/cache/src/test/kotlin/SingleFlightTest.kt @@ -9,6 +9,7 @@ import org.junit.jupiter.api.DisplayName import org.junit.jupiter.api.Nested import org.junit.jupiter.api.Test import java.util.concurrent.atomic.AtomicInteger +import kotlin.time.Duration.Companion.milliseconds class SingleFlightTest { @@ -24,7 +25,7 @@ class SingleFlightTest { val loader: suspend () -> String = { callCounter.incrementAndGet() - delay(50) + delay(50.milliseconds) "Shared Result" } @@ -96,7 +97,7 @@ class SingleFlightTest { val failingLoader: suspend () -> String = { callCounter.incrementAndGet() - delay(30) + delay(30.milliseconds) throw IllegalStateException("Loader failure") } diff --git a/core/src/main/kotlin/TimeUtils.kt b/core/src/main/kotlin/TimeUtils.kt index 6448c60..14ba23e 100644 --- a/core/src/main/kotlin/TimeUtils.kt +++ b/core/src/main/kotlin/TimeUtils.kt @@ -8,12 +8,14 @@ import kotlin.time.Clock /** * Provides utilities for timestamps. */ +@Suppress("unused") // Public utility consumed by downstream projects. object TimeUtils { /** * Returns the current date and time used to track row creation and updates. * * @return the current date and time in UTC. */ + @Suppress("unused") // Public utility consumed by downstream projects. fun now(): LocalDateTime = Clock.System.now().toLocalDateTime(TimeZone.UTC) } diff --git a/exposed/src/main/kotlin/DatabaseWrapper.kt b/exposed/src/main/kotlin/DatabaseWrapper.kt index d08a986..cb39f43 100644 --- a/exposed/src/main/kotlin/DatabaseWrapper.kt +++ b/exposed/src/main/kotlin/DatabaseWrapper.kt @@ -77,6 +77,7 @@ class DatabaseWrapper( * @param tables tables managed by [initializeSchema] and [dropSchema]. * @return `true` when the collection of registered tables changed. */ + @Suppress("unused") // Public schema API consumed by downstream projects. fun addTables(vararg tables: Table) = this.tables.addAll(tables) diff --git a/exposed/src/main/kotlin/repositories/EntityRepository.kt b/exposed/src/main/kotlin/repositories/EntityRepository.kt index d4a2978..6b1ccb2 100644 --- a/exposed/src/main/kotlin/repositories/EntityRepository.kt +++ b/exposed/src/main/kotlin/repositories/EntityRepository.kt @@ -12,6 +12,7 @@ import org.jetbrains.exposed.v1.jdbc.SizedIterable * @param ENTITY type of entity managed by this repository. * @property entityClass Exposed DAO class used to create and query entities. */ +@Suppress("unused") // Public base class consumed by downstream repositories. abstract class EntityRepository>( entityClass: EntityClass ) : UpsertRepository(entityClass) { @@ -20,6 +21,7 @@ abstract class EntityRepository>( * * @return a lazily evaluated Exposed entity sequence containing every entity. */ + @Suppress("unused") // Public repository API consumed by downstream projects. fun findAll(): SizedIterable = entityClass.all() @@ -29,6 +31,7 @@ abstract class EntityRepository>( * @param ids identifiers to search for. * @return a lazily evaluated Exposed entity sequence containing matching entities. */ + @Suppress("unused") // Public repository API consumed by downstream projects. fun findAllByIds(ids: List): SizedIterable = entityClass.find { entityClass.table.id inList ids } @@ -38,6 +41,7 @@ abstract class EntityRepository>( * @param id identifier to search for. * @return the matching entity, or `null` when it does not exist. */ + @Suppress("unused") // Public repository API consumed by downstream projects. fun findById(id: T): ENTITY? = entityClass.findById(id) } \ No newline at end of file diff --git a/exposed/src/main/kotlin/repositories/UpsertRepository.kt b/exposed/src/main/kotlin/repositories/UpsertRepository.kt index 3794c01..f4af8ea 100644 --- a/exposed/src/main/kotlin/repositories/UpsertRepository.kt +++ b/exposed/src/main/kotlin/repositories/UpsertRepository.kt @@ -25,6 +25,7 @@ abstract class UpsertRepository>( * @param id identifier of the referenced entity. * @return the typed entity reference of [id]. */ + @Suppress("unused") // Public repository API consumed by downstream projects. fun getReference(id: T): EntityID = EntityID(id, entityClass.table) diff --git a/gradle.properties b/gradle.properties index 3b12fcc..e5288bb 100644 --- a/gradle.properties +++ b/gradle.properties @@ -7,5 +7,5 @@ org.gradle.caching=true org.gradle.configuration-cache=true # Framework version -version=0.0.26-SNAPSHOT +version=0.0.27-SNAPSHOT diff --git a/guidelines/SESSION_TRACKING.md b/guidelines/SESSION_TRACKING.md new file mode 100644 index 0000000..c54fbab --- /dev/null +++ b/guidelines/SESSION_TRACKING.md @@ -0,0 +1,79 @@ +# Session Tracking + +The `ktor` module ships a persistent session journal, enabled by default for every project using +`configureDefaultModules()`. It answers one question: what happened in one user session, including +across a restart and across services. + +## Identifiers + +- `X-Session-Id` — long-lived session identifier. A client generates it or keeps the one returned + by the first response. It is a correlation token, **never** an authorization credential. +- `X-Request-Id` — identifier of one request. It is generated when absent, echoed on the response, + and the parent link of every outgoing call. +- `X-Parent-Request-Id` — set by `createHttpClient()` on an outgoing call; the called service + records it on its `IN` event. + +Only `[A-Za-z0-9_-]{8,64}` is accepted. A malformed value is replaced by a generated identifier +instead of failing the request, and a header value never becomes a file name without validation. + +## Storage + +`FileSessionStore` writes one append-only JSONL file per session and per service: + +``` +/sessions//.jsonl +``` + +Default directory: `$SHIKKANIME_LOGS_DIR/sessions`, or `/logs/sessions`. +Default service: `MODULE_NAME`, then `SHIKKANIME_SERVICE`, then `default`. + +Events are `IN` (request received), `OUT` (request sent by `createHttpClient`) and `EVENT` +(application-defined). They carry status, duration, path without query string, target host, error +type/message and response size. Bodies, query strings and personal identifiers are never stored. + +Retention deletes a file 7 days after its last write by default; configure `SessionConfig.retention`. +A partial last line (crash) is ignored on read and repaired on the next append. + +## Performance contract + +- `SessionStore.append` is called on the request thread and must not block. `FileSessionStore` + offers the event to a bounded `ArrayBlockingQueue` and returns immediately. +- A single daemon thread serializes and writes the events, flushes when the queue drains, and runs + retention once a day. A full queue drops the event (`droppedEvents`) instead of slowing a request. +- Reads are bounded by `limit` and run on `Dispatchers.IO`; the trace endpoint is not on the request + path. +- The writer thread runs at minimum priority. The journal is a diagnostic side channel, not part of + the response path. + +## Read API + +`GET /v1/session-traces/{sessionId}` and `GET /v1/session-traces` are registered only when +`SHIKKANIME_SESSION_TOKEN` (or `SessionConfig.apiToken`) is set. Both require +`Authorization: Bearer `. An unknown session is 404, an invalid identifier 400, a missing or +wrong token 401. The endpoints are excluded from tracking. + +## Configuration and extension + +```kotlin +configureDefaultModules( + session = SessionConfig( + enabled = true, + service = "api", + retention = Duration.ofDays(7), + apiToken = System.getenv("SHIKKANIME_SESSION_TOKEN") + ) +) +``` + +A project that needs a database implements `SessionStore` (the implementation can live in a module +that depends on Exposed) and passes it through `SessionConfig.store`. The framework then leaves the +store lifecycle to the caller; the `ktor` module keeps no dependency on `exposed` or `cache`. + +`SessionContext.current()` exposes the current scope to a controller or service, +`SessionContext.event` appends an application-defined event, and `SessionContext.with` opens a +scope around a background job. + +## Out of scope + +Log correlation (session identifiers in every log line) is a separate layer. The journal is the +timeline; the log files are the detail. diff --git a/koin-exposed/src/main/kotlin/TransactionalAutoProxy.kt b/koin-exposed/src/main/kotlin/TransactionalAutoProxy.kt index 20972ab..d350022 100644 --- a/koin-exposed/src/main/kotlin/TransactionalAutoProxy.kt +++ b/koin-exposed/src/main/kotlin/TransactionalAutoProxy.kt @@ -69,6 +69,7 @@ fun applyTransactionalProxies(koin: Koin): Int { * @return number of definitions wrapped by a transactional proxy. * @see applyTransactionalProxies */ +@Suppress("unused") // Public convenience overload consumed by downstream projects. fun applyTransactionalProxies(): Int = applyTransactionalProxies(KoinPlatformTools.defaultContext().get()) diff --git a/ktor/AGENTS.md b/ktor/AGENTS.md index aee049a..317e3c5 100644 --- a/ktor/AGENTS.md +++ b/ktor/AGENTS.md @@ -3,7 +3,7 @@ This file contains specific rules for the `ktor` submodule. All agents working within this submodule must strictly adhere to these guidelines in addition to the root [`AGENTS.md`](../AGENTS.md). ## Module Purpose & Scope -The `ktor` module provides web framework integration for Ktor (server **and client**). It handles `@RestController` discovery, route mapping (`@GetMapping`, `@PostMapping`, `@PatchMapping`), request argument resolution (`@QueryParam`, `@PathParam`, `@RequestBody`), automatic `@Valid` validation, `ResponseEntity` packaging, `MessageDto` response serialization, and OpenAPI documentation generation, as well as framework-aware HTTP clients (e.g. a preconfigured `HttpClient`). +The `ktor` module provides web framework integration for Ktor (server **and client**). It handles `@RestController` discovery, route mapping (`@GetMapping`, `@PostMapping`, `@PatchMapping`), request argument resolution (`@QueryParam`, `@PathParam`, `@RequestBody`), automatic `@Valid` validation, `ResponseEntity` packaging, `MessageDto` response serialization, OpenAPI documentation generation, a persistent per-session journal, and framework-aware HTTP clients (e.g. a preconfigured `HttpClient`). ## Submodule-Specific Rules @@ -28,3 +28,13 @@ The `ktor` module provides web framework integration for Ktor (server **and clie - Methods should return responses wrapped in `ResponseEntity` or direct serializable objects. - Error payloads must utilize `MessageDto.error(message)` or `ErrorMessageDto`. - Document endpoints with `@Operation`, `@ApiResponses`, and `@ApiResponse` to generate compliant OpenAPI documentation. + +6. **Session Tracking**: + - `configureDefaultModules()` installs the session journal by default. `SessionStore.append` + must stay non-blocking: never move JSON serialization or file I/O to the request thread. + - The read API is registered only when `SessionConfig.apiToken` is set, and every endpoint is + bearer-protected. An unprotected session endpoint is a review blocker. + - Session and request identifiers become file names: validate them with `SessionIds`, never + interpolate a raw header into a path. + - See [Session Tracking](../guidelines/SESSION_TRACKING.md) for the storage and performance + contract. diff --git a/ktor/src/main/kotlin/Client.kt b/ktor/src/main/kotlin/Client.kt index 1ba4dbf..f1c6813 100644 --- a/ktor/src/main/kotlin/Client.kt +++ b/ktor/src/main/kotlin/Client.kt @@ -1,5 +1,6 @@ package fr.shikkanime.ktor +import fr.shikkanime.ktor.session.installSessionPropagation import io.ktor.client.* import io.ktor.client.engine.okhttp.* import io.ktor.client.plugins.* @@ -12,14 +13,15 @@ import kotlinx.serialization.ExperimentalSerializationApi /** * Creates a preconfigured [HttpClient] using the OkHttp engine. * - * Configures HTTP timeouts (30s request, 10s connect, 30s socket) and registers content negotiation - * for JSON, Protobuf, and CBOR using [defaultJson], [defaultProtoBuf], and [defaultCbor]. + * Configures HTTP timeouts (30s request, 10s connect, 30s socket), registers content negotiation + * for JSON, Protobuf, and CBOR using [defaultJson], [defaultProtoBuf], and [defaultCbor], and + * forwards the current session to outgoing calls. * * @return a new preconfigured [HttpClient] instance. */ @OptIn(ExperimentalSerializationApi::class) -fun createHttpClient(): HttpClient = - HttpClient(OkHttp) { +fun createHttpClient(): HttpClient { + val client = HttpClient(OkHttp) { install(HttpTimeout) { requestTimeoutMillis = 30_000 connectTimeoutMillis = 10_000 @@ -31,4 +33,7 @@ fun createHttpClient(): HttpClient = protobuf(defaultProtoBuf) cbor(defaultCbor) } - } \ No newline at end of file + } + client.installSessionPropagation() + return client +} \ No newline at end of file diff --git a/ktor/src/main/kotlin/ControllerRequestHandler.kt b/ktor/src/main/kotlin/ControllerRequestHandler.kt index 8b68189..002a309 100644 --- a/ktor/src/main/kotlin/ControllerRequestHandler.kt +++ b/ktor/src/main/kotlin/ControllerRequestHandler.kt @@ -5,7 +5,6 @@ import fr.shikkanime.ktor.dtos.MessageDto import fr.shikkanime.validator.Validator import fr.shikkanime.validator.exceptions.ObjectNotValidException import io.ktor.http.* -import io.ktor.server.request.* import io.ktor.util.* import io.ktor.server.response.* import io.ktor.server.routing.* @@ -15,6 +14,14 @@ import kotlin.reflect.KFunction import kotlin.reflect.full.callSuspendBy import kotlin.reflect.full.hasAnnotation +/** + * Call attribute holding the exception that failed a controller invocation. + * + * The session journal and the future log layer read it to describe the failure without parsing the + * response body. + */ +internal val ErrorAttribute: AttributeKey = AttributeKey("Error") + /** * Creates Ktor handlers that invoke controller functions and translate their results to responses. */ @@ -63,6 +70,7 @@ internal object ControllerRequestHandler { } } catch (exception: Exception) { val cause = (exception as? InvocationTargetException)?.targetException ?: exception + call.attributes.put(ErrorAttribute, cause) cause.printStackTrace() when (cause) { diff --git a/ktor/src/main/kotlin/HealthController.kt b/ktor/src/main/kotlin/HealthController.kt index a9afe9e..c4bbeb3 100644 --- a/ktor/src/main/kotlin/HealthController.kt +++ b/ktor/src/main/kotlin/HealthController.kt @@ -5,6 +5,7 @@ package fr.shikkanime.ktor */ @RestController("/health") class HealthController : IController { + @Suppress("unused") // Bound reflectively by ControllerBinder. @Operation(summary = "Health check", description = "Returns OK when the service is UP") @ApiResponses([ApiResponse(status = 200, description = "Service is UP", responseType = String::class)]) @GetMapping diff --git a/ktor/src/main/kotlin/Modules.kt b/ktor/src/main/kotlin/Modules.kt index 9f13db0..4299d8d 100644 --- a/ktor/src/main/kotlin/Modules.kt +++ b/ktor/src/main/kotlin/Modules.kt @@ -1,5 +1,12 @@ package fr.shikkanime.ktor +import fr.shikkanime.ktor.session.FileSessionStore +import fr.shikkanime.ktor.session.SessionConfig +import fr.shikkanime.ktor.session.SessionHeaders +import fr.shikkanime.ktor.session.SessionTraceController +import fr.shikkanime.ktor.session.installSessionTracking +import fr.shikkanime.ktor.session.resolvedDirectory +import fr.shikkanime.ktor.session.resolvedService import io.ktor.http.* import io.ktor.openapi.* import io.ktor.serialization.kotlinx.cbor.* @@ -16,18 +23,49 @@ import kotlinx.serialization.json.Json import kotlinx.serialization.modules.SerializersModule /** - * Configures the default content negotiation and CORS modules, and registers the built-in [HealthController]. + * Configures the default content negotiation, CORS, session tracking, and the built-in + * [HealthController]. * - * Content negotiation supports JSON, Protobuf, and CBOR formats using standard framework configurations, - * while CORS accepts requests from any host using Ktor's default HTTP methods. A `GET /health` route - * returning `OK` is bound automatically so every consumer exposes the same UP endpoint. + * Content negotiation supports JSON, Protobuf, and CBOR formats using standard framework + * configurations, while CORS accepts requests from any host using Ktor's default HTTP methods and + * exposes the session headers to browser clients. A `GET /health` route returning `OK` is bound + * automatically so every consumer exposes the same UP endpoint. + * + * When [session] is enabled, every request is journaled under + * `logs/sessions//.jsonl`, and the read API is registered when a token is set. + * + * @param extraModules kotlinx serialization modules merged into the JSON converter. + * @param session session tracking configuration. */ -fun Application.configureDefaultModules(vararg extraModules: SerializersModule = emptyArray()) { +fun Application.configureDefaultModules( + vararg extraModules: SerializersModule = emptyArray(), + session: SessionConfig = SessionConfig() +) { install(ResponseSizeRecorder) configureContentNegotiation(*extraModules) configureCORS() + + val store = if (session.enabled) { + session.store ?: FileSessionStore( + directory = session.resolvedDirectory(), + service = session.resolvedService(), + retention = session.retention + ) + } else { + null + } + + if (store != null) { + installSessionTracking(session, store, closeStoreOnStop = session.store == null) + } + routing { - ControllerBinder.register(this, listOf(HealthController())) + val controllers = mutableListOf(HealthController()) + val token = session.apiToken?.takeIf { it.isNotBlank() } + if (store != null && token != null) { + controllers += SessionTraceController(store, token) + } + ControllerBinder.register(this, controllers) } } @@ -74,11 +112,19 @@ internal fun Application.configureContentNegotiation(vararg extraModules: Serial /** * Configures CORS on the server to accept requests from any host using default HTTP methods. + * + * The session correlation headers are allowed in preflight requests and exposed on responses, so a + * browser client can read the identifiers the server assigned to its session. */ internal fun Application.configureCORS() { install(CORS) { anyHost() HttpMethod.DefaultMethods.forEach(::allowMethod) + allowHeader(SessionHeaders.SESSION_ID) + allowHeader(SessionHeaders.REQUEST_ID) + allowHeader(SessionHeaders.PARENT_REQUEST_ID) + exposeHeader(SessionHeaders.SESSION_ID) + exposeHeader(SessionHeaders.REQUEST_ID) } } diff --git a/ktor/src/main/kotlin/ResponseSizePlugin.kt b/ktor/src/main/kotlin/ResponseSizePlugin.kt index 3793395..72f5671 100644 --- a/ktor/src/main/kotlin/ResponseSizePlugin.kt +++ b/ktor/src/main/kotlin/ResponseSizePlugin.kt @@ -3,9 +3,8 @@ package fr.shikkanime.ktor import io.ktor.http.ContentType import io.ktor.http.Headers import io.ktor.http.content.OutgoingContent -import io.ktor.utils.io.readRemaining +import io.ktor.utils.io.readBuffer import kotlinx.io.readByteArray -import java.io.InputStream import io.ktor.server.application.ApplicationPlugin import io.ktor.server.application.createApplicationPlugin import io.ktor.server.application.hooks.ResponseBodyReadyForSend @@ -45,7 +44,7 @@ val ResponseSizeRecorder: ApplicationPlugin = // and send the very same bytes. val materialized = (content as OutgoingContent.ReadChannelContent) .readFrom() - .readRemaining() + .readBuffer() .readByteArray() call.attributes.put(ResponseSizeAttribute, materialized.size.toLong()) diff --git a/ktor/src/main/kotlin/content/BinaryContent.kt b/ktor/src/main/kotlin/content/BinaryContent.kt index bf3ea88..f367784 100644 --- a/ktor/src/main/kotlin/content/BinaryContent.kt +++ b/ktor/src/main/kotlin/content/BinaryContent.kt @@ -14,8 +14,8 @@ import io.ktor.http.content.OutgoingContent * * @property bytes raw payload. * @property contentType media type of the payload. - * @property etag validator identifying this exact payload, when the caller publishes one. - * @property cacheControl `Cache-Control` directive of the response, when the caller publishes one. + * @param etag validator identifying this exact payload, when the caller publishes one. + * @param cacheControl `Cache-Control` directive of the response, when the caller publishes one. * @see fr.shikkanime.ktor.ResponseEntity.binary */ class BinaryContent( diff --git a/ktor/src/main/kotlin/session/FileSessionStore.kt b/ktor/src/main/kotlin/session/FileSessionStore.kt new file mode 100644 index 0000000..8178199 --- /dev/null +++ b/ktor/src/main/kotlin/session/FileSessionStore.kt @@ -0,0 +1,170 @@ +package fr.shikkanime.ktor.session + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.io.RandomAccessFile +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.Files +import java.nio.file.Path +import java.time.Clock +import java.time.Duration + +private const val DEFAULT_QUEUE_CAPACITY = 16_384 +private const val LAST_LINE_WINDOW_BYTES = 4_096L +private const val FIRST_LINE_ATTEMPTS = 3 + +/** + * Session store writing one append-only JSONL file per session and per service. + * + * Files live under `//.jsonl`. Writes go through [SessionWriter], + * which keeps them off the request thread; reads are bounded by the requested limit and happen on + * [Dispatchers.IO]. This is the default store used by `configureDefaultModules`. + * + * @param directory root directory of the session journals. + * @param service service name used as the subdirectory. + * @param retention how long an inactive session file is kept. + * @param queueCapacity maximum number of events waiting for the writer. + * @param clock clock used for retention. + */ +class FileSessionStore( + directory: Path, + private val service: String, + retention: Duration = Duration.ofDays(7), + queueCapacity: Int = DEFAULT_QUEUE_CAPACITY, + clock: Clock = Clock.systemUTC() +) : SessionStore { + private val serviceDirectory: Path = directory.resolve(service) + private val writer: SessionWriter + + /** + * Number of events dropped because the writer queue was full. + * + * A non-zero value means the service produced events faster than the single writer could drain + * them; the request path itself was never blocked. + */ + val droppedEvents: Long + get() = writer.droppedEvents() + + init { + require(SessionIds.isValidService(service)) { "Invalid session service name: $service" } + require(queueCapacity > 0) { "The session queue capacity must be positive" } + Files.createDirectories(serviceDirectory) + writer = SessionWriter(serviceDirectory, retention, queueCapacity, clock) + } + + override fun append(event: SessionEvent) = + writer.append(event) + + override suspend fun find(sessionId: String, limit: Int): SessionTrace? = + withContext(Dispatchers.IO) { readTrace(sessionId, limit) } + + override suspend fun sessions(limit: Int): List = + withContext(Dispatchers.IO) { readSummaries(limit) } + + override fun flush() = + writer.flush() + + override fun close() = + writer.close() + + private fun readTrace(sessionId: String, limit: Int): SessionTrace? { + if (!SessionIds.isValid(sessionId)) return null + require(limit > 0) { "The session event limit must be positive" } + val file = serviceDirectory.resolve(sessionId + SESSION_FILE_EXTENSION) + if (!Files.isRegularFile(file)) return null + val events = ArrayDeque(limit) + var first: String? = null + var last: String? = null + var count = 0 + Files.newBufferedReader(file, UTF_8).useLines { lines -> + lines.forEach { line -> + val event = decode(line) ?: return@forEach + if (first == null) first = event.timestamp + last = event.timestamp + count++ + if (events.size == limit) events.removeFirst() + events.addLast(event) + } + } + val firstSeen = first ?: return null + val lastSeen = last ?: return null + return SessionTrace( + sessionId = sessionId, + service = service, + firstSeen = firstSeen, + lastSeen = lastSeen, + eventCount = count, + truncated = count > limit, + events = events.toList() + ) + } + + private fun readSummaries(limit: Int): List { + require(limit > 0) { "The session summary limit must be positive" } + val files = Files.list(serviceDirectory).use { stream -> + stream.filter { Files.isRegularFile(it) } + .filter { it.fileName.toString().endsWith(SESSION_FILE_EXTENSION) } + .sorted( + Comparator + .comparing { file -> Files.getLastModifiedTime(file).toMillis() } + .reversed() + ) + .limit(limit.toLong()) + .toList() + } + return files.map(::summary) + } + + private fun summary(file: Path): SessionSummary = + SessionSummary( + sessionId = file.fileName.toString().removeSuffix(SESSION_FILE_EXTENSION), + service = service, + firstSeen = readFirstEvent(file)?.timestamp, + lastSeen = readLastLine(file)?.let(::decode)?.timestamp, + sizeBytes = Files.size(file) + ) + + private fun readFirstEvent(file: Path): SessionEvent? = + Files.newBufferedReader(file, UTF_8).use { reader -> + repeat(FIRST_LINE_ATTEMPTS) { + val line = reader.readLine() ?: return null + decode(line)?.let { return it } + } + null + } + + private fun readLastLine(file: Path): String? { + RandomAccessFile(file.toFile(), "r").use { fileHandle -> + var end = fileHandle.length() + if (end == 0L) return null + var window = LAST_LINE_WINDOW_BYTES + while (end > 0) { + val start = maxOf(0L, end - window) + val bytes = ByteArray((end - start).toInt()) + fileHandle.seek(start) + fileHandle.readFully(bytes) + val text = String(bytes, UTF_8) + val newline = text.lastIndexOf('\n', text.length - 2) + if (newline >= 0) { + val line = text.substring(newline + 1).trim() + if (line.isNotEmpty()) return line + end = start + newline + window = LAST_LINE_WINDOW_BYTES + } else if (start == 0L) { + return text.trim().takeIf { it.isNotEmpty() } + } else { + end = start + window *= 2 + } + } + } + return null + } + + private fun decode(line: String): SessionEvent? = + try { + SESSION_JSON.decodeFromString(line) + } catch (_: Exception) { + null + } +} diff --git a/ktor/src/main/kotlin/session/SessionAttributes.kt b/ktor/src/main/kotlin/session/SessionAttributes.kt new file mode 100644 index 0000000..0e03e3d --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionAttributes.kt @@ -0,0 +1,13 @@ +package fr.shikkanime.ktor.session + +import io.ktor.util.AttributeKey + +/** + * Call attribute holding the [SessionScope] resolved for the request. + */ +internal val SessionScopeAttribute: AttributeKey = AttributeKey("SessionScope") + +/** + * Call attribute holding the [System.nanoTime] at which the request started. + */ +internal val SessionStartAttribute: AttributeKey = AttributeKey("SessionStart") diff --git a/ktor/src/main/kotlin/session/SessionClientPlugin.kt b/ktor/src/main/kotlin/session/SessionClientPlugin.kt new file mode 100644 index 0000000..7ac6fd8 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionClientPlugin.kt @@ -0,0 +1,64 @@ +package fr.shikkanime.ktor.session + +import io.ktor.client.* +import io.ktor.client.plugins.* +import java.time.Instant + +/** + * Forwards the current session to every outgoing HTTP call. + * + * The client replaces any session headers set by the caller with the identifiers of the current + * request, so a service cannot accidentally detach its outgoing calls from the session it serves. + * The call is recorded as an `OUT` event when a store is available. + */ +internal fun HttpClient.installSessionPropagation() { + plugin(HttpSend).intercept { request -> + val scope = SessionContext.current() ?: return@intercept execute(request) + val childRequestId = SessionIds.newRequestId() + request.headers[SessionHeaders.SESSION_ID] = scope.sessionId + request.headers[SessionHeaders.REQUEST_ID] = childRequestId + request.headers[SessionHeaders.PARENT_REQUEST_ID] = scope.requestId + val method = request.method.value + val path = request.url.encodedPathSegments + .filter { it.isNotEmpty() } + .joinToString("/", prefix = "/") + val target = request.url.host + val startedAt = System.nanoTime() + try { + val call = execute(request) + recordOutbound(scope, childRequestId, method, path, target, call.response.status.value, startedAt) + call + } catch (exception: Exception) { + recordOutbound(scope, childRequestId, method, path, target, null, startedAt, exception) + throw exception + } + } +} + +private fun recordOutbound( + scope: SessionScope, + requestId: String, + method: String, + path: String, + target: String, + status: Int?, + startedAt: Long, + error: Throwable? = null +) { + scope.store?.append( + SessionEvent( + timestamp = Instant.now().toString(), + sessionId = scope.sessionId, + requestId = requestId, + parentRequestId = scope.requestId, + service = scope.service, + kind = SessionEventKind.OUT, + method = method, + path = path, + target = target, + status = status, + durationMs = (System.nanoTime() - startedAt) / NANOS_PER_MILLI, + error = error?.toSessionError() + ) + ) +} diff --git a/ktor/src/main/kotlin/session/SessionConfig.kt b/ktor/src/main/kotlin/session/SessionConfig.kt new file mode 100644 index 0000000..37f8ac8 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionConfig.kt @@ -0,0 +1,61 @@ +package fr.shikkanime.ktor.session + +import java.nio.file.Path +import java.nio.file.Paths +import java.time.Duration + +private const val LOG_DIRECTORY_ENV = "SHIKKANIME_LOGS_DIR" +private const val SERVICE_ENV = "MODULE_NAME" +private const val SERVICE_ENV_FALLBACK = "SHIKKANIME_SERVICE" +private const val SESSION_TOKEN_ENV = "SHIKKANIME_SESSION_TOKEN" +private const val SESSIONS_DIRECTORY = "sessions" +private val DEFAULT_EXCLUDED_PATHS = setOf("/health", "/swagger", "/openapi", "/v1/session-traces") + +/** + * Configuration of the session tracking installed by `configureDefaultModules`. + * + * @property enabled whether requests are tracked at all. Disable it in tests or in an application + * that must not write a journal. + * @property directory directory holding the per-service session directories. When `null`, the + * framework uses `$SHIKKANIME_LOGS_DIR/sessions`, or `/logs/sessions`. + * @property service name of this service in the journal. When `null`, the framework reads + * `MODULE_NAME`, then `SHIKKANIME_SERVICE`, and falls back to `default`. + * @property retention how long a session file is kept after its last event. + * @property generateIfMissing whether a request without a session header starts a new session. + * When `false`, only requests carrying a valid header are tracked. + * @property store custom persistent store. When set, [directory] and [retention] are ignored and + * the caller owns the store lifecycle. + * @property apiToken bearer token protecting the read API. When `null` or blank, the API is not + * registered. Defaults to `SHIKKANIME_SESSION_TOKEN`. + * @property excludedPathPrefixes request paths ignored by tracking, compared by prefix. + */ +data class SessionConfig( + val enabled: Boolean = true, + val directory: Path? = null, + val service: String? = null, + val retention: Duration = Duration.ofDays(7), + val generateIfMissing: Boolean = true, + val store: SessionStore? = null, + val apiToken: String? = System.getenv(SESSION_TOKEN_ENV), + val excludedPathPrefixes: Set = DEFAULT_EXCLUDED_PATHS +) + +/** + * Resolves the directory receiving the per-service session directories. + * + * @return the configured directory, or the framework default. + */ +internal fun SessionConfig.resolvedDirectory(): Path { + directory?.let { return it } + val base = System.getenv(LOG_DIRECTORY_ENV)?.takeIf { it.isNotBlank() } + ?: Paths.get(System.getProperty("user.dir"), "logs").toString() + return Paths.get(base).resolve(SESSIONS_DIRECTORY) +} + +/** + * Resolves the service name written in every event. + * + * @return a safe service name, never blank. + */ +internal fun SessionConfig.resolvedService(): String = + SessionIds.service(service ?: System.getenv(SERVICE_ENV) ?: System.getenv(SERVICE_ENV_FALLBACK)) diff --git a/ktor/src/main/kotlin/session/SessionConstants.kt b/ktor/src/main/kotlin/session/SessionConstants.kt new file mode 100644 index 0000000..994cb45 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionConstants.kt @@ -0,0 +1,22 @@ +package fr.shikkanime.ktor.session + +import kotlinx.serialization.json.Json + +/** Nanoseconds in one millisecond. */ +internal const val NANOS_PER_MILLI = 1_000_000L + +/** Extension of a session journal file. */ +internal const val SESSION_FILE_EXTENSION = ".jsonl" + +/** Maximum length of an exception summary written in a journal event. */ +internal const val MAX_ERROR_LENGTH = 512 + +/** + * JSON format of one journal line: compact, without default or `null` properties, and tolerant to + * properties added by a newer framework version. + */ +internal val SESSION_JSON: Json = Json { + encodeDefaults = false + explicitNulls = false + ignoreUnknownKeys = true +} diff --git a/ktor/src/main/kotlin/session/SessionContext.kt b/ktor/src/main/kotlin/session/SessionContext.kt new file mode 100644 index 0000000..bda3577 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionContext.kt @@ -0,0 +1,87 @@ +package fr.shikkanime.ktor.session + +import java.time.Instant + +/** + * Holds the [SessionScope] of the request running on the current thread. + * + * Ktor installs the scope for the whole call pipeline; code running inside a controller, a service + * or a use case can read it, and the framework's HTTP client forwards it to outgoing calls. A + * coroutine keeps the scope across thread switches through [SessionContextElement]. + */ +object SessionContext { + private val holder = ThreadLocal() + + /** + * Returns the scope of the current request, or `null` outside a tracked request. + * + * @return the current scope, if any. + */ + fun current(): SessionScope? = + holder.get() + + /** + * Installs a scope on the current thread and returns the previous one. + * + * @param scope scope to install. + * @return the scope that was installed before, or `null`. + */ + internal fun install(scope: SessionScope): SessionScope? { + val previous = holder.get() + holder.set(scope) + return previous + } + + /** + * Restores the scope installed before [install]. + * + * @param previous scope returned by [install]. + */ + internal fun restore(previous: SessionScope?) { + if (previous == null) holder.remove() else holder.set(previous) + } + + /** + * Runs [block] with [scope] installed, restoring the previous scope afterwards. + * + * This is the entry point for a background job that calls other services and wants its calls + * correlated under one session. + * + * @param scope scope to install for the duration of [block]. + * @param block code to run. + * @param T result type of [block]. + * @return the result of [block]. + */ + fun with(scope: SessionScope, block: () -> T): T { + val previous = install(scope) + return try { + block() + } finally { + restore(previous) + } + } + + /** + * Appends an application-defined event to the journal of the current session. + * + * The call is a no-op outside a tracked request or when no store is configured. + * + * @param name event name, for example `cache.miss`. + * @param data event attributes. + */ + fun event(name: String, data: Map = emptyMap()) { + val scope = current() ?: return + scope.store?.append( + SessionEvent( + timestamp = Instant.now().toString(), + sessionId = scope.sessionId, + requestId = scope.requestId, + parentRequestId = scope.parentRequestId, + service = scope.service, + kind = SessionEventKind.EVENT, + name = name, + data = data + ) + ) + } +} diff --git a/ktor/src/main/kotlin/session/SessionContextElement.kt b/ktor/src/main/kotlin/session/SessionContextElement.kt new file mode 100644 index 0000000..f7f8b1e --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionContextElement.kt @@ -0,0 +1,25 @@ +package fr.shikkanime.ktor.session + +import kotlinx.coroutines.ThreadContextElement +import kotlin.coroutines.AbstractCoroutineContextElement +import kotlin.coroutines.CoroutineContext + +/** + * Carries the [SessionScope] across coroutine thread switches. + * + * The element sets the [SessionContext] thread-local before a coroutine resumes on a thread and + * restores the previous value after it suspends, which keeps the scope visible to code that runs + * on a different dispatcher. + */ +internal class SessionContextElement( + private val scope: SessionScope +) : ThreadContextElement, AbstractCoroutineContextElement(Key) { + /** Coroutine context key of the element. */ + companion object Key : CoroutineContext.Key + + override fun updateThreadContext(context: CoroutineContext): SessionScope? = + SessionContext.install(scope) + + override fun restoreThreadContext(context: CoroutineContext, oldState: SessionScope?) = + SessionContext.restore(oldState) +} diff --git a/ktor/src/main/kotlin/session/SessionError.kt b/ktor/src/main/kotlin/session/SessionError.kt new file mode 100644 index 0000000..e50ffcd --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionError.kt @@ -0,0 +1,15 @@ +package fr.shikkanime.ktor.session + +/** + * Formats a throwable as the bounded summary stored in a session event. + * + * The stack trace is deliberately omitted: the journal is a timeline, and a full trace belongs to + * the log files. + * + * @return exception type and message, truncated to [MAX_ERROR_LENGTH] characters. + */ +internal fun Throwable.toSessionError(): String { + val type = this::class.qualifiedName ?: this::class.simpleName ?: "Error" + val detail = this.message?.let { ": $it" } ?: "" + return (type + detail).take(MAX_ERROR_LENGTH) +} diff --git a/ktor/src/main/kotlin/session/SessionEvent.kt b/ktor/src/main/kotlin/session/SessionEvent.kt new file mode 100644 index 0000000..aa699ee --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionEvent.kt @@ -0,0 +1,41 @@ +package fr.shikkanime.ktor.session + +import kotlinx.serialization.Serializable + +/** + * One entry of a session journal, serialized as a single JSON line. + * + * @property timestamp event time, in ISO-8601 UTC. + * @property sessionId session this event belongs to. + * @property requestId identifier of the request that produced the event. + * @property parentRequestId identifier of the request that triggered an outgoing call, when any. + * @property service name of the service that wrote the event. + * @property kind nature of the event. + * @property method HTTP method, for request events. + * @property path request path without its query string. + * @property target host of an outgoing call. + * @property status HTTP status, when the event has one. + * @property durationMs handler duration in milliseconds. + * @property size response payload size in bytes, when known. + * @property error exception type and message, bounded by the store. + * @property name application-defined event name. + * @property data application-defined event attributes. + */ +@Serializable +data class SessionEvent( + val timestamp: String, + val sessionId: String, + val requestId: String, + val parentRequestId: String? = null, + val service: String, + val kind: SessionEventKind, + val method: String? = null, + val path: String? = null, + val target: String? = null, + val status: Int? = null, + val durationMs: Long? = null, + val size: Long? = null, + val error: String? = null, + val name: String? = null, + val data: Map = emptyMap() +) diff --git a/ktor/src/main/kotlin/session/SessionEventKind.kt b/ktor/src/main/kotlin/session/SessionEventKind.kt new file mode 100644 index 0000000..a786a9d --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionEventKind.kt @@ -0,0 +1,18 @@ +package fr.shikkanime.ktor.session + +import kotlinx.serialization.Serializable + +/** + * Nature of one entry in a session journal. + */ +@Serializable +enum class SessionEventKind { + /** A request received by this service. */ + IN, + + /** A request sent by this service to another service. */ + OUT, + + /** An application-defined event attached to the current session. */ + EVENT +} diff --git a/ktor/src/main/kotlin/session/SessionHeaders.kt b/ktor/src/main/kotlin/session/SessionHeaders.kt new file mode 100644 index 0000000..2ac42f6 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionHeaders.kt @@ -0,0 +1,18 @@ +package fr.shikkanime.ktor.session + +/** + * Names of the HTTP headers carrying the session correlation identifiers. + * + * The session identifier is a correlation token, never an authentication credential: a service + * must not authorize a request from it. + */ +object SessionHeaders { + /** Long-lived identifier of a client session. */ + const val SESSION_ID = "X-Session-Id" + + /** Identifier of one request, generated by the service receiving it. */ + const val REQUEST_ID = "X-Request-Id" + + /** Identifier of the request that triggered an outgoing call, when there is one. */ + const val PARENT_REQUEST_ID = "X-Parent-Request-Id" +} diff --git a/ktor/src/main/kotlin/session/SessionIds.kt b/ktor/src/main/kotlin/session/SessionIds.kt new file mode 100644 index 0000000..368aa8a --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionIds.kt @@ -0,0 +1,88 @@ +package fr.shikkanime.ktor.session + +import java.util.UUID +import java.util.concurrent.ThreadLocalRandom + +private const val MIN_ID_LENGTH = 8 +private const val MAX_ID_LENGTH = 64 +private const val MIN_SERVICE_LENGTH = 1 +private const val MAX_SERVICE_LENGTH = 32 +private const val DEFAULT_SERVICE = "default" +private const val VERSION_MASK = -0xF001L +private const val VERSION_4 = 0x0000000000004000L +private const val VARIANT_MASK = 0x3FFFFFFFFFFFFFFFL + +/** + * Validation and generation of the correlation identifiers. + * + * Identifiers become file names in [FileSessionStore], so only a strict, path-safe alphabet is + * accepted. A rejected value is replaced by a generated identifier instead of being refused: a + * malformed correlation header must not fail the business request. + */ +internal object SessionIds { + /** + * Tells whether [value] can be used as a session or request identifier. + * + * @param value candidate identifier. + * @return `true` when the value is 8 to 64 path-safe characters. + */ + fun isValid(value: String?): Boolean = + value != null && value.length in MIN_ID_LENGTH..MAX_ID_LENGTH && value.all(::isAllowed) + + /** + * Tells whether [value] can be used as a service name. + * + * @param value candidate service name. + * @return `true` when the value is 1 to 32 path-safe characters. + */ + fun isValidService(value: String?): Boolean = + value != null && value.length in MIN_SERVICE_LENGTH..MAX_SERVICE_LENGTH && value.all(::isAllowed) + + /** + * Returns [value] when it is a valid session identifier, or a new secure identifier. + * + * @param value identifier sent by the client. + * @return a valid session identifier. + */ + fun resolveSession(value: String?): String = + value?.takeIf(::isValid) ?: UUID.randomUUID().toString() + + /** + * Returns [value] when it is a valid request identifier, or a new fast identifier. + * + * Request identifiers need uniqueness, not unpredictability: the secure generator is reserved + * for session identifiers, which can be exposed to a client. + * + * @param value identifier sent by the client. + * @return a valid request identifier. + */ + fun resolveRequest(value: String?): String = + value?.takeIf(::isValid) ?: newRequestId() + + /** + * Returns [value] when it is a valid service name, or `default`. + * + * @param value candidate service name. + * @return a valid service name. + */ + fun service(value: String?): String = + value?.takeIf(::isValidService) ?: DEFAULT_SERVICE + + /** + * Generates a version 4 UUID without the contention of [UUID.randomUUID]. + * + * @return a new request identifier. + */ + fun newRequestId(): String { + val random = ThreadLocalRandom.current() + val most = random.nextLong() and VERSION_MASK or VERSION_4 + val least = random.nextLong() and VARIANT_MASK or Long.MIN_VALUE + return UUID(most, least).toString() + } + + private fun isAllowed(character: Char): Boolean = + when (character) { + in 'a'..'z', in 'A'..'Z', in '0'..'9', '-', '_' -> true + else -> false + } +} diff --git a/ktor/src/main/kotlin/session/SessionScope.kt b/ktor/src/main/kotlin/session/SessionScope.kt new file mode 100644 index 0000000..7645110 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionScope.kt @@ -0,0 +1,18 @@ +package fr.shikkanime.ktor.session + +/** + * Correlation identifiers of the request currently executing. + * + * @property sessionId long-lived session identifier. + * @property requestId identifier of the current request. + * @property parentRequestId identifier of the request that triggered this one, when any. + * @property service name of the service executing the request. + * @property store journal receiving the events, when tracking is enabled. + */ +data class SessionScope( + val sessionId: String, + val requestId: String, + val parentRequestId: String? = null, + val service: String, + val store: SessionStore? = null +) diff --git a/ktor/src/main/kotlin/session/SessionStore.kt b/ktor/src/main/kotlin/session/SessionStore.kt new file mode 100644 index 0000000..e819a28 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionStore.kt @@ -0,0 +1,49 @@ +package fr.shikkanime.ktor.session + +/** + * Persistent store of session events. + * + * Implementations must make [append] cheap and non-blocking: it is called from the request thread. + * The framework ships [FileSessionStore]; a project that needs a database implements this interface + * and passes the instance through [SessionConfig.store]. + */ +interface SessionStore { + /** + * Appends one event. + * + * @param event event to persist. + */ + fun append(event: SessionEvent) + + /** + * Reads the most recent events of one session. + * + * @param sessionId session to read. + * @param limit maximum number of events returned, from the end of the journal. + * @return the session timeline, or `null` when the session has no readable event. + */ + suspend fun find(sessionId: String, limit: Int = DEFAULT_EVENT_LIMIT): SessionTrace? + + /** + * Lists the most recently active sessions. + * + * @param limit maximum number of sessions returned. + * @return session summaries, most recent first. + */ + suspend fun sessions(limit: Int = DEFAULT_SESSION_LIMIT): List + + /** Flushes the events queued by [append]. */ + fun flush() + + /** Flushes and releases the resources of the store. */ + fun close() + + /** Defaults used by the read API. */ + companion object { + /** Default number of events returned by [find]. */ + const val DEFAULT_EVENT_LIMIT = 500 + + /** Default number of sessions returned by [sessions]. */ + const val DEFAULT_SESSION_LIMIT = 50 + } +} diff --git a/ktor/src/main/kotlin/session/SessionSummary.kt b/ktor/src/main/kotlin/session/SessionSummary.kt new file mode 100644 index 0000000..5c0e4a2 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionSummary.kt @@ -0,0 +1,21 @@ +package fr.shikkanime.ktor.session + +import kotlinx.serialization.Serializable + +/** + * Lightweight description of one session, used by the list API. + * + * @property sessionId identifier of the session. + * @property service service that produced the events. + * @property firstSeen timestamp of the oldest event, when the journal has a readable event. + * @property lastSeen timestamp of the newest event, when the journal has a readable event. + * @property sizeBytes size of the journal file in bytes. + */ +@Serializable +data class SessionSummary( + val sessionId: String, + val service: String, + val firstSeen: String?, + val lastSeen: String?, + val sizeBytes: Long +) diff --git a/ktor/src/main/kotlin/session/SessionTokenVerifier.kt b/ktor/src/main/kotlin/session/SessionTokenVerifier.kt new file mode 100644 index 0000000..62fd2d8 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionTokenVerifier.kt @@ -0,0 +1,24 @@ +package fr.shikkanime.ktor.session + +import java.nio.charset.StandardCharsets.UTF_8 +import java.security.MessageDigest + +private const val BEARER_PREFIX = "Bearer " + +/** + * Verifies the bearer token protecting the session read API. + */ +internal object SessionTokenVerifier { + /** + * Compares the configured token with an `Authorization` header in constant time. + * + * @param expected configured token. + * @param authorization value of the `Authorization` header, when present. + * @return `true` when the header carries the expected bearer token. + */ + fun matches(expected: String, authorization: String?): Boolean { + if (authorization == null || !authorization.startsWith(BEARER_PREFIX)) return false + val actual = authorization.substring(BEARER_PREFIX.length) + return MessageDigest.isEqual(expected.toByteArray(UTF_8), actual.toByteArray(UTF_8)) + } +} diff --git a/ktor/src/main/kotlin/session/SessionTrace.kt b/ktor/src/main/kotlin/session/SessionTrace.kt new file mode 100644 index 0000000..454e174 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionTrace.kt @@ -0,0 +1,25 @@ +package fr.shikkanime.ktor.session + +import kotlinx.serialization.Serializable + +/** + * Timeline of one session, as returned by the read API. + * + * @property sessionId identifier of the session. + * @property service service that produced the events. + * @property firstSeen timestamp of the oldest event. + * @property lastSeen timestamp of the newest event. + * @property eventCount number of events in the journal, including the ones outside [events]. + * @property truncated whether older events were omitted because of the requested limit. + * @property events most recent events, in chronological order. + */ +@Serializable +data class SessionTrace( + val sessionId: String, + val service: String, + val firstSeen: String, + val lastSeen: String, + val eventCount: Int, + val truncated: Boolean, + val events: List +) diff --git a/ktor/src/main/kotlin/session/SessionTraceController.kt b/ktor/src/main/kotlin/session/SessionTraceController.kt new file mode 100644 index 0000000..f22eb7c --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionTraceController.kt @@ -0,0 +1,117 @@ +package fr.shikkanime.ktor.session + +import fr.shikkanime.ktor.ApiResponse +import fr.shikkanime.ktor.ApiResponses +import fr.shikkanime.ktor.GetMapping +import fr.shikkanime.ktor.HttpHeader +import fr.shikkanime.ktor.IController +import fr.shikkanime.ktor.Operation +import fr.shikkanime.ktor.PathParam +import fr.shikkanime.ktor.QueryParam +import fr.shikkanime.ktor.ResponseEntity +import fr.shikkanime.ktor.RestController +import fr.shikkanime.ktor.dtos.ErrorMessageDto +import fr.shikkanime.ktor.dtos.MessageDto +import io.ktor.http.HttpStatusCode + +private const val MAX_EVENT_LIMIT = 1_000 +private const val MAX_SESSION_LIMIT = 200 + +/** + * Read API of the session journals. + * + * The controller is registered only when a token is configured; every endpoint requires + * `Authorization: Bearer `. + * + * @param store journal read by the endpoints. + * @param token bearer token accepted by the endpoints. + */ +@RestController("/v1/session-traces") +class SessionTraceController( + private val store: SessionStore, + private val token: String +) : IController { + /** + * Returns the timeline of one session. + * + * @param sessionId session to read. + * @param limit maximum number of events returned, from the end of the journal. + * @param authorization bearer token. + * @return the session timeline, or an error payload. + */ + @Suppress("unused") // Bound reflectively by ControllerBinder. + @GetMapping("/{sessionId}") + @Operation( + summary = "Read a session trace", + description = "Returns the most recent events recorded for one session" + ) + @ApiResponses([ + ApiResponse(status = 200, description = "Session found", responseType = SessionTrace::class), + ApiResponse(status = 400, description = "Invalid session id", responseType = ErrorMessageDto::class), + ApiResponse( + status = 401, + description = "Missing or invalid token", + responseType = ErrorMessageDto::class + ), + ApiResponse(status = 404, description = "Session not found", responseType = ErrorMessageDto::class) + ]) + suspend fun trace( + @PathParam("sessionId") sessionId: String, + @QueryParam("limit", required = false) limit: Int?, + @HttpHeader("Authorization", required = false) authorization: String? + ): ResponseEntity { + if (!SessionTokenVerifier.matches(token, authorization)) return unauthorized() + if (!SessionIds.isValid(sessionId)) return badRequest("Invalid session id") + val trace = store.find(sessionId, resolveEventLimit(limit)) ?: return notFound() + return ResponseEntity.ok(trace) + } + + /** + * Lists the most recently active sessions. + * + * @param limit maximum number of sessions returned. + * @param authorization bearer token. + * @return session summaries, or an error payload. + */ + @Suppress("unused") // Bound reflectively by ControllerBinder. + @GetMapping + @Operation( + summary = "List session traces", + description = "Returns the sessions with the most recent activity" + ) + @ApiResponses([ + ApiResponse( + status = 200, + description = "Sessions found", + responseType = List::class, + typeArguments = [SessionSummary::class] + ), + ApiResponse( + status = 401, + description = "Missing or invalid token", + responseType = ErrorMessageDto::class + ) + ]) + suspend fun sessions( + @QueryParam("limit", required = false) limit: Int?, + @HttpHeader("Authorization", required = false) authorization: String? + ): ResponseEntity { + if (!SessionTokenVerifier.matches(token, authorization)) return unauthorized() + return ResponseEntity.ok(store.sessions(resolveSessionLimit(limit))) + } + + private fun resolveEventLimit(limit: Int?): Int = + limit?.coerceIn(1, MAX_EVENT_LIMIT) ?: SessionStore.DEFAULT_EVENT_LIMIT + + private fun resolveSessionLimit(limit: Int?): Int = + limit?.coerceIn(1, MAX_SESSION_LIMIT) ?: SessionStore.DEFAULT_SESSION_LIMIT + + private fun unauthorized(): ResponseEntity = + ResponseEntity(MessageDto.error("Unauthorized"), HttpStatusCode.Unauthorized) + + private fun badRequest(message: String): ResponseEntity = + ResponseEntity(MessageDto.error(message), HttpStatusCode.BadRequest) + + private fun notFound(): ResponseEntity = + ResponseEntity(MessageDto.error("Session not found"), HttpStatusCode.NotFound) +} diff --git a/ktor/src/main/kotlin/session/SessionTracking.kt b/ktor/src/main/kotlin/session/SessionTracking.kt new file mode 100644 index 0000000..d7631ad --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionTracking.kt @@ -0,0 +1,101 @@ +package fr.shikkanime.ktor.session + +import fr.shikkanime.ktor.ErrorAttribute +import fr.shikkanime.ktor.ResponseSizeAttribute +import io.ktor.http.HttpMethod +import io.ktor.server.application.* +import io.ktor.server.application.hooks.ResponseBodyReadyForSend +import io.ktor.server.request.* +import kotlinx.coroutines.withContext +import java.time.Instant + +/** + * Installs the session tracking plugin on this application. + * + * @param config session tracking configuration. + * @param store journal receiving the events. + * @param closeStoreOnStop whether the store is owned by the framework and must be closed when the + * application stops. + */ +internal fun Application.installSessionTracking( + config: SessionConfig, + store: SessionStore, + closeStoreOnStop: Boolean +) { + install(sessionTrackingPlugin(config, store, closeStoreOnStop)) +} + +private fun sessionTrackingPlugin( + config: SessionConfig, + store: SessionStore, + closeStoreOnStop: Boolean +): ApplicationPlugin = + createApplicationPlugin(name = "SessionTracking") { + val service = config.resolvedService() + application.intercept(ApplicationCallPipeline.Setup) { + if (isExcluded(call.request, config)) { + proceed() + return@intercept + } + val scope = resolveScope(call, config, store, service) + if (scope == null) { + proceed() + return@intercept + } + call.attributes.put(SessionScopeAttribute, scope) + call.attributes.put(SessionStartAttribute, System.nanoTime()) + call.response.headers.append(SessionHeaders.SESSION_ID, scope.sessionId) + call.response.headers.append(SessionHeaders.REQUEST_ID, scope.requestId) + withContext(SessionContextElement(scope)) { proceed() } + } + on(ResponseBodyReadyForSend) { _, call, _ -> + recordInbound(call, store) + } + if (closeStoreOnStop) { + application.monitor.subscribe(ApplicationStopped) { store.close() } + } + } + +private fun isExcluded(request: ApplicationRequest, config: SessionConfig): Boolean { + if (request.httpMethod == HttpMethod.Options) return true + val path = request.path() + return config.excludedPathPrefixes.any { path.startsWith(it) } +} + +private fun resolveScope( + call: ApplicationCall, + config: SessionConfig, + store: SessionStore, + service: String +): SessionScope? { + val requestedSession = call.request.headers[SessionHeaders.SESSION_ID] + if (!config.generateIfMissing && !SessionIds.isValid(requestedSession)) return null + return SessionScope( + sessionId = SessionIds.resolveSession(requestedSession), + requestId = SessionIds.resolveRequest(call.request.headers[SessionHeaders.REQUEST_ID]), + parentRequestId = call.request.headers[SessionHeaders.PARENT_REQUEST_ID]?.takeIf(SessionIds::isValid), + service = service, + store = store + ) +} + +private fun recordInbound(call: ApplicationCall, store: SessionStore) { + val scope = call.attributes.getOrNull(SessionScopeAttribute) ?: return + val startedAt = call.attributes.getOrNull(SessionStartAttribute) ?: return + store.append( + SessionEvent( + timestamp = Instant.now().toString(), + sessionId = scope.sessionId, + requestId = scope.requestId, + parentRequestId = scope.parentRequestId, + service = scope.service, + kind = SessionEventKind.IN, + method = call.request.httpMethod.value, + path = call.request.path(), + status = call.response.status()?.value, + durationMs = (System.nanoTime() - startedAt) / NANOS_PER_MILLI, + size = call.attributes.getOrNull(ResponseSizeAttribute), + error = call.attributes.getOrNull(ErrorAttribute)?.toSessionError() + ) + ) +} diff --git a/ktor/src/main/kotlin/session/SessionWriteCommand.kt b/ktor/src/main/kotlin/session/SessionWriteCommand.kt new file mode 100644 index 0000000..da09983 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionWriteCommand.kt @@ -0,0 +1,17 @@ +package fr.shikkanime.ktor.session + +import java.util.concurrent.CountDownLatch + +/** + * Command processed by the [SessionWriter] thread. + */ +internal sealed interface SessionWriteCommand { + /** Append one event. */ + data class Event(val event: SessionEvent) : SessionWriteCommand + + /** Flush every open file, then release [latch]. */ + data class Flush(val latch: CountDownLatch) : SessionWriteCommand + + /** Flush, close every file, and stop the writer. */ + data object Close : SessionWriteCommand +} diff --git a/ktor/src/main/kotlin/session/SessionWriter.kt b/ktor/src/main/kotlin/session/SessionWriter.kt new file mode 100644 index 0000000..97feb96 --- /dev/null +++ b/ktor/src/main/kotlin/session/SessionWriter.kt @@ -0,0 +1,189 @@ +package fr.shikkanime.ktor.session + +import java.io.BufferedWriter +import java.io.IOException +import java.io.RandomAccessFile +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.StandardOpenOption.APPEND +import java.nio.file.StandardOpenOption.CREATE +import java.time.Clock +import java.time.Duration +import java.time.LocalDate +import java.util.concurrent.ArrayBlockingQueue +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicLong + +private const val MAX_OPEN_WRITERS = 64 +private const val CLOSE_TIMEOUT_MILLIS = 5_000L +private const val CLOSE_OFFER_ATTEMPTS = 5_000 +private const val THREAD_NAME = "shikkanime-session-writer" + +/** + * Single-threaded append writer for one service directory. + * + * The request thread only offers an event to a bounded queue; serialization and disk writes happen + * on the writer thread. A full queue drops the event instead of slowing the request down. The queue + * is flushed when it drains, so a burst is written in batches and an idle service flushes each + * event. + * + * @param directory service directory receiving the journal files. + * @param retention how long an inactive journal file is kept. + * @param queueCapacity maximum number of events waiting for the writer. + * @param clock clock used by the retention policy. + */ +internal class SessionWriter( + private val directory: Path, + private val retention: Duration, + queueCapacity: Int, + private val clock: Clock +) { + private val queue = ArrayBlockingQueue(queueCapacity) + private val writers = object : LinkedHashMap(MAX_OPEN_WRITERS, 0.75f, true) { + override fun removeEldestEntry(eldest: MutableMap.MutableEntry): Boolean { + if (size <= MAX_OPEN_WRITERS) return false + eldest.value.close() + return true + } + } + private val thread = Thread(::run, THREAD_NAME).apply { + isDaemon = true + // The writer is I/O-bound background work: it must yield the CPU to request threads. + priority = Thread.MIN_PRIORITY + } + private val droppedEvents = AtomicLong() + private var lastPurge: LocalDate? = null + + @Volatile private var accepting = true + + init { + thread.start() + } + + /** + * Offers one event to the writer. + * + * @param event event to write. + */ + fun append(event: SessionEvent) { + if (!accepting || !SessionIds.isValid(event.sessionId)) return + if (!queue.offer(SessionWriteCommand.Event(event))) droppedEvents.incrementAndGet() + } + + /** + * Returns the number of events dropped because the queue was full. + * + * @return the number of dropped events since the writer started. + */ + fun droppedEvents(): Long = + droppedEvents.get() + + /** Waits until every queued event is written and every open file is flushed. */ + fun flush() { + if (!accepting || !thread.isAlive) return + val latch = CountDownLatch(1) + if (!queue.offer(SessionWriteCommand.Flush(latch))) return + latch.await(CLOSE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS) + } + + /** Writes the queued events, closes every file and stops the writer thread. */ + fun close() { + if (!accepting) return + accepting = false + var attempts = 0 + while (!queue.offer(SessionWriteCommand.Close)) { + if (!thread.isAlive || attempts++ >= CLOSE_OFFER_ATTEMPTS) break + try { + Thread.sleep(1) + } catch (_: InterruptedException) { + Thread.currentThread().interrupt() + break + } + } + thread.join(CLOSE_TIMEOUT_MILLIS) + if (thread.isAlive) thread.interrupt() + } + + private fun run() { + try { + purgeIfNeeded() + while (true) { + when (val command = queue.take()) { + is SessionWriteCommand.Event -> { + write(command.event) + if (queue.isEmpty()) flushAll() + } + + is SessionWriteCommand.Flush -> { + flushAll() + command.latch.countDown() + } + + SessionWriteCommand.Close -> { + flushAll() + closeAll() + return + } + } + purgeIfNeeded() + } + } catch (_: InterruptedException) { + Thread.currentThread().interrupt() + } finally { + flushAll() + closeAll() + } + } + + private fun write(event: SessionEvent) { + try { + val writer = writers.getOrPut(event.sessionId) { open(event.sessionId) } + writer.write(SESSION_JSON.encodeToString(SessionEvent.serializer(), event)) + writer.newLine() + } catch (_: Exception) { + // A failed event must never stop the writer or fail the request that produced it. + } + } + + private fun open(sessionId: String): BufferedWriter { + val file = directory.resolve(sessionId + SESSION_FILE_EXTENSION) + val needsNewline = Files.exists(file) && Files.size(file) > 0 && !endsWithNewline(file) + val writer = Files.newBufferedWriter(file, UTF_8, CREATE, APPEND) + if (needsNewline) writer.newLine() + return writer + } + + private fun endsWithNewline(file: Path): Boolean = + RandomAccessFile(file.toFile(), "r").use { fileHandle -> + fileHandle.seek(fileHandle.length() - 1) + fileHandle.read() == '\n'.code + } + + private fun flushAll() { + writers.values.forEach { writer -> runCatching { writer.flush() } } + } + + private fun closeAll() { + writers.values.forEach { writer -> runCatching { writer.close() } } + writers.clear() + } + + private fun purgeIfNeeded() { + val today = LocalDate.now(clock) + if (lastPurge == today) return + lastPurge = today + val cutoff = clock.millis() - retention.toMillis() + try { + Files.list(directory).use { stream -> + stream.filter { Files.isRegularFile(it) } + .filter { it.fileName.toString().endsWith(SESSION_FILE_EXTENSION) } + .filter { Files.getLastModifiedTime(it).toMillis() < cutoff } + .forEach { file -> runCatching { Files.deleteIfExists(file) } } + } + } catch (_: IOException) { + // Retention is best-effort: it must never stop the writer. + } + } +} diff --git a/ktor/src/test/kotlin/ClientTest.kt b/ktor/src/test/kotlin/ClientTest.kt index 8c5036e..0a9d1ea 100644 --- a/ktor/src/test/kotlin/ClientTest.kt +++ b/ktor/src/test/kotlin/ClientTest.kt @@ -1,5 +1,6 @@ package fr.shikkanime.ktor +import fr.shikkanime.ktor.session.SessionConfig import io.ktor.client.plugins.HttpTimeout import io.ktor.client.plugins.contentnegotiation.ContentNegotiation import io.ktor.client.plugins.pluginOrNull @@ -27,6 +28,7 @@ class ClientTest { @Serializable data class TestData(val id: Int, val name: String, val active: Boolean = true) + @Suppress("unused") // Controller methods are bound reflectively by ControllerBinder. @RestController("/api/client-test") class ClientTestController : IController { @GetMapping @@ -64,7 +66,7 @@ class ClientTest { testApplication { // Given application { - configureDefaultModules() + configureDefaultModules(session = SessionConfig(enabled = false)) routing { ControllerBinder.register(this, listOf(ClientTestController())) } @@ -90,7 +92,7 @@ class ClientTest { testApplication { // Given application { - configureDefaultModules() + configureDefaultModules(session = SessionConfig(enabled = false)) routing { ControllerBinder.register(this, listOf(ClientTestController())) } diff --git a/ktor/src/test/kotlin/ContentNegotiationTest.kt b/ktor/src/test/kotlin/ContentNegotiationTest.kt index 9021ffc..7296f82 100644 --- a/ktor/src/test/kotlin/ContentNegotiationTest.kt +++ b/ktor/src/test/kotlin/ContentNegotiationTest.kt @@ -1,5 +1,6 @@ package fr.shikkanime.ktor +import fr.shikkanime.ktor.session.SessionConfig import io.ktor.client.request.* import io.ktor.client.statement.* import io.ktor.http.* @@ -39,6 +40,7 @@ class ContentNegotiationTest { val extra: String = "extra" ) + @Suppress("unused") // Controller methods are bound reflectively by ControllerBinder. @RestController("/api/test-negotiation") class NegotiationTestController : IController { @GetMapping @@ -168,7 +170,7 @@ class ContentNegotiationTest { testApplication { // Given application { - configureDefaultModules() + configureDefaultModules(session = SessionConfig(enabled = false)) routing { ControllerBinder.register(this, listOf(NegotiationTestController())) } diff --git a/ktor/src/test/kotlin/ControllerArgumentResolverTest.kt b/ktor/src/test/kotlin/ControllerArgumentResolverTest.kt index 7f8d848..cf833b6 100644 --- a/ktor/src/test/kotlin/ControllerArgumentResolverTest.kt +++ b/ktor/src/test/kotlin/ControllerArgumentResolverTest.kt @@ -14,7 +14,7 @@ import org.junit.jupiter.api.Test class ControllerArgumentResolverTest { - enum class TestStatus { ACTIVE, INACTIVE } + enum class TestStatus { ACTIVE } class SampleController { fun sampleEndpoint( @@ -175,7 +175,7 @@ class ControllerArgumentResolverTest { } @Test - suspend fun `should refuse an absent required header`() { + fun `should refuse an absent required header`() { // Given val call = mockk(relaxed = true) every { call.request.headers["If-None-Match"] } returns null diff --git a/ktor/src/test/kotlin/ControllerRegistrationTest.kt b/ktor/src/test/kotlin/ControllerRegistrationTest.kt index 4750e4e..4babcac 100644 --- a/ktor/src/test/kotlin/ControllerRegistrationTest.kt +++ b/ktor/src/test/kotlin/ControllerRegistrationTest.kt @@ -1,5 +1,6 @@ package fr.shikkanime.ktor +import fr.shikkanime.ktor.session.SessionConfig import fr.shikkanime.validator.NotBlank import fr.shikkanime.validator.exceptions.ObjectNotValidException import io.ktor.client.request.accept @@ -35,6 +36,7 @@ class ControllerRegistrationTest { val title: String ) + @Suppress("unused") // Controller methods are bound reflectively by ControllerBinder. @RestController("/api/books") class BookController : IController { @GetMapping("/{id}") @@ -100,12 +102,14 @@ class ControllerRegistrationTest { /** Not annotated with [RestController]: must not be registered. */ + @Suppress("unused") // Controller methods are bound reflectively by ControllerBinder. class PlainController { fun get(): ResponseEntity = ResponseEntity.ok("nope") } /** [RestController] with a method that has no mapping annotation: must not be bound. */ + @Suppress("unused") // Controller methods are bound reflectively by ControllerBinder. @RestController("/api/extra") class ExtraController : IController { fun unannotated(): String = @@ -114,7 +118,7 @@ class ControllerRegistrationTest { private fun ApplicationTestBuilder.registerControllers(controllers: List) { application { - configureDefaultModules() + configureDefaultModules(session = SessionConfig(enabled = false)) routing { ControllerBinder.register(this, controllers) } } } diff --git a/ktor/src/test/kotlin/HealthControllerTest.kt b/ktor/src/test/kotlin/HealthControllerTest.kt index 0ebaab6..f0d92e4 100644 --- a/ktor/src/test/kotlin/HealthControllerTest.kt +++ b/ktor/src/test/kotlin/HealthControllerTest.kt @@ -1,5 +1,6 @@ package fr.shikkanime.ktor +import fr.shikkanime.ktor.session.SessionConfig import io.ktor.client.request.get import io.ktor.client.statement.bodyAsText import io.ktor.http.HttpStatusCode @@ -13,7 +14,7 @@ import org.junit.jupiter.api.Test class HealthControllerTest { private fun ApplicationTestBuilder.configureApplication() { application { - configureDefaultModules() + configureDefaultModules(session = SessionConfig(enabled = false)) } } diff --git a/ktor/src/test/kotlin/OpenApiTest.kt b/ktor/src/test/kotlin/OpenApiTest.kt index 965e291..4621ae7 100644 --- a/ktor/src/test/kotlin/OpenApiTest.kt +++ b/ktor/src/test/kotlin/OpenApiTest.kt @@ -1,5 +1,6 @@ package fr.shikkanime.ktor +import fr.shikkanime.ktor.session.SessionConfig import io.ktor.client.request.get import io.ktor.client.statement.bodyAsText import io.ktor.http.HttpStatusCode @@ -17,6 +18,7 @@ class OpenApiTest { @Serializable data class MovieDto(val id: Int, val title: String) + @Suppress("unused") // Controller methods are bound reflectively by ControllerBinder. @RestController("/api/movies") class MovieController : IController { @@ -55,7 +57,7 @@ class OpenApiTest { testApplication { // Given application { - configureDefaultModules() + configureDefaultModules(session = SessionConfig(enabled = false)) routing { configureSwaggerRoute("Movie API", "1.0.0") ControllerBinder.register(this, listOf(MovieController())) diff --git a/ktor/src/test/kotlin/session/FileSessionStoreTest.kt b/ktor/src/test/kotlin/session/FileSessionStoreTest.kt new file mode 100644 index 0000000..6836cb5 --- /dev/null +++ b/ktor/src/test/kotlin/session/FileSessionStoreTest.kt @@ -0,0 +1,201 @@ +package fr.shikkanime.ktor.session + +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.encodeToString +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.attribute.FileTime +import java.time.Duration +import java.time.Instant +import java.time.temporal.ChronoUnit +import java.util.concurrent.Executors + +class FileSessionStoreTest { + + @TempDir + lateinit var tempDir: Path + + private fun event(sessionId: String, requestId: String = "request-1"): SessionEvent = + SessionEvent( + timestamp = Instant.now().toString(), + sessionId = sessionId, + requestId = requestId, + service = "api", + kind = SessionEventKind.IN, + method = "GET", + path = "/api/ok", + status = 200 + ) + + private fun store( + retention: Duration = Duration.ofDays(1), + queueCapacity: Int = 4_096 + ): FileSessionStore = + FileSessionStore(tempDir, "api", retention = retention, queueCapacity = queueCapacity) + + @Nested + @DisplayName("tests for the append and read cycle") + inner class ReadWriteTests { + + @Test + fun `should read the events written by a previous store instance`() = + runBlocking { + // Given + val first = store() + first.append(event("session-1")) + first.flush() + first.close() + + // When + val second = store() + second.append(event("session-1", requestId = "request-2")) + second.flush() + val trace = second.find("session-1") + second.close() + + // Then + assertNotNull(trace) + assertEquals(2, trace!!.eventCount) + assertEquals(listOf("request-1", "request-2"), trace.events.map { it.requestId }) + } + + @Test + fun `should return only the requested tail of a session`() = + runBlocking { + // Given + val store = store() + repeat(3) { index -> store.append(event("session-1", requestId = "request-$index")) } + store.flush() + + // When + val trace = store.find("session-1", 2) + store.close() + + // Then + assertNotNull(trace) + assertEquals(3, trace!!.eventCount) + assertTrue(trace.truncated) + assertEquals(listOf("request-1", "request-2"), trace.events.map { it.requestId }) + } + + @Test + fun `should skip a truncated line after a restart`() = + runBlocking { + // Given + val directory = tempDir.resolve("api") + Files.createDirectories(directory) + val file = directory.resolve("session-1.jsonl") + Files.writeString(file, """{"timestamp":"2026-01-01T00:00:00Z","sessionId":"session-1""") + + // When + val store = store() + store.append(event("session-1")) + store.flush() + val trace = store.find("session-1") + store.close() + + // Then + assertNotNull(trace) + assertEquals(1, trace!!.eventCount) + } + + @Test + fun `should ignore a session id that is not path safe`() { + // Given + val store = store() + + // When + store.append(event("../escape")) + store.flush() + store.close() + + // Then + assertFalse(Files.exists(tempDir.resolve("escape.jsonl"))) + assertFalse(Files.list(tempDir.resolve("api")).use { it.findAny().isPresent }) + } + } + + @Nested + @DisplayName("tests for concurrent writers") + inner class ConcurrencyTests { + + @Test + fun `should write every event offered by concurrent requests`() = + runBlocking { + // Given + val store = store(queueCapacity = 4_096) + val executor = Executors.newFixedThreadPool(8) + + // When + val futures = (1..200).map { index -> + executor.submit { + store.append(event("session-1", requestId = "request-$index")) + } + } + futures.forEach { it.get() } + executor.shutdown() + store.flush() + val trace = store.find("session-1", 500) + store.close() + + // Then + assertNotNull(trace) + assertEquals(200, trace!!.eventCount) + assertEquals(0, store.droppedEvents) + } + } + + @Nested + @DisplayName("tests for retention and listings") + inner class MaintenanceTests { + + @Test + fun `should delete a journal older than the retention`() { + // Given + val directory = tempDir.resolve("api") + Files.createDirectories(directory) + val oldFile = directory.resolve("oldsession1.jsonl") + Files.writeString(oldFile, SESSION_JSON.encodeToString(event("oldsession1")) + "\n") + Files.setLastModifiedTime(oldFile, FileTime.from(Instant.now().minus(30, ChronoUnit.DAYS))) + + // When + val store = store(Duration.ofDays(1)) + store.append(event("newsession1")) + store.flush() + store.close() + + // Then + assertFalse(Files.exists(oldFile)) + } + + @Test + fun `should list the most recently modified sessions first`() = + runBlocking { + // Given + val directory = tempDir.resolve("api") + Files.createDirectories(directory) + val first = directory.resolve("sessionone1.jsonl") + val second = directory.resolve("sessiontwo2.jsonl") + Files.writeString(first, SESSION_JSON.encodeToString(event("sessionone1")) + "\n") + Files.writeString(second, SESSION_JSON.encodeToString(event("sessiontwo2")) + "\n") + Files.setLastModifiedTime(first, FileTime.from(Instant.now().minus(2, ChronoUnit.HOURS))) + Files.setLastModifiedTime(second, FileTime.from(Instant.now().minus(1, ChronoUnit.HOURS))) + + // When + val store = store() + val sessions = store.sessions(2) + store.close() + + // Then + assertEquals(listOf("sessiontwo2", "sessionone1"), sessions.map { it.sessionId }) + } + } +} diff --git a/ktor/src/test/kotlin/session/SessionClientPluginTest.kt b/ktor/src/test/kotlin/session/SessionClientPluginTest.kt new file mode 100644 index 0000000..2b49ca0 --- /dev/null +++ b/ktor/src/test/kotlin/session/SessionClientPluginTest.kt @@ -0,0 +1,96 @@ +package fr.shikkanime.ktor.session + +import io.ktor.client.HttpClient +import io.ktor.client.engine.mock.MockEngine +import io.ktor.client.engine.mock.respond +import io.ktor.client.request.get +import io.ktor.http.Headers +import io.ktor.http.HttpHeaders +import io.ktor.http.HttpStatusCode +import io.ktor.http.headersOf +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.util.concurrent.CopyOnWriteArrayList +import java.util.concurrent.atomic.AtomicReference + +class SessionClientPluginTest { + + private class RecordingStore : SessionStore { + val events = CopyOnWriteArrayList() + + override fun append(event: SessionEvent) { + events.add(event) + } + + override suspend fun find(sessionId: String, limit: Int): SessionTrace? = + null + + override suspend fun sessions(limit: Int): List = + emptyList() + + override fun flush() = Unit + + override fun close() = Unit + } + + @Test + fun `should forward the session and record the outgoing call`() = + runBlocking { + // Given + val store = RecordingStore() + val capturedHeaders = AtomicReference() + val engine = MockEngine { request -> + capturedHeaders.set(request.headers) + respond("ok", HttpStatusCode.OK, headersOf(HttpHeaders.ContentType, "text/plain")) + } + val client = HttpClient(engine) + client.installSessionPropagation() + val scope = SessionScope("session-1", "request-1", service = "test", store = store) + + // When + withContext(SessionContextElement(scope)) { + client.get("http://catalog.test/v1/items/42") + } + client.close() + + // Then + val headers = capturedHeaders.get() + assertEquals("session-1", headers[SessionHeaders.SESSION_ID]) + assertEquals("request-1", headers[SessionHeaders.PARENT_REQUEST_ID]) + val requestId = headers[SessionHeaders.REQUEST_ID] + assertTrue(SessionIds.isValid(requestId)) + val event = store.events.single() + assertEquals(SessionEventKind.OUT, event.kind) + assertEquals("session-1", event.sessionId) + assertEquals("request-1", event.parentRequestId) + assertEquals(requestId, event.requestId) + assertEquals("GET", event.method) + assertEquals("/v1/items/42", event.path) + assertEquals("catalog.test", event.target) + assertEquals(200, event.status) + } + + @Test + fun `should leave an untracked call untouched`() = + runBlocking { + // Given + val capturedHeaders = AtomicReference() + val engine = MockEngine { request -> + capturedHeaders.set(request.headers) + respond("ok", HttpStatusCode.OK) + } + val client = HttpClient(engine) + client.installSessionPropagation() + + // When + client.get("http://catalog.test/v1/items/42") + client.close() + + // Then + assertNull(capturedHeaders.get()[SessionHeaders.SESSION_ID]) + } +} diff --git a/ktor/src/test/kotlin/session/SessionContextTest.kt b/ktor/src/test/kotlin/session/SessionContextTest.kt new file mode 100644 index 0000000..ff3a7f9 --- /dev/null +++ b/ktor/src/test/kotlin/session/SessionContextTest.kt @@ -0,0 +1,102 @@ +package fr.shikkanime.ktor.session + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test +import java.util.concurrent.CopyOnWriteArrayList + +class SessionContextTest { + + private class RecordingStore : SessionStore { + val events = CopyOnWriteArrayList() + + override fun append(event: SessionEvent) { + events.add(event) + } + + override suspend fun find(sessionId: String, limit: Int): SessionTrace? = + null + + override suspend fun sessions(limit: Int): List = + emptyList() + + override fun flush() = Unit + + override fun close() = Unit + } + + private fun scope(store: SessionStore? = null): SessionScope = + SessionScope("session-1", "request-1", service = "test", store = store) + + @Nested + @DisplayName("tests for the thread-local scope") + inner class ThreadLocalTests { + + @Test + fun `should start without a scope`() { + // Given / When / Then + assertNull(SessionContext.current()) + } + + @Test + fun `should restore the previous scope after a nested block`() { + // Given + val outer = scope() + val inner = outer.copy(requestId = "request-2") + + // When / Then + SessionContext.with(outer) { + assertSame(outer, SessionContext.current()) + SessionContext.with(inner) { + assertSame(inner, SessionContext.current()) + } + assertSame(outer, SessionContext.current()) + } + assertNull(SessionContext.current()) + } + + @Test + fun `should append an event to the current session`() { + // Given + val store = RecordingStore() + + // When + SessionContext.with(scope(store)) { + SessionContext.event("cache.miss", mapOf("key" to "anime-1")) + } + + // Then + val event = store.events.single() + assertEquals(SessionEventKind.EVENT, event.kind) + assertEquals("cache.miss", event.name) + assertEquals("anime-1", event.data["key"]) + assertEquals("session-1", event.sessionId) + } + } + + @Nested + @DisplayName("tests for coroutine propagation") + inner class CoroutineTests { + + @Test + fun `should keep the scope across a dispatcher switch`() = + runBlocking { + // Given + val scope = scope() + + // When / Then + withContext(SessionContextElement(scope)) { + withContext(Dispatchers.IO) { + assertSame(scope, SessionContext.current()) + } + } + assertNull(SessionContext.current()) + } + } +} diff --git a/ktor/src/test/kotlin/session/SessionIdsTest.kt b/ktor/src/test/kotlin/session/SessionIdsTest.kt new file mode 100644 index 0000000..12f8ab5 --- /dev/null +++ b/ktor/src/test/kotlin/session/SessionIdsTest.kt @@ -0,0 +1,95 @@ +package fr.shikkanime.ktor.session + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNotEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test +import java.util.UUID + +class SessionIdsTest { + + @Nested + @DisplayName("tests for identifier validation") + inner class ValidationTests { + + @Test + fun `should accept a UUID and a short path-safe identifier`() { + // Given / When / Then + assertTrue(SessionIds.isValid(UUID.randomUUID().toString())) + assertTrue(SessionIds.isValid("session-1")) + assertTrue(SessionIds.isValid("a".repeat(8))) + assertTrue(SessionIds.isValid("a".repeat(64))) + } + + @Test + fun `should reject short, long, and path-unsafe identifiers`() { + // Given / When / Then + assertFalse(SessionIds.isValid("short")) + assertFalse(SessionIds.isValid("a".repeat(65))) + assertFalse(SessionIds.isValid("../../etc/passwd")) + assertFalse(SessionIds.isValid("session.id")) + assertFalse(SessionIds.isValid("session 1")) + assertFalse(SessionIds.isValid(null)) + } + } + + @Nested + @DisplayName("tests for identifier generation") + inner class GenerationTests { + + @Test + fun `should keep a valid session identifier`() { + // Given + val value = "session-1" + + // When + val resolved = SessionIds.resolveSession(value) + + // Then + assertEquals(value, resolved) + } + + @Test + fun `should replace an unsafe session identifier`() { + // Given / When + val resolved = SessionIds.resolveSession("../escape") + + // Then + assertNotEquals("../escape", resolved) + assertTrue(SessionIds.isValid(resolved)) + } + + @Test + fun `should generate a version 4 request identifier`() { + // Given / When + val identifier = UUID.fromString(SessionIds.newRequestId()) + + // Then + assertEquals(4, identifier.version()) + assertEquals(2, identifier.variant()) + } + } + + @Nested + @DisplayName("tests for service names") + inner class ServiceTests { + + @Test + fun `should keep a simple service name`() { + // Given / When / Then + assertEquals("api", SessionIds.service("api")) + assertEquals("api_v1", SessionIds.service("api_v1")) + } + + @Test + fun `should fall back to default for an unsafe service name`() { + // Given / When / Then + assertEquals("default", SessionIds.service("api/v1")) + assertEquals("default", SessionIds.service("a".repeat(33))) + assertEquals("default", SessionIds.service(null)) + } + } +} diff --git a/ktor/src/test/kotlin/session/SessionTrackingTest.kt b/ktor/src/test/kotlin/session/SessionTrackingTest.kt new file mode 100644 index 0000000..cc0aad7 --- /dev/null +++ b/ktor/src/test/kotlin/session/SessionTrackingTest.kt @@ -0,0 +1,287 @@ +package fr.shikkanime.ktor.session + +import fr.shikkanime.ktor.ControllerBinder +import fr.shikkanime.ktor.GetMapping +import fr.shikkanime.ktor.IController +import fr.shikkanime.ktor.ResponseEntity +import fr.shikkanime.ktor.RestController +import fr.shikkanime.ktor.configureDefaultModules +import io.ktor.client.request.get +import io.ktor.client.request.header +import io.ktor.client.request.options +import io.ktor.client.statement.bodyAsText +import io.ktor.http.HttpHeaders +import io.ktor.http.HttpStatusCode +import io.ktor.server.routing.routing +import io.ktor.server.testing.ApplicationTestBuilder +import io.ktor.server.testing.testApplication +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Nested +import org.junit.jupiter.api.Test +import java.util.concurrent.CopyOnWriteArrayList + +class SessionTrackingTest { + + class RecordingStore : SessionStore { + val events = CopyOnWriteArrayList() + + override fun append(event: SessionEvent) { + events.add(event) + } + + override suspend fun find(sessionId: String, limit: Int): SessionTrace? { + val matching = events.filter { it.sessionId == sessionId } + if (matching.isEmpty()) return null + val selected = matching.takeLast(limit) + return SessionTrace( + sessionId = sessionId, + service = "test", + firstSeen = matching.first().timestamp, + lastSeen = matching.last().timestamp, + eventCount = matching.size, + truncated = selected.size < matching.size, + events = selected + ) + } + + override suspend fun sessions(limit: Int): List = + events.groupBy { it.sessionId }.entries.take(limit).map { (sessionId, events) -> + SessionSummary( + sessionId = sessionId, + service = "test", + firstSeen = events.first().timestamp, + lastSeen = events.last().timestamp, + sizeBytes = events.size.toLong() + ) + } + + override fun flush() = Unit + + override fun close() = Unit + } + + @Suppress("unused") // Controller methods are bound reflectively by ControllerBinder. + @RestController("/api") + class TrackingController : IController { + @GetMapping("/ok") + fun ok(): ResponseEntity = + ResponseEntity.ok("ok") + + @GetMapping("/fail") + fun fail(): String = + throw IllegalStateException("boom") + + @GetMapping("/session") + fun session(): ResponseEntity = + ResponseEntity.ok(SessionContext.current()?.sessionId ?: "none") + } + + private fun ApplicationTestBuilder.configure( + store: SessionStore, + config: SessionConfig = SessionConfig(store = store, service = "test", apiToken = "secret") + ) { + application { + configureDefaultModules(session = config) + routing { ControllerBinder.register(this, listOf(TrackingController())) } + } + } + + @Nested + @DisplayName("tests for inbound correlation") + inner class InboundTests { + + @Test + fun `should reuse the session and request identifiers of the client`() = + testApplication { + // Given + val store = RecordingStore() + configure(store) + + // When + val response = client.get("/api/ok") { + header(SessionHeaders.SESSION_ID, "session-1") + header(SessionHeaders.REQUEST_ID, "request-1") + } + + // Then + assertEquals("session-1", response.headers[SessionHeaders.SESSION_ID]) + assertEquals("request-1", response.headers[SessionHeaders.REQUEST_ID]) + val event = store.events.single() + assertEquals(SessionEventKind.IN, event.kind) + assertEquals("session-1", event.sessionId) + assertEquals("request-1", event.requestId) + assertEquals("test", event.service) + assertEquals("GET", event.method) + assertEquals("/api/ok", event.path) + assertEquals(200, event.status) + assertNotNull(event.durationMs) + assertNotNull(event.size) + } + + @Test + fun `should generate identifiers when the client sends none`() = + testApplication { + // Given + val store = RecordingStore() + configure(store) + + // When + val response = client.get("/api/ok") + + // Then + val sessionId = response.headers[SessionHeaders.SESSION_ID] + val requestId = response.headers[SessionHeaders.REQUEST_ID] + assertNotNull(sessionId) + assertNotNull(requestId) + assertTrue(SessionIds.isValid(sessionId)) + assertTrue(SessionIds.isValid(requestId)) + assertEquals(sessionId, store.events.single().sessionId) + } + + @Test + fun `should replace an unsafe session identifier`() = + testApplication { + // Given + val store = RecordingStore() + configure(store) + + // When + val response = client.get("/api/ok") { + header(SessionHeaders.SESSION_ID, "../../etc/passwd") + } + + // Then + val sessionId = response.headers[SessionHeaders.SESSION_ID] + assertNotNull(sessionId) + assertTrue(SessionIds.isValid(sessionId)) + assertNotEquals("../../etc/passwd", sessionId) + assertEquals(sessionId, store.events.single().sessionId) + } + + @Test + fun `should expose the scope inside the controller`() = + testApplication { + // Given + val store = RecordingStore() + configure(store) + + // When + val response = client.get("/api/session") { + header(SessionHeaders.SESSION_ID, "session-1") + } + + // Then + assertEquals("session-1", response.bodyAsText().trim('"')) + } + + @Test + fun `should record the exception of a failing controller`() = + testApplication { + // Given + val store = RecordingStore() + configure(store) + + // When + val response = client.get("/api/fail") { + header(SessionHeaders.SESSION_ID, "session-1") + } + + // Then + assertEquals(HttpStatusCode.InternalServerError, response.status) + val event = store.events.single() + assertEquals(500, event.status) + assertNotNull(event.error) + assertTrue(event.error!!.contains("IllegalStateException")) + assertTrue(event.error!!.contains("boom")) + } + } + + @Nested + @DisplayName("tests for excluded requests") + inner class ExclusionTests { + + @Test + fun `should ignore health and preflight requests`() = + testApplication { + // Given + val store = RecordingStore() + configure(store) + + // When + assertEquals(HttpStatusCode.OK, client.get("/health").status) + client.options("/api/ok") + + // Then + assertTrue(store.events.isEmpty()) + } + + @Test + fun `should not track a request without a session when generation is disabled`() = + testApplication { + // Given + val store = RecordingStore() + configure( + store, + SessionConfig(store = store, service = "test", generateIfMissing = false) + ) + + // When + val response = client.get("/api/ok") + + // Then + assertNull(response.headers[SessionHeaders.SESSION_ID]) + assertTrue(store.events.isEmpty()) + } + } + + @Nested + @DisplayName("tests for the session read API") + inner class ReadApiTests { + + @Test + fun `should protect and serve the session read API`() = + testApplication { + // Given + val store = RecordingStore() + configure(store) + client.get("/api/ok") { header(SessionHeaders.SESSION_ID, "session-1") } + + // When / Then + assertEquals( + HttpStatusCode.Unauthorized, + client.get("/v1/session-traces/session-1").status + ) + + val trace = client.get("/v1/session-traces/session-1") { + header(HttpHeaders.Authorization, "Bearer secret") + } + assertEquals(HttpStatusCode.OK, trace.status) + assertTrue(trace.bodyAsText().contains(""""sessionId":"session-1"""")) + + val sessions = client.get("/v1/session-traces") { + header(HttpHeaders.Authorization, "Bearer secret") + } + assertEquals(HttpStatusCode.OK, sessions.status) + assertTrue(sessions.bodyAsText().contains("session-1")) + + assertEquals( + HttpStatusCode.NotFound, + client.get("/v1/session-traces/unknown-session") { + header(HttpHeaders.Authorization, "Bearer secret") + }.status + ) + assertEquals( + HttpStatusCode.BadRequest, + client.get("/v1/session-traces/bad!id") { + header(HttpHeaders.Authorization, "Bearer secret") + }.status + ) + assertEquals(1, store.events.size) + } + } +}