184 lines
11 KiB
Java
184 lines
11 KiB
Java
/*
|
|
* Copyright 2025 Google. This software is provided as-is, without warranty or representation for any use or purpose.
|
|
* Your use of it is subject to your agreement with Google.
|
|
*/
|
|
|
|
package com.example.service.notification;
|
|
|
|
import com.example.dto.dialogflow.notification.ExternalNotRequestDTO;
|
|
import com.example.dto.dialogflow.base.DetectIntentRequestDTO;
|
|
import com.example.dto.dialogflow.base.DetectIntentResponseDTO;
|
|
import com.example.dto.dialogflow.conversation.ConversationEntryDTO;
|
|
import com.example.dto.dialogflow.conversation.ConversationMessageDTO;
|
|
import com.example.dto.dialogflow.conversation.ConversationSessionDTO;
|
|
import com.example.dto.dialogflow.notification.NotificationDTO;
|
|
import com.example.mapper.conversation.ConversationEntryMapper;
|
|
import com.example.mapper.notification.ExternalNotRequestMapper;
|
|
import com.example.service.base.DialogflowClientService;
|
|
import com.example.service.conversation.DataLossPrevention;
|
|
import com.example.service.conversation.FirestoreConversationService;
|
|
import com.example.service.conversation.MemoryStoreConversationService;
|
|
import com.example.util.SessionIdGenerator;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
import org.springframework.stereotype.Service;
|
|
import reactor.core.publisher.Mono;
|
|
import java.time.Instant;
|
|
import java.util.HashMap;
|
|
import java.util.Map;
|
|
import java.util.Objects;
|
|
|
|
@Service
|
|
public class NotificationManagerService {
|
|
|
|
private static final Logger logger = LoggerFactory.getLogger(NotificationManagerService.class);
|
|
private static final String eventName = "notificacion";
|
|
private static final String PREFIX_PO_PARAM = "notification_po_";
|
|
|
|
private final DialogflowClientService dialogflowClientService;
|
|
private final FirestoreNotificationService firestoreNotificationService;
|
|
private final MemoryStoreNotificationService memoryStoreNotificationService;
|
|
private final ExternalNotRequestMapper externalNotRequestMapper;
|
|
private final MemoryStoreConversationService memoryStoreConversationService;
|
|
private final FirestoreConversationService firestoreConversationService;
|
|
private final DataLossPrevention dataLossPrevention;
|
|
private final String dlpTemplateCompleteFlow;
|
|
private final ConversationEntryMapper conversationEntryMapper;
|
|
|
|
@Value("${dialogflow.default-language-code:es}")
|
|
private String defaultLanguageCode;
|
|
|
|
public NotificationManagerService(
|
|
DialogflowClientService dialogflowClientService,
|
|
FirestoreNotificationService firestoreNotificationService,
|
|
MemoryStoreNotificationService memoryStoreNotificationService,
|
|
MemoryStoreConversationService memoryStoreConversationService,
|
|
FirestoreConversationService firestoreConversationService,
|
|
|
|
ExternalNotRequestMapper externalNotRequestMapper,
|
|
DataLossPrevention dataLossPrevention,
|
|
ConversationEntryMapper conversationEntryMapper,
|
|
@Value("${google.cloud.dlp.dlpTemplateCompleteFlow}") String dlpTemplateCompleteFlow) {
|
|
|
|
this.dialogflowClientService = dialogflowClientService;
|
|
this.firestoreNotificationService = firestoreNotificationService;
|
|
this.memoryStoreNotificationService = memoryStoreNotificationService;
|
|
this.externalNotRequestMapper = externalNotRequestMapper;
|
|
this.dataLossPrevention = dataLossPrevention;
|
|
this.dlpTemplateCompleteFlow = dlpTemplateCompleteFlow;
|
|
this.memoryStoreConversationService = memoryStoreConversationService;
|
|
this.firestoreConversationService = firestoreConversationService;
|
|
this.conversationEntryMapper = conversationEntryMapper;
|
|
}
|
|
|
|
public Mono<DetectIntentResponseDTO> processNotification(ExternalNotRequestDTO externalRequest) {
|
|
Objects.requireNonNull(externalRequest, "ExternalNotRequestDTO cannot be null.");
|
|
|
|
String telefono = externalRequest.phoneNumber();
|
|
if (telefono == null || telefono.isBlank()) {
|
|
logger.warn("No phone number provided in ExternalNotRequestDTO. Cannot process notification.");
|
|
return Mono.error(new IllegalArgumentException("Phone number is required."));
|
|
}
|
|
|
|
return dataLossPrevention.getObfuscatedString(externalRequest.text(), dlpTemplateCompleteFlow)
|
|
.flatMap(obfuscatedMessage -> {
|
|
ExternalNotRequestDTO obfuscatedRequest = new ExternalNotRequestDTO(
|
|
obfuscatedMessage,
|
|
externalRequest.phoneNumber(),
|
|
externalRequest.hiddenParameters()
|
|
);
|
|
|
|
String newNotificationId = SessionIdGenerator.generateStandardSessionId();
|
|
Map<String, Object> parameters = new HashMap<>();
|
|
if (obfuscatedRequest.hiddenParameters() != null) {
|
|
obfuscatedRequest.hiddenParameters().forEach((key, value) -> parameters.put(PREFIX_PO_PARAM + key, value));
|
|
}
|
|
|
|
NotificationDTO newNotificationEntry = new NotificationDTO(newNotificationId, telefono, Instant.now(),
|
|
obfuscatedRequest.text(), eventName, defaultLanguageCode, parameters, "active");
|
|
Mono<Void> persistenceMono = memoryStoreNotificationService.saveOrAppendNotificationEntry(newNotificationEntry)
|
|
.doOnSuccess(v -> {
|
|
logger.info("Notification for phone {} cached. Kicking off async Firestore write-back.", telefono);
|
|
firestoreNotificationService.saveOrAppendNotificationEntry(newNotificationEntry)
|
|
.subscribe(
|
|
ignored -> logger.debug(
|
|
"Background: Notification entry persistence initiated for phone {} in Firestore.", telefono),
|
|
e -> logger.error(
|
|
"Background: Error during notification entry persistence for phone {} in Firestore: {}",
|
|
telefono, e.getMessage(), e));
|
|
});
|
|
|
|
Mono<ConversationSessionDTO> sessionMono = memoryStoreConversationService.getSessionByTelefono(telefono)
|
|
.doOnNext(session -> logger.info("Found existing conversation session {} for phone number {}",
|
|
session.sessionId(), telefono))
|
|
.flatMap(session -> {
|
|
Map<String, Object> prefixedParameters = new HashMap<>();
|
|
if (obfuscatedRequest.hiddenParameters() != null) {
|
|
obfuscatedRequest.hiddenParameters()
|
|
.forEach((key, value) -> prefixedParameters.put(PREFIX_PO_PARAM + key, value));
|
|
}
|
|
ConversationEntryDTO systemEntry = ConversationEntryDTO.forSystem(obfuscatedRequest.text(),
|
|
prefixedParameters);
|
|
return persistConversationTurn(session, systemEntry)
|
|
.thenReturn(session);
|
|
})
|
|
.switchIfEmpty(Mono.defer(() -> {
|
|
String newSessionId = SessionIdGenerator.generateStandardSessionId();
|
|
logger.info("No existing conversation session found for phone number {}. Creating new session: {}",
|
|
telefono, newSessionId);
|
|
String userId = "user_by_phone_" + telefono;
|
|
Map<String, Object> prefixedParameters = new HashMap<>();
|
|
if (obfuscatedRequest.hiddenParameters() != null) {
|
|
obfuscatedRequest.hiddenParameters()
|
|
.forEach((key, value) -> prefixedParameters.put(PREFIX_PO_PARAM + key, value));
|
|
}
|
|
ConversationEntryDTO systemEntry = ConversationEntryDTO.forSystem(obfuscatedRequest.text(),
|
|
prefixedParameters);
|
|
ConversationSessionDTO newSession = ConversationSessionDTO.create(newSessionId, userId, telefono);
|
|
return persistConversationTurn(newSession, systemEntry)
|
|
.then(Mono.just(newSession));
|
|
}));
|
|
|
|
return persistenceMono.then(sessionMono)
|
|
.flatMap(session -> {
|
|
final String sessionId = session.sessionId();
|
|
logger.info("Sending notification text to Dialogflow using conversation session: {}", sessionId);
|
|
|
|
DetectIntentRequestDTO detectIntentRequest = externalNotRequestMapper.map(obfuscatedRequest);
|
|
|
|
return dialogflowClientService.detectIntent(sessionId, detectIntentRequest);
|
|
})
|
|
.doOnSuccess(response -> logger
|
|
.info("Finished processing notification. Dialogflow response received for phone {}.", telefono))
|
|
.doOnError(e -> logger.error("Overall error in NotificationManagerService: {}", e.getMessage(), e));
|
|
});
|
|
}
|
|
|
|
private Mono<Void> persistConversationTurn(ConversationSessionDTO session, ConversationEntryDTO entry) {
|
|
logger.debug("Starting Write-Back persistence for session {}. Type: {}. Writing to Redis first.", session.sessionId(),
|
|
entry.type().name());
|
|
ConversationMessageDTO message = conversationEntryMapper.toConversationMessageDTO(entry);
|
|
ConversationSessionDTO updatedSession = session.withLastMessage(message.text());
|
|
|
|
return memoryStoreConversationService.saveSession(updatedSession)
|
|
.then(memoryStoreConversationService.saveMessage(session.sessionId(), message))
|
|
.doOnSuccess(v -> {
|
|
logger.info(
|
|
"Entry saved to Redis for session {}. Type: {}. Kicking off async Firestore write-back.",
|
|
session.sessionId(), entry.type().name());
|
|
|
|
firestoreConversationService.saveSession(updatedSession)
|
|
.then(firestoreConversationService.saveMessage(session.sessionId(), message))
|
|
.subscribe(
|
|
fsVoid -> logger.debug(
|
|
"Asynchronously (Write-Back): Entry successfully saved to Firestore for session {}. Type: {}.",
|
|
session.sessionId(), entry.type().name()),
|
|
fsError -> logger.error(
|
|
"Asynchronously (Write-Back): Failed to save entry to Firestore for session {}. Type: {}: {}",
|
|
session.sessionId(), entry.type().name(), fsError.getMessage(), fsError));
|
|
})
|
|
.doOnError(e -> logger.error("Error during primary Redis write for session {}. Type: {}: {}", session.sessionId(),
|
|
entry.type().name(), e.getMessage(), e));
|
|
}
|
|
} |