Refactor SessionsResource for better memory usage and performance

Closes #45727

Signed-off-by: Pedro Ruivo <1492066+pruivo@users.noreply.github.com>
Signed-off-by: Alexander Schwartz <alexander.schwartz@ibm.com>
Co-authored-by: Pedro Ruivo <1492066+pruivo@users.noreply.github.com>
Co-authored-by: Alexander Schwartz <alexander.schwartz@ibm.com>

docs/documentation/upgrading/topics/changes/changes-26_6_0.adoc

@@ -40,11 +40,11 @@ See the link:{upgradingguide_link}[{upgradingguide_name}] for details on how to
 
 OIDC specification has multiple client authentication methods. Two of them `client_secret_basic` and `client_secret_post` are implemented in {project_name} by **Client Id and Secret**
 client authenticator. Until now, when the OIDC client registration request was sent with the attribute `token_endpoint_auth_method` set to `client_secret_basic` or `client_secret_post`,
-the {project_name} allowed to authenticate client with both `client_secret_basic` and `client_secret_post` methods and did not preserved the single method specified in the OIDC client
+the {project_name} allowed to authenticate client with both `client_secret_basic` and `client_secret_post` methods and did not preserve the single method specified in the OIDC client
 registration request. Now the specified method is respected, so for example when the OIDC client registration is sent with `token_endpoint_auth_method` set to `client_secret_post`, then
 it is required to authenticate the client really with the client secret sent as a parameter in the POST request body.
 
-It is still possible to make the OIDC client to allow both methods and clients migrated from previous versions are set by default to allow authentication with both methods.
+It is still possible to make the OIDC client allow both methods and clients migrated from previous versions are set by default to allow authentication with both methods.
 
 === Usage of virtual threads for embedded caches
 
@@ -56,11 +56,11 @@ This change should prevent deadlocks due to pinned virtual threads.
 
 Specific sensitive information is omitted from the HTTP Access log, such as the value of the `Authorization` HTTP header. Moreover, values of specific sensitive {project_name} cookies, such as `KEYCLOAK_SESSION`, `KEYCLOAK_IDENTITY`, or `AUTH_SESSION_ID`, are also omitted.
 
-=== Accepting URL paths without a semicolon
+=== Rejecting URL paths with semicolons
 
 Previously {project_name} accepted HTTP requests with paths containing a semicolon (`;`).
-When processing them, it handled them as of RFC 3986 as a separator for matrix parameters, which basically ignored those parts.
+When processing them, it handled them according to RFC 3986 as a separator for matrix parameters, which basically ignored those parts.
-As this has led to a hard-to-configure URL filtering, for example, in reverse proxies, this is now disabled, and {project_name} responds with an HTTP 400 response code.
+As this has led to hard-to-configure URL filtering, for example, in reverse proxies, this is now disabled, and {project_name} responds with an HTTP 400 response code.
 
 To analyze rejected requests in the server log, enable debug logging for `org.keycloak.quarkus.runtime.services.RejectNonNormalizedPathFilter`.
 
@@ -70,7 +70,7 @@ With this configuration, enable and review the HTTP access log to identify probl
 === Adjusted default clock-skew for not-before JWT token checks
 
 Previous versions of {project_name} tested the issued-at timestamp with a clock-skew of zero seconds by default if this value was provided in a token.
-Be default, access tokens issued by {project_name} used for the admin API or for the user info endpoint by default do not contain a not-before claim in the token, but customized setups might have it.
+By default, access tokens issued by {project_name} used for the admin API or for the user info endpoint do not contain a not-before claim in the token, but customized setups might have it.
 In a setup with such a claim configured and with clocks that were not fully synchronized, this could lead to rejecting tokens on a node that lagged behind, as it would consider the tokens not valid yet.
 
 Starting with this version, {project_name} will honor a clock skew of 10 seconds for the issued-at timestamp by default which is aligned with the best practices for example of the https://openid.net/specs/fapi-security-profile-2_0-final.html#section-5.3.2.1-6[FAPI 2.0 Security Profile].
@@ -84,11 +84,21 @@ All the `base` themes are now tagged as **abstract**, and they are not listed in
 
 === New database indexes on the `OFFLINE_CLIENT_SESSION` table
 
-The `OFFLINE_CLIENT_SESSION` table now contains two additional index `IDX_OFFLINE_CSS_BY_CLIENT_AND_REALM` and `IDX_OFFLINE_CSS_BY_USER_SESSION_AND_OFFLINE` to improve performance.
+The `OFFLINE_CLIENT_SESSION` table now contains two additional indexes `IDX_OFFLINE_CSS_BY_CLIENT_AND_REALM` and `IDX_OFFLINE_CSS_BY_USER_SESSION_AND_OFFLINE` to improve performance.
 
 If the table contains more than 300000 entries, {project_name} will skip the index creation by default during the automatic schema migration and instead log the SQL statement on the console during migration to be applied manually after {project_name}'s startup.
 See the link:{upgradingguide_link}[{upgradingguide_name}] for details on how to configure a different limit.
 
+=== `UserSessionProvider` API changes
+
+The internal `UserSessionProvider` interface received new methods for efficient streaming of user sessions, in both speed and memory.
+The streaming instances are read-only, meaning they will not change.
+The previous methods have been deprecated.
+Custom providers should implement these new methods to avoid performance penalties.
+They should avoid keeping the returned items in memory by for example binding them to the current transaction, in case the invoker needs to analyze all data.
+
+A similar change was done for the `UserSessionPersisterProvider`.
+
 // ------------------------ Deprecated features ------------------------ //
 == Deprecated features
 
@@ -97,7 +107,7 @@ The following sections provide details on deprecated features.
 === Deprecation of specific tracing properties in Keycloak CR
 
 The `tracing.serviceName`, and `tracing.resourceAttributes` fields of the Keycloak CR, are now deprecated.
-You should use the new `telemetry.serviceName`, and `telemetry.resourceAttributes` fields citizens that are shared among all OpenTelemetry components - logs, metrics, and traces.
+You should use the new `telemetry.serviceName`, and `telemetry.resourceAttributes` fields that are shared among all OpenTelemetry components - logs, metrics, and traces.
 
 // ------------------------ Removed features ------------------------ //
 == Removed features

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/ImmutableClientSession.java

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2026 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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
+ *
+ * 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.keycloak.models.sessions.infinispan;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserSessionModel;
+
+import static org.keycloak.models.sessions.infinispan.ImmutableSession.readOnly;
+
+/**
+ * An immutable {@link AuthenticatedClientSessionModel} implementation.
+ * <p>
+ * All setters throw a {@link UnsupportedOperationException}.
+ */
+record ImmutableClientSession(
+        String id,
+        ClientModel client,
+        ImmutableUserSessionModel userSessionModel,
+        Map<String, String> notes,
+        String redirectUri,
+        String action,
+        String protocol,
+        int timestamp,
+        int started
+) implements AuthenticatedClientSessionModel {
+
+
+    @Override
+    public String getId() {
+        return id;
+    }
+
+    @Override
+    public int getTimestamp() {
+        return timestamp;
+    }
+
+    @Override
+    public void setTimestamp(int timestamp) {
+        readOnly();
+    }
+
+    @Override
+    public int getStarted() {
+        return started;
+    }
+
+    @Override
+    public void detachFromUserSession() {
+        readOnly();
+    }
+
+    @Override
+    public UserSessionModel getUserSession() {
+        return userSessionModel;
+    }
+
+    @Override
+    public String getNote(String name) {
+        return notes.get(name);
+    }
+
+    @Override
+    public void setNote(String name, String value) {
+        readOnly();
+    }
+
+    @Override
+    public void removeNote(String name) {
+        readOnly();
+    }
+
+    @Override
+    public Map<String, String> getNotes() {
+        return Collections.unmodifiableMap(notes);
+    }
+
+    @Override
+    public String getRedirectUri() {
+        return redirectUri;
+    }
+
+    @Override
+    public void setRedirectUri(String uri) {
+        readOnly();
+    }
+
+    @Override
+    public RealmModel getRealm() {
+        return userSessionModel().getRealm();
+    }
+
+    @Override
+    public ClientModel getClient() {
+        return client;
+    }
+
+    @Override
+    public String getAction() {
+        return action;
+    }
+
+    @Override
+    public void setAction(String action) {
+        readOnly();
+    }
+
+    @Override
+    public String getProtocol() {
+        return protocol;
+    }
+
+    @Override
+    public void setProtocol(String method) {
+        readOnly();
+    }
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/ImmutableSession.java

@@ -0,0 +1,207 @@
+/*
+ * Copyright 2026 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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
+ *
+ * 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.keycloak.models.sessions.infinispan;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.sessions.infinispan.changes.SessionEntityWrapper;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.ClientSessionKey;
+import org.keycloak.models.sessions.infinispan.entities.EmbeddedClientSessionKey;
+import org.keycloak.models.sessions.infinispan.entities.RemoteAuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.RemoteUserSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
+import org.keycloak.models.sessions.infinispan.query.ClientSessionQueries;
+import org.keycloak.models.sessions.infinispan.query.QueryHelper;
+import org.keycloak.models.sessions.infinispan.util.SessionExpirationPredicates;
+import org.keycloak.utils.StreamsUtil;
+
+import org.infinispan.Cache;
+import org.infinispan.client.hotrod.RemoteCache;
+
+/**
+ * Helper class to map a list of user and client sessions, from Infinispan caches, into an immutable session.
+ * <p>
+ * It copies the data to a new instance to prevent observing changes made by other threads to the underlying cached
+ * instances.
+ */
+public final class ImmutableSession {
+
+    public static void readOnly() {
+        throw new UnsupportedOperationException("this instance is read-only");
+    }
+
+    public static Stream<UserSessionModel> copyOf(KeycloakSession session,
+                                                  Collection<UserSessionEntity> entityList,
+                                                  SessionExpirationPredicates expiration,
+                                                  Cache<EmbeddedClientSessionKey, SessionEntityWrapper<AuthenticatedClientSessionEntity>> cache) {
+        var clientSessionKeys = new HashSet<EmbeddedClientSessionKey>();
+        var userSessionMap = new LinkedHashMap<String, ImmutableUserSessionModel>();
+        var users = session.users();
+        entityList.forEach(entity -> {
+            if (entity == null) {
+                return;
+            }
+            if (!Objects.equals(entity.getRealmId(), expiration.realm().getId())) {
+                return;
+            }
+            if (expiration.isUserSessionExpired(entity)) {
+                return;
+            }
+            var user = users.getUserById(expiration.realm(), entity.getUser());
+            if (user == null) {
+                return;
+            }
+            var copy = new ImmutableUserSessionModel(
+                    entity.getId(),
+                    expiration.realm(),
+                    user,
+                    entity.getBrokerSessionId(),
+                    entity.getBrokerUserId(),
+                    entity.getLoginUsername(),
+                    entity.getIpAddress(),
+                    entity.getAuthMethod(),
+                    new HashMap<>(), // to break cyclic dependency between user and client session
+                    Map.copyOf(entity.getNotes()),
+                    entity.getState(),
+                    entity.getStarted(),
+                    entity.getLastSessionRefresh(),
+                    entity.isRememberMe(),
+                    expiration.offline()
+            );
+            entity.getClientSessions().forEach(clientId -> clientSessionKeys.add(new EmbeddedClientSessionKey(copy.id(), clientId)));
+            userSessionMap.put(copy.id(), copy);
+        });
+
+        populateClientSessions(userSessionMap, clientSessionKeys, expiration, cache);
+        return userSessionMap.values().stream()
+                .map(UserSessionModel.class::cast);
+    }
+
+    public static Stream<UserSessionModel> copyOf(KeycloakSession session,
+                                                  Collection<RemoteUserSessionEntity> entityList,
+                                                  SessionExpirationPredicates expiration,
+                                                  RemoteCache<ClientSessionKey, RemoteAuthenticatedClientSessionEntity> cache,
+                                                  int batchSize) {
+        var userSessionMap = new LinkedHashMap<String, ImmutableUserSessionModel>();
+        var users = session.users();
+        entityList.forEach(entity -> {
+            if (entity == null) {
+                return;
+            }
+            if (!Objects.equals(entity.getRealmId(), expiration.realm().getId())) {
+                return;
+            }
+            if (expiration.isUserSessionExpired(entity)) {
+                return;
+            }
+            var user = users.getUserById(expiration.realm(), entity.getUserId());
+            if (user == null) {
+                return;
+            }
+            var copy = new ImmutableUserSessionModel(
+                    entity.getUserSessionId(),
+                    expiration.realm(),
+                    user,
+                    entity.getBrokerSessionId(),
+                    entity.getBrokerUserId(),
+                    entity.getLoginUsername(),
+                    entity.getIpAddress(),
+                    entity.getAuthMethod(),
+                    new HashMap<>(), // to break cyclic dependency between user and client session
+                    Map.copyOf(entity.getNotes()),
+                    entity.getState(),
+                    entity.getStarted(),
+                    entity.getLastSessionRefresh(),
+                    entity.isRememberMe(),
+                    expiration.offline()
+            );
+            userSessionMap.put(copy.id(), copy);
+        });
+
+
+        populateClientSessions(userSessionMap, expiration, cache, batchSize);
+        return userSessionMap.values().stream().map(UserSessionModel.class::cast);
+    }
+
+    private static void populateClientSessions(Map<String, ImmutableUserSessionModel> userSessionMap, Set<EmbeddedClientSessionKey> clientSessionKeys, SessionExpirationPredicates expirationPredicates, Cache<EmbeddedClientSessionKey, SessionEntityWrapper<AuthenticatedClientSessionEntity>> cache) {
+        StreamsUtil.closing(cache.entrySet().stream()
+                        .filterKeys(clientSessionKeys))
+                .iterator()
+                .forEachRemaining(entry -> {
+                    var clientSession = entry.getValue().getEntity();
+                    var userSession = userSessionMap.get(entry.getKey().userSessionId());
+                    var client = expirationPredicates.realm().getClientById(entry.getKey().clientId());
+                    if (client == null || userSession == null) {
+                        return;
+                    }
+                    if (expirationPredicates.isClientSessionExpired(clientSession, userSession.rememberMe(), client)) {
+                        return;
+                    }
+                    var copy = new ImmutableClientSession(
+                            entry.getKey().toId(),
+                            client,
+                            userSession,
+                            Map.copyOf(clientSession.getNotes()),
+                            clientSession.getRedirectUri(),
+                            clientSession.getAction(),
+                            clientSession.getAuthMethod(),
+                            clientSession.getTimestamp(),
+                            clientSession.getStarted()
+                    );
+                    userSession.clientSessions().put(entry.getKey().clientId(), copy);
+                });
+    }
+
+    private static void populateClientSessions(Map<String, ImmutableUserSessionModel> userSessionMap, SessionExpirationPredicates expirationPredicates, RemoteCache<ClientSessionKey, RemoteAuthenticatedClientSessionEntity> cache, int batchSize) {
+        var query = ClientSessionQueries.fetchClientSessions(cache, userSessionMap.keySet());
+        QueryHelper.streamAll(query, batchSize, Function.identity()).forEach(entity -> {
+            var userSession = userSessionMap.get(entity.getUserSessionId());
+            var client = expirationPredicates.realm().getClientById(entity.getClientId());
+            if (client == null || userSession == null) {
+                return;
+            }
+            if (expirationPredicates.isClientSessionExpired(entity, userSession.started(), userSession.rememberMe(), client)) {
+                return;
+            }
+            var copy = new ImmutableClientSession(
+                    entity.createId(),
+                    client,
+                    userSession,
+                    Map.copyOf(entity.getNotes()),
+                    entity.getRedirectUri(),
+                    entity.getAction(),
+                    entity.getProtocol(),
+                    entity.getTimestamp(),
+                    entity.getStarted()
+            );
+            userSession.clientSessions().put(entity.getClientId(), copy);
+        });
+    }
+
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/ImmutableUserSessionModel.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2026 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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
+ *
+ * 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.keycloak.models.sessions.infinispan;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserModel;
+import org.keycloak.models.UserSessionModel;
+
+import static org.keycloak.models.sessions.infinispan.ImmutableSession.readOnly;
+
+/**
+ * An immutable {@link UserSessionModel} implementation.
+ * <p>
+ * All setters throw a {@link UnsupportedOperationException}.
+ */
+record ImmutableUserSessionModel(
+        String id,
+        RealmModel realm,
+        UserModel user,
+        String brokerSessionId,
+        String brokerUserId,
+        String loginUserName,
+        String ipAddress,
+        String authMethod,
+        Map<String, AuthenticatedClientSessionModel> clientSessions,
+        Map<String, String> notes,
+        State state,
+        int started,
+        int lastSessionRefresh,
+        boolean rememberMe,
+        boolean offline
+) implements UserSessionModel {
+
+    @Override
+    public String getId() {
+        return id;
+    }
+
+    @Override
+    public RealmModel getRealm() {
+        return realm;
+    }
+
+    @Override
+    public String getBrokerSessionId() {
+        return brokerSessionId;
+    }
+
+    @Override
+    public String getBrokerUserId() {
+        return brokerUserId;
+    }
+
+    @Override
+    public UserModel getUser() {
+        return user;
+    }
+
+    @Override
+    public String getLoginUsername() {
+        return loginUserName;
+    }
+
+    @Override
+    public String getIpAddress() {
+        return ipAddress;
+    }
+
+    @Override
+    public String getAuthMethod() {
+        return authMethod;
+    }
+
+    @Override
+    public boolean isRememberMe() {
+        return rememberMe;
+    }
+
+    @Override
+    public int getStarted() {
+        return started;
+    }
+
+    @Override
+    public int getLastSessionRefresh() {
+        return lastSessionRefresh;
+    }
+
+    @Override
+    public void setLastSessionRefresh(int seconds) {
+        readOnly();
+    }
+
+    @Override
+    public boolean isOffline() {
+        return offline;
+    }
+
+    @Override
+    public Map<String, AuthenticatedClientSessionModel> getAuthenticatedClientSessions() {
+        return Collections.unmodifiableMap(clientSessions);
+    }
+
+    @Override
+    public void removeAuthenticatedClientSessions(Collection<String> removedClientUUIDS) {
+        readOnly();
+    }
+
+    @Override
+    public String getNote(String name) {
+        return notes.get(name);
+    }
+
+    @Override
+    public void setNote(String name, String value) {
+        readOnly();
+    }
+
+    @Override
+    public void removeNote(String name) {
+        readOnly();
+    }
+
+    @Override
+    public Map<String, String> getNotes() {
+        return Collections.unmodifiableMap(notes);
+    }
+
+    @Override
+    public State getState() {
+        return state;
+    }
+
+    @Override
+    public void setState(State state) {
+        readOnly();
+    }
+
+    @Override
+    public void restartSession(RealmModel realm, UserModel user, String loginUsername, String ipAddress, String authMethod, boolean rememberMe, String brokerSessionId, String brokerUserId) {
+        readOnly();
+    }
+}

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/InfinispanUserSessionProvider.java

@@ -70,9 +70,11 @@ import org.keycloak.models.sessions.infinispan.stream.Mappers;
 import org.keycloak.models.sessions.infinispan.stream.SessionWrapperPredicate;
 import org.keycloak.models.sessions.infinispan.stream.UserSessionPredicate;
 import org.keycloak.models.sessions.infinispan.util.FuturesHelper;
+import org.keycloak.models.sessions.infinispan.util.SessionExpirationPredicates;
 import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
 import org.keycloak.utils.StreamsUtil;
 
+import io.reactivex.rxjava3.core.Flowable;
 import org.infinispan.Cache;
 import org.infinispan.commons.api.AsyncCache;
 import org.infinispan.commons.util.concurrent.CompletionStages;
@@ -466,6 +468,48 @@ public class InfinispanUserSessionProvider implements UserSessionProvider, Sessi
                 .sorted(Comparator.comparing(UserSessionModel::getLastSessionRefresh))), firstResult, maxResults);
     }
 
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm) {
+        var expiration = new SessionExpirationPredicates(realm, true, Time.currentTime());
+        return session.getProvider(UserSessionPersisterProvider.class).readOnlyUserSessionStream(realm, true)
+                .filter(Predicate.not(expiration::isUserSessionExpired));
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm) {
+        return readOnlyStreamFromCache(UserSessionPredicate.create(realm.getId()), realm);
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm, ClientModel client) {
+        var expiration = new SessionExpirationPredicates(realm, true, Time.currentTime());
+        return session.getProvider(UserSessionPersisterProvider.class).readOnlyUserSessionStream(realm, client, true)
+                .filter(Predicate.not(expiration::isUserSessionExpired));
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm, ClientModel client) {
+        return readOnlyStreamFromCache(UserSessionPredicate.create(realm.getId()).client(client.getId()), realm);
+    }
+
+    private Stream<UserSessionModel> readOnlyStreamFromCache(UserSessionPredicate cachePredicate, RealmModel realm) {
+        var predicates = new SessionExpirationPredicates(realm, false, Time.currentTime());
+        var clientSessionCache = getClientSessionCache(false);
+
+        // not great, distributed sort not supported, and we're sorting everything locally
+        // follow-up, iterate by segment and sort the sessions in that segment.
+        var stream = StreamsUtil.closing(getCache(false).entrySet().stream()
+                .filter(cachePredicate)
+                .map(Mappers.userSessionEntity())
+                .sorted(Comparator.comparing(UserSessionEntity::getId)));
+
+        return Flowable.fromIterable(stream::iterator)
+                .buffer(512)
+                .map(us -> ImmutableSession.copyOf(session, us, predicates, clientSessionCache))
+                .blockingStream()
+                .flatMap(Function.identity());
+    }
+
     @Override
     public UserSessionModel getUserSessionWithPredicate(RealmModel realm, String id, boolean offline, Predicate<UserSessionModel> predicate) {
         UserSessionModel userSession = getUserSession(realm, id, offline);

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/PersistentUserSessionProvider.java

@@ -75,6 +75,7 @@ import org.keycloak.models.sessions.infinispan.stream.RemoveKeyConsumer;
 import org.keycloak.models.sessions.infinispan.stream.SessionWrapperPredicate;
 import org.keycloak.models.sessions.infinispan.stream.UserSessionPredicate;
 import org.keycloak.models.sessions.infinispan.util.FuturesHelper;
+import org.keycloak.models.sessions.infinispan.util.SessionExpirationPredicates;
 import org.keycloak.models.sessions.infinispan.util.SessionTimeouts;
 import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.models.utils.UserModelDelegate;
@@ -414,6 +415,38 @@ public class PersistentUserSessionProvider implements UserSessionProvider, Sessi
         session.getProvider(UserSessionPersisterProvider.class).removeUserSessions(realm);
     }
 
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm) {
+        return readOnlyStream(realm, false);
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm) {
+        return readOnlyStream(realm, true);
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm, ClientModel client) {
+        return readOnlyStream(realm, client, true);
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm, ClientModel client) {
+        return readOnlyStream(realm, client, false);
+    }
+
+    private Stream<UserSessionModel> readOnlyStream(RealmModel realm, boolean offline) {
+        var expiration = new SessionExpirationPredicates(realm, offline, Time.currentTime());
+        return session.getProvider(UserSessionPersisterProvider.class).readOnlyUserSessionStream(realm, offline)
+                .filter(Predicate.not(expiration::isUserSessionExpired));
+    }
+
+    private Stream<UserSessionModel> readOnlyStream(RealmModel realm, ClientModel client, boolean offline) {
+        var expiration = new SessionExpirationPredicates(realm, offline, Time.currentTime());
+        return session.getProvider(UserSessionPersisterProvider.class).readOnlyUserSessionStream(realm, client, offline)
+                .filter(Predicate.not(expiration::isUserSessionExpired));
+    }
+
     protected void onRemoveUserSessionsEvent(String realmId) {
         removeLocalUserSessions(realmId, false);
         removeLocalUserSessions(realmId, true);

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/query/ClientSessionQueries.java

@@ -17,6 +17,8 @@
 
 package org.keycloak.models.sessions.infinispan.query;
 
+import java.util.Collection;
+
 import org.keycloak.marshalling.Marshalling;
 import org.keycloak.models.sessions.infinispan.entities.ClientSessionKey;
 import org.keycloak.models.sessions.infinispan.entities.RemoteAuthenticatedClientSessionEntity;
@@ -38,6 +40,7 @@ public final class ClientSessionQueries {
     private static final String PER_CLIENT_COUNT = "SELECT e.clientId, count(e.clientId) FROM %s as e WHERE e.realmId = :realmId GROUP BY e.clientId ORDER BY e.clientId".formatted(CLIENT_SESSION);
     private static final String CLIENT_SESSION_COUNT = "SELECT count(e) FROM %s as e WHERE e.realmId = :realmId && e.clientId = :clientId".formatted(CLIENT_SESSION);
     private static final String FROM_USER_SESSION = "FROM %s as e WHERE e.userSessionId = :userSessionId ORDER BY e.clientId".formatted(CLIENT_SESSION);
+    private static final String FROM_MULTI_USER_SESSION = "FROM %s as e WHERE e.userSessionId IN (:userSessionIds) ORDER BY e.clientId".formatted(CLIENT_SESSION);
     private static final String IDS_FROM_USER_SESSION = "SELECT e.clientId FROM %s as e WHERE e.userSessionId = :userSessionId ORDER BY e.clientId".formatted(CLIENT_SESSION);
 
     /**
@@ -58,7 +61,7 @@ public final class ClientSessionQueries {
     }
 
     /**
-     * Returns a projection with the sum of all client session belonging to the client ID.
+     * Returns a projection with the sum of all client sessions belonging to the client ID.
      */
     public static Query<Object[]> countClientSessions(RemoteCache<ClientSessionKey, RemoteAuthenticatedClientSessionEntity> cache, String realmId, String clientId) {
         return cache.<Object[]>query(CLIENT_SESSION_COUNT)
@@ -67,8 +70,7 @@ public final class ClientSessionQueries {
     }
 
     /**
-     * Returns a projection with the client session, and the version of all client sessions belonging to the user
+     * Returns the client sessions belonging to the user session ID.
-     * session ID.
      */
     public static Query<RemoteAuthenticatedClientSessionEntity> fetchClientSessions(RemoteCache<ClientSessionKey, RemoteAuthenticatedClientSessionEntity> cache, String userSessionId) {
         return cache.<RemoteAuthenticatedClientSessionEntity>query(FROM_USER_SESSION)
@@ -85,5 +87,12 @@ public final class ClientSessionQueries {
                 .setParameter("userSessionId", userSessionId);
     }
 
+    /**
+     * Returns all the client sessions belonging to all user session IDs.
+     */
+    public static Query<RemoteAuthenticatedClientSessionEntity> fetchClientSessions(RemoteCache<ClientSessionKey, RemoteAuthenticatedClientSessionEntity> cache, Collection<String> userSessionIds) {
+        return cache.<RemoteAuthenticatedClientSessionEntity>query(FROM_MULTI_USER_SESSION)
+                .setParameter("userSessionIds", userSessionIds);
+    }
 
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/query/UserSessionQueries.java

@@ -37,10 +37,10 @@ public final class UserSessionQueries {
     private static final String BY_BROKER_SESSION_ID = BASE_QUERY + "WHERE e.realmId = :realmId && e.brokerSessionId = :brokerSessionId ORDER BY e.userSessionId";
     private static final String BY_USER_ID = BASE_QUERY + "WHERE e.realmId = :realmId && e.userId = :userId ORDER BY e.userSessionId";
     private static final String BY_BROKER_USER_ID = BASE_QUERY + "WHERE e.realmId = :realmId && e.brokerUserId = :brokerUserId ORDER BY e.userSessionId";
+    private static final String BY_REALM = BASE_QUERY + "WHERE e.realmId = :realmId ORDER BY e.userSessionId";
 
     /**
-     * Returns a projection with the user session, and the version of all user sessions belonging to the broker session
+     * Returns all user sessions belonging to the broker session ID.
-     * ID.
      */
     public static Query<RemoteUserSessionEntity> searchByBrokerSessionId(RemoteCache<String, RemoteUserSessionEntity> cache, String realmId, String brokerSessionId) {
         return cache.<RemoteUserSessionEntity>query(BY_BROKER_SESSION_ID)
@@ -49,7 +49,7 @@ public final class UserSessionQueries {
     }
 
     /**
-     * Returns a projection with the user session, and the version of all user sessions belonging to the user ID.
+     * Returns all user sessions belonging to the user ID.
      */
     public static Query<RemoteUserSessionEntity> searchByUserId(RemoteCache<String, RemoteUserSessionEntity> cache, String realmId, String userId) {
         return cache.<RemoteUserSessionEntity>query(BY_USER_ID)
@@ -58,12 +58,19 @@ public final class UserSessionQueries {
     }
 
     /**
-     * Returns a projection with the user session, and the version of all user sessions belonging to the broker user
+     * Returns all user sessions belonging to the broker user ID.
-     * ID.
      */
     public static Query<RemoteUserSessionEntity> searchByBrokerUserId(RemoteCache<String, RemoteUserSessionEntity> cache, String realmId, String brokerUserId) {
         return cache.<RemoteUserSessionEntity>query(BY_BROKER_USER_ID)
                 .setParameter("realmId", realmId)
                 .setParameter("brokerUserId", brokerUserId);
     }
+
+    /**
+     * Returns all the user sessions belonging to the Realm.
+     */
+    public static Query<RemoteUserSessionEntity> searchByRealm(RemoteCache<String, RemoteUserSessionEntity> cache, String realmId) {
+        return cache.<RemoteUserSessionEntity>query(BY_REALM)
+                .setParameter("realmId", realmId);
+    }
 }

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/remote/RemoteUserSessionProvider.java

@@ -34,6 +34,7 @@ import java.util.stream.Stream;
 import org.keycloak.cluster.ClusterProvider;
 import org.keycloak.common.Profile;
 import org.keycloak.common.util.SecretGenerator;
+import org.keycloak.common.util.Time;
 import org.keycloak.models.AuthenticatedClientSessionModel;
 import org.keycloak.models.ClientModel;
 import org.keycloak.models.KeycloakSession;
@@ -44,6 +45,7 @@ import org.keycloak.models.UserSessionModel;
 import org.keycloak.models.UserSessionProvider;
 import org.keycloak.models.light.LightweightUserAdapter;
 import org.keycloak.models.session.UserSessionPersisterProvider;
+import org.keycloak.models.sessions.infinispan.ImmutableSession;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.BaseUpdater;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.client.AuthenticatedClientSessionUpdater;
 import org.keycloak.models.sessions.infinispan.changes.remote.updater.user.AuthenticatedClientSessionMapping;
@@ -57,10 +59,12 @@ import org.keycloak.models.sessions.infinispan.query.UserSessionQueries;
 import org.keycloak.models.sessions.infinispan.remote.transaction.ClientSessionChangeLogTransaction;
 import org.keycloak.models.sessions.infinispan.remote.transaction.UserSessionChangeLogTransaction;
 import org.keycloak.models.sessions.infinispan.remote.transaction.UserSessionTransaction;
+import org.keycloak.models.sessions.infinispan.util.SessionExpirationPredicates;
 import org.keycloak.models.utils.KeycloakModelUtils;
 import org.keycloak.utils.StreamsUtil;
 
 import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
 import org.infinispan.client.hotrod.RemoteCache;
 import org.infinispan.commons.util.concurrent.CompletionStages;
 import org.jboss.logging.Logger;
@@ -278,6 +282,49 @@ public class RemoteUserSessionProvider implements UserSessionProvider {
 
     }
 
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm) {
+        return readOnlyStream(realm, true);
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm) {
+        return readOnlyStream(realm, false);
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm, ClientModel client) {
+        return readOnlyStream(realm, client, true);
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm, ClientModel client) {
+        return readOnlyStream(realm, client, false);
+    }
+
+    private Stream<UserSessionModel> readOnlyStream(RealmModel realm, boolean offline) {
+        var expiration = new SessionExpirationPredicates(realm, offline, Time.currentTime());
+        var query = UserSessionQueries.searchByRealm(getUserSessionTransaction(offline).getCache(), realm.getId());
+        var clientSessionCache = getClientSessionTransaction(offline).getCache();
+        //not very efficient at all.
+        return Flowable.fromStream(QueryHelper.streamAll(query, batchSize, Function.identity()))
+                .buffer(batchSize)
+                .blockingStream()
+                .flatMap(us -> ImmutableSession.copyOf(session, us, expiration, clientSessionCache, batchSize));
+    }
+
+    private Stream<UserSessionModel> readOnlyStream(RealmModel realm, ClientModel client, boolean offline) {
+        var expiration = new SessionExpirationPredicates(realm, offline, Time.currentTime());
+        var userSessionIdQuery = ClientSessionQueries.fetchUserSessionIdForClientId(getClientSessionTransaction(offline).getCache(), realm.getId(), client.getId());
+        var userSessionCache = getUserSessionTransaction(offline).getCache();
+        var clientSessionCache = getClientSessionTransaction(offline).getCache();
+        return Flowable.fromIterable(QueryHelper.toCollection(userSessionIdQuery, QueryHelper.SINGLE_PROJECTION_TO_STRING))
+                .buffer(batchSize)
+                .flatMapMaybe(sessionId -> Maybe.fromCompletionStage(userSessionCache.getAllAsync(Set.copyOf(sessionId))), false, MAX_CONCURRENT_REQUESTS)
+                .blockingStream(batchSize)
+                .flatMap(entity -> ImmutableSession.copyOf(session, entity.values(), expiration, clientSessionCache, batchSize));
+    }
+
     private void migrateUserSessions(boolean offline) {
         log.info("Migrate user sessions from database to the remote cache");
 

model/infinispan/src/main/java/org/keycloak/models/sessions/infinispan/util/SessionExpirationPredicates.java

@@ -0,0 +1,65 @@
+package org.keycloak.models.sessions.infinispan.util;
+
+import org.keycloak.models.AuthenticatedClientSessionModel;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserSessionModel;
+import org.keycloak.models.sessions.infinispan.entities.AuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.RemoteAuthenticatedClientSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.RemoteUserSessionEntity;
+import org.keycloak.models.sessions.infinispan.entities.UserSessionEntity;
+import org.keycloak.models.utils.SessionExpirationUtils;
+
+/**
+ * Utility record to check if a user or client session is expired. It handles all the current entities, from JPA or from
+ * caching.
+ *
+ * @param realm       The {@link RealmModel} to fetch the max-idle and lifespan settings.
+ * @param offline     Indicates whether the sessions are offline.
+ * @param currentTime The current time value.
+ */
+public record SessionExpirationPredicates(RealmModel realm, boolean offline, int currentTime) {
+
+    public boolean isUserSessionExpired(UserSessionModel model) {
+        return isUserSessionExpired(model.isRememberMe(), model.getStarted(), model.getLastSessionRefresh());
+    }
+
+    public boolean isUserSessionExpired(UserSessionEntity entity) {
+        return isUserSessionExpired(entity.isRememberMe(), entity.getStarted(), entity.getLastSessionRefresh());
+    }
+
+    public boolean isUserSessionExpired(RemoteUserSessionEntity entity) {
+        return isUserSessionExpired(entity.isRememberMe(), entity.getStarted(), entity.getLastSessionRefresh());
+    }
+
+    public boolean isClientSessionExpired(AuthenticatedClientSessionModel model) {
+        return isClientSessionExpired(model.getUserSession().isRememberMe(), model.getStarted(), model.getUserSessionStarted(), model.getTimestamp(), model.getClient());
+    }
+
+    public boolean isClientSessionExpired(AuthenticatedClientSessionEntity entity, boolean rememberMe, ClientModel client) {
+        return isClientSessionExpired(rememberMe, entity.getStarted(), entity.getUserSessionStarted(), entity.getTimestamp(), client);
+    }
+
+    public boolean isClientSessionExpired(RemoteAuthenticatedClientSessionEntity entity, int userSessionStarted, boolean rememberMe, ClientModel client) {
+        return isClientSessionExpired(rememberMe, entity.getStarted(), userSessionStarted, entity.getTimestamp(), client);
+    }
+
+    private boolean isUserSessionExpired(boolean rememberMe, long started, long lastRefresh) {
+        var lifespan = SessionExpirationUtils.calculateUserSessionMaxLifespanTimestamp(offline, rememberMe, started, realm);
+        var maxIdle = SessionExpirationUtils.calculateUserSessionIdleTimestamp(offline, rememberMe, lastRefresh, realm);
+        return isExpired(lifespan, maxIdle);
+    }
+
+    private boolean isClientSessionExpired(boolean rememberMe, long started, long userSessionStarted, long lastRefresh, ClientModel client) {
+        var lifespan = SessionExpirationUtils.calculateClientSessionMaxLifespanTimestamp(offline, rememberMe, started, userSessionStarted, realm, client);
+        var maxIdle = SessionExpirationUtils.calculateClientSessionIdleTimestamp(offline, rememberMe, lastRefresh, realm, client);
+        return isExpired(lifespan, maxIdle);
+    }
+
+    private boolean isExpired(long lifespanTimestamp, long maxIdleTimestamp) {
+        var maxIdleExpired = maxIdleTimestamp - currentTime <= 0;
+        return lifespanTimestamp == -1 ?
+                maxIdleExpired :
+                maxIdleExpired || lifespanTimestamp - currentTime <= 0;
+    }
+}

model/jpa/src/main/java/org/keycloak/models/jpa/session/ImmutablePersistentClientSessionEntity.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2026 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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
+ *
+ * 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.keycloak.models.jpa.session;
+
+import org.keycloak.models.session.PersistentClientSessionModel;
+import org.keycloak.storage.StorageId;
+
+/**
+ * An immutable {@link PersistentClientSessionEntity} to optimize read-only queries.
+ */
+public record ImmutablePersistentClientSessionEntity(
+        String userSessionId,
+        String clientId,
+        String clientStorageProvider,
+        String externalClientId,
+        String offline,
+        String data,
+        String realmId,
+        int timestamp
+) implements PersistentClientSessionModel {
+    @Override
+    public String getUserSessionId() {
+        return userSessionId;
+    }
+
+    @Override
+    public void setUserSessionId(String userSessionId) {
+        readOnly();
+    }
+
+    @Override
+    public String getClientId() {
+        return externalClientId.equals(PersistentClientSessionEntity.LOCAL) ?
+                clientId :
+                new StorageId(clientStorageProvider, externalClientId).getId();
+    }
+
+    @Override
+    public void setClientId(String clientId) {
+        readOnly();
+    }
+
+    @Override
+    public int getTimestamp() {
+        return timestamp;
+    }
+
+    @Override
+    public void setTimestamp(int timestamp) {
+        readOnly();
+    }
+
+    @Override
+    public String getData() {
+        return data;
+    }
+
+    @Override
+    public void setData(String data) {
+        readOnly();
+    }
+
+    private static void readOnly() {
+        throw new UnsupportedOperationException("this instance is read-only");
+    }
+}

model/jpa/src/main/java/org/keycloak/models/jpa/session/ImmutablePersistentUserSessionEntity.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2026 Red Hat, Inc. and/or its affiliates
+ * and other contributors as indicated by the @author tags.
+ *
+ * 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
+ *
+ * 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.keycloak.models.jpa.session;
+
+import org.keycloak.models.session.PersistentUserSessionModel;
+
+/**
+ * An immutable {@link PersistentUserSessionEntity} to optimize read-only queries.
+ */
+public record ImmutablePersistentUserSessionEntity(
+        String userSessionId,
+        String realmId,
+        String userId,
+        int createOn,
+        int lastSessionRefresh,
+        String brokerSessionId,
+        String offline,
+        String data,
+        Boolean rememberMe
+) implements PersistentUserSessionModel {
+    @Override
+    public String getUserSessionId() {
+        return userSessionId;
+    }
+
+    @Override
+    public void setUserSessionId(String userSessionId) {
+        readOnly();
+    }
+
+    @Override
+    public int getStarted() {
+        return createOn;
+    }
+
+    @Override
+    public void setStarted(int started) {
+        readOnly();
+    }
+
+    @Override
+    public int getLastSessionRefresh() {
+        return lastSessionRefresh;
+    }
+
+    @Override
+    public void setLastSessionRefresh(int lastSessionRefresh) {
+        readOnly();
+    }
+
+    @Override
+    public boolean isOffline() {
+        return JpaSessionUtil.offlineFromString(offline);
+    }
+
+    @Override
+    public void setOffline(boolean offline) {
+        readOnly();
+    }
+
+    @Override
+    public String getData() {
+        return data;
+    }
+
+    @Override
+    public void setData(String data) {
+        readOnly();
+    }
+
+    @Override
+    public void setRealmId(String realmId) {
+        readOnly();
+    }
+
+    @Override
+    public void setUserId(String userId) {
+        readOnly();
+    }
+
+    @Override
+    public void setBrokerSessionId(String brokerSessionId) {
+        readOnly();
+    }
+
+    @Override
+    public boolean isRememberMe() {
+        return rememberMe == Boolean.TRUE;
+    }
+
+    @Override
+    public void setRememberMe(boolean rememberMe) {
+        readOnly();
+    }
+
+    private static void readOnly() {
+        throw new UnsupportedOperationException("this instance is read-only");
+    }
+}

model/jpa/src/main/java/org/keycloak/models/jpa/session/JpaUserSessionPersisterProvider.java

@@ -54,6 +54,7 @@ import org.keycloak.models.utils.SessionExpirationUtils;
 import org.keycloak.storage.StorageId;
 import org.keycloak.utils.StreamsUtil;
 
+import org.hibernate.jpa.HibernateHints;
 import org.jboss.logging.Logger;
 
 import static org.keycloak.models.jpa.PaginationUtils.paginateQuery;
@@ -412,6 +413,35 @@ public class JpaUserSessionPersisterProvider implements UserSessionPersisterProv
                 .orElse(null);
     }
 
+    @Override
+    public Stream<UserSessionModel> readOnlyUserSessionStream(RealmModel realm, boolean offline) {
+        var query = em.createNamedQuery("findUserSessionsByRealmAndTypeReadOnly", ImmutablePersistentUserSessionEntity.class)
+                .setHint(HibernateHints.HINT_READ_ONLY, true)
+                .setHint(HibernateHints.HINT_FETCH_SIZE, 512)
+                .setParameter("realmId", realm.getId())
+                .setParameter("offline", offlineToString(offline))
+                .setParameter("lastSessionRefresh", calculateOldestSessionTime(realm, offline));
+        return readOnlyLoadExactUserSessionsWithClientSessions(query, realm, offlineToString(offline));
+    }
+
+
+    @Override
+    public Stream<UserSessionModel> readOnlyUserSessionStream(RealmModel realm, ClientModel client, boolean offline) {
+        var clientStorageId = new StorageId(client.getId());
+        var query = clientStorageId.isLocal() ?
+                em.createNamedQuery("findUserSessionsByClientIdReadOnly", ImmutablePersistentUserSessionEntity.class)
+                        .setParameter("clientId", client.getId()) :
+                em.createNamedQuery("findUserSessionsByExternalClientIdReadOnly", ImmutablePersistentUserSessionEntity.class)
+                        .setParameter("clientStorageProvider", clientStorageId.getProviderId())
+                        .setParameter("externalClientId", clientStorageId.getExternalId());
+        query.setParameter("offline", offlineToString(offline))
+                .setParameter("realmId", realm.getId())
+                .setParameter("lastSessionRefresh", calculateOldestSessionTime(realm, offline))
+                .setHint(HibernateHints.HINT_READ_ONLY, true)
+                .setHint(HibernateHints.HINT_FETCH_SIZE, 512);
+        return readOnlyLoadExactUserSessionsWithClientSessions(query, realm, offlineToString(offline));
+    }
+
     /**
      * Only client sessions from the user sessions obtained from the {@code query} are loaded.
      */
@@ -436,7 +466,50 @@ public class JpaUserSessionPersisterProvider implements UserSessionPersisterProv
 
                     removedClientUUIDs.forEach(this::onClientRemoved);
 
-                    logger.tracef("Loaded %d batch of user sessions (offline=%s, sessionIds=%s)", batchedUserSessions.size(), offlineStr, sessionsById.keySet());
+                    if (logger.isTraceEnabled()) {
+                        logger.tracef("Loaded %d batch of user sessions (offline=%s, sessionIds=%s)", batchedUserSessions.size(), offlineStr, sessionsById.keySet());
+                    }
+
+                    return batchedUserSessions.stream();
+                }).map(UserSessionModel.class::cast));
+    }
+
+    private Stream<UserSessionModel> readOnlyLoadExactUserSessionsWithClientSessions(TypedQuery<ImmutablePersistentUserSessionEntity> query, RealmModel realm, String offlineStr) {
+        // Take the results returned by the database in chunks and enrich them.
+        // The chunking avoids loading all the entries, as the caller usually adds pagination for the frontend.
+        var stream = StreamsUtil.closing(query.getResultStream())
+                .map(entity -> new PersistentUserSessionAdapter(session, entity, realm, entity.userId(), new HashMap<>()));
+        return closing(StreamsUtil.chunkedStream(stream, 100)
+                .flatMap(batchedUserSessions -> {
+
+                    var sessionsById = batchedUserSessions.stream()
+                            .collect(Collectors.toMap(UserSessionModel::getId, Function.identity()));
+
+                    var queryClientSessions = em.createNamedQuery("findClientSessionsOrderedByIdExactReadOnly", ImmutablePersistentClientSessionEntity.class)
+                            .setParameter("offline", offlineStr)
+                            .setParameter("userSessionIds", sessionsById.keySet())
+                            .setHint(HibernateHints.HINT_READ_ONLY, true)
+                            .setHint(HibernateHints.HINT_FETCH_SIZE, 512);
+                    var clientStream = closing(queryClientSessions.getResultStream());
+
+                    clientStream.forEach(clientSessionEntity -> {
+                        var userSession = sessionsById.get(clientSessionEntity.getUserSessionId());
+                        // check if we have a user session for the client session
+                        if (userSession == null) {
+                            return;
+                        }
+                        var client = realm.getClientById(clientSessionEntity.getClientId());
+                        if (client == null) {
+                            return;
+                        }
+
+                        var adapter = new PersistentAuthenticatedClientSessionAdapter(session, clientSessionEntity, realm, client, userSession);
+                        userSession.getAuthenticatedClientSessions().put(clientSessionEntity.getClientId(), adapter);
+                    });
+
+                    if (logger.isTraceEnabled()) {
+                        logger.tracef("Loaded %d batch of user sessions (offline=%s, sessionIds=%s)", batchedUserSessions.size(), offlineStr, sessionsById.keySet());
+                    }
 
                     return batchedUserSessions.stream();
                 }).map(UserSessionModel.class::cast));

model/jpa/src/main/java/org/keycloak/models/jpa/session/PersistentClientSessionEntity.java

@@ -56,6 +56,11 @@ import org.hibernate.annotations.DynamicUpdate;
                 " FROM PersistentClientSessionEntity sess" +
                 " WHERE sess.offline = :offline AND sess.realmId = :realmId" +
                 " GROUP BY sess.clientId, sess.externalClientId, sess.clientStorageProvider"),
+        @NamedQuery(name = "findClientSessionsOrderedByIdExactReadOnly",
+                query = "SELECT new org.keycloak.models.jpa.session.ImmutablePersistentClientSessionEntity(sess.userSessionId, sess.clientId, sess.clientStorageProvider, sess.externalClientId, sess.offline, sess.data, sess.realmId, sess.timestamp)" +
+                        " FROM PersistentClientSessionEntity sess" +
+                        " WHERE sess.offline = :offline AND sess.userSessionId IN (:userSessionIds)"
+        ),
 })
 @Table(name="OFFLINE_CLIENT_SESSION")
 @Entity

model/jpa/src/main/java/org/keycloak/models/jpa/session/PersistentUserSessionEntity.java

@@ -95,6 +95,21 @@ import org.hibernate.annotations.DynamicUpdate;
                 query = "SELECT sess.userSessionId, sess.userId" +
                         " FROM PersistentUserSessionEntity sess" +
                         " WHERE sess.realmId = :realmId AND sess.offline = '0' AND sess.rememberMe = true"),
+        @NamedQuery(name = "findUserSessionsByRealmAndTypeReadOnly",
+                query = "SELECT new org.keycloak.models.jpa.session.ImmutablePersistentUserSessionEntity(sess.userSessionId, sess.realmId, sess.userId, sess.createdOn, sess.lastSessionRefresh, sess.brokerSessionId, sess.offline, sess.data, sess.rememberMe)" +
+                        " FROM PersistentUserSessionEntity sess" +
+                        " WHERE sess.realmId = :realmId AND sess.offline = :offline AND sess.lastSessionRefresh >= :lastSessionRefresh" +
+                        " ORDER BY sess.userSessionId"),
+        @NamedQuery(name = "findUserSessionsByClientIdReadOnly",
+                query = "SELECT new org.keycloak.models.jpa.session.ImmutablePersistentUserSessionEntity(sess.userSessionId, sess.realmId, sess.userId, sess.createdOn, sess.lastSessionRefresh, sess.brokerSessionId, sess.offline, sess.data, sess.rememberMe)" +
+                        " FROM PersistentUserSessionEntity sess INNER JOIN PersistentClientSessionEntity clientSess " +
+                        " ON sess.userSessionId = clientSess.userSessionId AND sess.offline = clientSess.offline AND clientSess.clientId = :clientId WHERE sess.offline = :offline " +
+                        " AND sess.realmId = :realmId AND sess.lastSessionRefresh >= :lastSessionRefresh ORDER BY sess.userSessionId"),
+        @NamedQuery(name = "findUserSessionsByExternalClientIdReadOnly",
+                query = "SELECT new org.keycloak.models.jpa.session.ImmutablePersistentUserSessionEntity(sess.userSessionId, sess.realmId, sess.userId, sess.createdOn, sess.lastSessionRefresh, sess.brokerSessionId, sess.offline, sess.data, sess.rememberMe)" +
+                        " FROM PersistentUserSessionEntity sess INNER JOIN PersistentClientSessionEntity clientSess " +
+                        " ON sess.userSessionId = clientSess.userSessionId AND clientSess.clientStorageProvider = :clientStorageProvider AND sess.offline = clientSess.offline AND clientSess.externalClientId = :externalClientId WHERE sess.offline = :offline " +
+                        " AND sess.realmId = :realmId AND sess.lastSessionRefresh >= :lastSessionRefresh ORDER BY sess.userSessionId"),
 
 })
 @Table(name="OFFLINE_USER_SESSION")

model/storage-private/src/main/java/org/keycloak/models/session/DisabledUserSessionPersisterProvider.java

@@ -150,4 +150,14 @@ public class DisabledUserSessionPersisterProvider implements UserSessionPersiste
     public Map<String, Long> getUserSessionsCountsByClients(RealmModel realm, boolean offline) {
         return Collections.emptyMap();
     }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyUserSessionStream(RealmModel realm, boolean offline) {
+        return Stream.empty();
+    }
+
+    @Override
+    public Stream<UserSessionModel> readOnlyUserSessionStream(RealmModel realm, ClientModel client, boolean offline) {
+        return Stream.empty();
+    }
 }

model/storage-private/src/main/java/org/keycloak/models/session/UserSessionPersisterProvider.java

@@ -155,4 +155,28 @@ public interface UserSessionPersisterProvider extends Provider {
         removeUserSessions(realm, false);
     }
 
+    /**
+     * Stream all the sessions belonging to the realm.
+     * <p>
+     * The returned {@link UserSessionModel} instances are immutable. More precisely, the entity is not tracked by the JPA and any
+     * modification may throw an {@link UnsupportedOperationException}.
+     *
+     * @param realm   The {@link RealmModel} instance.
+     * @param offline If {@code true}, it streams the offline sessions, otherwise the regular sessions.
+     * @return A {@link Stream} for all the sessions in the realm.
+     */
+    Stream<UserSessionModel> readOnlyUserSessionStream(RealmModel realm, boolean offline);
+
+    /**
+     * Stream all the sessions belonging to the realm and having a client session from the client.
+     * <p>
+     * The returned {@link UserSessionModel} instances are immutable. More precisely, the entity is not tracked by the JPA and any
+     * modification may throw an {@link UnsupportedOperationException}.
+     *
+     * @param realm   The {@link RealmModel} instance.
+     * @param client  The {@link ClientModel} instance.
+     * @param offline If {@code true}, it streams the offline sessions, otherwise the regular sessions.
+     * @return A {@link Stream} for all the sessions matching the parameters.
+     */
+    Stream<UserSessionModel> readOnlyUserSessionStream(RealmModel realm, ClientModel client, boolean offline);
 }

rest/admin-ui-ext/src/main/java/org/keycloak/admin/ui/rest/SessionsResource.java

@@ -1,6 +1,5 @@
 package org.keycloak.admin.ui.rest;
 
-import java.util.Map;
 import java.util.stream.Stream;
 
 import jakarta.ws.rs.Consumes;
@@ -10,7 +9,6 @@ import jakarta.ws.rs.Path;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.QueryParam;
 
-import org.keycloak.admin.ui.rest.model.ClientIdSessionType;
 import org.keycloak.admin.ui.rest.model.ClientIdSessionType.SessionType;
 import org.keycloak.admin.ui.rest.model.SessionRepresentation;
 import org.keycloak.common.util.Time;
@@ -29,7 +27,6 @@ import org.eclipse.microprofile.openapi.annotations.media.Content;
 import org.eclipse.microprofile.openapi.annotations.media.Schema;
 import org.eclipse.microprofile.openapi.annotations.responses.APIResponse;
 
-import static org.keycloak.admin.ui.rest.model.ClientIdSessionType.SessionType.ALL;
 import static org.keycloak.admin.ui.rest.model.ClientIdSessionType.SessionType.OFFLINE;
 import static org.keycloak.admin.ui.rest.model.ClientIdSessionType.SessionType.REGULAR;
 
@@ -67,36 +64,13 @@ public class SessionsResource {
                                                        @QueryParam("max") @DefaultValue("10") int max) {
         auth.realm().requireViewRealm();
 
-        Stream<ClientIdSessionType> sessionIdStream = Stream.<ClientIdSessionType>builder().build();
+        var stream = switch (type) {
-        if (type == ALL || type == REGULAR) {
+            case OFFLINE -> streamOffline();
-            final Map<String, Long> clientSessionStats = session.sessions().getActiveClientSessionStats(realm, false);
+            case REGULAR -> streamRegular();
-            sessionIdStream = Stream.concat(sessionIdStream, clientSessionStats
+            case ALL -> Stream.concat(streamRegular(), streamOffline());
-                    .keySet().stream().map(i -> new ClientIdSessionType(i, REGULAR)));
+        };
-        }
-        if (type == ALL || type == OFFLINE) {
-            sessionIdStream = Stream.concat(sessionIdStream, session.sessions().getActiveClientSessionStats(realm, true)
-                    .keySet().stream().map(i -> new ClientIdSessionType(i, OFFLINE)));
-        }
 
-        Stream<SessionRepresentation> result = sessionIdStream.flatMap((clientIdSessionType) -> {
+        return applySearch(search, stream).skip(first).limit(max);
-            ClientModel clientModel = realm.getClientById(clientIdSessionType.getClientId());
-            if (clientModel == null) {
-                // client has been removed in the meantime
-                return Stream.empty();
-            }
-            switch (clientIdSessionType.getType()) {
-                case REGULAR:
-                    return session.sessions().getUserSessionsStream(realm, clientModel)
-                            .map(s -> toRepresentation(s, REGULAR));
-                case OFFLINE:
-                    return session.sessions()
-                            .getOfflineUserSessionsStream(realm, clientModel, null, null)
-                            .map(s -> toRepresentation(s, OFFLINE));
-            }
-            return Stream.<SessionRepresentation>builder().build();
-        });
-
-        return applySearch(search, result).distinct().skip(first).limit(max);
     }
 
     @GET
@@ -125,18 +99,12 @@ public class SessionsResource {
         ClientModel clientModel = realm.getClientById(clientId);
         auth.clients().requireView(clientModel);
 
-        Stream<SessionRepresentation> result = Stream.<SessionRepresentation>builder().build();
+        var stream = switch (type) {
-        if (type == ALL || type == REGULAR) {
+            case OFFLINE -> streamOffline(clientModel);
-            result = Stream.concat(result, session.sessions()
+            case REGULAR -> streamRegular(clientModel);
-                    .getUserSessionsStream(clientModel.getRealm(), clientModel).map(s -> toRepresentation(s, REGULAR)));
+            case ALL -> Stream.concat(streamRegular(clientModel), streamOffline(clientModel));
-        }
+        };
-        if (type == ALL || type == OFFLINE) {
+        return applySearch(search, stream).skip(first).limit(max);
-            result = Stream.concat(result, session.sessions()
-                    .getOfflineUserSessionsStream(clientModel.getRealm(), clientModel, null, null)
-                    .map(s -> toRepresentation(s, OFFLINE)));
-        }
-
-        return applySearch(search, result).distinct().skip(first).limit(max);
     }
 
     private Stream<SessionRepresentation> applySearch(String search, Stream<SessionRepresentation> result) {
@@ -151,6 +119,26 @@ public class SessionsResource {
         return result;
     }
 
+    private Stream<SessionRepresentation> streamRegular() {
+        return session.sessions().readOnlyStreamUserSessions(realm)
+                .map(s -> toRepresentation(s, REGULAR));
+    }
+
+    private Stream<SessionRepresentation> streamOffline() {
+        return session.sessions().readOnlyStreamOfflineUserSessions(realm)
+                .map(s -> toRepresentation(s, OFFLINE));
+    }
+
+    private Stream<SessionRepresentation> streamRegular(ClientModel client) {
+        return session.sessions().readOnlyStreamUserSessions(realm, client)
+                .map(s -> toRepresentation(s, REGULAR));
+    }
+
+    private Stream<SessionRepresentation> streamOffline(ClientModel client) {
+        return session.sessions().readOnlyStreamOfflineUserSessions(realm, client)
+                .map(s -> toRepresentation(s, OFFLINE));
+    }
+
     private static SessionRepresentation toRepresentation(UserSessionModel session, SessionType type) {
         SessionRepresentation rep = new SessionRepresentation();
         rep.setId(session.getId());

server-spi/src/main/java/org/keycloak/models/UserSessionProvider.java

@@ -278,4 +278,66 @@ public interface UserSessionProvider extends Provider {
     default UserSessionModel getUserSessionIfClientExists(RealmModel realm, String userSessionId, boolean offline, String clientUUID) {
         return getUserSessionWithPredicate(realm, userSessionId, offline, userSession -> userSession.getAuthenticatedClientSessionByClient(clientUUID) != null);
     }
+
+    /**
+     * Stream all the regular sessions in the realm.
+     * <p>
+     * The returned {@link UserSessionModel} instances are immutable. More precisely, the entity is not tracked by the transaction and any
+     * modification may throw an {@link UnsupportedOperationException}.
+     *
+     * @param realm The {@link RealmModel} instance.
+     * @return A {@link Stream} for all the sessions in the realm.
+     */
+    default Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm) {
+        return getActiveClientSessionStats(realm, false)
+                .keySet()
+                .stream()
+                .map(realm::getClientById)
+                .flatMap((client) -> getUserSessionsStream(realm, client));
+    }
+
+    /**
+     * Stream all the offline sessions in the realm.
+     * <p>
+     * The returned {@link UserSessionModel} instances are immutable. More precisely, the entity is not tracked by the transaction and any
+     * modification may throw an {@link UnsupportedOperationException}.
+     *
+     * @param realm The {@link RealmModel} instance.
+     * @return A {@link Stream} for all the sessions in the realm.
+     */
+    default Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm) {
+        return getActiveClientSessionStats(realm, true)
+                .keySet()
+                .stream()
+                .map(realm::getClientById)
+                .flatMap((client) -> getOfflineUserSessionsStream(realm, client, null, null));
+    }
+
+    /**
+     * Stream all the regular sessions belonging to the realm and having a client session from the client.
+     * <p>
+     * The returned {@link UserSessionModel} instances are immutable. More precisely, the entity is not tracked by the transaction and any
+     * modification may throw an {@link UnsupportedOperationException}.
+     *
+     * @param realm  The {@link RealmModel} instance.
+     * @param client The {@link ClientModel} instance.
+     * @return A {@link Stream} for all the sessions matching the parameters.
+     */
+    default Stream<UserSessionModel> readOnlyStreamUserSessions(RealmModel realm, ClientModel client) {
+        return getUserSessionsStream(realm, client);
+    }
+
+    /**
+     * Stream all the offline sessions belonging to the realm and having a client session from the client.
+     * <p>
+     * The returned {@link UserSessionModel} instances are immutable. More precisely, the entity is not tracked by the transaction and any
+     * modification may throw an {@link UnsupportedOperationException}.
+     *
+     * @param realm   The {@link RealmModel} instance.
+     * @param client  The {@link ClientModel} instance.
+     * @return A {@link Stream} for all the sessions matching the parameters.
+     */
+    default Stream<UserSessionModel> readOnlyStreamOfflineUserSessions(RealmModel realm, ClientModel client) {
+        return getOfflineUserSessionsStream(realm, client, null, null);
+    }
 }

tests/base/src/test/java/org/keycloak/tests/model/UserSessionProviderTest.java

@@ -886,6 +886,25 @@ public class UserSessionProviderTest {
         runOnServer.run(UserSessionProviderTest::testOnUserRemovedLazyUserAttributesAreLoaded);
     }
 
+    @TestOnServer
+    public void testReadOnlyStreams(KeycloakSession session) {
+        var sessions = KeycloakModelUtils.runJobInTransactionWithResult(session.getKeycloakSessionFactory(), UserSessionProviderTest::createSessions);
+
+        KeycloakModelUtils.runJobInTransaction(session.getKeycloakSessionFactory(), kcSession -> {
+            RealmModel realm = kcSession.realms().getRealmByName("test");
+            var readOnlySessionList = kcSession.sessions().readOnlyStreamUserSessions(realm).toList();
+            assertSessions(readOnlySessionList, sessions);
+
+            // all sessions have client sessions from test-app
+            readOnlySessionList = kcSession.sessions().readOnlyStreamUserSessions(realm, realm.getClientByClientId("test-app")).toList();
+            assertSessions(readOnlySessionList, sessions);
+
+            // only the first one have a client session form third-party
+            readOnlySessionList = kcSession.sessions().readOnlyStreamUserSessions(realm, realm.getClientByClientId("third-party")).toList();
+            assertSessions(readOnlySessionList, sessions[0]);
+        });
+    }
+
     public static void testOnUserRemovedLazyUserAttributesAreLoaded(KeycloakSession session) {
         RealmModel realm = session.realms().getRealmByName("test");
         UserModel user1 = session.users().getUserByUsername(realm, "user1");