Merge branch 'feature/SUPPORT-9416_listen_kafka' into develop

backend/src/main/java/ru/micord/ervu/account_applications/component/rpc/UserApplicationListRpcService.java

@@ -6,6 +6,8 @@ import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import ru.micord.ervu.account_applications.security.context.SecurityContext;
+import ru.micord.ervu.account_applications.security.model.UserSession;
 import ru.micord.ervu.account_applications.service.AccountFetchService;
 import ru.micord.ervu.account_applications.service.UserApplicationListService;
 
@@ -24,16 +26,14 @@ public class UserApplicationListRpcService extends Behavior {
   private UserApplicationListService applicationListService;
   @Autowired
   private AccountFetchService accountService;
+  @Autowired
+  private SecurityContext securityContext;
 
   @RpcCall
   public void saveError(long appNumber, String errorMsg) {
+    UserSession userSession = securityContext.getUserSession();
     LOGGER.error("error for application = {}, message: {}", appNumber, errorMsg);
-    applicationListService.saveError(appNumber, errorMsg);
+    applicationListService.saveError(appNumber, errorMsg, userSession.name(), userSession.userId());
-  }
-
-  @RpcCall
-  public void saveAcceptedStatus(long appNumber) {
-    applicationListService.saveAcceptedStatus(appNumber);
   }
 
   @RpcCall

backend/src/main/java/ru/micord/ervu/account_applications/controller/UserApplicationController.java

@@ -1,58 +1,35 @@
 package ru.micord.ervu.account_applications.controller;
 
-import org.slf4j.Logger;
+import java.util.List;
-import org.slf4j.LoggerFactory;
+import java.util.Map;
-import org.springframework.http.MediaType;
+import java.util.stream.Collectors;
-import org.springframework.util.StringUtils;
+
-import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
-import ru.micord.ervu.account_applications.security.service.EncryptionService;
+import ru.micord.ervu.account_applications.model.StatusResponse;
 import ru.micord.ervu.account_applications.service.UserApplicationListService;
-import ru.micord.ervu.account_applications.websocket.dto.ProcessErrorMsg;
-import ru.micord.ervu.account_applications.websocket.dto.ProcessResponseDto;
 
 /**
  * @author gulnaz
  */
 @RestController
 public class UserApplicationController {
-  private static final Logger LOGGER = LoggerFactory.getLogger(UserApplicationController.class);
-
   private final UserApplicationListService applicationService;
-  private final EncryptionService encryptionService;
 
-  public UserApplicationController(UserApplicationListService applicationService,
+  public UserApplicationController(UserApplicationListService applicationService) {
-                                   EncryptionService encryptionService) {
     this.applicationService = applicationService;
-    this.encryptionService = encryptionService;
   }
 
-  @PutMapping(value = "/status", consumes = MediaType.APPLICATION_JSON_VALUE)
-  public Long updateStatus(@RequestBody ProcessResponseDto data) {
-    Long appNumber = data.body().applicationNumber();
 
-    switch (data.className()) {
+  @PostMapping("/status/batch")
-      case UPDATE -> {
+  public List<StatusResponse> getBatchStatuses(@RequestBody List<Long> appNumbers) {
-        LOGGER.info("update by appNumber = {}", appNumber);
+    Map<Long, String> statuses = applicationService.getStatusesBatch(appNumbers);
-        String tempPass = data.body().tempPass();
+    return appNumbers.stream()
-
+        .map(appNumber -> new StatusResponse(
-        if (StringUtils.hasText(tempPass)) {
+            appNumber,
-          String encryptedPassword = encryptionService.encrypt(tempPass);
+            statuses.get(appNumber)
-          applicationService.savePassword(appNumber, encryptedPassword);
+        ))
-        }
+        .collect(Collectors.toList());
-        else {
-          applicationService.saveAcceptedStatus(appNumber);
-        }
-      }
-      case PROCESS_ERROR -> {
-        ProcessErrorMsg errorMsg = data.body().msg();
-        String msg = errorMsg == null ? "unknown error" : errorMsg.message();
-        LOGGER.error("error by appNumber = {}, message: {}", appNumber, msg);
-        applicationService.saveError(appNumber, msg);
-      }
-    }
-
-    return appNumber;
   }
 }

backend/src/main/java/ru/micord/ervu/account_applications/dao/UserApplicationListDao.java

@@ -2,6 +2,9 @@ package ru.micord.ervu.account_applications.dao;
 
 import java.sql.Timestamp;
 import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.jooq.DSLContext;
 import org.springframework.stereotype.Repository;
@@ -23,6 +26,22 @@ public class UserApplicationListDao {
     this.dslContext = dslContext;
   }
 
+  public Map<Long, String> getStatusesBatch(List<Long> appNumbers) {
+    if (appNumbers == null || appNumbers.isEmpty()) {
+      return Map.of();
+    }
+
+    return dslContext.select(USER_APPLICATION_LIST.NUMBER_APP, USER_APPLICATION_LIST.APPLICATION_STATUS)
+        .from(USER_APPLICATION_LIST)
+        .where(USER_APPLICATION_LIST.NUMBER_APP.in(appNumbers))
+        .fetch()
+        .stream()
+        .collect(Collectors.toMap(
+            record -> record.get(USER_APPLICATION_LIST.NUMBER_APP),
+            record -> record.get(USER_APPLICATION_LIST.APPLICATION_STATUS)
+        ));
+  }
+
   public void savePassword(Long appNumber, String encodedPass) {
     dslContext.update(USER_APPLICATION_LIST)
         .set(USER_APPLICATION_LIST.APPLICATION_STATUS, ACCEPTED.name())

backend/src/main/java/ru/micord/ervu/account_applications/kafka/DeclarationStatusListener.java

@@ -0,0 +1,69 @@
+package ru.micord.ervu.account_applications.kafka;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+import ru.micord.ervu.account_applications.kafka.model.DeclarationStatus;
+import ru.micord.ervu.account_applications.security.model.jwt.UserClaims;
+import ru.micord.ervu.account_applications.security.service.EncryptionService;
+import ru.micord.ervu.account_applications.security.service.JwtTokenService;
+import ru.micord.ervu.account_applications.service.UserApplicationListService;
+
+/**
+ * @author Adel Kalimullin
+ */
+@Component
+public class DeclarationStatusListener {
+  private static final Logger LOGGER = LoggerFactory.getLogger(DeclarationStatusListener.class);
+  private final ObjectMapper mapper;
+  private final UserApplicationListService applicationService;
+  private final EncryptionService encryptionService;
+  private final JwtTokenService jwtTokenService;
+
+  public DeclarationStatusListener(ObjectMapper mapper,
+      UserApplicationListService applicationService, EncryptionService encryptionService,
+      JwtTokenService jwtTokenService) {
+    this.mapper = mapper;
+    this.applicationService = applicationService;
+    this.encryptionService = encryptionService;
+    this.jwtTokenService = jwtTokenService;
+  }
+
+  @KafkaListener(id = "${kafka.application.status.group.id}",
+      topics = "${kafka.application.status}")
+  public void listenKafkaDomain(String kafkaMessage, @Header("authorization") String authHeader) {
+    try {
+      String token = authHeader.replace("Bearer ", "");
+      UserClaims userClaims = jwtTokenService.getUserClaims(token);
+      String name = userClaims.name();
+      String userId = userClaims.userId();
+
+      DeclarationStatus declarationStatus = mapper.readValue(kafkaMessage, DeclarationStatus.class);
+      Long applicationNumber = declarationStatus.applicationNumber();
+      if (declarationStatus.status()) {
+        LOGGER.info("update by appNumber = {}", applicationNumber);
+        String tempPass = declarationStatus.password();
+        if (StringUtils.hasText(tempPass)) {
+          String encryptedPassword = encryptionService.encrypt(tempPass);
+          applicationService.savePassword(applicationNumber, encryptedPassword, name, userId);
+        }
+        else {
+          applicationService.saveAcceptedStatus(applicationNumber, name, userId);
+        }
+      }
+      else {
+        String errorMsg = declarationStatus.errorMsg();
+        LOGGER.error("error by appNumber = {}, message: {}", applicationNumber, errorMsg);
+        applicationService.saveError(applicationNumber, errorMsg, name, userId);
+      }
+    }
+    catch (JsonProcessingException e) {
+      LOGGER.error("Failed to deserialize message: {}", kafkaMessage, e);
+    }
+  }
+}

backend/src/main/java/ru/micord/ervu/account_applications/kafka/model/DeclarationStatus.java

@@ -0,0 +1,14 @@
+package ru.micord.ervu.account_applications.kafka.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * @author Adel Kalimullin
+ */
+public record DeclarationStatus(
+    @JsonProperty("applicationNumber") Long applicationNumber,
+    @JsonProperty("password") String password,
+    @JsonProperty("status") boolean status,
+    @JsonProperty("userName") String userName,
+    @JsonProperty("description") String errorMsg
+)  {}

backend/src/main/java/ru/micord/ervu/account_applications/model/StatusResponse.java

@@ -0,0 +1,6 @@
+package ru.micord.ervu.account_applications.model;
+
+/**
+ * @author Adel Kalimullin
+ */
+public record StatusResponse(Long appNumber, String status) {}

backend/src/main/java/ru/micord/ervu/account_applications/service/UserApplicationListService.java

@@ -3,12 +3,12 @@ package ru.micord.ervu.account_applications.service;
 import java.sql.Timestamp;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import ru.micord.ervu.account_applications.component.dao.AuditDao;
 import ru.micord.ervu.account_applications.dao.UserApplicationListDao;
-import ru.micord.ervu.account_applications.security.context.SecurityContext;
-import ru.micord.ervu.account_applications.security.model.UserSession;
 import utils.DateTimeUtil;
 
 import static ru.micord.ervu.account_applications.enums.ApplicationStatus.ACCEPTED;
@@ -22,38 +22,40 @@ public class UserApplicationListService {
 
   private final UserApplicationListDao dao;
   private final AuditDao auditDao;
-  private final SecurityContext securityContext;
 
-  public UserApplicationListService(UserApplicationListDao dao, AuditDao auditDao, SecurityContext securityContext) {
+  public UserApplicationListService(UserApplicationListDao dao, AuditDao auditDao) {
     this.dao = dao;
     this.auditDao = auditDao;
-    this.securityContext = securityContext;
   }
 
-  public void savePassword(Long appNumber, String encodedPass) {
+  public Map<Long, String> getStatusesBatch(List<Long> appNumbers) {
+    return dao.getStatusesBatch(appNumbers);
+  }
+
+  @Transactional
+  public void savePassword(Long appNumber, String encodedPass, String name, String userId) {
     dao.savePassword(appNumber, encodedPass);
-    saveAuditStatusByAppNumber(appNumber, ACCEPTED.name());
+    saveAuditStatusByAppNumber(appNumber, ACCEPTED.name(), name, userId);
   }
 
   public boolean userExists(String login){
     return dao.userExists(login);
   }
 
-  public void saveAcceptedStatus(long appNumber) {
+  @Transactional
+  public void saveAcceptedStatus(long appNumber, String name, String userId) {
     dao.saveAcceptedStatus(appNumber);
-    saveAuditStatusByAppNumber(appNumber, ACCEPTED.name());
+    saveAuditStatusByAppNumber(appNumber, ACCEPTED.name(), name, userId);
   }
 
-  public void saveError(long appNumber, String errorMsg) {
+  @Transactional
+  public void saveError(long appNumber, String errorMsg, String name, String userId) {
     dao.saveError(appNumber, errorMsg);
-    saveAuditStatusByAppNumber(appNumber, AGREED.name());
+    saveAuditStatusByAppNumber(appNumber, AGREED.name(), name, userId);
   }
 
-  private void saveAuditStatusByAppNumber(long appNumber, String status) {
+  private void saveAuditStatusByAppNumber(long appNumber, String status, String name, String userId) {
     List<Long> appIds = auditDao.selectAppListIdsByAppNumber(appNumber);
-    UserSession userSession = securityContext.getUserSession();
-    String name = userSession.name();
-    String userId = userSession.userId();
     appIds.forEach(id -> {
       auditDao.insert(id, name, userId, status, Timestamp.valueOf(
           DateTimeUtil.dateToLocalDateTimeUtc(new Date())));

backend/src/main/java/ru/micord/ervu/account_applications/websocket/dto/ProcessErrorMsg.java

@@ -1,10 +0,0 @@
-package ru.micord.ervu.account_applications.websocket.dto;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-
-/**
- * @author gulnaz
- */
-@JsonIgnoreProperties(ignoreUnknown = true)
-public record ProcessErrorMsg(String message) {
-}

backend/src/main/java/ru/micord/ervu/account_applications/websocket/dto/ProcessResponseBody.java

@@ -1,13 +0,0 @@
-package ru.micord.ervu.account_applications.websocket.dto;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-/**
- * @author gulnaz
- */
-@JsonIgnoreProperties(ignoreUnknown = true)
-public record ProcessResponseBody(String type, Long applicationNumber, String userName,
-                                  @JsonProperty("value") String tempPass,
-                                  String secretLink, ProcessErrorMsg msg) {
-}

backend/src/main/java/ru/micord/ervu/account_applications/websocket/dto/ProcessResponseDto.java

@@ -1,11 +0,0 @@
-package ru.micord.ervu.account_applications.websocket.dto;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import ru.micord.ervu.account_applications.websocket.enums.ClassName;
-
-/**
- * @author gulnaz
- */
-@JsonIgnoreProperties(ignoreUnknown = true)
-public record ProcessResponseDto(String forUser, ClassName className, ProcessResponseBody body) {
-}

backend/src/main/java/ru/micord/ervu/account_applications/websocket/enums/ClassName.java

@@ -1,16 +0,0 @@
-package ru.micord.ervu.account_applications.websocket.enums;
-
-import com.fasterxml.jackson.annotation.JsonEnumDefaultValue;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-/**
- * @author gulnaz
- */
-public enum ClassName {
-  @JsonProperty("update")
-  UPDATE,
-  @JsonProperty("processError")
-  PROCESS_ERROR,
-  @JsonEnumDefaultValue
-  SKIP
-}

frontend/src/ts/account_applications/component/button/UserManagementService.ts

@@ -7,7 +7,7 @@ import {
   TextField,
   Visible
 } from "@webbpm/base-package";
-import {HttpClient, HttpErrorResponse, HttpResponse} from "@angular/common/http";
+import {HttpClient} from "@angular/common/http";
 import {FormField} from "../field/FormField";
 import {AuthorizationService} from "../../../modules/app/service/authorization.service";
 import {ApplicationKind} from "../enum/ApplicationKind";
@@ -202,10 +202,12 @@ export class UserManagementService extends Behavior {
     this.httpClient.post(url, request).toPromise()
         .then((response: ProcessResponse) => {
           let code = response.code;
-
           if (code !== '200') {
             this.saveError(appNumber, response.msg);
           }
+          else {
+            this.statusUpdateService.trackDeclaration(appNumber);
+          }
         })
         .catch(reason => {
           console.error("Error while executing request:", reason.toString());

frontend/src/ts/modules/app/service/authorization.service.ts

@@ -1,9 +1,6 @@
-import {Injectable, OnDestroy} from "@angular/core";
+import {Injectable} from "@angular/core";
 import {Subject} from "rxjs";
 import {HttpClient} from "@angular/common/http";
-import {WebsocketService} from "../websocket/websocket.service";
-import {StatusUpdateService} from "./status-update.service";
-import {ErvuPermission} from "../enum/ErvuPermission";
 
 export interface UserSession {
   userId: string,
@@ -14,14 +11,13 @@ export interface UserSession {
 }
 
 @Injectable({providedIn: 'root'})
-export class AuthorizationService implements OnDestroy {
+export class AuthorizationService{
 
   private session: UserSession;
 
   public onSessionUpdate: Subject<UserSession> = new Subject<UserSession>();
 
-  constructor(protected httpClient: HttpClient, protected websocketService: WebsocketService,
+  constructor(protected httpClient: HttpClient) {}
-              protected statusUpdateService: StatusUpdateService) {}
 
   public getCurrentSession(): Promise<any> {
     if (this.session) return new Promise(resolve => resolve(this.session));
@@ -30,18 +26,6 @@ export class AuthorizationService implements OnDestroy {
         .then((session: UserSession) => {
           this.session = session;
           this.onSessionUpdate.next(session);
-
-          if (this.hasPermission(ErvuPermission.APPROVER)) {
-            this.websocketService.subscribe(({data}) => {
-              let parsedObj = JSON.parse(data);
-
-              if (parsedObj && parsedObj.body && parsedObj.body.applicationNumber) {
-                if (parsedObj.className === 'update' || parsedObj.className === 'processError') {
-                  this.statusUpdateService.update(parsedObj);
-                }
-              }
-            });
-          }
           return session;
         })
   }
@@ -77,8 +61,4 @@ export class AuthorizationService implements OnDestroy {
   getPermissions(): string[] {
     return this.isAuthorized() ? this.session.permissions : null;
   }
-
-  ngOnDestroy(): void {
-    this.websocketService.unsubscribe();
-  }
 }

frontend/src/ts/modules/app/service/status-update.service.ts

@@ -2,6 +2,11 @@ import {Injectable} from "@angular/core";
 import {BehaviorSubject} from "rxjs";
 import {HttpClient} from "@angular/common/http";
 
+export interface StatusResponse {
+  appNumber: number;
+  status: string;
+}
+
 enum ApplicationStatus {
   AGREED = 'Согласована',
   ACCEPTED = 'Исполнена'
@@ -14,19 +19,82 @@ export class StatusUpdateService {
   }
 
   public statusMessage = new BehaviorSubject<any>(null);
+  private pendingDeclarations = new Map<number, number>();
+  private pollingInterval: any;
+  private readonly MAX_ATTEMPTS = 12;
 
-  public update(responseObj: any): void {
+  public trackDeclaration(appNumber: number): void {
-    this.httpClient.put<number>('status', responseObj)
+    if (!this.pendingDeclarations.has(appNumber)) {
-        .toPromise()
+      this.pendingDeclarations.set(appNumber, 0);
-        .then(appNumber => this.publishStatus(appNumber, responseObj.className === 'update'))
+      this.startPolling();
-        .catch(err => console.log('failed to update application status', err));
+    }
   }
 
   public publishStatus(appNumber: number, accepted: boolean) {
-    this.statusMessage.next(
+    this.statusMessage.next({
-        {
+                              appNumber: appNumber,
-          appNumber: appNumber,
+                              status: accepted
-          status: accepted ? ApplicationStatus.ACCEPTED : ApplicationStatus.AGREED
+                                      ? ApplicationStatus.ACCEPTED
+                                      : ApplicationStatus.AGREED
+                            });
+  }
+
+  private startPolling(): void {
+    if (this.pendingDeclarations.size === 0 || this.pollingInterval) {
+      return;
+    }
+
+    this.pollingInterval = setInterval(() => {
+      this.checkPendingStatuses();
+    }, 5000);
+  }
+
+  private stopPolling(): void {
+    if (this.pollingInterval) {
+      clearInterval(this.pollingInterval);
+      this.pollingInterval = null;
+    }
+  }
+
+  private checkPendingStatuses(): void {
+    const appNumbers = Array.from(this.pendingDeclarations.keys());
+    if (appNumbers.length === 0) {
+      this.stopPolling();
+      return;
+    }
+
+    this.httpClient.post<StatusResponse[]>('status/batch', appNumbers)
+        .toPromise()
+        .then(responses => {
+          responses.forEach(response => {
+            const attemptCount = (this.pendingDeclarations.get(response.appNumber) || 0) + 1;
+            this.pendingDeclarations.set(response.appNumber, attemptCount);
+
+            if (response.status !== 'SENT') {
+              this.pendingDeclarations.delete(response.appNumber);
+              this.publishStatus(response.appNumber, response.status === 'ACCEPTED');
+            }
+            else if (attemptCount >= this.MAX_ATTEMPTS) {
+              this.pendingDeclarations.delete(response.appNumber);
+              console.warn(`Max attempts exceeded for declaration ${response.appNumber}`);
+            }
+          });
+
+          if (this.pendingDeclarations.size === 0) {
+            this.stopPolling();
+          }
+        })
+        .catch(err => {
+          console.error('Failed to check statuses', err);
+          appNumbers.forEach(appNumber => {
+            const attemptCount = (this.pendingDeclarations.get(appNumber) || 0) + 1;
+            this.pendingDeclarations.set(appNumber, attemptCount);
+
+            if (attemptCount >= this.MAX_ATTEMPTS) {
+              this.pendingDeclarations.delete(appNumber);
+              console.warn(`Max attempts exceeded for declaration ${appNumber} due to errors`);
+            }
+          });
         });
   }
-}
+}

frontend/src/ts/modules/app/websocket/websocket.service.ts

@@ -1,37 +0,0 @@
-import {Injectable} from "@angular/core";
-
-@Injectable({providedIn: 'root'})
-export class WebsocketService {
-
-  private initialData;
-
-  public subscribe(fn: Function): void {
-    let property = Object.getOwnPropertyDescriptor(MessageEvent.prototype, "data");
-    const data = property.get;
-    this.initialData = data;
-
-    // wrapper that replaces getter
-    function lookAtMessage() {
-      let socket = this.currentTarget instanceof WebSocket;
-
-      if (!socket || !this.currentTarget.url.endsWith('notifier.message.send.push')) {
-        return data.call(this);
-      }
-      let msg = data.call(this);
-      Object.defineProperty(this, "data", { value: msg }); //anti-loop
-      fn({ data: msg, socket: this.currentTarget, event: this });
-      return msg;
-    }
-    property.get = lookAtMessage;
-    Object.defineProperty(MessageEvent.prototype, "data", property);
-  }
-
-  public unsubscribe(): void {
-    let property = Object.getOwnPropertyDescriptor(MessageEvent.prototype, "data");
-
-    if (this.initialData) {
-      property.get = this.initialData;
-      Object.defineProperty(MessageEvent.prototype, "data", property);
-    }
-  }
-}