va/int-layer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UPDATE code 20-Ago

Browse Source
This commit is contained in:
PAVEL PALMA
2025-08-21 00:24:21 -06:00
parent 82eee5f7c0
commit 404f152097
54 changed files with 1851 additions and 597 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/main/java/com/example/service/base/ConvSessionCloserService.java
View File

@@ -1,5 +0,0 @@
package com.example.service.base;
public class ConvSessionCloserService {
}

355
src/main/java/com/example/service/conversation/ConversationManagerService.java
View File

@@ -5,193 +5,290 @@
package com.example.service.conversation;
import com.example.mapper.conversation.ExternalConvRequestMapper;
import com.example.service.base.DialogflowClientService;
import com.example.util.SessionIdGenerator;
import com.example.dto.dialogflow.base.DetectIntentRequestDTO;
import com.example.dto.dialogflow.base.DetectIntentResponseDTO;
import com.example.dto.dialogflow.conversation.ConversationContext;
import com.example.dto.dialogflow.conversation.ConversationEntryDTO;
import com.example.dto.dialogflow.conversation.ConversationEntryEntity;
import com.example.dto.dialogflow.conversation.ConversationEntryType;
import com.example.dto.dialogflow.conversation.ConversationSessionDTO;
import com.example.dto.dialogflow.conversation.ExternalConvRequestDTO;
import com.example.dto.dialogflow.notification.NotificationDTO;
import com.example.mapper.conversation.ExternalConvRequestMapper;
import com.example.mapper.messagefilter.ConversationContextMapper;
import com.example.mapper.messagefilter.NotificationContextMapper;
import com.example.service.base.DialogflowClientService;
import com.example.service.base.MessageEntryFilter;
import com.example.service.notification.MemoryStoreNotificationService;
import com.example.service.quickreplies.QuickRepliesManagerService;
import com.example.util.SessionIdGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
import java.time.Instant;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
/**
* Service for orchestrating the end-to-end conversation flow.
* It manages user sessions, creating new ones or reusing existing ones
* based on a session reset threshold. The service handles the entire
* conversation turn, from mapping an external request to calling Dialogflow,
* and then persists both user and agent messages using a write-back strategy
* to a primary cache (Redis) and an asynchronous write to Firestore.
*/
@Service
public class ConversationManagerService {
private static final Logger logger = LoggerFactory.getLogger(ConversationManagerService.class);
private static final long SESSION_RESET_THRESHOLD_HOURS = 24;
private static final String CURRENT_PAGE_PARAM = "currentPage";
private static final Logger logger = LoggerFactory.getLogger(ConversationManagerService.class);
private final ExternalConvRequestMapper externalRequestToDialogflowMapper;
private final DialogflowClientService dialogflowServiceClient;
private final FirestoreConversationService firestoreConversationService;
private final MemoryStoreConversationService memoryStoreConversationService;
private final QuickRepliesManagerService quickRepliesManagerService;
private final MessageEntryFilter messageEntryFilter;
private final MemoryStoreNotificationService memoryStoreNotificationService;
private final NotificationContextMapper notificationContextMapper;
private final ConversationContextMapper conversationContextMapper;
private static final long SESSION_RESET_THRESHOLD_HOURS = 24;
private static final String CURRENT_PAGE_PARAM = "currentPage";
private final ExternalConvRequestMapper externalRequestToDialogflowMapper;
private final DialogflowClientService dialogflowServiceClient;
private final FirestoreConversationService firestoreConversationService;
private final MemoryStoreConversationService memoryStoreConversationService;
public ConversationManagerService(
DialogflowClientService dialogflowServiceClient,
FirestoreConversationService firestoreConversationService,
MemoryStoreConversationService memoryStoreConversationService,
ExternalConvRequestMapper externalRequestToDialogflowMapper) {
public ConversationManagerService(
DialogflowClientService dialogflowServiceClient,
FirestoreConversationService firestoreConversationService,
MemoryStoreConversationService memoryStoreConversationService,
ExternalConvRequestMapper externalRequestToDialogflowMapper,
QuickRepliesManagerService quickRepliesManagerService,
MessageEntryFilter messageEntryFilter,
MemoryStoreNotificationService memoryStoreNotificationService,
NotificationContextMapper notificationContextMapper,
ConversationContextMapper conversationContextMapper) {
this.dialogflowServiceClient = dialogflowServiceClient;
this.firestoreConversationService = firestoreConversationService;
this.memoryStoreConversationService = memoryStoreConversationService;
this.externalRequestToDialogflowMapper = externalRequestToDialogflowMapper;
}
this.quickRepliesManagerService = quickRepliesManagerService;
this.messageEntryFilter = messageEntryFilter;
this.memoryStoreNotificationService = memoryStoreNotificationService;
this.notificationContextMapper = notificationContextMapper;
this.conversationContextMapper = conversationContextMapper;
}
public Mono<DetectIntentResponseDTO>manageConversation(ExternalConvRequestDTO Externalrequest) {
final DetectIntentRequestDTO request;
try {
request = externalRequestToDialogflowMapper.mapExternalRequestToDetectIntentRequest(Externalrequest);
logger.debug("Successfully pre-mapped ExternalRequestDTO to DetectIntentRequestDTO");
} catch (IllegalArgumentException e) {
logger.error("Error during pre-mapping: {}", e.getMessage());
return Mono.error(new IllegalArgumentException("Failed to process external request due to mapping error: " + e.getMessage(), e));
public Mono<DetectIntentResponseDTO> manageConversation(ExternalConvRequestDTO externalrequest) {
return memoryStoreConversationService.getSessionByTelefono(externalrequest.user().telefono())
.flatMap(session -> {
if (session != null && !session.entries().isEmpty()) {
ConversationEntryDTO lastEntry = session.entries().get(session.entries().size() - 1);
if (lastEntry.entity() == ConversationEntryEntity.SISTEMA && lastEntry.type() == ConversationEntryType.INICIO) {
logger.info("Detected 'SISTEMA' and 'INICIO' values in last session entry. Delegating to QuickRepliesManagerService.");
ExternalConvRequestDTO updatedRequest = new ExternalConvRequestDTO(
externalrequest.message(),
externalrequest.user(),
externalrequest.channel(),
externalrequest.tipo(),
session.pantallaContexto()
);
return quickRepliesManagerService.manageConversation(updatedRequest);
}
}
return continueManagingConversation(externalrequest);
})
.switchIfEmpty(continueManagingConversation(externalrequest));
}
final ConversationContext context;
try {
context = resolveAndValidateRequest(request);
} catch (IllegalArgumentException e) {
logger.error("Validation error for incoming request: {}", e.getMessage());
return Mono.error(e);
private Mono<DetectIntentResponseDTO> continueManagingConversation(ExternalConvRequestDTO externalrequest) {
final DetectIntentRequestDTO request;
try {
request = externalRequestToDialogflowMapper.mapExternalRequestToDetectIntentRequest(externalrequest);
logger.debug("Successfully pre-mapped ExternalRequestDTO to DetectIntentRequestDTO");
} catch (IllegalArgumentException e) {
logger.error("Error during pre-mapping: {}", e.getMessage());
return Mono.error(new IllegalArgumentException(
"Failed to process external request due to mapping error: " + e.getMessage(), e));
}
final ConversationContext context;
try {
context = resolveAndValidateRequest(request);
} catch (IllegalArgumentException e) {
logger.error("Validation error for incoming request: {}", e.getMessage());
return Mono.error(e);
}
return handleMessageClassification(context, request);
}
private Mono<DetectIntentResponseDTO> handleMessageClassification(ConversationContext context, DetectIntentRequestDTO request) {
final String userPhoneNumber = context.primaryPhoneNumber();
final String userMessageText = context.userMessageText();
return memoryStoreNotificationService.getNotificationIdForPhone(userPhoneNumber)
.flatMap(notificationId -> memoryStoreNotificationService.getCachedNotificationSession(notificationId))
.map(notificationSession -> notificationSession.notificaciones().stream()
.filter(notification -> "active".equalsIgnoreCase(notification.status()))
.max(java.util.Comparator.comparing(NotificationDTO::timestampCreacion))
.orElse(null))
.filter(Objects::nonNull)
.flatMap((NotificationDTO notification) -> {
String notificationText = notificationContextMapper.toText(notification);
return memoryStoreConversationService.getSessionByTelefono(userPhoneNumber)
.map(conversationContextMapper::toText)
.defaultIfEmpty("")
.flatMap(conversationHistory -> {
String classification = messageEntryFilter.classifyMessage(userMessageText, notificationText, conversationHistory);
if (MessageEntryFilter.CATEGORY_NOTIFICATION.equals(classification)) {
return startNotificationConversation(context, request, notification);
} else {
return continueConversationFlow(context, request);
}
});
})
.switchIfEmpty(continueConversationFlow(context, request));
}
private Mono<DetectIntentResponseDTO> continueConversationFlow(ConversationContext context, DetectIntentRequestDTO request) {
final String userId = context.userId();
final String userMessageText = context.userMessageText();
final String userPhoneNumber = context.primaryPhoneNumber();
if (userPhoneNumber == null || userPhoneNumber.isBlank()) {
logger.warn("No phone number provided in request. Cannot manage conversation session without it.");
return Mono.error(new IllegalArgumentException("Phone number is required to manage conversation sessions."));
}
logger.info("Primary Check (MemoryStore): Looking up session for phone number: {}", userPhoneNumber);
return memoryStoreConversationService.getSessionByTelefono(userPhoneNumber)
.flatMap(session -> {
Instant now = Instant.now();
if (Duration.between(session.lastModified(), now).toHours() < SESSION_RESET_THRESHOLD_HOURS) {
logger.info("Recent Session Found: Session {} is within the 24-hour threshold. Proceeding to Dialogflow.", session.sessionId());
return processDialogflowRequest(session, request, userId, userMessageText, userPhoneNumber, false);
} else {
logger.info("Old Session Found: Session {} is older than the threshold. Proceeding to full lookup.", session.sessionId());
return fullLookupAndProcess(session, request, userId, userMessageText, userPhoneNumber);
}
})
.switchIfEmpty(Mono.defer(() -> {
logger.info("No session found in MemoryStore. Performing full lookup to Firestore.");
return fullLookupAndProcess(null, request, userId, userMessageText, userPhoneNumber);
}))
.onErrorResume(e -> {
logger.error("Overall error handling conversation in ConversationManagerService: {}", e.getMessage(), e);
return Mono.error(new RuntimeException("Failed to process conversation due to an internal error.", e));
});
}
private Mono<DetectIntentResponseDTO> fullLookupAndProcess(ConversationSessionDTO oldSession, DetectIntentRequestDTO request, String userId, String userMessageText, String userPhoneNumber) {
return firestoreConversationService.getSessionByTelefono(userPhoneNumber)
.map(conversationContextMapper::toTextWithLimits)
.defaultIfEmpty("")
.flatMap(conversationHistory -> {
String newSessionId = SessionIdGenerator.generateStandardSessionId();
logger.info("Creating new session {} after full lookup.", newSessionId);
ConversationSessionDTO newSession = ConversationSessionDTO.create(newSessionId, userId, userPhoneNumber);
DetectIntentRequestDTO newRequest = request.withParameter(CURRENT_PAGE_PARAM, conversationHistory);
return processDialogflowRequest(newSession, newRequest, userId, userMessageText, userPhoneNumber, true);
});
}
private Mono<DetectIntentResponseDTO> processDialogflowRequest(ConversationSessionDTO session, DetectIntentRequestDTO request, String userId, String userMessageText, String userPhoneNumber, boolean newSession) {
final String finalSessionId = session.sessionId();
ConversationEntryDTO userEntry = ConversationEntryDTO.forUser(userMessageText);
return this.persistConversationTurn(userId, finalSessionId, userEntry, userPhoneNumber)
.doOnSuccess(v -> logger.debug("User entry successfully persisted for session {}. Proceeding to Dialogflow...", finalSessionId))
.doOnError(e -> logger.error("Error during user entry persistence for session {}: {}", finalSessionId, e.getMessage(), e))
.then(Mono.defer(() -> dialogflowServiceClient.detectIntent(finalSessionId, request)
.flatMap(response -> {
logger.debug("Received Dialogflow CX response for session {}. Initiating agent response persistence.", finalSessionId);
ConversationEntryDTO agentEntry = ConversationEntryDTO.forAgent(response.queryResult());
return persistConversationTurn(userId, finalSessionId, agentEntry, userPhoneNumber)
.thenReturn(response);
})
.doOnError(error -> logger.error("Overall error during conversation management for session {}: {}", finalSessionId, error.getMessage(), error))
));
}
private Mono<DetectIntentResponseDTO> startNotificationConversation(ConversationContext context, DetectIntentRequestDTO request, NotificationDTO notification) {
final String userId = context.userId();
final String userMessageText = context.userMessageText();
final String userPhoneNumber = context.primaryPhoneNumber();
Mono<ConversationSessionDTO> sessionMono;
if (userPhoneNumber != null && !userPhoneNumber.isBlank()) {
logger.info("Checking for existing session for phone number: {}", userPhoneNumber);
sessionMono = memoryStoreConversationService.getSessionByTelefono(userPhoneNumber)
.doOnNext(session -> logger.info("Found existing session {} for phone number {}", session.sessionId(), userPhoneNumber))
return memoryStoreNotificationService.getSessionByTelefono(userPhoneNumber)
.switchIfEmpty(Mono.defer(() -> {
String newSessionId = SessionIdGenerator.generateStandardSessionId();
logger.info("No existing session found for phone number {}. Creating new session: {}", userPhoneNumber, newSessionId);
return Mono.just(ConversationSessionDTO.create(newSessionId, userId, userPhoneNumber));
}));
} else {
logger.warn("No phone number provided in request. Cannot manage conversation session without it.");
return Mono.error(new IllegalArgumentException("Phone number is required to manage conversation sessions."));
String newSessionId = SessionIdGenerator.generateStandardSessionId();
logger.info("No existing notification session found for phone number {}. Creating new session: {}",
userPhoneNumber, newSessionId);
return Mono.just(ConversationSessionDTO.create(newSessionId, userId, userPhoneNumber));
}))
.flatMap(session -> {
final String sessionId = session.sessionId();
ConversationEntryDTO userEntry = ConversationEntryDTO.forUser(userMessageText);
return memoryStoreNotificationService.saveEntry(userId, sessionId, userEntry, userPhoneNumber)
.then(dialogflowServiceClient.detectIntent(sessionId, request)
.doOnSuccess(response -> {
ConversationEntryDTO agentEntry = ConversationEntryDTO.forAgent(response.queryResult());
memoryStoreNotificationService.saveEntry(userId, sessionId, agentEntry, userPhoneNumber).subscribe();
}));
});
}
return sessionMono.flatMap(session -> {
final String finalSessionId = session.sessionId();
private Mono<Void> persistConversationTurn(String userId, String sessionId, ConversationEntryDTO entry,String userPhoneNumber) {
logger.debug("Starting Write-Back persistence for session {}. Type: {}. Writing to Redis first.", sessionId,
entry.type().name());
logger.info("Managing conversation for resolved session: {}", finalSessionId);
ConversationEntryDTO userEntry = ConversationEntryDTO.forUser(userMessageText);
final DetectIntentRequestDTO requestToDialogflow;
Instant currentInteractionTimestamp = userEntry.timestamp();
if (session.lastModified() != null &&
Duration.between(session.lastModified(), currentInteractionTimestamp).toHours() >= SESSION_RESET_THRESHOLD_HOURS) {
logger.info("Session {} (last modified: {}) is older than {} hours. Adding '{}' parameter to Dialogflow request.",
session.sessionId(), session.lastModified(), SESSION_RESET_THRESHOLD_HOURS, CURRENT_PAGE_PARAM);
requestToDialogflow = request.withParameter(CURRENT_PAGE_PARAM, true);
} else {
requestToDialogflow = request;
}
return this.persistConversationTurn(userId, finalSessionId, userEntry, userPhoneNumber)
.doOnSuccess(v -> logger.debug("User entry successfully persisted for session {}. Proceeding to Dialogflow...", finalSessionId))
.doOnError(e -> logger.error("Error during user entry persistence for session {}: {}", finalSessionId, e.getMessage(), e))
.then(Mono.defer(() -> {
return dialogflowServiceClient.detectIntent(finalSessionId, requestToDialogflow)
.doOnSuccess(response -> {
logger.debug("Received Dialogflow CX response for session {}. Initiating agent response persistence.", finalSessionId);
ConversationEntryDTO agentEntry = ConversationEntryDTO.forAgent(response.queryResult());
this.persistConversationTurn(userId, finalSessionId, agentEntry, userPhoneNumber).subscribe(
v -> logger.debug("Background: Agent entry persistence initiated for session {}.", finalSessionId),
e -> logger.error("Background: Error during agent entry persistence for session {}: {}", finalSessionId, e.getMessage(), e)
);
})
.doOnError(error -> logger.error("Overall error during conversation management for session {}: {}", finalSessionId, error.getMessage(), error));
}));
})
.onErrorResume(e -> {
logger.error("Overall error handling conversation in ConversationManagerService: {}", e.getMessage(), e);
return Mono.error(new RuntimeException("Failed to process conversation due to an internal error.", e));
})
.subscribeOn(Schedulers.boundedElastic());
}
private Mono<Void> persistConversationTurn(String userId, String sessionId, ConversationEntryDTO entry, String userPhoneNumber) {
logger.debug("Starting Write-Back persistence for session {}. Type: {}. Writing to Redis first.", sessionId, entry.type().name());
return memoryStoreConversationService.saveEntry(userId, sessionId, entry, userPhoneNumber)
.doOnSuccess(v -> {
logger.info("Entry saved to Redis for session {}. Type: {}. Kicking off async Firestore write-back.", sessionId, entry.type().name());
firestoreConversationService.saveEntry(userId, sessionId, entry, userPhoneNumber)
.subscribe(
fsVoid -> logger.debug("Asynchronously (Write-Back): Entry successfully saved to Firestore for session {}. Type: {}.",
sessionId, entry.type().name()),
fsError -> logger.error("Asynchronously (Write-Back): Failed to save entry to Firestore for session {}. Type: {}: {}",
sessionId, entry.type().name(), fsError.getMessage(), fsError)
);
})
.doOnError(e -> logger.error("Error during primary Redis write for session {}. Type: {}: {}", sessionId, entry.type().name(), e.getMessage(), e));
}
.doOnSuccess(v -> logger.info(
"Entry saved to Redis for session {}. Type: {}. Kicking off async Firestore write-back.",
sessionId, entry.type().name()))
.then(firestoreConversationService.saveEntry(userId, sessionId, entry, userPhoneNumber)
.doOnSuccess(fsVoid -> logger.debug(
"Asynchronously (Write-Back): Entry successfully saved to Firestore for session {}. Type: {}.",
sessionId, entry.type().name()))
.doOnError(fsError -> logger.error(
"Asynchronously (Write-Back): Failed to save entry to Firestore for session {}. Type: {}: {}",
sessionId, entry.type().name(), fsError.getMessage(), fsError)))
.doOnError(e -> logger.error("Error during primary Redis write for session {}. Type: {}: {}", sessionId,
entry.type().name(), e.getMessage(), e));
}
private ConversationContext resolveAndValidateRequest(DetectIntentRequestDTO request) {
private ConversationContext resolveAndValidateRequest(DetectIntentRequestDTO request) {
Map<String, Object> params = Optional.ofNullable(request.queryParams())
.map(queryParamsDTO -> queryParamsDTO.parameters())
.orElse(Collections.emptyMap());
.map(queryParamsDTO -> queryParamsDTO.parameters())
.orElse(Collections.emptyMap());
String primaryPhoneNumber = null;
Object telefonoObj = params.get("telefono"); // Get from map
if (telefonoObj instanceof String) {
primaryPhoneNumber = (String) telefonoObj;
primaryPhoneNumber = (String) telefonoObj;
} else if (telefonoObj != null) {
logger.warn("Parameter 'telefono' in queryParams is not a String (type: {}). Expected String.", telefonoObj.getClass().getName());
logger.warn("Parameter 'telefono' in queryParams is not a String (type: {}). Expected String.",
telefonoObj.getClass().getName());
}
if (primaryPhoneNumber == null || primaryPhoneNumber.trim().isEmpty()) {
throw new IllegalArgumentException("Phone number (telefono) is required in query parameters for conversation management.");
throw new IllegalArgumentException(
"Phone number (telefono) is required in query parameters for conversation management.");
}
String resolvedUserId = null;
Object userIdObj = params.get("usuario_id");
if (userIdObj instanceof String) {
resolvedUserId = (String) userIdObj;
resolvedUserId = (String) userIdObj;
} else if (userIdObj != null) {
logger.warn("Parameter 'userId' in queryParams is not a String (type: {}). Expected String.", userIdObj.getClass().getName());
}
if (resolvedUserId == null || resolvedUserId.trim().isEmpty()) {
resolvedUserId = "user_by_phone_" + primaryPhoneNumber.replaceAll("[^0-9]", "");
logger.warn("User ID not provided in query parameters. Using derived ID from phone number: {}", resolvedUserId);
logger.warn("Parameter 'userId' in query_params is not a String (type: {}). Expected String.",
userIdObj.getClass().getName());
}
if (request.queryInput() == null || request.queryInput().text() == null ||
request.queryInput().text().text() == null || request.queryInput().text().text().trim().isEmpty()) {
throw new IllegalArgumentException("Dialogflow query input text is required.");
if (resolvedUserId == null || resolvedUserId.trim().isEmpty()) {
resolvedUserId = "user_by_phone_" + primaryPhoneNumber.replaceAll("[^0-9]", "");
logger.warn("User ID not provided in query parameters. Using derived ID from phone number: {}",
resolvedUserId);
}
if (request.queryInput() == null || request.queryInput().text() == null ||
request.queryInput().text().text() == null || request.queryInput().text().text().trim().isEmpty()) {
throw new IllegalArgumentException("Dialogflow query input text is required.");
}
String userMessageText = request.queryInput().text().text();
return new ConversationContext(resolvedUserId, null, userMessageText, primaryPhoneNumber);
}
}
}

23
src/main/java/com/example/service/conversation/FirestoreConversationService.java
View File

@@ -43,6 +43,10 @@ public class FirestoreConversationService {
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber) {
return saveEntry(userId, sessionId, newEntry, userPhoneNumber, null);
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber, String pantallaContexto) {
logger.info("Attempting to save conversation entry to Firestore for session {}. Entity: {}", sessionId, newEntry.entity().name());
return Mono.fromRunnable(() -> {
DocumentReference sessionDocRef = getSessionDocumentReference(sessionId);
@@ -52,12 +56,15 @@ public class FirestoreConversationService {
if (firestoreBaseRepository.documentExists(sessionDocRef)) {
// Update: Append the new entry using arrayUnion and update lastModified
Map<String, Object> updates = firestoreConversationMapper.createUpdateMapForSingleEntry(newEntry);
if (pantallaContexto != null) {
updates.put("pantallaContexto", pantallaContexto);
}
batch.update(sessionDocRef, updates);
logger.info("Appending entry to existing conversation session for user {} and session {}. Entity: {}", userId, sessionId, newEntry.entity().name());
} else {
// Create: Start a new session with the first entry.
// Pass userId and userPhoneNumber to the mapper to be stored as fields in the document.
Map<String, Object> newSessionMap = firestoreConversationMapper.createNewSessionMapForSingleEntry(sessionId, userId, userPhoneNumber, newEntry);
Map<String, Object> newSessionMap = firestoreConversationMapper.createNewSessionMapForSingleEntry(sessionId, userId, userPhoneNumber, newEntry, pantallaContexto);
batch.set(sessionDocRef, newSessionMap);
logger.info("Creating new conversation session with first entry for user {} and session {}. Entity: {}", userId, sessionId, newEntry.entity().name());
}
@@ -94,6 +101,20 @@ public class FirestoreConversationService {
}).subscribeOn(Schedulers.boundedElastic());
}
public Mono<ConversationSessionDTO> getSessionByTelefono(String userPhoneNumber) {
logger.info("Attempting to retrieve conversation session for phone number {}.", userPhoneNumber);
return firestoreBaseRepository.getDocumentsByField(getConversationCollectionPath(), "userPhoneNumber", userPhoneNumber)
.map(documentSnapshot -> {
if (documentSnapshot != null && documentSnapshot.exists()) {
ConversationSessionDTO sessionDTO = firestoreConversationMapper.mapFirestoreDocumentToConversationSessionDTO(documentSnapshot);
logger.info("Successfully retrieved and mapped conversation session for session {}.", sessionDTO.sessionId());
return sessionDTO;
}
logger.info("Conversation session not found for phone number {}.", userPhoneNumber);
return null;
});
}
private String getConversationCollectionPath() {
return String.format(CONVERSATION_COLLECTION_PATH_FORMAT, firestoreBaseRepository.getAppId());
}

28
src/main/java/com/example/service/conversation/MemoryStoreConversationService.java
View File

@@ -29,28 +29,48 @@ public class MemoryStoreConversationService {
private static final Duration SESSION_TTL = Duration.ofHours(24);
private final ReactiveRedisTemplate<String, ConversationSessionDTO> redisTemplate;
private final ReactiveRedisTemplate<String, String> stringRedisTemplate;
@Autowired
public MemoryStoreConversationService(
ReactiveRedisTemplate<String, ConversationSessionDTO> redisTemplate,
ReactiveRedisTemplate<String, String> stringRedisTemplate) {
ReactiveRedisTemplate<String, String> stringRedisTemplate,
FirestoreConversationService firestoreConversationService) {
this.redisTemplate = redisTemplate;
this.stringRedisTemplate = stringRedisTemplate;
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber) {
return saveEntry(userId, sessionId, newEntry, userPhoneNumber, null);
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber, String pantallaContexto) {
String sessionKey = SESSION_KEY_PREFIX + sessionId;
String phoneToSessionKey = PHONE_TO_SESSION_KEY_PREFIX + userPhoneNumber;
logger.info("Attempting to save entry to Redis for session {}. Entity: {}", sessionId, newEntry.entity().name());
return redisTemplate.opsForValue().get(sessionKey)
.defaultIfEmpty(ConversationSessionDTO.create(sessionId, userId, userPhoneNumber))
.flatMap(session -> {
ConversationSessionDTO sessionWithUpdatedTelefono = session.withTelefono(userPhoneNumber);
ConversationSessionDTO updatedSession = sessionWithUpdatedTelefono.withAddedEntry(newEntry);
ConversationSessionDTO sessionWithPantallaContexto = (pantallaContexto != null) ? sessionWithUpdatedTelefono.withPantallaContexto(pantallaContexto) : sessionWithUpdatedTelefono;
ConversationSessionDTO updatedSession = sessionWithPantallaContexto.withAddedEntry(newEntry);
logger.info("Attempting to set updated session {} with new entry entity {} in Redis.", sessionId, newEntry.entity().name());
return redisTemplate.opsForValue().set(sessionKey, updatedSession, SESSION_TTL)
.then(stringRedisTemplate.opsForValue().set(phoneToSessionKey, sessionId, SESSION_TTL))
.then(); // <--- ADD THIS .then() WITHOUT ARGUMENTS
.then();
})
.doOnSuccess(success -> {
logger.info("Successfully saved updated session and phone mapping to Redis for session {}. Entity Type: {}", sessionId, newEntry.entity().name());
})
.doOnSuccess(success -> logger.info("Successfully saved updated session and phone mapping to Redis for session {}. Entity Type: {}", sessionId, newEntry.entity().name()))
.doOnError(e -> logger.error("Error appending entry to Redis for session {}: {}", sessionId, e.getMessage(), e));
}
public Mono<ConversationSessionDTO> getSessionByTelefono(String telefono) {

107
src/main/java/com/example/service/notification/FirestoreNotificationConvService.java Normal file
View File

@@ -0,0 +1,107 @@
/*
* 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.conversation.ConversationEntryDTO;
import com.example.dto.dialogflow.conversation.ConversationSessionDTO;
import com.example.exception.FirestorePersistenceException;
import com.example.mapper.conversation.FirestoreConversationMapper;
import com.example.repository.FirestoreBaseRepository;
import com.google.cloud.firestore.DocumentReference;
import com.google.cloud.firestore.DocumentSnapshot;
import com.google.cloud.firestore.WriteBatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.util.Map;
import java.util.concurrent.ExecutionException;
/**
* Service for managing notification conversation sessions in Firestore.
* It handles the persistence of conversation entries, either by creating
* a new document for a new session or appending an entry to an existing
* session document using a Firestore batch. The service also provides
* methods for retrieving a complete conversation session from Firestore.
*/
@Service
public class FirestoreNotificationConvService {
private static final Logger logger = LoggerFactory.getLogger(FirestoreNotificationConvService.class);
private static final String CONVERSATION_COLLECTION_PATH_FORMAT = "artifacts/%s/conversation-notifications";
private final FirestoreBaseRepository firestoreBaseRepository;
private final FirestoreConversationMapper firestoreConversationMapper;
public FirestoreNotificationConvService(FirestoreBaseRepository firestoreBaseRepository, FirestoreConversationMapper firestoreConversationMapper) {
this.firestoreBaseRepository = firestoreBaseRepository;
this.firestoreConversationMapper = firestoreConversationMapper;
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber) {
logger.info("Attempting to save conversation entry to Firestore for session {}. Entity: {}", sessionId, newEntry.entity().name());
return Mono.fromRunnable(() -> {
DocumentReference sessionDocRef = getSessionDocumentReference(sessionId);
// Synchronize on the session ID to prevent race conditions when creating a new session.
synchronized (sessionId.intern()) {
WriteBatch batch = firestoreBaseRepository.createBatch();
try {
if (firestoreBaseRepository.documentExists(sessionDocRef)) {
// Update: Append the new entry using arrayUnion and update lastModified
Map<String, Object> updates = firestoreConversationMapper.createUpdateMapForSingleEntry(newEntry);
batch.update(sessionDocRef, updates);
logger.info("Appending entry to existing conversation session for user {} and session {}. Entity: {}", userId, sessionId, newEntry.entity().name());
} else {
// Create: Start a new session with the first entry.
// Pass userId and userPhoneNumber to the mapper to be stored as fields in the document.
Map<String, Object> newSessionMap = firestoreConversationMapper.createNewSessionMapForSingleEntry(sessionId, userId, userPhoneNumber, newEntry);
batch.set(sessionDocRef, newSessionMap);
logger.info("Creating new conversation session with first entry for user {} and session {}. Entity: {}", userId, sessionId, newEntry.entity().name());
}
firestoreBaseRepository.commitBatch(batch);
logger.info("Successfully committed batch for session {} to Firestore.", sessionId);
} catch (ExecutionException e) {
logger.error("Error saving conversation entry to Firestore for session {}: {}", sessionId, e.getMessage(), e);
throw new FirestorePersistenceException("Failed to save conversation entry to Firestore for session " + sessionId, e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.error("Thread interrupted while saving conversation entry to Firestore for session {}: {}", sessionId, e.getMessage(), e);
throw new FirestorePersistenceException("Saving conversation entry was interrupted for session " + sessionId, e);
}
}
}).subscribeOn(Schedulers.boundedElastic()).then();
}
public Mono<ConversationSessionDTO> getConversationSession(String userId, String sessionId) {
logger.info("Attempting to retrieve conversation session for session {} (user ID {} for context).", sessionId, userId);
return Mono.fromCallable(() -> {
DocumentReference sessionDocRef = getSessionDocumentReference(sessionId);
try {
DocumentSnapshot documentSnapshot = firestoreBaseRepository.getDocumentSnapshot(sessionDocRef);
if (documentSnapshot != null && documentSnapshot.exists()) {
ConversationSessionDTO sessionDTO = firestoreConversationMapper.mapFirestoreDocumentToConversationSessionDTO(documentSnapshot);
logger.info("Successfully retrieved and mapped conversation session for session {}.", sessionId);
return sessionDTO;
}
logger.info("Conversation session not found for session {}.", sessionId);
return null; // Or Mono.empty() if this method returned Mono<Optional<ConversationSessionDTO>>
} catch (InterruptedException | ExecutionException e) {
logger.error("Error retrieving conversation session from Firestore for session {}: {}", sessionId, e.getMessage(), e);
throw new FirestorePersistenceException("Failed to retrieve conversation session from Firestore for session " + sessionId, e);
}
}).subscribeOn(Schedulers.boundedElastic());
}
private String getConversationCollectionPath() {
return String.format(CONVERSATION_COLLECTION_PATH_FORMAT, firestoreBaseRepository.getAppId());
}
private DocumentReference getSessionDocumentReference(String sessionId) {
String collectionPath = getConversationCollectionPath();
return firestoreBaseRepository.getDocumentReference(collectionPath, sessionId);
}
}

209
src/main/java/com/example/service/notification/FirestoreNotificationService.java
View File

@@ -1,3 +1,8 @@
/*
* 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.NotificationDTO;
@@ -7,91 +12,157 @@ import com.example.repository.FirestoreBaseRepository;
import com.google.cloud.Timestamp;
import com.google.cloud.firestore.DocumentReference;
import com.google.cloud.firestore.FieldValue;
import java.time.Instant;
import java.util.Collections;
import java.util.Map;
import java.util.List;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.time.Instant;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ExecutionException;
@Service
public class FirestoreNotificationService {
private static final Logger logger = LoggerFactory.getLogger(FirestoreNotificationService.class);
private static final String NOTIFICATION_COLLECTION_PATH_FORMAT = "artifacts/%s/notifications";
private static final String FIELD_MESSAGES = "notificaciones";
private static final String FIELD_LAST_UPDATED = "ultimaActualizacion";
private static final String FIELD_PHONE_NUMBER = "telefono";
private static final String FIELD_NOTIFICATION_ID = "sessionId";
private static final Logger logger = LoggerFactory.getLogger(FirestoreNotificationService.class);
private static final String NOTIFICATION_COLLECTION_PATH_FORMAT = "artifacts/%s/notifications";
private static final String FIELD_MESSAGES = "notificaciones";
private static final String FIELD_LAST_UPDATED = "ultimaActualizacion";
private static final String FIELD_PHONE_NUMBER = "telefono";
private static final String FIELD_NOTIFICATION_ID = "sessionId";
private final FirestoreBaseRepository firestoreBaseRepository;
private final FirestoreNotificationMapper firestoreNotificationMapper;
private final FirestoreBaseRepository firestoreBaseRepository;
private final FirestoreNotificationMapper firestoreNotificationMapper;
public FirestoreNotificationService(
FirestoreBaseRepository firestoreBaseRepository,
FirestoreNotificationMapper firestoreNotificationMapper) {
this.firestoreBaseRepository = firestoreBaseRepository;
this.firestoreNotificationMapper = firestoreNotificationMapper;
}
public FirestoreNotificationService(
FirestoreBaseRepository firestoreBaseRepository,
FirestoreNotificationMapper firestoreNotificationMapper,
MemoryStoreNotificationService memoryStoreNotificationService) {
this.firestoreBaseRepository = firestoreBaseRepository;
this.firestoreNotificationMapper = firestoreNotificationMapper;
}
public Mono<Void> saveOrAppendNotificationEntry(NotificationDTO newEntry) {
return Mono.fromRunnable(
() -> {
String phoneNumber = newEntry.telefono();
if (phoneNumber == null || phoneNumber.isBlank()) {
throw new IllegalArgumentException(
"Phone number is required to manage notification entries.");
}
// Use the phone number as the document ID for the session.
String notificationSessionId = phoneNumber;
public Mono<Void> saveOrAppendNotificationEntry(NotificationDTO newEntry) {
return Mono.fromRunnable(() -> {
String phoneNumber = newEntry.telefono();
if (phoneNumber == null || phoneNumber.isBlank()) {
throw new IllegalArgumentException("Phone number is required to manage notification entries.");
}
// Synchronize on the notification session ID to prevent race conditions when creating a new session.
synchronized (notificationSessionId.intern()) {
DocumentReference notificationDocRef =
getNotificationDocumentReference(notificationSessionId);
Map<String, Object> entryMap =
firestoreNotificationMapper.mapNotificationDTOToMap(newEntry);
try {
// Check if the session document exists.
boolean docExists = firestoreBaseRepository.documentExists(notificationDocRef);
// Use the phone number as the document ID for the session.
String notificationSessionId = phoneNumber;
DocumentReference notificationDocRef = getNotificationDocumentReference(notificationSessionId);
Map<String, Object> entryMap = firestoreNotificationMapper.mapNotificationDTOToMap(newEntry);
try {
// Check if the session document exists.
boolean docExists = notificationDocRef.get().get().exists();
if (docExists) {
if (docExists) {
// If the document exists, append the new entry to the 'notificaciones' array.
Map<String, Object> updates = Map.of(
FIELD_MESSAGES, FieldValue.arrayUnion(entryMap),
FIELD_LAST_UPDATED, Timestamp.of(java.util.Date.from(Instant.now()))
);
notificationDocRef.update(updates).get();
logger.info("Successfully appended new entry to notification session {} in Firestore.", notificationSessionId);
} else {
Map<String, Object> updates =
Map.of(
FIELD_MESSAGES, FieldValue.arrayUnion(entryMap),
FIELD_LAST_UPDATED, Timestamp.of(java.util.Date.from(Instant.now())));
firestoreBaseRepository.updateDocument(notificationDocRef, updates);
logger.info(
"Successfully appended new entry to notification session {} in Firestore.",
notificationSessionId);
} else {
// If the document does not exist, create a new session document.
Map<String, Object> newSessionData = Map.of(
FIELD_NOTIFICATION_ID, notificationSessionId,
FIELD_PHONE_NUMBER, phoneNumber,
"fechaCreacion", Timestamp.of(java.util.Date.from(Instant.now())),
FIELD_LAST_UPDATED, Timestamp.of(java.util.Date.from(Instant.now())),
FIELD_MESSAGES, Collections.singletonList(entryMap)
);
notificationDocRef.set(newSessionData).get();
logger.info("Successfully created a new notification session {} in Firestore.", notificationSessionId);
Map<String, Object> newSessionData =
Map.of(
FIELD_NOTIFICATION_ID,
notificationSessionId,
FIELD_PHONE_NUMBER,
phoneNumber,
"fechaCreacion",
Timestamp.of(java.util.Date.from(Instant.now())),
FIELD_LAST_UPDATED,
Timestamp.of(java.util.Date.from(Instant.now())),
FIELD_MESSAGES,
Collections.singletonList(entryMap));
firestoreBaseRepository.setDocument(notificationDocRef, newSessionData);
logger.info(
"Successfully created a new notification session {} in Firestore.",
notificationSessionId);
}
} catch (ExecutionException e) {
logger.error(
"Error saving notification to Firestore for phone {}: {}",
phoneNumber,
e.getMessage(),
e);
throw new FirestorePersistenceException(
"Failed to save notification to Firestore for phone " + phoneNumber, e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.error(
"Thread interrupted while saving notification to Firestore for phone {}: {}",
phoneNumber,
e.getMessage(),
e);
throw new FirestorePersistenceException(
"Saving notification was interrupted for phone " + phoneNumber, e);
}
} catch (ExecutionException e) {
logger.error("Error saving notification to Firestore for phone {}: {}", phoneNumber, e.getMessage(), e);
throw new FirestorePersistenceException("Failed to save notification to Firestore for phone " + phoneNumber, e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.error("Thread interrupted while saving notification to Firestore for phone {}: {}", phoneNumber, e.getMessage(), e);
throw new FirestorePersistenceException("Saving notification was interrupted for phone " + phoneNumber, e);
}
}).subscribeOn(Schedulers.boundedElastic()).then();
}
private String getNotificationCollectionPath() {
return String.format(NOTIFICATION_COLLECTION_PATH_FORMAT, firestoreBaseRepository.getAppId());
}
}
})
.subscribeOn(Schedulers.boundedElastic())
.then();
}
private DocumentReference getNotificationDocumentReference(String notificationId) {
String collectionPath = getNotificationCollectionPath();
return firestoreBaseRepository.getDocumentReference(collectionPath, notificationId);
}
private String getNotificationCollectionPath() {
return String.format(NOTIFICATION_COLLECTION_PATH_FORMAT, firestoreBaseRepository.getAppId());
}
private DocumentReference getNotificationDocumentReference(String notificationId) {
String collectionPath = getNotificationCollectionPath();
return firestoreBaseRepository.getDocumentReference(collectionPath, notificationId);
}
@SuppressWarnings("unchecked")
public Mono<Void> updateNotificationStatus(String sessionId, String status) {
return Mono.fromRunnable(() -> {
DocumentReference notificationDocRef = getNotificationDocumentReference(sessionId);
try {
Map<String, Object> sessionData = firestoreBaseRepository.getDocument(notificationDocRef, Map.class);
if (sessionData != null) {
List<Map<String, Object>> notifications = (List<Map<String, Object>>) sessionData.get(FIELD_MESSAGES);
if (notifications != null) {
List<Map<String, Object>> updatedNotifications = new ArrayList<>();
for (Map<String, Object> notification : notifications) {
Map<String, Object> updatedNotification = new HashMap<>(notification);
updatedNotification.put("status", status);
updatedNotifications.add(updatedNotification);
}
Map<String, Object> updates = new HashMap<>();
updates.put(FIELD_MESSAGES, updatedNotifications);
updates.put(FIELD_LAST_UPDATED, Timestamp.of(java.util.Date.from(Instant.now())));
firestoreBaseRepository.updateDocument(notificationDocRef, updates);
logger.info("Successfully updated notification status to '{}' for session {} in Firestore.", status, sessionId);
}
} else {
logger.warn("Notification session {} not found in Firestore. Cannot update status.", sessionId);
}
} catch (ExecutionException e) {
logger.error("Error updating notification status in Firestore for session {}: {}", sessionId, e.getMessage(), e);
throw new FirestorePersistenceException("Failed to update notification status in Firestore for session " + sessionId, e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.error("Thread interrupted while updating notification status in Firestore for session {}: {}", sessionId, e.getMessage(), e);
throw new FirestorePersistenceException("Updating notification status was interrupted for session " + sessionId, e);
}
})
.subscribeOn(Schedulers.boundedElastic())
.then();
}
}

68
src/main/java/com/example/service/notification/MemoryStoreNotificationService.java
View File

@@ -1,5 +1,7 @@
package com.example.service.notification;
import com.example.dto.dialogflow.conversation.ConversationEntryDTO;
import com.example.dto.dialogflow.conversation.ConversationSessionDTO;
import com.example.dto.dialogflow.notification.NotificationDTO;
import com.example.dto.dialogflow.notification.NotificationSessionDTO;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -19,17 +21,25 @@ public class MemoryStoreNotificationService {
private static final Logger logger = LoggerFactory.getLogger(MemoryStoreNotificationService.class);
private final ReactiveRedisTemplate<String, NotificationSessionDTO> notificationRedisTemplate;
private final ReactiveRedisTemplate<String, ConversationSessionDTO> conversationRedisTemplate;
private final ReactiveRedisTemplate<String, String> stringRedisTemplate;
private final FirestoreNotificationConvService firestoreNotificationConvService;
private static final String NOTIFICATION_KEY_PREFIX = "notification:";
private static final String PHONE_TO_SESSION_KEY_PREFIX = "notification:phone_to_notification:";
private final Duration notificationTtl = Duration.ofHours(24);
private static final String PHONE_TO_NOTIFICATION_SESSION_KEY_PREFIX = "notification:phone_to_notification:";
private static final String CONVERSATION_SESSION_KEY_PREFIX = "conversation-notification:session:";
private static final String PHONE_TO_CONVERSATION_SESSION_KEY_PREFIX = "conversation-notification:phone_to_session:";
private final Duration notificationTtl = Duration.ofMinutes(5);
public MemoryStoreNotificationService(
ReactiveRedisTemplate<String, NotificationSessionDTO> notificationRedisTemplate,
ReactiveRedisTemplate<String, ConversationSessionDTO> conversationRedisTemplate,
ReactiveRedisTemplate<String, String> stringRedisTemplate,
FirestoreNotificationConvService firestoreNotificationConvService,
ObjectMapper objectMapper) {
this.notificationRedisTemplate = notificationRedisTemplate;
this.conversationRedisTemplate = conversationRedisTemplate;
this.stringRedisTemplate = stringRedisTemplate;
this.firestoreNotificationConvService = firestoreNotificationConvService;
}
@@ -38,7 +48,6 @@ public class MemoryStoreNotificationService {
if (phoneNumber == null || phoneNumber.isBlank()) {
return Mono.error(new IllegalArgumentException("Phone number is required to manage notification entries."));
}
//noote: Use the phone number as the session ID for notifications
String notificationSessionId = phoneNumber;
@@ -73,7 +82,7 @@ public class MemoryStoreNotificationService {
private Mono<Boolean> cacheNotificationSession(NotificationSessionDTO session) {
String key = NOTIFICATION_KEY_PREFIX + session.sessionId();
String phoneToSessionKey = PHONE_TO_SESSION_KEY_PREFIX + session.telefono();
String phoneToSessionKey = PHONE_TO_NOTIFICATION_SESSION_KEY_PREFIX + session.telefono();
return notificationRedisTemplate.opsForValue().set(key, session, notificationTtl)
.then(stringRedisTemplate.opsForValue().set(phoneToSessionKey, session.sessionId(), notificationTtl));
@@ -93,7 +102,7 @@ public class MemoryStoreNotificationService {
}
public Mono<String> getNotificationIdForPhone(String phone) {
String key = PHONE_TO_SESSION_KEY_PREFIX + phone;
String key = PHONE_TO_NOTIFICATION_SESSION_KEY_PREFIX + phone;
return stringRedisTemplate.opsForValue().get(key)
.doOnSuccess(sessionId -> {
if (sessionId != null) {
@@ -105,4 +114,53 @@ public class MemoryStoreNotificationService {
.doOnError(e -> logger.error("Error retrieving session ID for phone {} from MemoryStore: {}", phone,
e.getMessage(), e));
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber) {
String sessionKey = CONVERSATION_SESSION_KEY_PREFIX + sessionId;
String phoneToSessionKey = PHONE_TO_CONVERSATION_SESSION_KEY_PREFIX + userPhoneNumber;
logger.info("Attempting to save entry to Redis for session {}. Entity: {}", sessionId, newEntry.entity().name());
return conversationRedisTemplate.opsForValue().get(sessionKey)
.defaultIfEmpty(ConversationSessionDTO.create(sessionId, userId, userPhoneNumber))
.flatMap(session -> {
ConversationSessionDTO sessionWithUpdatedTelefono = session.withTelefono(userPhoneNumber);
ConversationSessionDTO updatedSession = sessionWithUpdatedTelefono.withAddedEntry(newEntry);
logger.info("Attempting to set updated session {} with new entry entity {} in Redis.", sessionId, newEntry.entity().name());
return conversationRedisTemplate.opsForValue().set(sessionKey, updatedSession, notificationTtl)
.then(stringRedisTemplate.opsForValue().set(phoneToSessionKey, sessionId, notificationTtl))
.then();
})
.doOnSuccess(success ->{
logger.info("Successfully saved updated session and phone mapping to Redis for session {}. Entity Type: {}", sessionId, newEntry.entity().name());
firestoreNotificationConvService.saveEntry(userId, sessionId, newEntry, userPhoneNumber)
.subscribe(
fsVoid -> logger.debug(
"Asynchronously (Write-Back): Entry successfully saved to Firestore for session {}. Type: {}.",
sessionId, newEntry.type().name()),
fsError -> logger.error(
"Asynchronously (Write-Back): Failed to save entry to Firestore for session {}. Type: {}: {}",
sessionId, newEntry.type().name(), fsError.getMessage(), fsError));
})
.doOnError(e -> logger.error("Error appending entry to Redis for session {}: {}", sessionId, e.getMessage(), e));
}
public Mono<ConversationSessionDTO> getSessionByTelefono(String telefono) {
if (telefono == null || telefono.isBlank()) {
return Mono.empty();
}
String phoneToSessionKey = PHONE_TO_CONVERSATION_SESSION_KEY_PREFIX + telefono;
logger.debug("Attempting to retrieve session ID for phone number {} from Redis.", telefono);
return stringRedisTemplate.opsForValue().get(phoneToSessionKey)
.flatMap(sessionId -> {
logger.debug("Found session ID {} for phone number {}. Retrieving session data.", sessionId, telefono);
return conversationRedisTemplate.opsForValue().get(CONVERSATION_SESSION_KEY_PREFIX + sessionId);
})
.doOnSuccess(session -> {
if (session != null) {
logger.info("Successfully retrieved session {} by phone number {}.", session.sessionId(), telefono);
} else {
logger.info("No session found in Redis for phone number {}.", telefono);
}
})
.doOnError(e -> logger.error("Error retrieving session by phone number {}: {}", telefono, e.getMessage(), e));
}
}

37
src/main/java/com/example/service/notification/NotificationExpirationListener.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* 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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Service;
@Service
public class NotificationExpirationListener implements MessageListener {
private static final Logger logger = LoggerFactory.getLogger(NotificationExpirationListener.class);
private final FirestoreNotificationService firestoreNotificationService;
private static final String NOTIFICATION_KEY_PREFIX = "notification:";
public NotificationExpirationListener(FirestoreNotificationService firestoreNotificationService) {
this.firestoreNotificationService = firestoreNotificationService;
}
@Override
public void onMessage(Message message, byte[] pattern) {
String expiredKey = new String(message.getBody());
logger.info("Expired key: " + expiredKey);
if (expiredKey.startsWith(NOTIFICATION_KEY_PREFIX)) {
String sessionId = expiredKey.substring(NOTIFICATION_KEY_PREFIX.length());
firestoreNotificationService.updateNotificationStatus(sessionId, "inactive")
.doOnSuccess(v -> logger.info("Notification status updated to inactive for session: " + sessionId))
.doOnError(e -> logger.error("Error updating notification status for session: " + sessionId, e))
.subscribe();
}
}
}

137
src/main/java/com/example/service/notification/NotificationManagerService.java
View File

@@ -3,12 +3,12 @@ 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.ConversationSessionDTO;
import com.example.dto.dialogflow.conversation.QueryInputDTO;
import com.example.dto.dialogflow.conversation.QueryParamsDTO;
import com.example.dto.dialogflow.notification.NotificationDTO;
import com.example.service.base.DialogflowClientService;
import com.example.service.conversation.MemoryStoreConversationService;
import com.example.util.SessionIdGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,85 +26,118 @@ import com.example.dto.dialogflow.conversation.TextInputDTO;
@Service
public class NotificationManagerService {
private static final Logger logger = LoggerFactory.getLogger(NotificationManagerService.class);
private static final String NOTIFICATION_TEXT_PARAM = "notificationText";
private static final String eventName = "notificacion";
private static final Logger logger = LoggerFactory.getLogger(NotificationManagerService.class);
private static final String NOTIFICATION_TEXT_PARAM = "notificationText";
private static final String eventName = "notificacion";
private final DialogflowClientService dialogflowClientService;
private final FirestoreNotificationService firestoreNotificationService;
private final MemoryStoreNotificationService memoryStoreNotificationService;
private final FirestoreNotificationConvService firestoreConversationService;
private final DialogflowClientService dialogflowClientService;
private final FirestoreNotificationService firestoreNotificationService;
private final MemoryStoreNotificationService memoryStoreNotificationService;
private final MemoryStoreConversationService memoryStoreConversationService;
@Value("${dialogflow.default-language-code:es}")
private String defaultLanguageCode;
@Value("${dialogflow.default-language-code:es}")
private String defaultLanguageCode;
public NotificationManagerService(
DialogflowClientService dialogflowClientService,
FirestoreNotificationService firestoreNotificationService,
MemoryStoreNotificationService memoryStoreNotificationService,
MemoryStoreConversationService memoryStoreConversationService) {
public NotificationManagerService(
DialogflowClientService dialogflowClientService,
FirestoreNotificationService firestoreNotificationService,
MemoryStoreNotificationService memoryStoreNotificationService,
FirestoreNotificationConvService firestoreConversationService) {
this.dialogflowClientService = dialogflowClientService;
this.firestoreNotificationService = firestoreNotificationService;
this.memoryStoreNotificationService = memoryStoreNotificationService;
this.memoryStoreConversationService = memoryStoreConversationService;
}
this.firestoreConversationService = firestoreConversationService;
}
public Mono<DetectIntentResponseDTO> processNotification(ExternalNotRequestDTO externalRequest) {
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."));
logger.warn("No phone number provided in ExternalNotRequestDTO. Cannot process notification.");
return Mono.error(new IllegalArgumentException("Phone number is required."));
}
// 1. Persist the incoming notification entry
String newNotificationId = SessionIdGenerator.generateStandardSessionId();
NotificationDTO newNotificationEntry = new NotificationDTO(newNotificationId,telefono, Instant.now(),
externalRequest.text(),eventName, defaultLanguageCode,Collections.emptyMap()
);
String newNotificationId = SessionIdGenerator.generateStandardSessionId();
NotificationDTO newNotificationEntry = new NotificationDTO(newNotificationId, telefono, Instant.now(),
externalRequest.text(), eventName, defaultLanguageCode, Collections.emptyMap(), "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)
);
});
.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));
});
// 2. Resolve or create a conversation session
Mono<ConversationSessionDTO> sessionMono = memoryStoreConversationService.getSessionByTelefono(telefono)
.doOnNext(session -> logger.info("Found existing conversation session {} for phone number {}", session.sessionId(), telefono))
.switchIfEmpty(Mono.defer(() -> {
Mono<ConversationSessionDTO> sessionMono = memoryStoreNotificationService.getSessionByTelefono(telefono)
.doOnNext(session -> logger.info("Found existing conversation session {} for phone number {}",
session.sessionId(), telefono))
.flatMap(session -> {
ConversationEntryDTO systemEntry = ConversationEntryDTO.forSystem(externalRequest.text());
return persistConversationTurn(session.userId(), session.sessionId(), systemEntry, telefono)
.thenReturn(session);
})
.switchIfEmpty(Mono.defer(() -> {
String newSessionId = SessionIdGenerator.generateStandardSessionId();
logger.info("No existing conversation session found for phone number {}. Creating new session: {}", telefono, newSessionId);
return Mono.just(ConversationSessionDTO.create(newSessionId, "user_by_phone_" + telefono, telefono));
}));
logger.info("No existing conversation session found for phone number {}. Creating new session: {}",telefono, newSessionId);
String userId = "user_by_phone_" + telefono;
ConversationEntryDTO systemEntry = ConversationEntryDTO.forSystem(externalRequest.text());
return persistConversationTurn(userId, newSessionId, systemEntry, telefono)
.then(Mono.just(ConversationSessionDTO.create(newSessionId, userId, telefono)));
}));
// 3. Send notification text to Dialogflow using the resolved conversation session
// 3. Send notification text to Dialogflow using the resolved conversation
// session
return persistenceMono.then(sessionMono)
.flatMap(session -> {
.flatMap(session -> {
final String sessionId = session.sessionId();
logger.info("Sending notification text to Dialogflow using conversation session: {}", sessionId);
Map<String, Object> parameters = new HashMap<>();
parameters.put("telefono", telefono);
parameters.put(NOTIFICATION_TEXT_PARAM, newNotificationEntry.texto());
// Use a TextInputDTO to correctly build the QueryInputDTO
// Use a TextInputDTO to correctly build the QueryInputDTO
TextInputDTO textInput = new TextInputDTO(newNotificationEntry.texto());
QueryInputDTO queryInput = new QueryInputDTO(textInput, null, defaultLanguageCode);
DetectIntentRequestDTO detectIntentRequest = new DetectIntentRequestDTO(
queryInput,
new QueryParamsDTO(parameters)
);
queryInput,
new QueryParamsDTO(parameters));
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));
}
}
})
.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(String userId, String sessionId, ConversationEntryDTO entry,
String userPhoneNumber) {
logger.debug("Starting Write-Back persistence for session {}. Type: {}. Writing to Redis first.", sessionId,
entry.type().name());
return memoryStoreNotificationService.saveEntry(userId, sessionId, entry, userPhoneNumber)
.doOnSuccess(v -> {
logger.info(
"Entry saved to Redis for session {}. Type: {}. Kicking off async Firestore write-back.",
sessionId, entry.type().name());
firestoreConversationService.saveEntry(userId, sessionId, entry, userPhoneNumber)
.subscribe(
fsVoid -> logger.debug(
"Asynchronously (Write-Back): Entry successfully saved to Firestore for session {}. Type: {}.",
sessionId, entry.type().name()),
fsError -> logger.error(
"Asynchronously (Write-Back): Failed to save entry to Firestore for session {}. Type: {}: {}",
sessionId, entry.type().name(), fsError.getMessage(), fsError));
})
.doOnError(e -> logger.error("Error during primary Redis write for session {}. Type: {}: {}", sessionId,
entry.type().name(), e.getMessage(), e));
}
}

88
src/main/java/com/example/service/quickreplies/FirestoreQRService.java Normal file
View File

@@ -0,0 +1,88 @@
/*
* 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.quickreplies;
import com.example.dto.dialogflow.conversation.ConversationEntryDTO;
import com.example.dto.dialogflow.conversation.ConversationSessionDTO;
import com.example.exception.FirestorePersistenceException;
import com.example.mapper.quickreplies.FirestoreQuickReplyMapper;
import com.example.repository.FirestoreBaseRepository;
import com.google.cloud.firestore.DocumentReference;
import com.google.cloud.firestore.WriteBatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.util.Map;
import java.util.concurrent.ExecutionException;
@Service
public class FirestoreQRService {
private static final Logger logger = LoggerFactory.getLogger(FirestoreQRService.class);
private static final String CONVERSATION_COLLECTION_PATH_FORMAT = "artifacts/%s/quick-replies-conversations";
private final FirestoreBaseRepository firestoreBaseRepository;
private final FirestoreQuickReplyMapper firestoreQuickReplyMapper;
public FirestoreQRService(FirestoreBaseRepository firestoreBaseRepository, FirestoreQuickReplyMapper firestoreQuickReplyMapper) {
this.firestoreBaseRepository = firestoreBaseRepository;
this.firestoreQuickReplyMapper = firestoreQuickReplyMapper;
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber) {
logger.info("Attempting to save quick reply entry to Firestore for session {}. Entity: {}", sessionId, newEntry.entity().name());
return Mono.fromRunnable(() -> {
DocumentReference sessionDocRef = getSessionDocumentReference(sessionId);
WriteBatch batch = firestoreBaseRepository.createBatch();
try {
if (firestoreBaseRepository.documentExists(sessionDocRef)) {
Map<String, Object> updates = firestoreQuickReplyMapper.createUpdateMapForSingleEntry(newEntry);
batch.update(sessionDocRef, updates);
logger.info("Appending entry to existing quick reply session for user {} and session {}. Entity: {}", userId, sessionId, newEntry.entity().name());
} else {
Map<String, Object> newSessionMap = firestoreQuickReplyMapper.createNewSessionMapForSingleEntry(sessionId, userId, userPhoneNumber, newEntry);
batch.set(sessionDocRef, newSessionMap);
logger.info("Creating new quick reply session with first entry for user {} and session {}. Entity: {}", userId, sessionId, newEntry.entity().name());
}
firestoreBaseRepository.commitBatch(batch);
logger.info("Successfully committed batch for session {} to Firestore.", sessionId);
} catch (ExecutionException e) {
logger.error("Error saving quick reply entry to Firestore for session {}: {}", sessionId, e.getMessage(), e);
throw new FirestorePersistenceException("Failed to save quick reply entry to Firestore for session " + sessionId, e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.error("Thread interrupted while saving quick reply entry to Firestore for session {}: {}", sessionId, e.getMessage(), e);
throw new FirestorePersistenceException("Saving quick reply entry was interrupted for session " + sessionId, e);
}
}).subscribeOn(Schedulers.boundedElastic()).then();
}
public Mono<ConversationSessionDTO> getSessionByTelefono(String userPhoneNumber) {
logger.info("Attempting to retrieve quick reply session for phone number {}.", userPhoneNumber);
return firestoreBaseRepository.getDocumentsByField(getConversationCollectionPath(), "userPhoneNumber", userPhoneNumber)
.map(documentSnapshot -> {
if (documentSnapshot != null && documentSnapshot.exists()) {
ConversationSessionDTO sessionDTO = firestoreQuickReplyMapper.mapFirestoreDocumentToConversationSessionDTO(documentSnapshot);
logger.info("Successfully retrieved and mapped quick reply session for session {}.", sessionDTO.sessionId());
return sessionDTO;
}
logger.info("Quick reply session not found for phone number {}.", userPhoneNumber);
return null;
});
}
private String getConversationCollectionPath() {
return String.format(CONVERSATION_COLLECTION_PATH_FORMAT, firestoreBaseRepository.getAppId());
}
private DocumentReference getSessionDocumentReference(String sessionId) {
String collectionPath = getConversationCollectionPath();
return firestoreBaseRepository.getDocumentReference(collectionPath, sessionId);
}
}

80
src/main/java/com/example/service/quickreplies/MemoryStoreQRService.java Normal file
View File

@@ -0,0 +1,80 @@
/*
* 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.quickreplies;
import com.example.dto.dialogflow.conversation.ConversationEntryDTO;
import com.example.dto.dialogflow.conversation.ConversationSessionDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import java.time.Duration;
@Service
public class MemoryStoreQRService {
private static final Logger logger = LoggerFactory.getLogger(MemoryStoreQRService.class);
private static final String SESSION_KEY_PREFIX = "qr:session:";
private static final String PHONE_TO_SESSION_KEY_PREFIX = "qr:phone_to_session:";
private static final Duration SESSION_TTL = Duration.ofHours(24);
private final ReactiveRedisTemplate<String, ConversationSessionDTO> redisTemplate;
private final ReactiveRedisTemplate<String, String> stringRedisTemplate;
@Autowired
public MemoryStoreQRService(
ReactiveRedisTemplate<String, ConversationSessionDTO> redisTemplate,
ReactiveRedisTemplate<String, String> stringRedisTemplate) {
this.redisTemplate = redisTemplate;
this.stringRedisTemplate = stringRedisTemplate;
}
public Mono<Void> saveEntry(String userId, String sessionId, ConversationEntryDTO newEntry, String userPhoneNumber) {
String sessionKey = SESSION_KEY_PREFIX + sessionId;
String phoneToSessionKey = PHONE_TO_SESSION_KEY_PREFIX + userPhoneNumber;
logger.info("Attempting to save entry to Redis for quick reply session {}. Entity: {}", sessionId, newEntry.entity().name());
return redisTemplate.opsForValue().get(sessionKey)
.defaultIfEmpty(ConversationSessionDTO.create(sessionId, userId, userPhoneNumber))
.flatMap(session -> {
ConversationSessionDTO sessionWithUpdatedTelefono = session.withTelefono(userPhoneNumber);
ConversationSessionDTO updatedSession = sessionWithUpdatedTelefono.withAddedEntry(newEntry);
logger.info("Attempting to set updated quick reply session {} with new entry entity {} in Redis.", sessionId, newEntry.entity().name());
return redisTemplate.opsForValue().set(sessionKey, updatedSession, SESSION_TTL)
.then(stringRedisTemplate.opsForValue().set(phoneToSessionKey, sessionId, SESSION_TTL))
.then();
})
.doOnSuccess(success -> {
logger.info("Successfully saved updated quick reply session and phone mapping to Redis for session {}. Entity Type: {}", sessionId, newEntry.entity().name());
})
.doOnError(e -> logger.error("Error appending entry to Redis for quick reply session {}: {}", sessionId, e.getMessage(), e));
}
public Mono<ConversationSessionDTO> getSessionByTelefono(String telefono) {
if (telefono == null || telefono.isBlank()) {
return Mono.empty();
}
String phoneToSessionKey = PHONE_TO_SESSION_KEY_PREFIX + telefono;
logger.debug("Attempting to retrieve quick reply session ID for phone number {} from Redis.", telefono);
return stringRedisTemplate.opsForValue().get(phoneToSessionKey)
.flatMap(sessionId -> {
logger.debug("Found quick reply session ID {} for phone number {}. Retrieving session data.", sessionId, telefono);
return redisTemplate.opsForValue().get(SESSION_KEY_PREFIX + sessionId);
})
.doOnSuccess(session -> {
if (session != null) {
logger.info("Successfully retrieved quick reply session {} by phone number {}.", session.sessionId(), telefono);
} else {
logger.info("No quick reply session found in Redis for phone number {}.", telefono);
}
})
.doOnError(e -> logger.error("Error retrieving quick reply session by phone number {}: {}", telefono, e.getMessage(), e));
}
}

104
src/main/java/com/example/service/quickreplies/QuickRepliesManagerService.java Normal file
View File

@@ -0,0 +1,104 @@
/*
* 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.quickreplies;
import com.example.dto.dialogflow.base.DetectIntentResponseDTO;
import com.example.dto.dialogflow.conversation.ConversationEntryDTO;
import com.example.dto.dialogflow.conversation.ConversationEntryEntity;
import com.example.dto.dialogflow.conversation.ConversationEntryType;
import com.example.dto.dialogflow.conversation.ExternalConvRequestDTO;
import com.example.dto.quickreplies.QuickReplyScreenRequestDTO;
import com.example.service.conversation.FirestoreConversationService;
import com.example.service.conversation.MemoryStoreConversationService;
import com.example.util.SessionIdGenerator;
import java.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
@Service
public class QuickRepliesManagerService {
private static final Logger logger = LoggerFactory.getLogger(QuickRepliesManagerService.class);
private final MemoryStoreConversationService memoryStoreConversationService;
private final FirestoreConversationService firestoreConversationService;
private final QuickReplyContentService quickReplyContentService;
public QuickRepliesManagerService(
MemoryStoreConversationService memoryStoreConversationService,
FirestoreConversationService firestoreConversationService,
QuickReplyContentService quickReplyContentService) {
this.memoryStoreConversationService = memoryStoreConversationService;
this.firestoreConversationService = firestoreConversationService;
this.quickReplyContentService = quickReplyContentService;
}
public Mono<DetectIntentResponseDTO> startQuickReplySession(QuickReplyScreenRequestDTO externalRequest) {
String userPhoneNumber = externalRequest.user().telefono();
if (userPhoneNumber == null || userPhoneNumber.isBlank()) {
logger.warn("No phone number provided in request. Cannot manage conversation session without it.");
return Mono.error(new IllegalArgumentException("Phone number is required to manage conversation sessions."));
}
return memoryStoreConversationService.getSessionByTelefono(userPhoneNumber)
.flatMap(session -> Mono.just(session.sessionId()))
.switchIfEmpty(Mono.fromCallable(SessionIdGenerator::generateStandardSessionId))
.flatMap(sessionId -> {
String userId = "user_by_phone_" + userPhoneNumber.replaceAll("[^0-9]", "");
ConversationEntryDTO systemEntry = new ConversationEntryDTO(
ConversationEntryEntity.SISTEMA,
ConversationEntryType.INICIO,
Instant.now(),
"Pantalla :" + externalRequest.pantallaContexto() + " Agregada a la conversacion :",
null,
null
);
return persistConversationTurn(userId, sessionId, systemEntry, userPhoneNumber, externalRequest.pantallaContexto())
.then(quickReplyContentService.getQuickReplies(externalRequest.pantallaContexto()))
.map(quickReplyDTO -> new DetectIntentResponseDTO(sessionId, null, quickReplyDTO));
});
}
public Mono<DetectIntentResponseDTO> manageConversation(ExternalConvRequestDTO externalRequest) {
String userPhoneNumber = externalRequest.user().telefono();
if (userPhoneNumber == null || userPhoneNumber.isBlank()) {
logger.warn("No phone number provided in request. Cannot manage conversation session without it.");
return Mono.error(new IllegalArgumentException("Phone number is required to manage conversation sessions."));
}
return memoryStoreConversationService.getSessionByTelefono(userPhoneNumber)
.switchIfEmpty(Mono.error(new IllegalStateException("No quick reply session found for phone number: " + userPhoneNumber)))
.flatMap(session -> {
String userId = session.userId();
String sessionId = session.sessionId();
ConversationEntryDTO userEntry = ConversationEntryDTO.forUser(externalRequest.message());
return persistConversationTurn(userId, sessionId, userEntry, userPhoneNumber, session.pantallaContexto())
.then(quickReplyContentService.getQuickReplies(session.pantallaContexto()))
.flatMap(quickReplyDTO -> {
ConversationEntryDTO agentEntry = ConversationEntryDTO.forAgentWithMessage(quickReplyDTO.toString());
return persistConversationTurn(userId, sessionId, agentEntry, userPhoneNumber, session.pantallaContexto())
.thenReturn(new DetectIntentResponseDTO(sessionId, null, quickReplyDTO));
});
});
}
private Mono<Void> persistConversationTurn(String userId, String sessionId, ConversationEntryDTO entry, String userPhoneNumber, String pantallaContexto) {
logger.debug("Starting Write-Back persistence for quick reply session {}. Type: {}. Writing to Redis first.", sessionId, entry.type().name());
return memoryStoreConversationService.saveEntry(userId, sessionId, entry, userPhoneNumber, pantallaContexto)
.doOnSuccess(v -> logger.info("Entry saved to Redis for quick reply session {}. Type: {}. Kicking off async Firestore write-back.", sessionId, entry.type().name()))
.then(firestoreConversationService.saveEntry(userId, sessionId, entry, userPhoneNumber, pantallaContexto)
.doOnSuccess(fsVoid -> logger.debug("Asynchronously (Write-Back): Entry successfully saved to Firestore for quick reply session {}. Type: {}.", sessionId, entry.type().name()))
.doOnError(fsError -> logger.error("Asynchronously (Write-Back): Failed to save entry to Firestore for quick reply session {}. Type: {}: {}", sessionId, entry.type().name(), fsError.getMessage(), fsError)))
.doOnError(e -> logger.error("Error during primary Redis write for quick reply session {}. Type: {}: {}", sessionId, entry.type().name(), e.getMessage(), e));
}
}

69
src/main/java/com/example/service/quickreplies/QuickReplyContentService.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* 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.quickreplies;
import com.example.dto.quickreplies.QuestionDTO;
import com.example.dto.quickreplies.QuickReplyDTO;
import com.google.cloud.firestore.DocumentSnapshot;
import com.google.cloud.firestore.Firestore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import java.util.Collections;
import java.util.List;
@Service
public class QuickReplyContentService {
private static final Logger logger = LoggerFactory.getLogger(QuickReplyContentService.class);
private final Firestore firestore;
public QuickReplyContentService(Firestore firestore) {
this.firestore = firestore;
}
public Mono<QuickReplyDTO> getQuickReplies(String collectionId) {
logger.info("Fetching quick replies from Firestore for document: {}", collectionId);
if (collectionId == null || collectionId.isBlank()) {
logger.warn("collectionId is null or empty. Returning empty quick replies.");
return Mono.just(new QuickReplyDTO("empty", Collections.emptyList()));
}
return Mono.fromCallable(() -> {
try {
return firestore.collection("artifacts")
.document("default-app-id")
.collection("quick-replies")
.document(collectionId)
.get()
.get();
} catch (Exception e) {
throw new RuntimeException(e);
}
})
.filter(DocumentSnapshot::exists)
.map(document -> {
QuestionDTO pregunta = new QuestionDTO(document.getString("titulo"), document.getString("descripcion"));
return new QuickReplyDTO("preguntas sobre " + collectionId, List.of(pregunta));
})
.doOnSuccess(quickReplyDTO -> {
if (quickReplyDTO != null) {
logger.info("Successfully fetched {} quick replies for document: {}", quickReplyDTO.preguntas().size(), collectionId);
} else {
logger.info("No quick reply document found for id: {}", collectionId);
}
})
.doOnError(error -> logger.error("Error fetching quick replies from Firestore for document: {}", collectionId, error))
.switchIfEmpty(Mono.defer(() -> {
logger.info("No quick reply document found for id: {}", collectionId);
return Mono.empty();
}));
}
}