ChargePoint assessment.

2024-03-04 13:13:01 +01:00 · 2024-03-04 13:13:01 +01:00 · bd2f0a79a8
commit bd2f0a79a8
37 changed files with 1664 additions and 0 deletions
--- a/.gitignore
+++ b/.gitignore
@ -0,0 +1,40 @@
+HELP.md
+.gradle
+build/
+!gradle/wrapper/gradle-wrapper.jar
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+bin/
+!**/src/main/**/bin/
+!**/src/test/**/bin/
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+out/
+!**/src/main/**/out/
+!**/src/test/**/out/
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+
+### VS Code ###
+.vscode/
+
+### Kotlin ###
+.kotlin
--- a/7
+++ b/7
@ -0,0 +1,7 @@
+FROM gcr.io/distroless/java17:latest
+
+COPY ./build/libs/chargepoint-0.0.1-SNAPSHOT.jar  /usr/app/chargepoint.jar
+
+EXPOSE 8080
+
+ENTRYPOINT ["java","-jar","/usr/app/chargepoint.jar"]
--- a/README.md
+++ b/README.md
@ -0,0 +1,125 @@
+# Transaction Auth Service Assignment
+
+This is the basic readme for the ChargePoint assessment. The
+application is written in Spring Boot with Kotlin, with the build
+managed by Gradle-Kotlin, and the application can be deployed as a
+Docker container. It is very simple, with the code for both the web
+API and the authentication service being stored in a single source
+tree. In a real environment, I assume they would be in separate source
+trees, if not separate repos.
+
+Spring profiles are used to separate the configurations, so that the
+application can run as a separate web API and auth service backend
+(though they can also run in the same process). All communication
+between the web socket API and the "auth service" are done over Kafka,
+even if everything is running in the same server process.
+
+## Building / Running
+
+To run the application in a docker compose setup:
+
+```
+./gradlew build -x test
+docker compose up --build
+```
+
+This will start Kafka (and Zookeeper), the "authentication service,"
+and the websocket API. The test UI is available on
+http://localhost:8080/
+
+## Using
+
+To use the test UI, press the Connect button to activate the
+websocket. Then a station UUID and driver identifier can be filled in
+and sent to the service via the websocket connection.
+ - Any station UUID can be given, as long as it's a UUID.
+ - There are two identifiers in the hardcoded whitelist by default.
+   They are listed on the page itself.
+ - As this is just a demo application, the UI is extremely bare-bones
+   and nowhere close to what would be called "production ready" (and
+   also not designed for a charging station, obviously).
+ - Browser console logs connectivity status and incoming messages (as
+   well as outputs them to the table).
+
+## Running Tests
+
+```
+./gradlew test
+```
+
+This will run all the tests, both unit tests and integration tests.
+
+
+## Notes
+
+Notes and thoughts on development of the application:
+ - I think it would make sense to return more information in the
+   response, given the fully async communication. Like driver and
+   station IDs.
+   - By only returning the status, some information is lost, and it
+     can make it hard to determine which response belongs to which
+     request.
+   - This could be alleviated by passing a request ID or something
+     along (and it seems that's the case in the diagram in the PDF).
+ - It would be useful to return more detailed error messages in
+   responses instead of just "Invalid."
+
+Design:
+ - I used Arrow for typed error handling in the validation. The Kotlin
+   `Result` type is good, but has some limitations that make it
+   impractical for more specific error handling in many cases.
+ - Arrow has a very similar dev experience to ScalaZ for validation
+   (notably, `zipAndAccumulate`).
+ - I find typed error handling to be extremely useful, and probably
+   one of the best things in functional programming. Kotlin inherits
+   this slightly, and Arrow rounds it out.
+ - Web socket in-memory broker is used because apparently there is no
+   Kafka-backed WebSocket broker for Spring? Or maybe there is now,
+   and my info is out of date.
+ - Weird issue: When writing the integration test for the web socket
+   API, using a Jackson message converter would cause the responses to
+   be lost in the aether and the response to never be received.
+   Switching to GSON message converter made it work immediately.
+
+Development:
+ - Development started by running Kafka locally in a basic
+   docker-compose file (which later became the final docker-compose).
+ - Connection was initially basic HTTP request-response, then quickly
+   replaced with a functioning websocket connection and a stupidly
+   simple UI that I found on the internet as a test harness.
+ - Eventually, I split the application configuration into three
+   separate Spring profiles to allow running everything in one
+   process, or separating the web API and auth consumer into multiple
+   processes.
+ - Kafka communication started with a basic regular Kafka template,
+   but I quickly switched to the `ReplyingKafkaTemplate` to get fully
+   async request-reply semantics.
+ - The docker-compose runs two instances of the application: one for
+   the websocket API, and another for the Kafka-backed "auth service"
+   with the hardcoded whitelist.
+   - In a real environment, the auth service is almost assuredly a
+     completely separate application either in another repo, or in a
+     separate source folder.
+   - Even when running in 1 process, the application will communicate
+     over Kafka, and with in-memory methods.
+
+Software choices:
+ - I think I would prefer to use Kotlin coroutines instead of
+   `CompletableFuture` for the async communication, but since the
+   Kafka lib for Spring is built on `CompletableFuture`, it was easier
+   to make use of that API for this simple project.
+ - I think Arrow is a good choice for validation. It doesn't
+   necessarily need to be applied across the whole codebase,
+   especially because its more powerful uses are somewhat esoteric.
+ - Scaling: With the way the service is built, any number of instances
+   of the web API should be able to sit behind a load balancer, and
+   any number of auth service consumers should be able to connect to
+   Kafka.
+ - Having a specific annotation on a method for creating a Kafka
+   consumer listener is easy, but I feel like the consumer method
+   could get lost in a larger codebase. Maybe it's good (assuming it's
+   possible) to wire up a consumer directly in Spring config?
+   - I prefer explicit wiring over Spring's classpath scanning,
+     especially for services with complexity.
+   - Classpath scanning is easy, but too much magic can make it hard
+     to tell what's going on and why beans are defined or not.
--- a/build.gradle.kts
+++ b/build.gradle.kts
@ -0,0 +1,49 @@
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
+
+plugins {
+	id("org.springframework.boot") version "3.2.3"
+	id("io.spring.dependency-management") version "1.1.4"
+	kotlin("jvm") version "1.9.22"
+	kotlin("plugin.spring") version "1.9.22"
+}
+
+group = "net.thermetics"
+version = "0.0.1-SNAPSHOT"
+
+java {
+	sourceCompatibility = JavaVersion.VERSION_17
+}
+
+repositories {
+	mavenCentral()
+}
+
+dependencies {
+	val mockkVersion = "1.13.10"
+	val testcontainersVersion = "1.19.7"
+
+	// Main deps
+	implementation("org.springframework.boot:spring-boot-starter-web")
+	implementation("org.springframework.boot:spring-boot-starter-websocket")
+	implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
+	implementation("org.jetbrains.kotlin:kotlin-reflect")
+	implementation("org.springframework.kafka:spring-kafka")
+	implementation("io.arrow-kt:arrow-core:1.2.1")
+
+	// Tests
+	testImplementation("org.springframework.boot:spring-boot-starter-test")
+	testImplementation("org.springframework.kafka:spring-kafka-test")
+	testImplementation("com.google.code.gson:gson")
+	testImplementation("io.mockk:mockk:${mockkVersion}")
+}
+
+tasks.withType<KotlinCompile> {
+	kotlinOptions {
+		freeCompilerArgs += "-Xjsr305=strict"
+		jvmTarget = "17"
+	}
+}
+
+tasks.withType<Test> {
+	useJUnitPlatform()
+}
--- a/docker-compose.yml
+++ b/docker-compose.yml
@ -0,0 +1,57 @@
+version: '2'
+services:
+  api:
+    build: .
+    container_name: chargepoint_transaction_api
+    environment:
+      - SPRING_KAFKA_BOOTSTRAP_SERVERS=kafka:29092
+      - SPRING_PROFILES_ACTIVE=web-api,kafka-auth-client
+    depends_on:
+      - kafka
+      - auth
+    ports:
+      - "8080:8080"
+    networks:
+      - chargepoint
+
+  auth:
+    build: .
+    container_name: chargepoint_transaction_auth
+    environment:
+      - SPRING_KAFKA_BOOTSTRAP_SERVERS=kafka:29092
+      - SPRING_PROFILES_ACTIVE=kafka-auth-consumer
+    depends_on:
+      - kafka
+    networks:
+      - chargepoint
+
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.4.4
+    hostname: zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - 22181:2181
+    networks:
+      - chargepoint
+
+  kafka:
+    image: confluentinc/cp-kafka:7.4.4
+    hostname: kafka
+    depends_on:
+      - zookeeper
+    ports:
+      - 29092:29092
+    networks:
+      - chargepoint
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://kafka:29092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+networks:
+  chargepoint:
+    driver: bridge
--- a/gradle/wrapper/gradle-wrapper.jar
+++ b/gradle/wrapper/gradle-wrapper.jar
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@ -0,0 +1,7 @@
+distributionBase=GRADLE_USER_HOME
+distributionPath=wrapper/dists
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
+networkTimeout=10000
+validateDistributionUrl=true
+zipStoreBase=GRADLE_USER_HOME
+zipStorePath=wrapper/dists
--- a/249
+++ b/249
@ -0,0 +1,249 @@
+#!/bin/sh
+
+#
+# Copyright © 2015-2021 the original authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+##############################################################################
+#
+#   Gradle start up script for POSIX generated by Gradle.
+#
+#   Important for running:
+#
+#   (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is
+#       noncompliant, but you have some other compliant shell such as ksh or
+#       bash, then to run this script, type that shell name before the whole
+#       command line, like:
+#
+#           ksh Gradle
+#
+#       Busybox and similar reduced shells will NOT work, because this script
+#       requires all of these POSIX shell features:
+#         * functions;
+#         * expansions «$var», «${var}», «${var:-default}», «${var+SET}»,
+#           «${var#prefix}», «${var%suffix}», and «$( cmd )»;
+#         * compound commands having a testable exit status, especially «case»;
+#         * various built-in commands including «command», «set», and «ulimit».
+#
+#   Important for patching:
+#
+#   (2) This script targets any POSIX shell, so it avoids extensions provided
+#       by Bash, Ksh, etc; in particular arrays are avoided.
+#
+#       The "traditional" practice of packing multiple parameters into a
+#       space-separated string is a well documented source of bugs and security
+#       problems, so this is (mostly) avoided, by progressively accumulating
+#       options in "$@", and eventually passing that to Java.
+#
+#       Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS,
+#       and GRADLE_OPTS) rely on word-splitting, this is performed explicitly;
+#       see the in-line comments for details.
+#
+#       There are tweaks for specific operating systems such as AIX, CygWin,
+#       Darwin, MinGW, and NonStop.
+#
+#   (3) This script is generated from the Groovy template
+#       https://github.com/gradle/gradle/blob/HEAD/subprojects/plugins/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt
+#       within the Gradle project.
+#
+#       You can find Gradle at https://github.com/gradle/gradle/.
+#
+##############################################################################
+
+# Attempt to set APP_HOME
+
+# Resolve links: $0 may be a link
+app_path=$0
+
+# Need this for daisy-chained symlinks.
+while
+    APP_HOME=${app_path%"${app_path##*/}"}  # leaves a trailing /; empty if no leading path
+    [ -h "$app_path" ]
+do
+    ls=$( ls -ld "$app_path" )
+    link=${ls#*' -> '}
+    case $link in             #(
+      /*)   app_path=$link ;; #(
+      *)    app_path=$APP_HOME$link ;;
+    esac
+done
+
+# This is normally unused
+# shellcheck disable=SC2034
+APP_BASE_NAME=${0##*/}
+# Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036)
+APP_HOME=$( cd "${APP_HOME:-./}" > /dev/null && pwd -P ) || exit
+
+# Use the maximum available, or set MAX_FD != -1 to use that value.
+MAX_FD=maximum
+
+warn () {
+    echo "$*"
+} >&2
+
+die () {
+    echo
+    echo "$*"
+    echo
+    exit 1
+} >&2
+
+# OS specific support (must be 'true' or 'false').
+cygwin=false
+msys=false
+darwin=false
+nonstop=false
+case "$( uname )" in                #(
+  CYGWIN* )         cygwin=true  ;; #(
+  Darwin* )         darwin=true  ;; #(
+  MSYS* | MINGW* )  msys=true    ;; #(
+  NONSTOP* )        nonstop=true ;;
+esac
+
+CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+
+
+# Determine the Java command to use to start the JVM.
+if [ -n "$JAVA_HOME" ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+        # IBM's JDK on AIX uses strange locations for the executables
+        JAVACMD=$JAVA_HOME/jre/sh/java
+    else
+        JAVACMD=$JAVA_HOME/bin/java
+    fi
+    if [ ! -x "$JAVACMD" ] ; then
+        die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+    fi
+else
+    JAVACMD=java
+    if ! command -v java >/dev/null 2>&1
+    then
+        die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+
+Please set the JAVA_HOME variable in your environment to match the
+location of your Java installation."
+    fi
+fi
+
+# Increase the maximum file descriptors if we can.
+if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then
+    case $MAX_FD in #(
+      max*)
+        # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked.
+        # shellcheck disable=SC2039,SC3045
+        MAX_FD=$( ulimit -H -n ) ||
+            warn "Could not query maximum file descriptor limit"
+    esac
+    case $MAX_FD in  #(
+      '' | soft) :;; #(
+      *)
+        # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked.
+        # shellcheck disable=SC2039,SC3045
+        ulimit -n "$MAX_FD" ||
+            warn "Could not set maximum file descriptor limit to $MAX_FD"
+    esac
+fi
+
+# Collect all arguments for the java command, stacking in reverse order:
+#   * args from the command line
+#   * the main class name
+#   * -classpath
+#   * -D...appname settings
+#   * --module-path (only if needed)
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables.
+
+# For Cygwin or MSYS, switch paths to Windows format before running java
+if "$cygwin" || "$msys" ; then
+    APP_HOME=$( cygpath --path --mixed "$APP_HOME" )
+    CLASSPATH=$( cygpath --path --mixed "$CLASSPATH" )
+
+    JAVACMD=$( cygpath --unix "$JAVACMD" )
+
+    # Now convert the arguments - kludge to limit ourselves to /bin/sh
+    for arg do
+        if
+            case $arg in                                #(
+              -*)   false ;;                            # don't mess with options #(
+              /?*)  t=${arg#/} t=/${t%%/*}              # looks like a POSIX filepath
+                    [ -e "$t" ] ;;                      #(
+              *)    false ;;
+            esac
+        then
+            arg=$( cygpath --path --ignore --mixed "$arg" )
+        fi
+        # Roll the args list around exactly as many times as the number of
+        # args, so each arg winds up back in the position where it started, but
+        # possibly modified.
+        #
+        # NB: a `for` loop captures its iteration list before it begins, so
+        # changing the positional parameters here affects neither the number of
+        # iterations, nor the values presented in `arg`.
+        shift                   # remove old arg
+        set -- "$@" "$arg"      # push replacement arg
+    done
+fi
+
+
+# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
+
+# Collect all arguments for the java command:
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
+#     and any embedded shellness will be escaped.
+#   * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be
+#     treated as '${Hostname}' itself on the command line.
+
+set -- \
+        "-Dorg.gradle.appname=$APP_BASE_NAME" \
+        -classpath "$CLASSPATH" \
+        org.gradle.wrapper.GradleWrapperMain \
+        "$@"
+
+# Stop when "xargs" is not available.
+if ! command -v xargs >/dev/null 2>&1
+then
+    die "xargs is not available"
+fi
+
+# Use "xargs" to parse quoted args.
+#
+# With -n1 it outputs one arg per line, with the quotes and backslashes removed.
+#
+# In Bash we could simply go:
+#
+#   readarray ARGS < <( xargs -n1 <<<"$var" ) &&
+#   set -- "${ARGS[@]}" "$@"
+#
+# but POSIX shell has neither arrays nor command substitution, so instead we
+# post-process each arg (as a line of input to sed) to backslash-escape any
+# character that might be a shell metacharacter, then use eval to reverse
+# that process (while maintaining the separation between arguments), and wrap
+# the whole thing up as a single "set" statement.
+#
+# This will of course break if any of these variables contains a newline or
+# an unmatched quote.
+#
+
+eval "set -- $(
+        printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" |
+        xargs -n1 |
+        sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' |
+        tr '\n' ' '
+    )" '"$@"'
+
+exec "$JAVACMD" "$@"
--- a/gradlew.bat
+++ b/gradlew.bat
@ -0,0 +1,92 @@
+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem      https://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+
+@if "%DEBUG%"=="" @echo off
+@rem ##########################################################################
+@rem
+@rem  Gradle startup script for Windows
+@rem
+@rem ##########################################################################
+
+@rem Set local scope for the variables with windows NT shell
+if "%OS%"=="Windows_NT" setlocal
+
+set DIRNAME=%~dp0
+if "%DIRNAME%"=="" set DIRNAME=.
+@rem This is normally unused
+set APP_BASE_NAME=%~n0
+set APP_HOME=%DIRNAME%
+
+@rem Resolve any "." and ".." in APP_HOME to make it shorter.
+for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi
+
+@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
+
+@rem Find java.exe
+if defined JAVA_HOME goto findJavaFromJavaHome
+
+set JAVA_EXE=java.exe
+%JAVA_EXE% -version >NUL 2>&1
+if %ERRORLEVEL% equ 0 goto execute
+
+echo.
+echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH.
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:findJavaFromJavaHome
+set JAVA_HOME=%JAVA_HOME:"=%
+set JAVA_EXE=%JAVA_HOME%/bin/java.exe
+
+if exist "%JAVA_EXE%" goto execute
+
+echo.
+echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME%
+echo.
+echo Please set the JAVA_HOME variable in your environment to match the
+echo location of your Java installation.
+
+goto fail
+
+:execute
+@rem Setup the command line
+
+set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
+
+
+@rem Execute Gradle
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %*
+
+:end
+@rem End local scope for the variables with windows NT shell
+if %ERRORLEVEL% equ 0 goto mainEnd
+
+:fail
+rem Set variable GRADLE_EXIT_CONSOLE if you need the _script_ return code instead of
+rem the _cmd.exe /c_ return code!
+set EXIT_CODE=%ERRORLEVEL%
+if %EXIT_CODE% equ 0 set EXIT_CODE=1
+if not ""=="%GRADLE_EXIT_CONSOLE%" exit %EXIT_CODE%
+exit /b %EXIT_CODE%
+
+:mainEnd
+if "%OS%"=="Windows_NT" endlocal
+
+:omega
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@ -0,0 +1 @@
+rootProject.name = "chargepoint"
--- a/src/main/kotlin/net/thermetics/chargepoint/ChargepointAssignment.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/ChargepointAssignment.kt
@ -0,0 +1,11 @@
+package net.thermetics.chargepoint
+
+import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.runApplication
+
+@SpringBootApplication
+class ChargepointAssignment
+
+fun main(args: Array<String>) {
+	runApplication<ChargepointAssignment>(*args)
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/authorization/KafkaAuthorizationConsumer.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/authorization/KafkaAuthorizationConsumer.kt
@ -0,0 +1,31 @@
+package net.thermetics.chargepoint.authorization
+
+import net.thermetics.chargepoint.config.KAFKA_AUTH_REPLIES
+import net.thermetics.chargepoint.config.KAFKA_AUTH_REQUESTS
+import net.thermetics.chargepoint.models.AuthStatus
+import net.thermetics.chargepoint.models.KafkaAuthRequest
+import net.thermetics.chargepoint.models.KafkaAuthResponse
+import org.springframework.kafka.annotation.KafkaListener
+import org.springframework.messaging.handler.annotation.SendTo
+
+
+/**
+ * The [KafkaAuthorizationConsumer] would be a separate application in the real system,
+ * connected to the web frontend API over Kafka. In this demonstration application, it runs
+ * in the same process as the web API by default, but all communication is done via Kafka,
+ * and it thus can run as a separate application as well.
+ */
+class KafkaAuthorizationConsumer {
+    @KafkaListener(id = "transaction-auth-consumer", topics = [KAFKA_AUTH_REQUESTS])
+    @SendTo(KAFKA_AUTH_REPLIES)
+    fun authenticate(message: KafkaAuthRequest): KafkaAuthResponse {
+        println("message received over kafka: $message")
+        println("Sending back to ${message.identifier} over kafka reply template")
+
+        return when (Whitelist.AllowedCards[message.identifier]) {
+            true -> KafkaAuthResponse(AuthStatus.Accepted)
+            false -> KafkaAuthResponse(AuthStatus.Rejected)
+            null -> KafkaAuthResponse(AuthStatus.Unknown)
+        }
+    }
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/authorization/Whitelist.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/authorization/Whitelist.kt
@ -0,0 +1,8 @@
+package net.thermetics.chargepoint.authorization
+
+object Whitelist {
+    val AllowedCards = mapOf(
+        "accepted-driver1-with-short-identifier" to true,
+        "rejected-driver2-with-a-longer-identifier-with-more-characters-yes" to false
+    )
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/config/KafkaAuthConsumerConfig.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/config/KafkaAuthConsumerConfig.kt
@ -0,0 +1,16 @@
+package net.thermetics.chargepoint.config
+
+import net.thermetics.chargepoint.authorization.KafkaAuthorizationConsumer
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.Profile
+
+@Configuration
+@Profile("kafka-auth-consumer")
+class KafkaAuthConsumerConfig {
+    // In a real-world scenario, this would be defined in its own separate application. This
+    // is the authorization application/consumer/service that actually checks if a driver is
+    // allowed to charge.
+    @Bean
+    fun authorizationApplication() = KafkaAuthorizationConsumer()
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/config/KafkaConfig.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/config/KafkaConfig.kt
@ -0,0 +1,80 @@
+package net.thermetics.chargepoint.config
+
+import net.thermetics.chargepoint.authorization.KafkaAuthorizationConsumer
+import net.thermetics.chargepoint.services.KafkaAuthService
+import net.thermetics.chargepoint.models.KafkaAuthRequest
+import net.thermetics.chargepoint.models.KafkaAuthResponse
+import org.apache.kafka.clients.admin.NewTopic
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.Profile
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory
+import org.springframework.kafka.config.TopicBuilder
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.core.ProducerFactory
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer
+import org.springframework.kafka.requestreply.ReplyingKafkaTemplate
+import org.springframework.messaging.simp.SimpMessagingTemplate
+
+
+const val KAFKA_AUTH_REQUESTS = "kafka-auth-requests"
+const val KAFKA_AUTH_REPLIES = "kafka-auth-replies"
+
+/**
+ * A fully async request-reply workflow is used to communicate with the authorization
+ * application over Kafka. This requires defining a custom [ReplyingKafkaTemplate] to be
+ * used in the [KafkaAuthService]. Spring Boot auto-config is leveraged to define most of the
+ * required Kafka beans.
+ */
+@Configuration
+@Profile("kafka-auth-client")
+class KafkaConfig {
+    // Communicates with the authorization application over Kafka, from the web API.
+    @Bean
+    fun authService(
+        template: ReplyingKafkaTemplate<String, KafkaAuthRequest, KafkaAuthResponse>
+    ) = KafkaAuthService(template)
+
+    @Bean
+    fun replyingTemplate(
+        pf: ProducerFactory<String, KafkaAuthRequest>,
+        repliesContainer: ConcurrentMessageListenerContainer<String, KafkaAuthResponse>
+    ): ReplyingKafkaTemplate<String, KafkaAuthRequest, KafkaAuthResponse> =
+        ReplyingKafkaTemplate(pf, repliesContainer).also {
+            it.defaultTopic = KAFKA_AUTH_REQUESTS // Needed if we decide to use Spring Message<T> abstraction.
+        }
+
+    // This is required for the replying template (i.e. Spring will yell at you if you do not define this).
+    @Bean
+    fun baseTemplate(pf: ProducerFactory<String, KafkaAuthResponse>): KafkaTemplate<String, KafkaAuthResponse> {
+        return KafkaTemplate(pf)
+    }
+
+    @Bean
+    fun repliesContainer(
+        containerFactory: ConcurrentKafkaListenerContainerFactory<String, KafkaAuthResponse>,
+        replyTemplate: KafkaTemplate<String, KafkaAuthResponse>
+    ): ConcurrentMessageListenerContainer<String, KafkaAuthResponse> {
+        containerFactory.setReplyTemplate(replyTemplate)
+        val repliesContainer =
+            containerFactory.createContainer(KAFKA_AUTH_REPLIES)
+        repliesContainer.containerProperties.setGroupId("repliesGroup")
+        repliesContainer.isAutoStartup = false
+        return repliesContainer
+    }
+
+    @Bean
+    fun requestTopic(): NewTopic =
+        TopicBuilder.name(KAFKA_AUTH_REQUESTS)
+            .partitions(10)
+            .replicas(2)
+            .build()
+
+
+    @Bean
+    fun replyTopic(): NewTopic =
+        TopicBuilder.name(KAFKA_AUTH_REPLIES)
+            .partitions(10)
+            .replicas(2)
+            .build()
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/config/WebSocketConfig.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/config/WebSocketConfig.kt
@ -0,0 +1,47 @@
+package net.thermetics.chargepoint.config
+
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.context.annotation.Profile
+import org.springframework.messaging.simp.config.MessageBrokerRegistry
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer
+
+
+/**
+ * Configuration for the setup of the web socket connection. Individual controllers
+ * are auto-wired with the Spring Controller annotation. This wires up a simple in-memory
+ * websocket connection broker. Individual users are treated as "anonymous" sessions, so that
+ * we are not broadcasting results of a transaction auth request to everyone who might be
+ * connected to the websocket.
+ */
+@Configuration
+@Profile("web-api")
+@EnableWebSocketMessageBroker
+class WebSocketConfig(
+    @Value("\${chargepoint.websocket.topic-prefix}") val wsTopicPrefix: String,
+    @Value("\${chargepoint.websocket.auth-topic}") val wsAuthTopic: String,
+    @Value("\${chargepoint.websocket.app-prefix}") val appPrefix: String,
+    @Value("\${chargepoint.websocket.user-prefix}") val userPrefix: String
+) : WebSocketMessageBrokerConfigurer {
+    // While this is technically the only string bean in the config,
+    // I think it makes maintenance sense to define primitive-type beans
+    // with a name.
+    @Bean("websocketTopic")
+    fun websocketTopic(): String =
+        "${wsTopicPrefix}/${wsAuthTopic}"
+
+    override fun registerStompEndpoints(registry: StompEndpointRegistry) {
+        registry.addEndpoint("/ws")
+    }
+
+    override  fun configureMessageBroker(
+        registry: MessageBrokerRegistry
+    ) {
+        registry.setApplicationDestinationPrefixes(appPrefix)
+        registry.setUserDestinationPrefix(userPrefix) // per-session websocket endpoint
+        registry.enableSimpleBroker(wsTopicPrefix)
+    }
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/controllers/AuthController.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/controllers/AuthController.kt
@ -0,0 +1,50 @@
+package net.thermetics.chargepoint
+
+import net.thermetics.chargepoint.models.*
+import net.thermetics.chargepoint.services.AuthService
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.context.annotation.Profile
+import org.springframework.messaging.handler.annotation.MessageMapping
+import org.springframework.messaging.handler.annotation.Payload
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor
+import org.springframework.messaging.simp.SimpMessageType
+import org.springframework.messaging.simp.SimpMessagingTemplate
+import org.springframework.stereotype.Controller
+
+@Controller
+@Profile("web-api")
+class TransactionController(
+    val authService: AuthService,
+    val socket: SimpMessagingTemplate,
+    @Autowired @Qualifier("websocketTopic") val wsAuthTopic: String
+) {
+    fun createHeaders(sessionId: String) =
+        SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE).also {
+            it.sessionId = sessionId
+            it.setLeaveMutable(true)
+        }.messageHeaders
+
+    fun sendAuthStatus(header: SimpMessageHeaderAccessor, reply: TransactionAuthResponse) {
+        println("Sending response ${reply} back to session ${header.sessionId} on topic $wsAuthTopic")
+        socket.convertAndSendToUser(
+            header.sessionId!!,
+            wsAuthTopic,
+            reply,
+            createHeaders(header.sessionId!!)
+        )
+    }
+
+    fun authorize(req: TransactionAuthRequest, header: SimpMessageHeaderAccessor) {
+        authService.authorizeTransaction(req).thenApply { authResp ->
+            sendAuthStatus(header, authResp)
+        }
+    }
+
+    @MessageMapping("/transaction/authorize")
+    fun authorizeEndpoint(@Payload req: TransactionAuthRequest, header: SimpMessageHeaderAccessor) {
+        println("Message incoming from ${header.sessionId}")
+        authorize(req, header)
+    }
+
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/models/AuthStatus.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/models/AuthStatus.kt
@ -0,0 +1,5 @@
+package net.thermetics.chargepoint.models
+
+enum class AuthStatus {
+    Accepted, Unknown, Invalid, Rejected
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/models/DriverIdentifier.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/models/DriverIdentifier.kt
@ -0,0 +1,3 @@
+package net.thermetics.chargepoint.models
+
+data class DriverIdentifier(val id: String)
--- a/src/main/kotlin/net/thermetics/chargepoint/models/KafkaAuthRequest.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/models/KafkaAuthRequest.kt
@ -0,0 +1,10 @@
+package net.thermetics.chargepoint.models
+
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor
+
+data class KafkaAuthRequest(val stationUuid: String, val identifier: String) {
+    constructor(req: TransactionAuthRequest) : this(
+        stationUuid = req.stationUuid,
+        identifier = req.driverIdentifier.id
+    )
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/models/KafkaAuthResponse.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/models/KafkaAuthResponse.kt
@ -0,0 +1,3 @@
+package net.thermetics.chargepoint.models
+
+data class KafkaAuthResponse(val status: AuthStatus)
--- a/src/main/kotlin/net/thermetics/chargepoint/models/TransactionAuthRequest.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/models/TransactionAuthRequest.kt
@ -0,0 +1,3 @@
+package net.thermetics.chargepoint.models
+
+data class TransactionAuthRequest(val stationUuid: String, val driverIdentifier: DriverIdentifier)
--- a/src/main/kotlin/net/thermetics/chargepoint/models/TransactionAuthResponse.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/models/TransactionAuthResponse.kt
@ -0,0 +1,6 @@
+package net.thermetics.chargepoint.models
+
+data class TransactionAuthResponse(val authorizationStatus: String) {
+    constructor(kafkaAuthResponse: KafkaAuthResponse) : this(authorizationStatus = kafkaAuthResponse.status.name)
+    constructor(authStatus: AuthStatus) : this(authorizationStatus = authStatus.name)
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/models/Validation.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/models/Validation.kt
@ -0,0 +1,36 @@
+package net.thermetics.chargepoint.models
+
+import arrow.core.Either
+import arrow.core.EitherNel
+import arrow.core.raise.either
+import arrow.core.raise.ensure
+import java.util.*
+
+typealias ValidationResult = EitherNel<ValidationError, TransactionAuthRequest>
+
+sealed class ValidationError(override val message: String) : Error(message) {
+    data object InvalidStationUuid : ValidationError("Invalid station ID")
+    data object IdentifierTooShort : ValidationError("Invalid driver identifier - too short")
+    data object IdentifierTooLong : ValidationError("Invalid driver identifier - too long")
+}
+
+fun validateStationUuid(stationUuid: String) =
+    Either.catch { UUID.fromString(stationUuid) }
+        .mapLeft { ValidationError.InvalidStationUuid }
+
+fun validateIdentifierNotShort(driver: DriverIdentifier) = either {
+    ensure(driver.id.length >= 20) { ValidationError.IdentifierTooShort }
+    driver
+}
+
+fun validateIdentifierNotLong(driver: DriverIdentifier) = either {
+    ensure(driver.id.length <= 80) { ValidationError.IdentifierTooLong }
+    driver
+}
+
+fun validateRequest(req: TransactionAuthRequest): EitherNel<ValidationError, TransactionAuthRequest> =
+    Either.zipOrAccumulate(
+        validateStationUuid(req.stationUuid),
+        validateIdentifierNotShort(req.driverIdentifier),
+        validateIdentifierNotLong(req.driverIdentifier),
+    ) { _, _, _ -> req }
--- a/src/main/kotlin/net/thermetics/chargepoint/services/AuthService.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/services/AuthService.kt
@ -0,0 +1,9 @@
+package net.thermetics.chargepoint.services
+
+import net.thermetics.chargepoint.models.TransactionAuthRequest
+import net.thermetics.chargepoint.models.TransactionAuthResponse
+import java.util.concurrent.CompletableFuture
+
+interface AuthService {
+    fun authorizeTransaction(req: TransactionAuthRequest): CompletableFuture<TransactionAuthResponse>
+}
--- a/src/main/kotlin/net/thermetics/chargepoint/services/KafkaAuthService.kt
+++ b/src/main/kotlin/net/thermetics/chargepoint/services/KafkaAuthService.kt
@ -0,0 +1,36 @@
+package net.thermetics.chargepoint.services
+
+import net.thermetics.chargepoint.config.KAFKA_AUTH_REQUESTS
+import net.thermetics.chargepoint.models.*
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.springframework.kafka.requestreply.ReplyingKafkaTemplate
+import java.util.concurrent.CompletableFuture
+
+typealias ReplyingAuthTemplate = ReplyingKafkaTemplate<String, KafkaAuthRequest, KafkaAuthResponse>
+typealias AuthProducerRecord = ProducerRecord<String, KafkaAuthRequest>
+
+/**
+ * The [KafkaAuthService] communicates with the authentication service application over
+ * Kafka. Java futures are used due to the Kafka API and the integration with Spring + WebSockets.
+ */
+class KafkaAuthService(
+    val kafka: ReplyingAuthTemplate,
+) : AuthService {
+    private fun validateAsync(req: TransactionAuthRequest): CompletableFuture<ValidationResult> =
+        CompletableFuture.supplyAsync { validateRequest(req) }
+
+    private fun sendToKafka(req: TransactionAuthRequest): CompletableFuture<TransactionAuthResponse> =
+        AuthProducerRecord(KAFKA_AUTH_REQUESTS, KafkaAuthRequest(req)).let {
+            kafka.sendAndReceive(it).thenApply { resp ->
+                TransactionAuthResponse(resp.value())
+            }
+        }
+
+    override fun authorizeTransaction(req: TransactionAuthRequest): CompletableFuture<TransactionAuthResponse> =
+        validateAsync(req).thenCompose { res ->
+            res.fold(
+                ifRight = { sendToKafka(it) },
+                ifLeft = { CompletableFuture.supplyAsync { TransactionAuthResponse(AuthStatus.Invalid) } },
+            )
+        }
+}
--- a/src/main/resources/application.properties
+++ b/src/main/resources/application.properties
@ -0,0 +1,27 @@
+# Top-level settings that control what parts of the application actually run.
+# Can be used to run web API separately from the Kafka consumer.
+# Consumer can be run alone with:
+#     spring.main.web-application-type=none
+#     spring.profiles.active=kafka-auth-consumer
+# If this is done, the web API should be started with:
+#     spring.profiles.active=kafka-auth-client,web-api
+
+# This will run the Kafka "client"/producer, the consumer,
+# and the web API all at once.
+spring.profiles.active=web-api,kafka-auth-consumer,kafka-auth-client
+
+# Application-related
+# Do not put quotes around these!
+chargepoint.websocket.topic-prefix=/topic
+chargepoint.websocket.app-prefix=/app
+chargepoint.websocket.user-prefix=/anonymous-session
+chargepoint.websocket.auth-topic=auth
+
+# Kafka-related
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.bootstrap-servers=localhost:29092
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.properties[spring.json.trusted.packages]=java.util,java.lang,net.thermetics.*
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
--- a/src/main/resources/static/app.js
+++ b/src/main/resources/static/app.js
@ -0,0 +1,66 @@
+const stompClient = new StompJs.Client({
+    brokerURL: 'ws://localhost:8080/ws'
+});
+
+stompClient.onConnect = (frame) => {
+    setConnected(true);
+    console.log('Connected: ' + frame);
+    stompClient.subscribe('/anonymous-session/topic/auth', (response) => {
+        console.log(response);
+        addResponse(JSON.parse(response.body).authorizationStatus);
+    });
+};
+
+stompClient.onWebSocketError = (error) => {
+    console.error('Error with websocket', error);
+};
+
+stompClient.onStompError = (frame) => {
+    console.error('Broker reported error: ' + frame.headers['message']);
+    console.error('Additional details: ' + frame.body);
+};
+
+function setConnected(connected) {
+    $("#connect").prop("disabled", connected);
+    $("#disconnect").prop("disabled", !connected);
+    if (connected) {
+        $("#conversation").show();
+    }
+    else {
+        $("#conversation").hide();
+    }
+    $("#greetings").html("");
+}
+
+function connect() {
+    stompClient.activate();
+}
+
+function disconnect() {
+    stompClient.deactivate();
+    setConnected(false);
+    console.log("Disconnected");
+}
+
+function sendRequest() {
+    const stationUuid = $('#station-uuid').val()
+    const driverIdentifier = $('#driver-id').val()
+    stompClient.publish({
+        destination: "/app/transaction/authorize",
+        body: JSON.stringify({
+                'stationUuid': stationUuid,
+                'driverIdentifier': { 'id': driverIdentifier }
+              })
+    });
+}
+
+function addResponse(status) {
+    $("#responses").append("<tr><td>" + status + "</td></tr>");
+}
+
+$(function () {
+    $("form").on('submit', (e) => false);
+    $( "#connect" ).click(() => connect());
+    $( "#disconnect" ).click(() => disconnect());
+    $( "#send" ).click(() => sendRequest());
+});
--- a/src/main/resources/static/index.html
+++ b/src/main/resources/static/index.html
@ -0,0 +1,74 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <title>Transaction Auth</title>
+    <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css"
+          integrity="sha384-BVYiiSIFeK1dGmJRAkycuHAHRg32OmUcww7on3RYdg4Va+PmSTsz/K68vbdEjh4u" crossorigin="anonymous">
+    <link href="/main.css" rel="stylesheet">
+    <script src="https://code.jquery.com/jquery-3.1.1.min.js"></script>
+    <script src="https://cdn.jsdelivr.net/npm/@stomp/stompjs@7.0.0/bundles/stomp.umd.min.js"></script>
+    <script src="/app.js"></script>
+</head>
+<body>
+<noscript><h2 style="color: #ff0000">Seems your browser doesn't support Javascript! Websocket relies on Javascript being
+    enabled. Please enable
+    Javascript and reload this page!</h2></noscript>
+<div id="main-content" class="container">
+    <div class="row">
+        <div class="col-md-6">
+            This UI was lifted from some example tutorial. It doesn't have to look pretty for this, just be functional.
+            To start the WebSocket connection, press the connect button.
+        </div>
+    </div>
+    <div class="row">
+        <div class="col-md-6">
+            <form class="form-inline">
+                <div class="form-group">
+                    <label for="connect">WebSocket connection:</label>
+                    <button id="connect" class="btn btn-default" type="submit">Connect</button>
+                    <button id="disconnect" class="btn btn-default" type="submit" disabled="disabled">Disconnect
+                    </button>
+                </div>
+            </form>
+        </div>
+        <div class="col-md-6">
+            <form class="form-inline">
+                <div class="form-group">
+                    <label for="station-uuid">Station UUID</label>
+                    <input type="text" id="station-uuid" class="form-control" placeholder="Any UUID">
+                </div>
+                <div class="form-group">
+                    <label for="driver-id">Driver Identifier:</label>
+                    <input type="text" id="driver-id" class="form-control" placeholder="Driver Identifier">
+                </div>
+                <button id="send" class="btn btn-default" type="submit">Send</button>
+            </form>
+        </div>
+    </div>
+
+    <div class="row">
+        <div class="col-md-6">
+            Two identifiers are known to the whitelist by default:
+            <ul>
+                <li>accepted-driver1-with-short-identifier</li>
+                <li>rejected-driver2-with-a-longer-identifier-with-more-characters-yes</li>
+            </ul>
+        </div>
+    </div>
+
+    <div class="row">
+        <div class="col-md-12">
+            <table id="conversation" class="table table-striped">
+                <thead>
+                <tr>
+                    <th>Responses</th>
+                </tr>
+                </thead>
+                <tbody id="responses">
+                </tbody>
+            </table>
+        </div>
+    </div>
+</div>
+</body>
+</html>
--- a/src/test/kotlin/net/thermetics/chargepoint/TransactionControllerTest.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/TransactionControllerTest.kt
@ -0,0 +1,61 @@
+package net.thermetics.chargepoint
+
+import io.mockk.*
+import net.thermetics.chargepoint.models.AuthStatus
+import net.thermetics.chargepoint.models.DriverIdentifier
+import net.thermetics.chargepoint.models.TransactionAuthRequest
+import net.thermetics.chargepoint.models.TransactionAuthResponse
+import net.thermetics.chargepoint.services.AuthService
+import org.junit.jupiter.api.Test
+import org.springframework.messaging.simp.SimpMessagingTemplate
+import org.junit.jupiter.api.Assertions.*
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor
+import java.util.*
+import java.util.concurrent.CompletableFuture
+
+class TransactionControllerTest {
+
+    @Test
+    fun testCreateHeaders() {
+        val authService = mockk<AuthService>()
+        val socket = mockk<SimpMessagingTemplate>()
+        val controller = TransactionController(authService, socket, "ws-topic")
+
+        val sessionId = "session-id"
+
+        val headers = controller.createHeaders(sessionId)
+
+        assertNotNull(headers["simpSessionId"])
+        assertEquals(sessionId, headers["simpSessionId"])
+    }
+
+    @Test
+    fun testSendsAuthStatusToSocket() {
+        val authService = mockk<AuthService>()
+        val socket = mockk<SimpMessagingTemplate>()
+        val headerAccessor = mockk<SimpMessageHeaderAccessor>()
+        val topic = "ws-topic"
+        val sessionId = "session-id"
+
+        val controller = TransactionController(authService, socket, topic)
+
+        val req = TransactionAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            driverIdentifier = DriverIdentifier("i".repeat(30))
+        )
+
+        val resp = TransactionAuthResponse(AuthStatus.Accepted)
+        val headers = controller.createHeaders(sessionId)
+
+        every { headerAccessor.sessionId } returns sessionId
+        every { socket.convertAndSendToUser(sessionId, topic, resp, headers) } just runs
+
+        every { authService.authorizeTransaction(req) } returns CompletableFuture.supplyAsync {
+            resp
+        }
+
+        controller.authorize(req, headerAccessor)
+        verify { socket.convertAndSendToUser(sessionId, topic, resp, headers) }
+        confirmVerified(socket)
+    }
+}
--- a/src/test/kotlin/net/thermetics/chargepoint/authorization/KafkaAuthorizationConsumerTest.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/authorization/KafkaAuthorizationConsumerTest.kt
@ -0,0 +1,45 @@
+package net.thermetics.chargepoint.authorization
+
+import net.thermetics.chargepoint.models.AuthStatus
+import net.thermetics.chargepoint.models.KafkaAuthRequest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import java.util.*
+
+class KafkaAuthorizationConsumerTest {
+    @Test
+    fun testAcceptedIdentifier() {
+        val consumer = KafkaAuthorizationConsumer()
+        val req = KafkaAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            identifier = "accepted-driver1-with-short-identifier"
+        )
+
+        val resp = consumer.authenticate(req)
+        assertEquals(AuthStatus.Accepted, resp.status)
+    }
+
+    @Test
+    fun testRejectedIdentifier() {
+        val consumer = KafkaAuthorizationConsumer()
+        val req = KafkaAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            identifier = "rejected-driver2-with-a-longer-identifier-with-more-characters-yes"
+        )
+
+        val resp = consumer.authenticate(req)
+        assertEquals(AuthStatus.Rejected, resp.status)
+    }
+
+    @Test
+    fun testUnknownIdentifier() {
+        val consumer = KafkaAuthorizationConsumer()
+        val req = KafkaAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            identifier = "we-are-unknown"
+        )
+
+        val resp = consumer.authenticate(req)
+        assertEquals(AuthStatus.Unknown, resp.status)
+    }
+}
--- a/src/test/kotlin/net/thermetics/chargepoint/integration/KafkaAuthServiceIT.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/integration/KafkaAuthServiceIT.kt
@ -0,0 +1,65 @@
+package net.thermetics.chargepoint.integration
+
+import net.thermetics.chargepoint.models.DriverIdentifier
+import net.thermetics.chargepoint.models.TransactionAuthRequest
+import net.thermetics.chargepoint.services.KafkaAuthService
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Assertions.*
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.kafka.annotation.EnableKafka
+import org.springframework.kafka.test.context.EmbeddedKafka
+import java.util.*
+import java.util.concurrent.TimeUnit
+
+@SpringBootTest
+@EnableKafka
+@EmbeddedKafka(
+    partitions = 1,
+    controlledShutdown = false,
+    brokerProperties = [
+        "listeners=PLAINTEXT://localhost:29093",
+        "port=29093"
+    ]
+)
+class KafkaAuthServiceIT {
+    @Autowired
+    lateinit var clientService: KafkaAuthService
+
+    @Test
+    fun testAcceptedDriver() {
+        val req = TransactionAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            driverIdentifier = DriverIdentifier("accepted-driver1-with-short-identifier")
+        )
+
+        val resp = clientService.authorizeTransaction(req).get(30, TimeUnit.SECONDS)
+        assertEquals("Accepted", resp.authorizationStatus)
+    }
+
+    @Test
+    fun testRejectedDriver() {
+        val req = TransactionAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            driverIdentifier = DriverIdentifier(
+                "rejected-driver2-with-a-longer-identifier-with-more-characters-yes"
+            )
+        )
+
+        val resp = clientService.authorizeTransaction(req).get(30, TimeUnit.SECONDS)
+        assertEquals("Rejected", resp.authorizationStatus)
+    }
+
+    @Test
+    fun testUnknownDriver() {
+        val req = TransactionAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            driverIdentifier = DriverIdentifier(
+                "unknown-suspicious-individual"
+            )
+        )
+
+        val resp = clientService.authorizeTransaction(req).get(30, TimeUnit.SECONDS)
+        assertEquals("Unknown", resp.authorizationStatus)
+    }
+}
--- a/src/test/kotlin/net/thermetics/chargepoint/integration/TransactionControllerIT.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/integration/TransactionControllerIT.kt
@ -0,0 +1,121 @@
+package net.thermetics.chargepoint.integration
+
+import net.thermetics.chargepoint.models.DriverIdentifier
+import net.thermetics.chargepoint.models.TransactionAuthRequest
+import net.thermetics.chargepoint.models.TransactionAuthResponse
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.kafka.annotation.EnableKafka
+import org.springframework.kafka.test.context.EmbeddedKafka
+import org.springframework.messaging.converter.GsonMessageConverter
+import org.springframework.messaging.simp.stomp.StompFrameHandler
+import org.springframework.messaging.simp.stomp.StompHeaders
+import org.springframework.messaging.simp.stomp.StompSession
+import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter
+import org.springframework.web.socket.client.standard.StandardWebSocketClient
+import org.springframework.web.socket.messaging.WebSocketStompClient
+import java.lang.reflect.Type
+import java.util.*
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.TimeUnit
+
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@EnableKafka
+@EmbeddedKafka(
+    partitions = 1,
+    controlledShutdown = false,
+    brokerProperties = [
+        "listeners=PLAINTEXT://localhost:29093",
+        "port=29093"
+    ]
+)
+class TransactionControllerIT {
+    @Value("\${local.server.port}")
+    private var port: Int = 0
+
+    private fun createClient() =
+        WebSocketStompClient(StandardWebSocketClient()).also {
+            // For some reason the jackson converter does not work.
+            // It will just hang on waiting for response indefinitely.
+            // Gson works perfectly fine.
+            it.messageConverter = GsonMessageConverter()
+        }
+
+    fun createFrameHandler(fut: CompletableFuture<TransactionAuthResponse>) = object : StompFrameHandler {
+        override fun getPayloadType(headers: StompHeaders): Type =
+            TransactionAuthResponse::class.java
+
+        override fun handleFrame(headers: StompHeaders, payload: Any?) {
+            fut.complete(payload as TransactionAuthResponse)
+        }
+
+    }
+
+    fun createSessionHandler(
+        subscription: String,
+        endpoint: String,
+        frameHandler: StompFrameHandler,
+        req: TransactionAuthRequest
+    ) = object : StompSessionHandlerAdapter() {
+        override fun afterConnected(session: StompSession, connectedHeaders: StompHeaders) {
+            session.subscribe(subscription, frameHandler)
+            session.send(endpoint, req)
+        }
+    }
+
+    fun sendRequest(req: TransactionAuthRequest): CompletableFuture<TransactionAuthResponse> {
+        val subscription = "/anonymous-session/topic/auth"
+        val endpoint = "/app/transaction/authorize"
+        val url = "ws://localhost:${port}/ws"
+
+        val eventualResponse = CompletableFuture<TransactionAuthResponse>()
+        val frameHandler = createFrameHandler(eventualResponse)
+
+        val sessionHandler = createSessionHandler(subscription, endpoint, frameHandler, req)
+
+        createClient().connectAsync(url, sessionHandler).get(30, TimeUnit.SECONDS)
+
+        return eventualResponse
+    }
+
+    @Test
+    fun testInvalidRequest() {
+        val req =
+            TransactionAuthRequest(UUID.randomUUID().toString(), driverIdentifier = DriverIdentifier("blah"))
+        val eventualResponse = sendRequest(req)
+        val respFromServer = eventualResponse.get(30, TimeUnit.SECONDS)
+        assertEquals("Invalid", respFromServer.authorizationStatus)
+    }
+
+    @Test
+    fun testUnknownRequest() {
+        val req =
+            TransactionAuthRequest(UUID.randomUUID().toString(), driverIdentifier = DriverIdentifier("i".repeat(30)))
+        val eventualResponse = sendRequest(req)
+        val respFromServer = eventualResponse.get(30, TimeUnit.SECONDS)
+        assertEquals("Unknown", respFromServer.authorizationStatus)
+    }
+
+    @Test
+    fun testAcceptedRequest() {
+        val identifier = "accepted-driver1-with-short-identifier"
+        val req =
+            TransactionAuthRequest(UUID.randomUUID().toString(), driverIdentifier = DriverIdentifier(identifier))
+        val eventualResponse = sendRequest(req)
+        val respFromServer = eventualResponse.get(30, TimeUnit.SECONDS)
+        assertEquals("Accepted", respFromServer.authorizationStatus)
+    }
+
+    @Test
+    fun testRejectedRequest() {
+        val identifier = "rejected-driver2-with-a-longer-identifier-with-more-characters-yes"
+        val req =
+            TransactionAuthRequest(UUID.randomUUID().toString(), driverIdentifier = DriverIdentifier(identifier))
+        val eventualResponse = sendRequest(req)
+        val respFromServer = eventualResponse.get(30, TimeUnit.SECONDS)
+        assertEquals("Rejected", respFromServer.authorizationStatus)
+    }
+}
--- a/src/test/kotlin/net/thermetics/chargepoint/models/KafkaAuthRequestTest.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/models/KafkaAuthRequestTest.kt
@ -0,0 +1,21 @@
+package net.thermetics.chargepoint.models
+
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Assertions.*
+import java.util.*
+
+class KafkaAuthRequestTest {
+    @Test
+    fun testConvertWebRequest() {
+        val webRequest = TransactionAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            driverIdentifier = DriverIdentifier(
+                id = "a-driver-identifier-that-is-valid"
+            )
+        )
+
+        val kafkaRequest = KafkaAuthRequest(webRequest)
+        assertEquals(webRequest.stationUuid, kafkaRequest.stationUuid)
+        assertEquals(webRequest.driverIdentifier.id, kafkaRequest.identifier)
+    }
+}
--- a/src/test/kotlin/net/thermetics/chargepoint/models/TransactionAuthResponseTest.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/models/TransactionAuthResponseTest.kt
@ -0,0 +1,34 @@
+package net.thermetics.chargepoint.models
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+class TransactionAuthResponseTest {
+    @Test
+    fun testKafkaResponseConversion() {
+        assertEquals(
+            "Unknown",
+            TransactionAuthResponse(KafkaAuthResponse(AuthStatus.Unknown)).authorizationStatus
+        )
+        assertEquals(
+            "Rejected",
+            TransactionAuthResponse(KafkaAuthResponse(AuthStatus.Rejected)).authorizationStatus
+        )
+        assertEquals(
+            "Invalid",
+            TransactionAuthResponse(KafkaAuthResponse(AuthStatus.Invalid)).authorizationStatus
+        )
+        assertEquals(
+            "Accepted",
+            TransactionAuthResponse(KafkaAuthResponse(AuthStatus.Accepted)).authorizationStatus
+        )
+    }
+
+    @Test
+    fun testStatusConversion() {
+        assertEquals("Unknown", TransactionAuthResponse(AuthStatus.Unknown).authorizationStatus)
+        assertEquals("Rejected", TransactionAuthResponse(AuthStatus.Rejected).authorizationStatus)
+        assertEquals("Invalid", TransactionAuthResponse(AuthStatus.Invalid).authorizationStatus)
+        assertEquals("Accepted", TransactionAuthResponse(AuthStatus.Accepted).authorizationStatus)
+    }
+}
--- a/src/test/kotlin/net/thermetics/chargepoint/services/KafkaAuthServiceTest.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/services/KafkaAuthServiceTest.kt
@ -0,0 +1,58 @@
+package net.thermetics.chargepoint.services
+
+import io.mockk.*
+import io.mockk.InternalPlatformDsl.toStr
+import net.thermetics.chargepoint.config.KAFKA_AUTH_REPLIES
+import net.thermetics.chargepoint.config.KAFKA_AUTH_REQUESTS
+import net.thermetics.chargepoint.models.*
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.springframework.kafka.requestreply.RequestReplyFuture
+import java.util.*
+import java.util.concurrent.TimeUnit
+
+class KafkaAuthServiceTest {
+    @Test
+    fun testReturnsInvalidOnValidationFailure() {
+        val kafka = mockk<ReplyingAuthTemplate>()
+        val service = KafkaAuthService(kafka)
+
+        val invalidReq = TransactionAuthRequest(
+            stationUuid = "not-a-uuid",
+            driverIdentifier = DriverIdentifier("not-valid")
+        )
+
+        val expectedResp = TransactionAuthResponse(AuthStatus.Invalid)
+
+        val resp = service.authorizeTransaction(invalidReq).get(30, TimeUnit.SECONDS)
+        assertEquals(expectedResp, resp)
+    }
+
+    @Test
+    fun testSendsValidRequestsToKafka() {
+        val kafka = mockk<ReplyingAuthTemplate>()
+        val service = KafkaAuthService(kafka)
+
+        val req = TransactionAuthRequest(
+            stationUuid = UUID.randomUUID().toStr(),
+            driverIdentifier = DriverIdentifier("i".repeat(30))
+        )
+
+        val reqRecord = AuthProducerRecord(KAFKA_AUTH_REQUESTS, KafkaAuthRequest(req))
+
+        // Future has to be mocked directly.
+        val fut = RequestReplyFuture<String, KafkaAuthRequest, KafkaAuthResponse>()
+        val respRecord = ConsumerRecord(
+            KAFKA_AUTH_REPLIES, 1, 1, "", KafkaAuthResponse(AuthStatus.Accepted)
+        )
+
+        fut.complete(respRecord)
+        every { kafka.sendAndReceive(reqRecord) } returns fut
+
+        // Getting the result of the future required for the computation to complete
+        service.authorizeTransaction(req).get(30, TimeUnit.SECONDS)
+        verify { kafka.sendAndReceive(reqRecord) }
+        confirmVerified(kafka)
+    }
+}
--- a/src/test/kotlin/net/thermetics/chargepoint/validation/ValidationTest.kt
+++ b/src/test/kotlin/net/thermetics/chargepoint/validation/ValidationTest.kt
@ -0,0 +1,111 @@
+package net.thermetics.chargepoint.validation
+
+import arrow.core.nonEmptyListOf
+import net.thermetics.chargepoint.models.*
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import java.util.*
+
+class ValidationTest {
+
+    @Test
+    fun testValidateValidStationId() {
+        val stationId = UUID.randomUUID()
+        val result = validateStationUuid(stationId.toString())
+
+        assertTrue(result.isRight())
+        assertEquals(stationId, result.getOrNull())
+    }
+
+    @Test
+    fun testValidateInvalidStationId() {
+        val stationId = "not a uuid"
+        val result = validateStationUuid(stationId)
+
+        assertTrue(result.isLeft())
+        assertEquals(ValidationError.InvalidStationUuid, result.leftOrNull())
+    }
+
+    @Test
+    fun testValidateValidIdentifier() {
+        val driver = DriverIdentifier(id = "this is between 20 and 80 characters")
+        val result = validateIdentifierNotShort(driver)
+        assertTrue(result.isRight())
+        assertEquals(driver, result.getOrNull())
+    }
+
+    @Test
+    fun testValidateIdentifierTooShort() {
+        val driver = DriverIdentifier(id = "short id")
+        val result = validateIdentifierNotShort(driver)
+        assertTrue(result.isLeft())
+        assertEquals(ValidationError.IdentifierTooShort, result.leftOrNull())
+    }
+
+    @Test
+    fun testValidateIdentifierTooLong() {
+        val driver = DriverIdentifier(id = "i".repeat(90))
+        val result = validateIdentifierNotLong(driver)
+        assertTrue(result.isLeft())
+        assertEquals(ValidationError.IdentifierTooLong, result.leftOrNull())
+    }
+
+    @Test
+    fun testValidateIdentifierExactly20() {
+        val driver = DriverIdentifier(id = "i".repeat(20))
+
+        val notTooShort = validateIdentifierNotShort(driver)
+        assertTrue(notTooShort.isRight())
+        assertEquals(driver, notTooShort.getOrNull())
+
+        val notTooLong = validateIdentifierNotLong(driver)
+        assertTrue(notTooLong.isRight())
+        assertEquals(driver, notTooLong.getOrNull())
+    }
+
+    @Test
+    fun testValidateIdentifierExactly80() {
+        val driver = DriverIdentifier(id = "i".repeat(80))
+
+        val notTooShort = validateIdentifierNotShort(driver)
+        assertTrue(notTooShort.isRight())
+        assertEquals(driver, notTooShort.getOrNull())
+
+        val notTooLong = validateIdentifierNotLong(driver)
+        assertTrue(notTooLong.isRight())
+        assertEquals(driver, notTooLong.getOrNull())
+    }
+
+    @Test
+    fun testValidationComposition() {
+        val req = TransactionAuthRequest(
+            stationUuid = "not-a-uuid",
+            driverIdentifier = DriverIdentifier("not20chars")
+        )
+
+        val result = validateRequest(req)
+
+        assertTrue(result.isLeft())
+        assertEquals(
+            nonEmptyListOf(
+                ValidationError.InvalidStationUuid,
+                ValidationError.IdentifierTooShort
+            ),
+            result.leftOrNull()
+        )
+    }
+
+    @Test
+    fun testValidRequest() {
+        val req = TransactionAuthRequest(
+            stationUuid = UUID.randomUUID().toString(),
+            driverIdentifier = DriverIdentifier("i".repeat(30))
+        )
+
+        val result = validateRequest(req)
+
+        assertTrue(result.isRight())
+        assertEquals(req, result.getOrNull())
+    }
+}