.

src.bak/main/java/com/example/service/notification/FirestoreNotificationService.java

@@ -0,0 +1,188 @@
+/*
+ * 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;
+import com.example.exception.FirestorePersistenceException;
+import com.example.mapper.notification.FirestoreNotificationMapper;
+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;
+
+@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 final FirestoreBaseRepository firestoreBaseRepository;
+    private final 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;
+
+                    // 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);
+
+                            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())));
+                                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));
+                                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: {}",
+                                    e.getMessage(),
+                                    e);
+                            throw new FirestorePersistenceException(
+                                    "Failed to save notification to Firestore for phone ", 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 ", e);
+                        }
+                    }
+                })
+                .subscribeOn(Schedulers.boundedElastic())
+                .then();
+    }
+
+    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();
+    }
+
+    public Mono<Void> deleteNotification(String notificationId) {
+        logger.info("Attempting to delete notification session {} from Firestore.", notificationId);
+        return Mono.fromRunnable(() -> {
+            try {
+                DocumentReference notificationDocRef = getNotificationDocumentReference(notificationId);
+                firestoreBaseRepository.deleteDocument(notificationDocRef);
+                logger.info("Successfully deleted notification session {} from Firestore.", notificationId);
+            } catch (ExecutionException e) {
+                logger.error("Error deleting notification session {} from Firestore: {}", notificationId, e.getMessage(), e);
+                throw new FirestorePersistenceException("Failed to delete notification session " + notificationId, e);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                logger.error("Thread interrupted while deleting notification session {} from Firestore: {}", notificationId, e.getMessage(), e);
+                throw new FirestorePersistenceException("Deleting notification session was interrupted for " + notificationId, e);
+            }
+        }).subscribeOn(Schedulers.boundedElastic()).then();
+    }
+}

src.bak/main/java/com/example/service/notification/MemoryStoreNotificationService.java

@@ -0,0 +1,120 @@
+/*
+ * 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;
+import com.example.dto.dialogflow.notification.NotificationSessionDTO;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.core.ReactiveRedisTemplate;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.time.Instant;
+
+@Service
+public class MemoryStoreNotificationService {
+
+    private static final Logger logger = LoggerFactory.getLogger(MemoryStoreNotificationService.class);
+    private final ReactiveRedisTemplate<String, NotificationSessionDTO> notificationRedisTemplate;
+    private final ReactiveRedisTemplate<String, String> stringRedisTemplate;
+    private static final String NOTIFICATION_KEY_PREFIX = "notification:";
+    private static final String PHONE_TO_NOTIFICATION_SESSION_KEY_PREFIX = "notification:phone_to_notification:";
+    private final Duration notificationTtl = Duration.ofDays(30);
+
+    public MemoryStoreNotificationService(
+            ReactiveRedisTemplate<String, NotificationSessionDTO> notificationRedisTemplate,
+            ReactiveRedisTemplate<String, String> stringRedisTemplate,
+            ObjectMapper objectMapper) {
+        this.notificationRedisTemplate = notificationRedisTemplate;
+        this.stringRedisTemplate = stringRedisTemplate;
+    }
+
+    public Mono<Void> saveOrAppendNotificationEntry(NotificationDTO newEntry) {
+        String phoneNumber = newEntry.telefono();
+        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;
+
+        return getCachedNotificationSession(notificationSessionId)
+            .flatMap(existingSession -> {
+                // Session exists, append the new entry
+                List<NotificationDTO> updatedEntries = new ArrayList<>(existingSession.notificaciones());
+                updatedEntries.add(newEntry);
+                NotificationSessionDTO updatedSession = new NotificationSessionDTO(
+                    notificationSessionId,
+                    phoneNumber,
+                    existingSession.fechaCreacion(),
+                    Instant.now(),
+                    updatedEntries
+                );
+                return Mono.just(updatedSession);
+            })
+            .switchIfEmpty(Mono.defer(() -> {
+                // No session found, create a new one
+                NotificationSessionDTO newSession = new NotificationSessionDTO(
+                    notificationSessionId,
+                    phoneNumber,
+                    Instant.now(),
+                    Instant.now(),
+                    Collections.singletonList(newEntry)
+                );
+                return Mono.just(newSession);
+            }))
+            .flatMap(this::cacheNotificationSession)
+            .then();
+    }
+    
+    private Mono<Boolean> cacheNotificationSession(NotificationSessionDTO session) {
+        String key = NOTIFICATION_KEY_PREFIX + session.sessionId();
+        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));
+    }
+
+    public Mono<NotificationSessionDTO> getCachedNotificationSession(String sessionId) {
+        String key = NOTIFICATION_KEY_PREFIX + sessionId;
+        return notificationRedisTemplate.opsForValue().get(key)
+                .doOnSuccess(notification -> {
+                    if (notification != null) {
+                        logger.info("Notification session with ID {} retrieved from MemoryStore.", sessionId);
+                    } else {
+                        logger.debug("Notification session with ID {} not found in MemoryStore.", sessionId);
+                    }
+                })
+                .doOnError(e -> logger.error("Error retrieving notification session with ID {} from MemoryStore: {}", sessionId, e.getMessage(), e));
+    }
+
+    public Mono<String> getNotificationIdForPhone(String phone) {
+        String key = PHONE_TO_NOTIFICATION_SESSION_KEY_PREFIX + phone;
+        return stringRedisTemplate.opsForValue().get(key)
+                .doOnSuccess(sessionId -> {
+                    if (sessionId != null) {
+                        logger.info("Session ID {} found for phone.", sessionId);
+                    } else {
+                        logger.debug("Session ID not found for phone.");
+                    }
+                })
+                .doOnError(e -> logger.error("Error retrieving session ID for phone from MemoryStore: {}",
+                        e.getMessage(), e));
+    }
+
+    public Mono<Void> deleteNotificationSession(String phoneNumber) {
+        String notificationKey = NOTIFICATION_KEY_PREFIX + phoneNumber;
+        String phoneToNotificationKey = PHONE_TO_NOTIFICATION_SESSION_KEY_PREFIX + phoneNumber;
+        logger.info("Deleting notification session for phone number {}.", phoneNumber);
+        return notificationRedisTemplate.opsForValue().delete(notificationKey)
+                .then(stringRedisTemplate.opsForValue().delete(phoneToNotificationKey))
+                .then();
+    }
+}

src.bak/main/java/com/example/service/notification/NotificationManagerService.java

@@ -0,0 +1,184 @@
+/*
+ * 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));
+    }
+}