refactor(auth): move USER_LAST_ACTIVE_TIME write out of JwtAuthFilter

access-control-service/src/main/scala/org/apache/texera/service/AccessControlService.scala

@@ -26,6 +26,7 @@ import io.dropwizard.core.setup.{Bootstrap, Environment}
 import org.apache.texera.amber.config.StorageConfig
 import org.apache.texera.auth.{JwtAuthFilter, RequestLoggingFilter, SessionUser}
 import org.apache.texera.dao.SqlServer
+import org.apache.texera.service.activity.UserActivityEventListener
 import org.apache.texera.service.resource.{
   AccessControlResource,
   HealthCheckResource,
@@ -77,6 +78,12 @@ class AccessControlService extends Application[AccessControlServiceConfiguration
       new io.dropwizard.auth.AuthValueFactoryProvider.Binder(classOf[SessionUser])
     )
 
+    // Record USER_LAST_ACTIVE_TIME on every matched, completed request.
+    // Lives only in this service because authenticated client sessions
+    // contact access-control-service often enough to capture activity
+    // with high recall.
+    environment.jersey.register(new UserActivityEventListener())
+
     // Route request logs through SLF4J, controlled by TEXERA_SERVICE_LOG_LEVEL
     RequestLoggingFilter.register(environment.getApplicationContext)
   }

access-control-service/src/main/scala/org/apache/texera/service/activity/UserActivityEventListener.scala

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *   http://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.
+ */
+
+package org.apache.texera.service.activity
+
+import jakarta.ws.rs.ext.Provider
+import org.apache.texera.auth.{SessionUser, UserActivityTracker}
+import org.glassfish.jersey.server.monitoring.{
+  ApplicationEvent,
+  ApplicationEventListener,
+  RequestEvent,
+  RequestEventListener
+}
+
+/** Records user activity (USER_LAST_ACTIVE_TIME) once per matched, completed
+  * request. Intentionally NOT a ContainerRequestFilter:
+  *
+  *   - It cannot reject or transform a request — it only observes.
+  *   - It runs at Jersey's monitoring layer, not the auth pipeline, so
+  *     activity tracking is decoupled from authentication concerns.
+  *   - It listens for RESOURCE_METHOD_FINISHED only, so requests that
+  *     fail before reaching a handler (no auth, 404, 4xx in earlier
+  *     filters) do not count as user activity.
+  *
+  * The DB write itself is throttled per-uid by [[UserActivityTracker]].
+  *
+  * Lives in access-control-service because USER_LAST_ACTIVE_TIME is a
+  * user-management concern; the assumption is that any authenticated
+  * client session contacts this service often enough (UI navigation,
+  * permission checks, LiteLLM proxy) to capture activity with high
+  * recall, so other services do not need to mirror this listener.
+  */
+@Provider
+class UserActivityEventListener(track: Integer => Unit = UserActivityTracker.markActive)
+    extends ApplicationEventListener {
+
+  override def onEvent(event: ApplicationEvent): Unit = ()
+
+  // SAM-converted lambda: avoids an inner anonymous class so coverage
+  // tooling sees a flat method body. Logic lives in the companion's
+  // `handle` so tests can drive it directly.
+  override def onRequest(requestEvent: RequestEvent): RequestEventListener =
+    (event: RequestEvent) => UserActivityEventListener.handle(event, track)
+}
+
+object UserActivityEventListener {
+
+  /** Process a single Jersey request event. Public-package for tests so the
+    * per-request branching is exercised without a Jersey runtime.
+    */
+  private[activity] def handle(event: RequestEvent, track: Integer => Unit): Unit = {
+    // `eq` (reference equality) is correct here because Type is a Java enum
+    // — its constants are singletons. It also compiles to a single
+    // `if_acmpne`, sidestepping Scala's BoxesRunTime.equals branch fan-out.
+    if (!(event.getType eq RequestEvent.Type.RESOURCE_METHOD_FINISHED)) return
+    val sc = event.getContainerRequest.getSecurityContext
+    if (sc == null) return
+    sc.getUserPrincipal match {
+      case u: SessionUser if u.getUid != null => track(u.getUid)
+      case _                                  =>
+    }
+  }
+}

access-control-service/src/test/scala/org/apache/texera/service/AccessControlServiceRunSpec.scala

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *   http://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.
+ */
+
+package org.apache.texera.service
+
+import io.dropwizard.core.setup.Environment
+import io.dropwizard.jersey.setup.JerseyEnvironment
+import io.dropwizard.jetty.MutableServletContextHandler
+import io.dropwizard.jetty.setup.ServletEnvironment
+import org.apache.texera.service.activity.UserActivityEventListener
+import org.mockito.ArgumentMatchers.isA
+import org.mockito.Mockito.{mock, verify, when}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class AccessControlServiceRunSpec extends AnyFlatSpec with Matchers {
+
+  "AccessControlService.run" should "register UserActivityEventListener on the Jersey environment" in {
+    val jersey = mock(classOf[JerseyEnvironment])
+    val servlets = mock(classOf[ServletEnvironment])
+    val context = mock(classOf[MutableServletContextHandler])
+    val env = mock(classOf[Environment])
+    when(env.jersey).thenReturn(jersey)
+    when(env.servlets).thenReturn(servlets)
+    when(env.getApplicationContext).thenReturn(context)
+
+    val service = new AccessControlService
+    service.run(mock(classOf[AccessControlServiceConfiguration]), env)
+
+    verify(jersey).register(isA(classOf[UserActivityEventListener]))
+    verify(jersey).setUrlPattern("/api/*")
+  }
+}

access-control-service/src/test/scala/org/apache/texera/service/activity/UserActivityEventListenerSpec.scala

@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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
+ *
+ *   http://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.
+ */
+
+package org.apache.texera.service.activity
+
+import jakarta.ws.rs.core.SecurityContext
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
+import org.apache.texera.dao.jooq.generated.tables.pojos.User
+import org.glassfish.jersey.server.ContainerRequest
+import org.glassfish.jersey.server.monitoring.{ApplicationEvent, RequestEvent}
+import org.mockito.Mockito.{mock, when}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.security.Principal
+import java.util.concurrent.ConcurrentLinkedQueue
+
+class UserActivityEventListenerSpec extends AnyFlatSpec with Matchers {
+
+  private def sessionUser(uid: Integer): SessionUser = {
+    val u = new User(uid, "u", null, null, null, null, UserRoleEnum.REGULAR, null, null, null, null)
+    new SessionUser(u)
+  }
+
+  private def buildEvent(eventType: RequestEvent.Type, sc: SecurityContext): RequestEvent = {
+    val req = mock(classOf[ContainerRequest])
+    when(req.getSecurityContext).thenReturn(sc)
+    val event = mock(classOf[RequestEvent])
+    when(event.getType).thenReturn(eventType)
+    when(event.getContainerRequest).thenReturn(req)
+    event
+  }
+
+  private def buildSecurityContext(principal: Principal): SecurityContext = {
+    val sc = mock(classOf[SecurityContext])
+    when(sc.getUserPrincipal).thenReturn(principal)
+    sc
+  }
+
+  private def newRecorder(): ConcurrentLinkedQueue[Integer] = new ConcurrentLinkedQueue[Integer]()
+  private def trackTo(q: ConcurrentLinkedQueue[Integer]): Integer => Unit =
+    uid => { q.add(uid); () }
+
+  "UserActivityEventListener.handle" should "invoke the tracker on RESOURCE_METHOD_FINISHED with a SessionUser principal" in {
+    val recorded = newRecorder()
+    UserActivityEventListener.handle(
+      buildEvent(RequestEvent.Type.RESOURCE_METHOD_FINISHED, buildSecurityContext(sessionUser(42))),
+      trackTo(recorded)
+    )
+    recorded.size shouldBe 1
+    recorded.peek() shouldBe 42
+  }
+
+  it should "ignore RequestEvent types other than RESOURCE_METHOD_FINISHED" in {
+    val recorded = newRecorder()
+    val sc = buildSecurityContext(sessionUser(42))
+    UserActivityEventListener.handle(buildEvent(RequestEvent.Type.START, sc), trackTo(recorded))
+    UserActivityEventListener.handle(
+      buildEvent(RequestEvent.Type.RESOURCE_METHOD_START, sc),
+      trackTo(recorded)
+    )
+    UserActivityEventListener.handle(buildEvent(RequestEvent.Type.FINISHED, sc), trackTo(recorded))
+    recorded.isEmpty shouldBe true
+  }
+
+  it should "ignore non-SessionUser principals" in {
+    val recorded = newRecorder()
+    val anon: Principal = new Principal { override def getName: String = "anon" }
+    UserActivityEventListener.handle(
+      buildEvent(RequestEvent.Type.RESOURCE_METHOD_FINISHED, buildSecurityContext(anon)),
+      trackTo(recorded)
+    )
+    recorded.isEmpty shouldBe true
+  }
+
+  it should "ignore SessionUser with null uid" in {
+    val recorded = newRecorder()
+    UserActivityEventListener.handle(
+      buildEvent(
+        RequestEvent.Type.RESOURCE_METHOD_FINISHED,
+        buildSecurityContext(sessionUser(null))
+      ),
+      trackTo(recorded)
+    )
+    recorded.isEmpty shouldBe true
+  }
+
+  it should "ignore null SecurityContext" in {
+    val recorded = newRecorder()
+    UserActivityEventListener.handle(
+      buildEvent(RequestEvent.Type.RESOURCE_METHOD_FINISHED, null),
+      trackTo(recorded)
+    )
+    recorded.isEmpty shouldBe true
+  }
+
+  // Listener-level smoke tests: verify the SAM lambda + dispatch glue,
+  // not the per-event branching (which lives in `handle`).
+  "UserActivityEventListener" should "dispatch RequestEvent to the handle function" in {
+    val recorded = newRecorder()
+    val listener = new UserActivityEventListener(trackTo(recorded))
+    val rel = listener.onRequest(mock(classOf[RequestEvent]))
+    rel.onEvent(
+      buildEvent(RequestEvent.Type.RESOURCE_METHOD_FINISHED, buildSecurityContext(sessionUser(7)))
+    )
+    recorded.peek() shouldBe 7
+  }
+
+  it should "no-op on ApplicationEvent (lifecycle hook unused)" in {
+    val recorded = newRecorder()
+    val listener = new UserActivityEventListener(trackTo(recorded))
+    listener.onEvent(mock(classOf[ApplicationEvent]))
+    recorded.isEmpty shouldBe true
+  }
+
+  it should "construct with the default tracker without invoking it" in {
+    new UserActivityEventListener() should not be null
+  }
+}

common/auth/build.sbt

@@ -58,5 +58,6 @@ libraryDependencies ++= Seq(
   "org.bitbucket.b_c" % "jose4j" % "0.9.6",                             // for jwt parser
   "jakarta.ws.rs" % "jakarta.ws.rs-api" % "3.0.0",                      // for JwtAuthFilter
   "jakarta.servlet" % "jakarta.servlet-api" % "5.0.0" % "provided",    // for RequestLoggingFilter
-  "org.eclipse.jetty" % "jetty-servlet" % "11.0.24" % "provided"       // for FilterHolder
+  "org.eclipse.jetty" % "jetty-servlet" % "11.0.24" % "provided",      // for FilterHolder
+  "org.scalatest" %% "scalatest" % "3.2.17" % Test
 )

common/auth/src/main/scala/org/apache/texera/auth/JwtAuthFilter.scala

@@ -20,23 +20,16 @@
 package org.apache.texera.auth
 
 import com.typesafe.scalalogging.LazyLogging
-import jakarta.ws.rs.container.{ContainerRequestContext, ContainerRequestFilter, ResourceInfo}
-import jakarta.ws.rs.core.{Context, HttpHeaders, SecurityContext}
+import jakarta.ws.rs.container.{ContainerRequestContext, ContainerRequestFilter}
+import jakarta.ws.rs.core.{HttpHeaders, SecurityContext}
 import jakarta.ws.rs.ext.Provider
-import org.apache.texera.dao.SqlServer
-import org.apache.texera.dao.jooq.generated.Tables.USER_LAST_ACTIVE_TIME
 import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
 
 import java.security.Principal
-import java.time.OffsetDateTime
 
 @Provider
 class JwtAuthFilter extends ContainerRequestFilter with LazyLogging {
 
-  @Context
-  private var resourceInfo: ResourceInfo = _
-  private def ctx = SqlServer.getInstance().createDSLContext()
-
   override def filter(requestContext: ContainerRequestContext): Unit = {
     val authHeader = requestContext.getHeaderString(HttpHeaders.AUTHORIZATION)
 
@@ -46,16 +39,6 @@ class JwtAuthFilter extends ContainerRequestFilter with LazyLogging {
 
       if (userOpt.isPresent) {
         val user = userOpt.get()
-
-        ctx
-          .insertInto(USER_LAST_ACTIVE_TIME)
-          .set(USER_LAST_ACTIVE_TIME.UID, user.getUid)
-          .set(USER_LAST_ACTIVE_TIME.LAST_ACTIVE_TIME, OffsetDateTime.now())
-          .onConflict(USER_LAST_ACTIVE_TIME.UID) // conflict on primary key uid
-          .doUpdate()
-          .set(USER_LAST_ACTIVE_TIME.LAST_ACTIVE_TIME, OffsetDateTime.now())
-          .execute()
-
         requestContext.setSecurityContext(new SecurityContext {
           override def getUserPrincipal: Principal = user
           override def isUserInRole(role: String): Boolean =