SUPPORT-9165: add active session execute and refactor

backend/src/main/java/AppConfig.java

@@ -22,7 +22,6 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.EnableAspectJAutoProxy;
 import org.springframework.context.annotation.FilterType;
 import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
-import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.web.client.RestTemplate;
@@ -77,7 +76,7 @@ public class AppConfig {
 
   @Bean
   public LockProvider lockProvider(@Qualifier("datasource") DataSource dataSource) {
-    return new JdbcTemplateLockProvider(dataSource);
+    return new JdbcTemplateLockProvider(dataSource, "ervu_business_metrics.shedlock");
   }
 
   @Bean

backend/src/main/java/ervu_business_metrics/config/KafkaEnabledCondition.java

@@ -8,12 +8,12 @@ import org.springframework.core.type.AnnotatedTypeMetadata;
 /**
  * @author Adel Kalimullin
  */
-public class IdmReconcileEnabledCondition implements Condition {
-  private static final String ERVU_RECONCILE_ENABLED = "ervu.idm.reconcile.enabled";
+public class KafkaEnabledCondition implements Condition {
+  private static final String ERVU_KAFKA_ENABLED = "ervu.kafka.enabled";
 
   @Override
   public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
     Environment env = context.getEnvironment();
-    return Boolean.parseBoolean(env.getProperty(ERVU_RECONCILE_ENABLED, "true"));
+    return Boolean.parseBoolean(env.getProperty(ERVU_KAFKA_ENABLED, "true"));
   }
 }

backend/src/main/java/ervu_business_metrics/dao/ActiveSessionDao.java

@@ -0,0 +1,43 @@
+package ervu_business_metrics.dao;
+
+import java.util.List;
+
+import ervu_business_metrics.model.sso.UserAccount;
+import ervu_business_metrics.model.sso.UserSessionInfo;
+import org.jooq.DSLContext;
+import org.springframework.stereotype.Repository;
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.records.ActiveSessionRecord;
+
+import static ru.micord.webbpm.ervu.business_metrics.db_beans.auth.Tables.ACTIVE_SESSION;
+
+/**
+ * @author Adel Kalimullin
+ */
+@Repository
+public class ActiveSessionDao {
+  private final DSLContext dsl;
+
+  public ActiveSessionDao(DSLContext dsl) {
+    this.dsl = dsl;
+  }
+
+  public void saveAll(List<UserSessionInfo> sessions) {
+    List<ActiveSessionRecord> records = sessions.stream()
+        .map(session -> {
+          UserAccount userAccount = session.getUserIdentity().getUserAccount();
+          ActiveSessionRecord record = dsl.newRecord(ACTIVE_SESSION);
+          record.setSessionId(session.getSessionId());
+          record.setDomainId(userAccount.getDomain().getId());
+          record.setUserId(userAccount.getId());
+          record.setIpAddress(session.getIp());
+          return record;
+        })
+        .toList();
+
+    dsl.batchInsert(records).execute();
+  }
+
+  public void clearActiveSessions() {
+    dsl.truncate(ACTIVE_SESSION).execute();
+  }
+}

backend/src/main/java/ervu_business_metrics/dao/IdmDirectoriesDao.java

@@ -1,10 +1,9 @@
 package ervu_business_metrics.dao;
 
-import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
+import ervu_business_metrics.config.KafkaEnabledCondition;
 import org.jooq.DSLContext;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.stereotype.Repository;
@@ -22,7 +21,7 @@ import static ru.micord.webbpm.ervu.business_metrics.db_beans.idm_reconcile.Tabl
  * @author Adel Kalimullin
  */
 @Repository
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class IdmDirectoriesDao {
   private final DSLContext dsl;
 

backend/src/main/java/ervu_business_metrics/kafka/KafkaConfig.java

@@ -3,11 +3,13 @@ package ervu_business_metrics.kafka;
 import java.util.HashMap;
 import java.util.Map;
 
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
+import ervu_business_metrics.config.KafkaEnabledCondition;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
@@ -15,15 +17,14 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
-import org.springframework.kafka.core.ConsumerFactory;
-import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.*;
 
 /**
  * @author Adel Kalimullin
  */
 @Configuration
 @EnableKafka
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class KafkaConfig {
   @Value("${kafka.hosts}")
   private String bootstrapServers;
@@ -68,4 +69,22 @@ public class KafkaConfig {
     factory.setConsumerFactory(consumerFactory());
     return factory;
   }
+
+  @Bean
+  public ProducerFactory<String, String> producerFactory() {
+    Map<String, Object> configProps = new HashMap<>();
+    configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+    configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+    configProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
+    configProps.put(SaslConfigs.SASL_JAAS_CONFIG, loginModule + " required username=\""
+                                                  + username + "\" password=\"" + password + "\";");
+    configProps.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
+    return new DefaultKafkaProducerFactory<>(configProps);
+  }
+
+  @Bean()
+  public KafkaTemplate<String, String> kafkaTemplate() {
+    return new KafkaTemplate<>(producerFactory());
+  }
 }

backend/src/main/java/ervu_business_metrics/kafka/KafkaConsumerInitializer.java

@@ -2,7 +2,7 @@ package ervu_business_metrics.kafka;
 
 import javax.annotation.PostConstruct;
 
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
+import ervu_business_metrics.config.KafkaEnabledCondition;
 import ervu_business_metrics.service.IdmDirectoriesService;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.DependsOn;
@@ -13,7 +13,7 @@ import org.springframework.stereotype.Component;
  */
 @Component
 @DependsOn("idmDirectoriesListener")
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class KafkaConsumerInitializer {
   private final IdmDirectoriesService idmDirectoriesService;
 

backend/src/main/java/ervu_business_metrics/kafka/listener/IdmDirectoriesListener.java

@@ -1,9 +1,9 @@
 package ervu_business_metrics.kafka.listener;
 
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
-import ervu_business_metrics.model.AccountData;
-import ervu_business_metrics.model.DomainData;
-import ervu_business_metrics.model.RoleData;
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.model.idm.AccountData;
+import ervu_business_metrics.model.idm.DomainData;
+import ervu_business_metrics.model.idm.RoleData;
 import ervu_business_metrics.service.IdmDirectoriesService;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.kafka.annotation.KafkaListener;
@@ -13,7 +13,7 @@ import org.springframework.stereotype.Component;
  * @author Adel Kalimullin
  */
 @Component
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class IdmDirectoriesListener {
   private final IdmDirectoriesService idmDirectoriesService;
 

backend/src/main/java/ervu_business_metrics/kafka/listener/SessionResponseListener.java

@@ -0,0 +1,42 @@
+package ervu_business_metrics.kafka.listener;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.model.sso.SessionStateResponse;
+import ervu_business_metrics.service.SessionResponseHolder;
+import exception.JsonParseException;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author Adel Kalimullin
+ */
+@Component
+@Conditional(KafkaEnabledCondition.class)
+public class SessionResponseListener{
+  private final ObjectMapper objectMapper;
+  private final SessionResponseHolder holder;
+
+  public SessionResponseListener(ObjectMapper objectMapper,
+      SessionResponseHolder holder) {
+    this.objectMapper = objectMapper;
+    this.holder = holder;
+  }
+
+  @KafkaListener(id = "${kafka.session.group.id}", topics = "${kafka.sso.sessions.state.response}")
+  public void listenKafkaSession(String kafkaMessage) {
+    SessionStateResponse response;
+    try {
+      response = objectMapper.readValue(kafkaMessage, SessionStateResponse.class);
+    }
+    catch (JsonProcessingException e) {
+      throw new JsonParseException(e);
+    }
+    holder.initIfAbsent(response.getRequestKey(), response.getTotal());
+    if (response.getUserSessionInfo() != null) {
+      holder.add(response.getRequestKey(), response.getUserSessionInfo());
+    }
+  }
+}

backend/src/main/java/ervu_business_metrics/model/idm/AccountData.java

@@ -1,11 +1,12 @@
-package ervu_business_metrics.model;
+package ervu_business_metrics.model.idm;
 
 import java.util.List;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import ervu_business_metrics.model.deserializer.ReferenceEntityDeserializer;
+import ervu_business_metrics.model.ReferenceEntity;
+import ervu_business_metrics.model.idm.deserializer.ReferenceEntityDeserializer;
 
 /**
  * @author Adel Kalimullin

backend/src/main/java/ervu_business_metrics/model/idm/DomainData.java

@@ -1,4 +1,4 @@
-package ervu_business_metrics.model;
+package ervu_business_metrics.model.idm;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 

backend/src/main/java/ervu_business_metrics/model/idm/RoleData.java

@@ -1,4 +1,4 @@
-package ervu_business_metrics.model;
+package ervu_business_metrics.model.idm;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 

backend/src/main/java/ervu_business_metrics/model/idm/deserializer/ReferenceEntityDeserializer.java

@@ -1,4 +1,4 @@
-package ervu_business_metrics.model.deserializer;
+package ervu_business_metrics.model.idm.deserializer;
 
 import java.io.IOException;
 

backend/src/main/java/ervu_business_metrics/model/sso/SessionStateResponse.java

@@ -0,0 +1,46 @@
+package ervu_business_metrics.model.sso;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+/**
+ * @author Adel Kalimullin
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class SessionStateResponse{
+  private String requestKey;
+  private int total;
+  private int current;
+  private UserSessionInfo userSessionInfo;
+
+  public String getRequestKey() {
+    return requestKey;
+  }
+
+  public void setRequestKey(String requestKey) {
+    this.requestKey = requestKey;
+  }
+
+  public int getTotal() {
+    return total;
+  }
+
+  public void setTotal(int total) {
+    this.total = total;
+  }
+
+  public int getCurrent() {
+    return current;
+  }
+
+  public void setCurrent(int current) {
+    this.current = current;
+  }
+
+  public UserSessionInfo getUserSessionInfo() {
+    return userSessionInfo;
+  }
+
+  public void setUserSessionInfo(UserSessionInfo userSessionInfo) {
+    this.userSessionInfo = userSessionInfo;
+  }
+}

backend/src/main/java/ervu_business_metrics/model/sso/UserAccount.java

@@ -0,0 +1,33 @@
+package ervu_business_metrics.model.sso;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import ervu_business_metrics.model.ReferenceEntity;
+import ervu_business_metrics.model.idm.deserializer.ReferenceEntityDeserializer;
+
+
+/**
+ * @author Adel Kalimullin
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class UserAccount {
+  private String id;
+  @JsonDeserialize(using = ReferenceEntityDeserializer.class)
+  private ReferenceEntity domain;
+
+  public String getId() {
+    return id;
+  }
+
+  public void setId(String id) {
+    this.id = id;
+  }
+
+  public ReferenceEntity getDomain() {
+    return domain;
+  }
+
+  public void setDomain(ReferenceEntity domain) {
+    this.domain = domain;
+  }
+}

backend/src/main/java/ervu_business_metrics/model/sso/UserIdentity.java

@@ -0,0 +1,19 @@
+package ervu_business_metrics.model.sso;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+/**
+ * @author Adel Kalimullin
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class UserIdentity {
+  private UserAccount userAccount;
+
+  public UserAccount getUserAccount() {
+    return userAccount;
+  }
+
+  public void setUserAccount(UserAccount userAccount) {
+    this.userAccount = userAccount;
+  }
+}

backend/src/main/java/ervu_business_metrics/model/sso/UserSessionInfo.java

@@ -0,0 +1,37 @@
+package ervu_business_metrics.model.sso;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+/**
+ * @author Adel Kalimullin
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class UserSessionInfo {
+  private String sessionId;
+  private UserIdentity userIdentity;
+  private String ip;
+
+  public String getSessionId() {
+    return sessionId;
+  }
+
+  public void setSessionId(String sessionId) {
+    this.sessionId = sessionId;
+  }
+
+  public UserIdentity getUserIdentity() {
+    return userIdentity;
+  }
+
+  public void setUserIdentity(UserIdentity userIdentity) {
+    this.userIdentity = userIdentity;
+  }
+
+  public String getIp() {
+    return ip;
+  }
+
+  public void setIp(String ip) {
+    this.ip = ip;
+  }
+}

backend/src/main/java/ervu_business_metrics/service/IdmDirectoriesDaoService.java

@@ -5,7 +5,7 @@ import java.util.List;
 import java.util.Set;
 
 import ervu_business_metrics.dao.IdmDirectoriesDao;
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
+import ervu_business_metrics.config.KafkaEnabledCondition;
 import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.cache.annotation.Cacheable;
 import org.springframework.context.annotation.Conditional;
@@ -17,7 +17,7 @@ import ru.micord.webbpm.ervu.business_metrics.db_beans.idm_reconcile.tables.reco
  * @author Adel Kalimullin
  */
 @Repository
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class IdmDirectoriesDaoService {
   private final IdmDirectoriesDao idmDirectoriesDao;
 

backend/src/main/java/ervu_business_metrics/service/IdmDirectoriesService.java

@@ -7,12 +7,12 @@ import java.util.Map;
 
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
+import ervu_business_metrics.config.KafkaEnabledCondition;
 import ervu_business_metrics.kafka.model.DeleteKafkaMessage;
 import ervu_business_metrics.kafka.model.UpsertMessage;
-import ervu_business_metrics.model.AccountData;
-import ervu_business_metrics.model.DomainData;
-import ervu_business_metrics.model.RoleData;
+import ervu_business_metrics.model.idm.AccountData;
+import ervu_business_metrics.model.idm.DomainData;
+import ervu_business_metrics.model.idm.RoleData;
 import ervu_business_metrics.service.processor.impl.AccountDataProcessor;
 import ervu_business_metrics.service.processor.DataProcessor;
 import ervu_business_metrics.service.processor.impl.DomainDataProcessor;
@@ -38,7 +38,7 @@ import org.springframework.web.client.RestTemplate;
  */
 @Component
 @DependsOn("liquibase")
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class IdmDirectoriesService {
   private static final Logger LOGGER = LoggerFactory.getLogger(
       MethodHandles.lookup().lookupClass());

backend/src/main/java/ervu_business_metrics/service/SessionResponseHolder.java

@@ -0,0 +1,74 @@
+package ervu_business_metrics.service;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.model.sso.UserSessionInfo;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author Adel Kalimullin
+ */
+@Component
+@Conditional(KafkaEnabledCondition.class)
+public class SessionResponseHolder {
+  private final Map<String, SessionAccumulator> sessionsMap = new ConcurrentHashMap<>();
+
+  public void initIfAbsent(String requestKey, int totalExpected) {
+    sessionsMap.computeIfAbsent(requestKey, key -> new SessionAccumulator(totalExpected));
+  }
+
+  public void add(String requestKey, UserSessionInfo sessionInfo) {
+    SessionAccumulator acc = sessionsMap.get(requestKey);
+    if (acc != null) {
+      acc.addSession(sessionInfo);
+    }
+  }
+
+  public List<UserSessionInfo> awaitSessions(String requestKey, long timeout, TimeUnit unit)
+      throws ExecutionException, InterruptedException, TimeoutException {
+    SessionAccumulator acc = sessionsMap.get(requestKey);
+    if (acc == null) {
+      return List.of();
+    }
+    return acc.getFuture().get(timeout, unit);
+  }
+
+  public void remove(String requestKey) {
+    sessionsMap.remove(requestKey);
+  }
+
+  private static class SessionAccumulator {
+    private final List<UserSessionInfo> sessions = new CopyOnWriteArrayList<>();
+    private final CompletableFuture<List<UserSessionInfo>> future = new CompletableFuture<>();
+    private final AtomicInteger current = new AtomicInteger();
+    private final int totalExpected;
+
+    public SessionAccumulator(int totalExpected) {
+      this.totalExpected = totalExpected;
+      if (totalExpected == 0) {
+        future.complete(List.of());
+      }
+    }
+
+    public void addSession(UserSessionInfo session) {
+      sessions.add(session);
+      if (current.incrementAndGet() == totalExpected) {
+        future.complete(sessions);
+      }
+    }
+
+    public CompletableFuture<List<UserSessionInfo>> getFuture() {
+      return future;
+    }
+  }
+}

backend/src/main/java/ervu_business_metrics/service/SessionService.java

@@ -0,0 +1,69 @@
+package ervu_business_metrics.service;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.dao.ActiveSessionDao;
+import ervu_business_metrics.model.sso.UserSessionInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author Adel Kalimullin
+ */
+@Component
+@Conditional(KafkaEnabledCondition.class)
+public class SessionService {
+  private final static Logger LOGGER = LoggerFactory.getLogger(SessionService.class);
+  private final KafkaTemplate<String, String> kafkaTemplate;
+  private final SessionResponseHolder holder;
+  private final ActiveSessionDao activeSessionDao;
+  @Value("${session.fetch.timeout:15}")
+  private int timeout;
+  @Value("${kafka.sso.sessions.state.receive}")
+  private String sessionTopic;
+
+  public SessionService(
+      KafkaTemplate<String, String> kafkaTemplate,
+      SessionResponseHolder holder, ActiveSessionDao activeSessionDao) {
+    this.kafkaTemplate = kafkaTemplate;
+    this.holder = holder;
+    this.activeSessionDao = activeSessionDao;
+  }
+
+  public void syncSessions() {
+    activeSessionDao.clearActiveSessions();
+
+    List<UserSessionInfo> sessions = fetchSessions();
+    if (!sessions.isEmpty()) {
+      activeSessionDao.saveAll(sessions);
+      LOGGER.info("Синхронизировано {} сессий", sessions.size());
+    }
+    else {
+      LOGGER.info("Нет сессий для синхронизации");
+    }
+  }
+
+  private List<UserSessionInfo> fetchSessions() {
+    String requestKey = UUID.randomUUID().toString();
+    LOGGER.info("Отправка запроса с requestKey = {}", requestKey);
+    kafkaTemplate.send(sessionTopic, requestKey, "{}");
+
+    try {
+      return holder.awaitSessions(requestKey, timeout, TimeUnit.SECONDS);
+    }
+    catch (Exception e) {
+      LOGGER.error("Ошибка при получении сессий по ключу {}", requestKey, e);
+      return List.of();
+    }
+    finally {
+      holder.remove(requestKey);
+    }
+  }
+}

backend/src/main/java/ervu_business_metrics/service/processor/impl/AccountDataProcessor.java

@@ -6,8 +6,8 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
-import ervu_business_metrics.model.AccountData;
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.model.idm.AccountData;
 import ervu_business_metrics.service.IdmDirectoriesDaoService;
 import ervu_business_metrics.service.processor.DataProcessor;
 import org.springframework.context.annotation.Conditional;
@@ -19,7 +19,7 @@ import ru.micord.webbpm.ervu.business_metrics.db_beans.idm_reconcile.tables.reco
  * @author Adel Kalimullin
  */
 @Component
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class AccountDataProcessor implements DataProcessor<AccountData> {
   private final IdmDirectoriesDaoService idmDirectoriesDaoService;
 

backend/src/main/java/ervu_business_metrics/service/processor/impl/DomainDataProcessor.java

@@ -6,8 +6,8 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
-import ervu_business_metrics.model.DomainData;
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.model.idm.DomainData;
 import ervu_business_metrics.service.IdmDirectoriesDaoService;
 import ervu_business_metrics.service.processor.DataProcessor;
 import org.springframework.context.annotation.Conditional;
@@ -18,7 +18,7 @@ import ru.micord.webbpm.ervu.business_metrics.db_beans.idm_reconcile.tables.reco
  * @author Adel Kalimullin
  */
 @Component
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class DomainDataProcessor implements DataProcessor<DomainData> {
   private final IdmDirectoriesDaoService idmDirectoriesDaoService;
 

backend/src/main/java/ervu_business_metrics/service/processor/impl/RoleDataProcessor.java

@@ -6,8 +6,8 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
-import ervu_business_metrics.config.IdmReconcileEnabledCondition;
-import ervu_business_metrics.model.RoleData;
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.model.idm.RoleData;
 import ervu_business_metrics.service.IdmDirectoriesDaoService;
 import ervu_business_metrics.service.processor.DataProcessor;
 import org.springframework.context.annotation.Conditional;
@@ -18,7 +18,7 @@ import ru.micord.webbpm.ervu.business_metrics.db_beans.idm_reconcile.tables.reco
  * @author Adel Kalimullin
  */
 @Component
-@Conditional(IdmReconcileEnabledCondition.class)
+@Conditional(KafkaEnabledCondition.class)
 public class RoleDataProcessor implements DataProcessor<RoleData> {
   private final IdmDirectoriesDaoService idmDirectoriesDaoService;
 

backend/src/main/java/ervu_business_metrics/service/scheduler/SessionScheduledService.java

@@ -0,0 +1,38 @@
+package ervu_business_metrics.service.scheduler;
+
+import javax.annotation.PostConstruct;
+
+import ervu_business_metrics.config.KafkaEnabledCondition;
+import ervu_business_metrics.service.SessionService;
+import net.javacrumbs.shedlock.core.SchedulerLock;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+
+/**
+ * @author Adel Kalimullin
+ */
+@Component
+@Conditional(KafkaEnabledCondition.class)
+public class SessionScheduledService {
+  private final SessionService sessionService;
+
+  public SessionScheduledService(SessionService sessionService) {
+    this.sessionService = sessionService;
+  }
+
+  @PostConstruct
+  @Transactional
+  public void init() {
+      run();
+  }
+
+  @Scheduled(cron = "${session.sync.cron:0 */10 * * * *}")
+  @SchedulerLock(name = "syncSessions")
+  @Transactional
+  public void run() {
+    sessionService.syncSessions();
+  }
+}

backend/src/main/java/exception/JsonParseException.java

@@ -0,0 +1,11 @@
+package exception;
+
+/**
+ * @author Adel Kalimullin
+ */
+public class JsonParseException extends RuntimeException {
+
+  public JsonParseException(Throwable cause) {
+    super(cause);
+  }
+}

backend/src/main/java/ru/micord/webbpm/ervu/business_metrics/db_beans/DefaultCatalog.java

@@ -13,6 +13,7 @@ import org.jooq.impl.CatalogImpl;
 
 import ru.micord.webbpm.ervu.business_metrics.db_beans.actualization.Actualization;
 import ru.micord.webbpm.ervu.business_metrics.db_beans.admin_indicators.AdminIndicators;
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.Auth;
 import ru.micord.webbpm.ervu.business_metrics.db_beans.deregistration.Deregistration;
 import ru.micord.webbpm.ervu.business_metrics.db_beans.idm_reconcile.IdmReconcile;
 import ru.micord.webbpm.ervu.business_metrics.db_beans.init_registration_info.InitRegistrationInfo;
@@ -47,6 +48,11 @@ public class DefaultCatalog extends CatalogImpl {
      */
     public final AdminIndicators ADMIN_INDICATORS = AdminIndicators.ADMIN_INDICATORS;
 
+    /**
+     * The schema <code>auth</code>.
+     */
+    public final Auth AUTH = Auth.AUTH;
+
     /**
      * The schema <code>deregistration</code>.
      */
@@ -104,6 +110,7 @@ public class DefaultCatalog extends CatalogImpl {
         return Arrays.asList(
             Actualization.ACTUALIZATION,
             AdminIndicators.ADMIN_INDICATORS,
+            Auth.AUTH,
             Deregistration.DEREGISTRATION,
             IdmReconcile.IDM_RECONCILE,
             InitRegistrationInfo.INIT_REGISTRATION_INFO,

backend/src/main/java/ru/micord/webbpm/ervu/business_metrics/db_beans/auth/Auth.java

@@ -0,0 +1,55 @@
+/*
+ * This file is generated by jOOQ.
+ */
+package ru.micord.webbpm.ervu.business_metrics.db_beans.auth;
+
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.jooq.Catalog;
+import org.jooq.Table;
+import org.jooq.impl.SchemaImpl;
+
+import ru.micord.webbpm.ervu.business_metrics.db_beans.DefaultCatalog;
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.ActiveSession;
+
+
+/**
+ * This class is generated by jOOQ.
+ */
+@SuppressWarnings({ "all", "unchecked", "rawtypes" })
+public class Auth extends SchemaImpl {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * The reference instance of <code>auth</code>
+     */
+    public static final Auth AUTH = new Auth();
+
+    /**
+     * The table <code>auth.active_session</code>.
+     */
+    public final ActiveSession ACTIVE_SESSION = ActiveSession.ACTIVE_SESSION;
+
+    /**
+     * No further instances allowed
+     */
+    private Auth() {
+        super("auth", null);
+    }
+
+
+    @Override
+    public Catalog getCatalog() {
+        return DefaultCatalog.DEFAULT_CATALOG;
+    }
+
+    @Override
+    public final List<Table<?>> getTables() {
+        return Arrays.asList(
+            ActiveSession.ACTIVE_SESSION
+        );
+    }
+}

backend/src/main/java/ru/micord/webbpm/ervu/business_metrics/db_beans/auth/Keys.java

@@ -0,0 +1,28 @@
+/*
+ * This file is generated by jOOQ.
+ */
+package ru.micord.webbpm.ervu.business_metrics.db_beans.auth;
+
+
+import org.jooq.TableField;
+import org.jooq.UniqueKey;
+import org.jooq.impl.DSL;
+import org.jooq.impl.Internal;
+
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.ActiveSession;
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.records.ActiveSessionRecord;
+
+
+/**
+ * A class modelling foreign key relationships and constraints of tables in
+ * auth.
+ */
+@SuppressWarnings({ "all", "unchecked", "rawtypes" })
+public class Keys {
+
+    // -------------------------------------------------------------------------
+    // UNIQUE and PRIMARY KEY definitions
+    // -------------------------------------------------------------------------
+
+    public static final UniqueKey<ActiveSessionRecord> PK_ACTIVE_SESSION = Internal.createUniqueKey(ActiveSession.ACTIVE_SESSION, DSL.name("pk_active_session"), new TableField[] { ActiveSession.ACTIVE_SESSION.SESSION_ID, ActiveSession.ACTIVE_SESSION.DOMAIN_ID }, true);
+}

backend/src/main/java/ru/micord/webbpm/ervu/business_metrics/db_beans/auth/Tables.java

@@ -0,0 +1,20 @@
+/*
+ * This file is generated by jOOQ.
+ */
+package ru.micord.webbpm.ervu.business_metrics.db_beans.auth;
+
+
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.ActiveSession;
+
+
+/**
+ * Convenience access to all tables in auth.
+ */
+@SuppressWarnings({ "all", "unchecked", "rawtypes" })
+public class Tables {
+
+    /**
+     * The table <code>auth.active_session</code>.
+     */
+    public static final ActiveSession ACTIVE_SESSION = ActiveSession.ACTIVE_SESSION;
+}

backend/src/main/java/ru/micord/webbpm/ervu/business_metrics/db_beans/auth/tables/ActiveSession.java

@@ -0,0 +1,233 @@
+/*
+ * This file is generated by jOOQ.
+ */
+package ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables;
+
+
+import java.util.Collection;
+
+import org.jooq.Condition;
+import org.jooq.Field;
+import org.jooq.Name;
+import org.jooq.PlainSQL;
+import org.jooq.QueryPart;
+import org.jooq.SQL;
+import org.jooq.Schema;
+import org.jooq.Select;
+import org.jooq.Stringly;
+import org.jooq.Table;
+import org.jooq.TableField;
+import org.jooq.TableOptions;
+import org.jooq.UniqueKey;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.jooq.impl.TableImpl;
+
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.Auth;
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.Keys;
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.records.ActiveSessionRecord;
+
+
+/**
+ * This class is generated by jOOQ.
+ */
+@SuppressWarnings({ "all", "unchecked", "rawtypes" })
+public class ActiveSession extends TableImpl<ActiveSessionRecord> {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * The reference instance of <code>auth.active_session</code>
+     */
+    public static final ActiveSession ACTIVE_SESSION = new ActiveSession();
+
+    /**
+     * The class holding records for this type
+     */
+    @Override
+    public Class<ActiveSessionRecord> getRecordType() {
+        return ActiveSessionRecord.class;
+    }
+
+    /**
+     * The column <code>auth.active_session.session_id</code>.
+     */
+    public final TableField<ActiveSessionRecord, String> SESSION_ID = createField(DSL.name("session_id"), SQLDataType.VARCHAR(128).nullable(false), this, "");
+
+    /**
+     * The column <code>auth.active_session.domain_id</code>.
+     */
+    public final TableField<ActiveSessionRecord, String> DOMAIN_ID = createField(DSL.name("domain_id"), SQLDataType.VARCHAR(128).nullable(false), this, "");
+
+    /**
+     * The column <code>auth.active_session.user_id</code>.
+     */
+    public final TableField<ActiveSessionRecord, String> USER_ID = createField(DSL.name("user_id"), SQLDataType.VARCHAR(128).nullable(false), this, "");
+
+    /**
+     * The column <code>auth.active_session.ip_address</code>.
+     */
+    public final TableField<ActiveSessionRecord, String> IP_ADDRESS = createField(DSL.name("ip_address"), SQLDataType.VARCHAR(64), this, "");
+
+    private ActiveSession(Name alias, Table<ActiveSessionRecord> aliased) {
+        this(alias, aliased, (Field<?>[]) null, null);
+    }
+
+    private ActiveSession(Name alias, Table<ActiveSessionRecord> aliased, Field<?>[] parameters, Condition where) {
+        super(alias, null, aliased, parameters, DSL.comment(""), TableOptions.table(), where);
+    }
+
+    /**
+     * Create an aliased <code>auth.active_session</code> table reference
+     */
+    public ActiveSession(String alias) {
+        this(DSL.name(alias), ACTIVE_SESSION);
+    }
+
+    /**
+     * Create an aliased <code>auth.active_session</code> table reference
+     */
+    public ActiveSession(Name alias) {
+        this(alias, ACTIVE_SESSION);
+    }
+
+    /**
+     * Create a <code>auth.active_session</code> table reference
+     */
+    public ActiveSession() {
+        this(DSL.name("active_session"), null);
+    }
+
+    @Override
+    public Schema getSchema() {
+        return aliased() ? null : Auth.AUTH;
+    }
+
+    @Override
+    public UniqueKey<ActiveSessionRecord> getPrimaryKey() {
+        return Keys.PK_ACTIVE_SESSION;
+    }
+
+    @Override
+    public ActiveSession as(String alias) {
+        return new ActiveSession(DSL.name(alias), this);
+    }
+
+    @Override
+    public ActiveSession as(Name alias) {
+        return new ActiveSession(alias, this);
+    }
+
+    @Override
+    public ActiveSession as(Table<?> alias) {
+        return new ActiveSession(alias.getQualifiedName(), this);
+    }
+
+    /**
+     * Rename this table
+     */
+    @Override
+    public ActiveSession rename(String name) {
+        return new ActiveSession(DSL.name(name), null);
+    }
+
+    /**
+     * Rename this table
+     */
+    @Override
+    public ActiveSession rename(Name name) {
+        return new ActiveSession(name, null);
+    }
+
+    /**
+     * Rename this table
+     */
+    @Override
+    public ActiveSession rename(Table<?> name) {
+        return new ActiveSession(name.getQualifiedName(), null);
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    public ActiveSession where(Condition condition) {
+        return new ActiveSession(getQualifiedName(), aliased() ? this : null, null, condition);
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    public ActiveSession where(Collection<? extends Condition> conditions) {
+        return where(DSL.and(conditions));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    public ActiveSession where(Condition... conditions) {
+        return where(DSL.and(conditions));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    public ActiveSession where(Field<Boolean> condition) {
+        return where(DSL.condition(condition));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    @PlainSQL
+    public ActiveSession where(SQL condition) {
+        return where(DSL.condition(condition));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    @PlainSQL
+    public ActiveSession where(@Stringly.SQL String condition) {
+        return where(DSL.condition(condition));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    @PlainSQL
+    public ActiveSession where(@Stringly.SQL String condition, Object... binds) {
+        return where(DSL.condition(condition, binds));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    @PlainSQL
+    public ActiveSession where(@Stringly.SQL String condition, QueryPart... parts) {
+        return where(DSL.condition(condition, parts));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    public ActiveSession whereExists(Select<?> select) {
+        return where(DSL.exists(select));
+    }
+
+    /**
+     * Create an inline derived table from this table
+     */
+    @Override
+    public ActiveSession whereNotExists(Select<?> select) {
+        return where(DSL.notExists(select));
+    }
+}

backend/src/main/java/ru/micord/webbpm/ervu/business_metrics/db_beans/auth/tables/records/ActiveSessionRecord.java

@@ -0,0 +1,109 @@
+/*
+ * This file is generated by jOOQ.
+ */
+package ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.records;
+
+
+import org.jooq.Record2;
+import org.jooq.impl.UpdatableRecordImpl;
+
+import ru.micord.webbpm.ervu.business_metrics.db_beans.auth.tables.ActiveSession;
+
+
+/**
+ * This class is generated by jOOQ.
+ */
+@SuppressWarnings({ "all", "unchecked", "rawtypes" })
+public class ActiveSessionRecord extends UpdatableRecordImpl<ActiveSessionRecord> {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * Setter for <code>auth.active_session.session_id</code>.
+     */
+    public void setSessionId(String value) {
+        set(0, value);
+    }
+
+    /**
+     * Getter for <code>auth.active_session.session_id</code>.
+     */
+    public String getSessionId() {
+        return (String) get(0);
+    }
+
+    /**
+     * Setter for <code>auth.active_session.domain_id</code>.
+     */
+    public void setDomainId(String value) {
+        set(1, value);
+    }
+
+    /**
+     * Getter for <code>auth.active_session.domain_id</code>.
+     */
+    public String getDomainId() {
+        return (String) get(1);
+    }
+
+    /**
+     * Setter for <code>auth.active_session.user_id</code>.
+     */
+    public void setUserId(String value) {
+        set(2, value);
+    }
+
+    /**
+     * Getter for <code>auth.active_session.user_id</code>.
+     */
+    public String getUserId() {
+        return (String) get(2);
+    }
+
+    /**
+     * Setter for <code>auth.active_session.ip_address</code>.
+     */
+    public void setIpAddress(String value) {
+        set(3, value);
+    }
+
+    /**
+     * Getter for <code>auth.active_session.ip_address</code>.
+     */
+    public String getIpAddress() {
+        return (String) get(3);
+    }
+
+    // -------------------------------------------------------------------------
+    // Primary key information
+    // -------------------------------------------------------------------------
+
+    @Override
+    public Record2<String, String> key() {
+        return (Record2) super.key();
+    }
+
+    // -------------------------------------------------------------------------
+    // Constructors
+    // -------------------------------------------------------------------------
+
+    /**
+     * Create a detached ActiveSessionRecord
+     */
+    public ActiveSessionRecord() {
+        super(ActiveSession.ACTIVE_SESSION);
+    }
+
+    /**
+     * Create a detached, initialised ActiveSessionRecord
+     */
+    public ActiveSessionRecord(String sessionId, String domainId, String userId, String ipAddress) {
+        super(ActiveSession.ACTIVE_SESSION);
+
+        setSessionId(sessionId);
+        setDomainId(domainId);
+        setUserId(userId);
+        setIpAddress(ipAddress);
+        resetChangedOnNotNull();
+    }
+}

backend/src/main/resources/config/v_1.0/20250512-SUPPORT-9165-add_session_table.xml

@@ -0,0 +1,47 @@
+<databaseChangeLog
+    xmlns="http://www.liquibase.org/xml/ns/dbchangelog/1.9"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog/1.9
+                      http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-1.9.xsd">
+
+
+
+  <changeSet id="0001" author="adel.ka">
+    <comment>add schema auth</comment>
+    <sql>
+      CREATE SCHEMA IF NOT EXISTS auth;
+      ALTER SCHEMA auth OWNER TO ervu_business_metrics;
+    </sql>
+  </changeSet>
+
+  <changeSet id="0002" author="adel.ka">
+    <comment>add table active_session</comment>
+    <sql>
+      CREATE TABLE IF NOT EXISTS auth.active_session (
+      session_id VARCHAR(128) NOT NULL,
+      domain_id VARCHAR(128) NOT NULL,
+      user_id VARCHAR(128) NOT NULL,
+      ip_address VARCHAR(64),
+
+      CONSTRAINT pk_active_session PRIMARY KEY (session_id, domain_id)
+      );
+
+      ALTER TABLE auth.active_session OWNER TO ervu_business_metrics;
+    </sql>
+  </changeSet>
+
+  <changeSet id="0003" author="adel.ka">
+    <comment>add shedlock table</comment>
+    <sql>
+      CREATE TABLE ervu_business_metrics.shedlock (
+      name VARCHAR(255),
+      lock_until TIMESTAMP WITHOUT TIME ZONE,
+      locked_at TIMESTAMP WITHOUT TIME ZONE,
+      locked_by VARCHAR(255),
+
+      CONSTRAINT shedlock_pk PRIMARY KEY (name)
+      );
+    </sql>
+  </changeSet>
+
+</databaseChangeLog>

backend/src/main/resources/config/v_1.0/changelog-1.0.xml

@@ -31,6 +31,7 @@
 	<include file="20250423-db_changes.xml" relativeToChangelogFile="true"/>
 	<include file="20250505-db_changes.xml" relativeToChangelogFile="true"/>
 	<include file="20250507-db_changes.xml" relativeToChangelogFile="true"/>
+	<include file="20250512-SUPPORT-9165-add_session_table.xml" relativeToChangelogFile="true"/>
 
 
 </databaseChangeLog>

config/micord.env

@@ -31,5 +31,10 @@ KAFKA_DOMAIN_DELETED_GROUP_ID=ervu-business-metrics-backend-domain-deleted
 KAFKA_DOMAIN_DELETED=idmv2.domain.deleted
 KAFKA_ACCOUNT_DELETED_GROUP_ID=ervu-business-metrics-backend-account-deleted
 KAFKA_ACCOUNT_DELETED=idmv2.account.deleted
-ERVU_IDM_RECONCILE_ENABLED=false
-ERVU_IDM_URL=http://idm
+ERVU_KAFKA_ENABLED=false
+ERVU_IDM_URL=http://idm
+SESSION_SYNC_CRON=0 */10 * * * *
+KAFKA_SSO_SESSIONS_STATE_RECEIVE=sso.session.state.receive
+KAFKA_SSO_SESSIONS_STATE_RESPONSE=sso.sessions.state.response
+KAFKA_SESSION_GROUP_ID=ervu-business-metrics-backend-session
+SESSION_FETCH_TIMEOUT=15

resources/src/main/resources/database/datasource.xml

@@ -9,6 +9,7 @@
     <port>5432</port>
     <schemas>actualization</schemas>
     <schemas>admin_indicators</schemas>
+    <schemas>auth</schemas>
     <schemas>deregistration</schemas>
     <schemas>idm_reconcile</schemas>
     <schemas>init_registration_info</schemas>