Webflux Updates (#3)

* adds empty async controller to be used later

* basic webflux implementation

* remove logging from json encoder

* add lightweight benchmarking script to help with optimizing async setup

* benchmarking script tweak

* update readme

* adds geometry data to Address to demonstrate postgis integration.  also adds to testutils.

* better table enumeration for resetting persistence in tests

* update readme

* update readme

* adds benchmark endpoint

* adds more or less working register/login functionality, along with a partial jwt auth skeleton

* adds basic working authentication / authorization via Spring Security + JWT

Author: halfhp

README.md

@@ -3,20 +3,11 @@
 # Overview
 Demonstrates using Spring Framework with Scala 3.
 
-Since the initial release of Scala 3 I've considered upgrading various 
-non-trivial personal and professional projects from 2.13 to 3, but until recently (2025) always ultimately declined to upgrade.   
-
-After getting over the deprecation hurdles,
-The biggest headaches have been with the level of compatibility tools like IntellIJ offer, and the relatively small market share
-scala 3 currently has of Scala projects and libraries.
-Builds can also slow to a crawl or hang during non-trivial refactorings.  Compile times
-are also up, and builds will randomly fail, only to succeed after a second or third retry. 
+The biggest headaches of upgrading from Scala 2.13 to Scala 3 has been with the level of compatibility tools like IntellIJ offer, and the relatively small market share
+it has of Scala projects and libraries. Builds sometimes slow to a crawl or hang or randomly fail, only to succeed after a second or third retry.
 
 The jury is still out on whether the migration is worthwhile for established projects, but
-as of 2025 I do feel like Scala 3 is the way to go.
-
-I'll try to keep this project updated
-with things I learn as I go.
+as of 2025 I do feel like Scala 3 is the way to go for new projects.
 
 :speech_balloon: **Questions / comments / suggestions are welcome in the [discussions](https://github.com/halfhp/ScalaSpringExperiment/discussions), or feel free to [contact me](mailto:halfhp@gmail.com) directly.**
 
@@ -81,13 +72,13 @@ This means that whenever you are working with variables coming Spring, you gener
 One particular place to watch out for this is when using Spring's `@RequestParam` and `@PathVariable` annotations in controllers.
 
 ## Spring's ThreadLocal Context
-Much of Spring's async programming model relies on ThreadLocal context, particularly when using WebMVC.  This used to be a common pattern in Java, but not one that is used in Scala.
-This becomes particularly annoying when interfacing between things like controller entry points and services and utilities that are built
+Parts of Spring's architecture relies on ThreadLocal context, particularly when using WebMVC.
+This can be problematic when interfacing between things like controller entry points and services and utilities that are built
 around IO/Future/ZIO etc. monads.  Effectively, trying to access something like Spring Security's SecurityContext from these methods
 will not work.  Without going into too much detail WebFlux has the same basic problem, even though its not technically using ThreadLocal context.
 
-The best solution I have found is to pass the SecurityContext and any other ThreadLocal / pseudo global context data as an argument to
-these methods.  This is not ideal, but it is the best solution I have found so far.
+My preferred solution is to pass the SecurityContext and any other ThreadLocal / pseudo global context data as an argument to
+these methods.
 
 ## Async Programming 
 Spring has it's own mechanisms for async programming, and it takes some work to adapt it to be compatible with IO monads.

build.gradle

@@ -30,6 +30,8 @@ dependencies {
     implementation 'org.springframework.security:spring-security-test'
     implementation 'org.springframework.boot:spring-boot-starter-jdbc'
 
+    implementation 'com.github.jwt-scala:jwt-circe_3:10.0.4'
+
     implementation 'net.postgis:postgis-jdbc:2024.1.0'
 
     implementation 'io.circe:circe-core_3:0.14.12'

extras/taurus/benchmark_localhost.sh

@@ -11,17 +11,54 @@
 #      - "-Xmx512m"
 
 execution:
+  - concurrency: 1
+    name: "serial conc=1, 10ms, 1x80"
+    scenario: serial_10ms_1x80
+    ramp-up: 1s
+    hold-for: 30s
+    sequential: true
+  - concurrency: 1
+    name: "parallel conc=1, 10ms, 4x80"
+    scenario: parallel_10ms_4x80
+    ramp-up: 1s
+    hold-for: 30s
+    sequential: true
   - concurrency: 10
-    ramp-up: 30s
-    hold-for: 1m
-    scenario: simple
+    name: "serial conc=10, 10ms, 1x80"
+    scenario: serial_10ms_1x80
+    ramp-up: 1s
+    hold-for: 30s
+    sequential: true
+  - concurrency: 10
+    name: "parallel conc=10, 10ms, 4x80"
+    scenario: parallel_10ms_4x80
+    ramp-up: 1s
+    hold-for: 30s
+    sequential: true
+  - concurrency: 100
+    name: "serial conc=100, 10ms, 1x80"
+    scenario: serial_10ms_1x80
+    ramp-up: 1s
+    hold-for: 30s
+    sequential: true
+  - concurrency: 100
+    name: "parallel conc=100, 10ms, 4x80"
+    scenario: parallel_10ms_4x80
+    ramp-up: 5s
+    hold-for: 30s
+    sequential: true
 
 scenarios:
-  simple:
+  serial_10ms_1x80:
+    requests:
+      - url: http://host.docker.internal:8080/benchmark?count=80&durationMs=10&parallelism=1
+        method: GET
+  parallel_10ms_4x80:
     requests:
-      - url: http://host.docker.internal:8080/
+      - url: http://host.docker.internal:8080/benchmark?count=80&durationMs=10&parallelism=4
         method: GET
 
+
 #settings:
 #  artifacts-dir: /output
 

extras/taurus/tests/test.yml

@@ -44,3 +44,22 @@ CREATE TRIGGER set_address_timestamps
     BEFORE INSERT OR UPDATE ON address
     FOR EACH ROW
 EXECUTE FUNCTION set_timestamp_fields();
+
+CREATE TABLE registered_user (
+    id BIGSERIAL PRIMARY KEY,
+    date_created timestamp not null,
+    last_updated timestamp not null,
+    email varchar unique not null,
+    email_verified boolean not null,
+    roles varchar[] not null,
+    password_hash varchar not null,
+    person_id BIGINT not null
+        constraint registered_user_person_id_fk
+        references person
+        on delete cascade
+);
+
+CREATE TRIGGER set_registered_user_timestamps
+    BEFORE INSERT OR UPDATE ON registered_user
+    FOR EACH ROW
+EXECUTE FUNCTION set_timestamp_fields();

src/main/resources/db/migration/V1.0.sql

@@ -2,53 +2,77 @@ package com.example.scalaspringexperiment
 
 import cats.effect.unsafe.IORuntime
 import cats.effect.{IO, Resource}
+import com.example.scalaspringexperiment.auth.{JwtAuthManager, JwtServerAuthConverter}
 import com.example.scalaspringexperiment.util.{CirceJsonDecoder, CirceJsonEncoder}
 import doobie.{DataSourceTransactor, ExecutionContexts}
 import doobie.util.transactor.Transactor
-import org.springframework.context.annotation.{Bean, Configuration}
+import org.springframework.context.annotation.{Bean, Configuration, Primary}
+import org.springframework.http.HttpStatus
 import org.springframework.http.codec.ServerCodecConfigurer
+import org.springframework.security.authentication.{DelegatingReactiveAuthenticationManager, UsernamePasswordAuthenticationToken}
 import org.springframework.security.config.Customizer
 import org.springframework.security.config.annotation.method.configuration.EnableReactiveMethodSecurity
 import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity
-import org.springframework.security.config.web.server.ServerHttpSecurity
+import org.springframework.security.config.web.server.{SecurityWebFiltersOrder, ServerHttpSecurity}
 import org.springframework.security.web.server.SecurityWebFilterChain
-import org.springframework.security.web.server.context.NoOpServerSecurityContextRepository
+import org.springframework.security.web.server.authentication.AuthenticationWebFilter
+import org.springframework.security.web.server.context.{NoOpServerSecurityContextRepository, WebSessionServerSecurityContextRepository}
 import org.springframework.web.reactive.config.WebFluxConfigurer
 
 import javax.sql.DataSource
 
+
 @Configuration
-@EnableWebFluxSecurity
-@EnableReactiveMethodSecurity
 class SpringConfig(
   dataSource: DataSource,
 ) {
 
-  @Bean
-  def catsEffectIORuntime(): IORuntime = {
-    cats.effect.unsafe.implicits.global
-  }
-
   @Bean
   def doobieTransactor(): Resource[IO, DataSourceTransactor[IO]] = {
     for {
       ce <- ExecutionContexts.fixedThreadPool[IO](32) // our connect EC
     } yield Transactor.fromDataSource[IO](dataSource, ce)
   }
+}
 
+@Configuration
+@EnableWebFluxSecurity
+@EnableReactiveMethodSecurity
+class SecurityConfig(
+  jwtAuthManager: JwtAuthManager,
+) {
   @Bean
-  def securityFilterChain(http: ServerHttpSecurity): SecurityWebFilterChain = {
+  def securityFilterChain(
+    http: ServerHttpSecurity,
+    jwtAuthFilter: AuthenticationWebFilter,
+  ): SecurityWebFilterChain = {
     http
       .cors(Customizer.withDefaults())
-      .csrf(csrf => csrf.disable()) // Stateless app using JWT
-      .securityContextRepository(NoOpServerSecurityContextRepository.getInstance()) // optional, disables session caching
-      .authorizeExchange(authz =>
-        authz.anyExchange().permitAll()
-      )
-//      .httpBasic().disable() // or leave enabled if using basic auth
-//      .formLogin().disable()
+      .csrf(csrf => csrf.disable())
+      .authorizeExchange(_.anyExchange().permitAll())
+      .addFilterAt(jwtAuthFilter, SecurityWebFiltersOrder.AUTHENTICATION)
+      .securityContextRepository(NoOpServerSecurityContextRepository.getInstance()) // stateless auth
       .build()
   }
+
+    @Bean
+    def jwtAuthFilter(
+      jwtAuthManager: JwtAuthManager
+    ): AuthenticationWebFilter = {
+      val filter = new AuthenticationWebFilter(jwtAuthManager)
+      filter.setServerAuthenticationConverter(new JwtServerAuthConverter)
+      filter.setSecurityContextRepository(NoOpServerSecurityContextRepository.getInstance())
+      filter
+    }
+}
+
+@Configuration(proxyBeanMethods = false)
+class CatsEffectConfig {
+
+  @Bean
+  def catsEffectIORuntime(): IORuntime = {
+    cats.effect.unsafe.implicits.global
+  }
 }
 
 @Configuration

src/main/scala/com/example/scalaspringexperiment/SpringConfig.scala

@@ -0,0 +1,164 @@
+package com.example.scalaspringexperiment.auth
+
+import cats.data.EitherT
+import cats.effect.IO
+import cats.effect.unsafe.IORuntime
+import com.example.scalaspringexperiment.auth.JwtAuthManager.{AuthError, InvalidCredentials, LoginSuccess, ROLE_USER, RegisterSuccess, UserExists, UserNotFound}
+import com.example.scalaspringexperiment.entity.{Person, RegisteredUser}
+import com.example.scalaspringexperiment.service.{PersonService, RegisteredUserService}
+import org.springframework.context.annotation.Lazy
+import org.springframework.security.authentication.{ReactiveAuthenticationManager, UsernamePasswordAuthenticationToken}
+import org.springframework.security.core.authority.SimpleGrantedAuthority
+import org.springframework.security.core.{Authentication, GrantedAuthority}
+import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder
+import org.springframework.stereotype.Component
+import pdi.jwt.{JwtAlgorithm, JwtCirce, JwtClaim}
+import reactor.core.publisher.Mono
+
+import java.security.SecureRandom
+import java.time.Instant
+import scala.jdk.CollectionConverters.*
+
+object JwtAuthManager {
+  val ROLE_USER = "ROLE_USER"
+  val ROLE_ADMIN = "ROLE_ADMIN"
+  sealed trait AuthError {
+    val message: String
+  }
+
+  case class InvalidCredentials(
+    message: String
+  ) extends AuthError
+
+  case class UserNotFound(
+    message: String
+  ) extends AuthError
+
+  case class UserExists(
+    message: String
+  ) extends AuthError
+
+  case class LoginSuccess(
+    registeredUser: RegisteredUser,
+    person: Person,
+    jwtToken: String
+  )
+
+  case class RegisterSuccess(
+    registeredUser: RegisteredUser,
+    person: Person,
+    jwtToken: String
+  )
+}
+
+@Component
+class JwtAuthManager(
+  @Lazy personService: PersonService,
+  @Lazy registeredUserService: RegisteredUserService,
+  runtime: IORuntime,
+) extends ReactiveAuthenticationManager {
+
+  private val hasher = new BCryptPasswordEncoder(10, new SecureRandom())
+
+  given theRuntime: IORuntime = runtime
+
+  // TODO: this is very bad:
+  private val secretKey: String = "secretKey"
+  private val algo = JwtAlgorithm.HS256
+
+  override def authenticate(authentication: Authentication): Mono[Authentication] = {
+    println(s"[AUTH DEBUG] Called with credentials: ${authentication.getCredentials}")
+    try {
+      val token = authentication.getCredentials.toString
+      val claim = JwtCirce.decode(token, secretKey, Seq(algo)).get
+      val authorities = Seq(new SimpleGrantedAuthority(ROLE_USER))
+      val auth = new UsernamePasswordAuthenticationToken(
+        claim.subject.get,
+        null,
+        authorities.asJava
+      )
+      Mono.just(auth)
+    } catch {
+      case e: Exception => Mono.empty()
+    }
+  }
+
+
+  def generateTokenForRegisteredUser(
+    user: RegisteredUser,
+    expiration: Instant
+  ): IO[String] = IO {
+    val claim = JwtClaim(
+      subject = Some(user.id.toString),
+      expiration = Some(expiration.getEpochSecond),
+      issuedAt = Some(Instant.now.getEpochSecond)
+    )
+    JwtCirce.encode(claim, secretKey, algo)
+  }
+
+  def validatePassword(
+    password: String,
+    passwordHash: String
+  ): IO[Boolean] = {
+    IO {
+      hasher.matches(password, passwordHash)
+    }.handleError { ex =>
+      false
+    }
+  }
+
+  def login(
+    email: String,
+    password: String
+  ): IO[Either[AuthError, LoginSuccess]] = {
+    (for {
+      registeredUser <- EitherT(registeredUserService.findByEmail(email).map {
+        case Some(user) => Right(user)
+        case None => Left(UserNotFound("User not found"))
+      })
+      isValidPassword <- EitherT(validatePassword(password, registeredUser.passwordHash).map {
+        case true => Right(true)
+        case false => Left(InvalidCredentials("Invalid credentials"))
+      })
+      person <- EitherT(personService.findById(registeredUser.personId).map {
+        case Some(p) => Right(p)
+        case None => Left(UserNotFound("Person not found"))
+      })
+      jwtToken <- EitherT.liftF[IO, AuthError, String](generateTokenForRegisteredUser(registeredUser, Instant.now.plusSeconds(3600 * 30)))
+    } yield LoginSuccess(
+        registeredUser = registeredUser,
+        person = person,
+        jwtToken = jwtToken
+      )).value
+  }
+
+  def register(
+    name: String,
+    age: Int,
+    email: String,
+    password: String
+  ): IO[Either[AuthError, RegisterSuccess]] = {
+    (for {
+      existingUser <- EitherT(registeredUserService.findByEmail(email).map {
+        case Some(_) => Left(UserExists("User already exists"))
+        case None => Right(())
+      })
+      person <- EitherT.liftF(personService.insert(Person(
+        name = name,
+        age = age
+      )))
+      passwordHash <- EitherT.liftF(IO(hasher.encode(password)))
+      registeredUser <- EitherT.liftF(registeredUserService.insert(RegisteredUser(
+        email = email,
+        passwordHash = passwordHash,
+        personId = person.id,
+        roles = List() // TODO
+      )))
+      jwtToken <- EitherT.liftF(generateTokenForRegisteredUser(registeredUser, Instant.now.plusSeconds(3600 * 30))) // 30 days
+    } yield RegisterSuccess(
+      registeredUser = registeredUser,
+      person = person,
+      jwtToken = jwtToken
+    )).value
+  }
+}