finish off except for readme

Dockerfile

@@ -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"]

build.gradle.kts

@@ -19,13 +19,22 @@ repositories {
 }
 
 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> {

docker-compose.yml

@@ -1,23 +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://localhost:29092
+      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

src/main/kotlin/net/thermetics/chargepoint/authorization/KafkaAuthorizationConsumer.kt

@@ -16,11 +16,16 @@ import org.springframework.messaging.handler.annotation.SendTo
  * and it thus can run as a separate application as well.
  */
 class KafkaAuthorizationConsumer {
-    @KafkaListener(id = "myId", topics = [KAFKA_AUTH_REQUESTS])
+    @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 KafkaAuthResponse(AuthStatus.Accepted)
+
+        return when (Whitelist.AllowedCards[message.identifier]) {
+            true -> KafkaAuthResponse(AuthStatus.Accepted)
+            false -> KafkaAuthResponse(AuthStatus.Rejected)
+            null -> KafkaAuthResponse(AuthStatus.Unknown)
+        }
     }
 }

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
+    )
+}

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()
+}

src/main/kotlin/net/thermetics/chargepoint/config/KafkaConfig.kt

@@ -7,6 +7,7 @@ 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
@@ -26,6 +27,7 @@ const val KAFKA_AUTH_REPLIES = "kafka-auth-replies"
  * required Kafka beans.
  */
 @Configuration
+@Profile("kafka-auth-client")
 class KafkaConfig {
     // Communicates with the authorization application over Kafka, from the web API.
     @Bean
@@ -75,12 +77,4 @@ class KafkaConfig {
             .partitions(10)
             .replicas(2)
             .build()
-
-
-    // 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()
-
 }

src/main/kotlin/net/thermetics/chargepoint/config/WebSocketConfig.kt

@@ -3,6 +3,7 @@ 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
@@ -17,6 +18,7 @@ import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerCo
  * connected to the websocket.
  */
 @Configuration
+@Profile("web-api")
 @EnableWebSocketMessageBroker
 class WebSocketConfig(
     @Value("\${chargepoint.websocket.topic-prefix}") val wsTopicPrefix: String,

src/main/kotlin/net/thermetics/chargepoint/controllers/AuthController.kt

@@ -1,22 +1,20 @@
 package net.thermetics.chargepoint
 
-import net.thermetics.chargepoint.models.TransactionAuthRequest
-import net.thermetics.chargepoint.models.Validation
-import net.thermetics.chargepoint.models.ValidationError
-import net.thermetics.chargepoint.models.validateRequest
+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
-import java.util.concurrent.CompletableFuture
 
 @Controller
-class HelloController(
+@Profile("web-api")
+class TransactionController(
     val authService: AuthService,
     val socket: SimpMessagingTemplate,
     @Autowired @Qualifier("websocketTopic") val wsAuthTopic: String
@@ -27,38 +25,26 @@ class HelloController(
             it.setLeaveMutable(true)
         }.messageHeaders
 
-    fun validateAsync(req: TransactionAuthRequest): CompletableFuture<Validation<TransactionAuthRequest, ValidationError>> =
-        CompletableFuture.supplyAsync { validateRequest(req) }
-
-    fun authorize(req: TransactionAuthRequest, header: SimpMessageHeaderAccessor) {
-        authService.authorizeTransaction(req).thenApply { authResp ->
-            socket.convertAndSendToUser(
-                header.sessionId!!,
-                wsAuthTopic,
-                authResp,
-                createHeaders(header.sessionId!!)
-            )
-        }
-    }
-
-    fun sendError(error: ValidationError, header: SimpMessageHeaderAccessor) {
-        val payload = object { val message = error.message }
+    fun sendAuthStatus(header: SimpMessageHeaderAccessor, reply: TransactionAuthResponse) {
+        println("Sending response ${reply} back to session ${header.sessionId} on topic $wsAuthTopic")
         socket.convertAndSendToUser(
             header.sessionId!!,
             wsAuthTopic,
-            payload,
+            reply,
             createHeaders(header.sessionId!!)
         )
     }
 
-    @MessageMapping("/transaction/authorize")
-    fun authorizeEndpoint(@Payload req: TransactionAuthRequest, header: SimpMessageHeaderAccessor) {
-        validateAsync(req).thenApply { validatedReq ->
-            when (validatedReq) {
-                is Validation.Success -> authorize(validatedReq.value, header)
-                is Validation.Error -> sendError(validatedReq.error, header)
-            }
+    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)
+    }
+
 }

src/main/kotlin/net/thermetics/chargepoint/models/TransactionAuthResponse.kt

@@ -2,4 +2,5 @@ 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)
 }

src/main/kotlin/net/thermetics/chargepoint/models/Validation.kt

@@ -1,24 +1,36 @@
 package net.thermetics.chargepoint.models
 
-// Why not Result<T>? That catches Throwables also, and I want this
-// specifically to catch only validation errors. Plus catching Throwables
-// can lead to fun things like catching JVM out-of-memory errors, which you
-// generally don't want to do.
-// In production, would likely use Arrow or make sure Result is not catching Throwables.
-sealed class Validation<out T, out E> {
-    data class Success<out T, out E>(val value: T) : Validation<T, E>()
-    data class Error<out T, out E>(val error: E) : Validation<T, E>()
-}
+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 NoSessionId : ValidationError("No session ID")
-    data object InvalidIdentifier : ValidationError("Invalid driver identifier")
+    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 validateRequest(req: TransactionAuthRequest): Validation<TransactionAuthRequest, ValidationError> {
-    if (req.driverIdentifier.id.length < 20 || req.driverIdentifier.id.length > 80) {
-        return Validation.Error(ValidationError.InvalidIdentifier)
-    }
+fun validateStationUuid(stationUuid: String) =
+    Either.catch { UUID.fromString(stationUuid) }
+        .mapLeft { ValidationError.InvalidStationUuid }
 
-    return Validation.Success(req)
-}
+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 }

src/main/kotlin/net/thermetics/chargepoint/services/KafkaAuthService.kt

@@ -1,27 +1,36 @@
 package net.thermetics.chargepoint.services
 
 import net.thermetics.chargepoint.config.KAFKA_AUTH_REQUESTS
-import net.thermetics.chargepoint.models.KafkaAuthRequest
-import net.thermetics.chargepoint.models.KafkaAuthResponse
-import net.thermetics.chargepoint.models.TransactionAuthRequest
-import net.thermetics.chargepoint.models.TransactionAuthResponse
+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: ReplyingKafkaTemplate<String, KafkaAuthRequest, KafkaAuthResponse>,
+    val kafka: ReplyingAuthTemplate,
 ) : AuthService {
-    override fun authorizeTransaction(req: TransactionAuthRequest): CompletableFuture<TransactionAuthResponse> =
-        ProducerRecord<String, KafkaAuthRequest>(KAFKA_AUTH_REQUESTS, KafkaAuthRequest(req)).let {
+    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) } },
+            )
+        }
 }

src/main/resources/application.properties

@@ -1,3 +1,15 @@
+# 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

src/main/resources/static/app.js

@@ -5,9 +5,9 @@ const stompClient = new StompJs.Client({
 stompClient.onConnect = (frame) => {
     setConnected(true);
     console.log('Connected: ' + frame);
-    stompClient.subscribe('/anonymous-session/topic/auth', (greeting) => {
-        console.log(greeting);
-        showGreeting(JSON.parse(greeting.body).authorizationStatus);
+    stompClient.subscribe('/anonymous-session/topic/auth', (response) => {
+        console.log(response);
+        addResponse(JSON.parse(response.body).authorizationStatus);
     });
 };
 
@@ -42,20 +42,25 @@ function disconnect() {
     console.log("Disconnected");
 }
 
-function sendName() {
+function sendRequest() {
+    const stationUuid = $('#station-uuid').val()
+    const driverIdentifier = $('#driver-id').val()
     stompClient.publish({
         destination: "/app/transaction/authorize",
-        body: JSON.stringify({'stationUuid': 'mystation', 'driverIdentifier': { 'id': $("#name").val()} } )
+        body: JSON.stringify({
+                'stationUuid': stationUuid,
+                'driverIdentifier': { 'id': driverIdentifier }
+              })
     });
 }
 
-function showGreeting(message) {
-    $("#greetings").append("<tr><td>" + message + "</td></tr>");
+function addResponse(status) {
+    $("#responses").append("<tr><td>" + status + "</td></tr>");
 }
 
 $(function () {
-    $("form").on('submit', (e) => e.preventDefault());
+    $("form").on('submit', (e) => false);
     $( "#connect" ).click(() => connect());
     $( "#disconnect" ).click(() => disconnect());
-    $( "#send" ).click(() => sendName());
+    $( "#send" ).click(() => sendRequest());
 });

src/main/resources/static/index.html

@@ -1,8 +1,9 @@
 <!DOCTYPE html>
 <html>
 <head>
-    <title>Hello WebSocket</title>
-    <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" integrity="sha384-BVYiiSIFeK1dGmJRAkycuHAHRg32OmUcww7on3RYdg4Va+PmSTsz/K68vbdEjh4u" crossorigin="anonymous">
+    <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>
@@ -13,6 +14,12 @@
     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">
@@ -27,22 +34,37 @@
         <div class="col-md-6">
             <form class="form-inline">
                 <div class="form-group">
-                    <label for="name">What is your name?</label>
-                    <input type="text" id="name" class="form-control" placeholder="Your name here...">
+                    <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>Greetings</th>
+                    <th>Responses</th>
                 </tr>
                 </thead>
-                <tbody id="greetings">
+                <tbody id="responses">
                 </tbody>
             </table>
         </div>

src/test/kotlin/net/thermetics/chargepoint/ChargepointAssignmentTests.kt

@@ -1,13 +0,0 @@
-package net.thermetics.chargepoint
-
-import org.junit.jupiter.api.Test
-import org.springframework.boot.test.context.SpringBootTest
-
-@SpringBootTest
-class ChargepointAssignmentTests {
-
-	@Test
-	fun contextLoads() {
-	}
-
-}

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)
+    }
+}

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)
+    }
+}

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)
+    }
+}

src/test/kotlin/net/thermetics/chargepoint/integration/TransactionControllerIT.kt

@@ -0,0 +1,87 @@
+package net.thermetics.chargepoint.integration
+
+import net.thermetics.chargepoint.TransactionController
+import net.thermetics.chargepoint.models.DriverIdentifier
+import net.thermetics.chargepoint.models.TransactionAuthRequest
+import net.thermetics.chargepoint.models.TransactionAuthResponse
+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.converter.MappingJackson2MessageConverter
+import org.springframework.messaging.simp.stomp.*
+import org.springframework.scheduling.TaskScheduler
+import org.springframework.web.socket.client.standard.StandardWebSocketClient
+import org.springframework.web.socket.messaging.WebSocketStompClient
+import org.springframework.web.socket.sockjs.client.WebSocketTransport
+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 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)
+        }
+    }
+
+    @Test
+    fun testSocket() {
+        val subscription = "/anonymous-session/topic/auth"
+        val endpoint = "/app/transaction/authorize"
+        val url = "ws://localhost:${port}/ws"
+
+        val fut = CompletableFuture<TransactionAuthResponse>()
+        val frameHandler = createFrameHandler(fut)
+
+        val req =
+            TransactionAuthRequest(UUID.randomUUID().toString(), driverIdentifier = DriverIdentifier("blah"))
+
+        val sessionHandler = createSessionHandler(subscription, endpoint, frameHandler, req)
+
+        createClient().connectAsync(url, sessionHandler).get(30, TimeUnit.SECONDS)
+        val respFromServer = fut.get(30, TimeUnit.SECONDS)
+        println("Got $respFromServer")
+
+    }
+}

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)
+    }
+}

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)
+    }
+}

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)
+    }
+}

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())
+    }
+}