feat: improve source activity fetching and error handling in SourceActivityProvider

Author: mjabascal10

backend/src/main/java/com/park/utmstack/service/network_scan/SourceActivityProvider.java

@@ -10,7 +10,6 @@
 import com.park.utmstack.service.logstash_pipeline.response.statistic.StatisticDocument;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.opensearch.client.json.JsonData;
 import org.opensearch.client.opensearch._types.SortOrder;
 import org.opensearch.client.opensearch.core.SearchRequest;
 import org.opensearch.client.opensearch.core.SearchResponse;
@@ -37,7 +36,6 @@ public class SourceActivityProvider {
     public Map<String, StatisticDocument> fetchLatestSourceActivity() {
         UtmDataInputStatusCheckpoint checkpoint = getOrCreateCheckpoint();
 
-
         String fromTimestamp = checkpoint.getLastProcessedTimestamp()
                 .minus(OVERLAP_SECONDS, ChronoUnit.SECONDS)
                 .toString();
@@ -51,12 +49,15 @@ public Map<String, StatisticDocument> fetchLatestSourceActivity() {
             Map<String, StatisticDocument> activityMap = extractLatestHits(response);
 
             if (!activityMap.isEmpty()) {
+                log.debug("Fetched {} active sources from statistics index", activityMap.size());
                 updateCheckpoint(checkpoint, activityMap);
+            } else {
+                log.debug("No new source activity found since checkpoint: {}", checkpoint.getLastProcessedTimestamp());
             }
 
             return activityMap;
         } catch (Exception e) {
-            log.error("Error consultando telemetría en Elasticsearch: {}", e.getMessage());
+            log.error("Error fetching telemetry from Elasticsearch: {}", e.getMessage(), e);
             return Collections.emptyMap();
         }
     }
@@ -70,45 +71,84 @@ private SearchRequest buildActivityQuery(String fromTimestamp) {
         return SearchRequest.of(s -> s
             .index(Constants.STATISTICS_INDEX_PATTERN)
             .query(SearchUtil.toQuery(filters))
-            .collapse(c -> c
-                .field("dataSource.keyword")
-                .innerHits(ih -> ih
-                    .name("latest")
-                    .size(1)
-                    .sort(sort -> sort.field(f -> f.field("@timestamp").order(SortOrder.Desc)))
+            .aggregations("by_source", agg -> agg
+                .terms(t -> t
+                    .field("dataSource.keyword")
+                    .size(10000)
+                )
+                .aggregations("latest_doc", latestAgg -> latestAgg
+                    .topHits(th -> th
+                        .size(1)
+                        .sort(sort -> sort.field(f -> f.field("@timestamp").order(SortOrder.Desc)))
+                    )
                 )
             )
-            .size(10000)
+            .size(0)  // We don't need the main hits, only aggregations
         );
     }
 
     private Map<String, StatisticDocument> extractLatestHits(SearchResponse<StatisticDocument> response) {
         Map<String, StatisticDocument> results = new HashMap<>();
 
-        response.hits().hits().forEach(hit -> {
-            if (hit.innerHits() != null && hit.innerHits().containsKey("latest")) {
-                var innerHits = hit.innerHits().get("latest").hits().hits();
-                if (!innerHits.isEmpty()) {
-                    JsonData json = innerHits.get(0).source();
-                    if (json != null) {
-                        StatisticDocument doc = json.to(StatisticDocument.class);
-                        results.put(doc.getDataSource(), doc);
+        try {
+            var aggregations = response.aggregations();
+            if (aggregations == null || aggregations.get("by_source") == null) {
+                log.warn("No aggregation results found in response");
+                return results;
+            }
+
+            var bySourceAgg = aggregations.get("by_source").sterms();
+            if (bySourceAgg == null || bySourceAgg.buckets().array().isEmpty()) {
+                log.debug("No data source buckets found in aggregation");
+                return results;
+            }
+
+            bySourceAgg.buckets().array().forEach(bucket -> {
+                String dataSource = bucket.key();
+
+                var latestDocsAgg = bucket.aggregations().get("latest_doc");
+                if (latestDocsAgg != null) {
+                    var topHits = latestDocsAgg.topHits();
+                    if (topHits != null && !topHits.hits().hits().isEmpty()) {
+                        var hit = topHits.hits().hits().get(0);
+                        if (hit.source() != null) {
+                            StatisticDocument doc = hit.source().to(StatisticDocument.class);
+                            if (doc != null) {
+                                results.put(dataSource, doc);
+                                log.debug("Extracted latest activity for source: {} at {}", dataSource, doc.getTimestamp());
+                            }
+                        }
                     }
                 }
-            }
-        });
-        return results;
+            });
+
+            return results;
+        } catch (Exception e) {
+            log.error("Error extracting latest hits from aggregation response: {}", e.getMessage(), e);
+            return results;
+        }
     }
 
     private void updateCheckpoint(UtmDataInputStatusCheckpoint checkpoint, Map<String, StatisticDocument> activityMap) {
         activityMap.values().stream()
-                .map(doc -> Instant.parse(doc.getTimestamp()))
+                .map(doc -> {
+                    try {
+                        return Instant.parse(doc.getTimestamp());
+                    } catch (Exception e) {
+                        log.warn("Failed to parse timestamp '{}' for document: {}", doc.getTimestamp(), e.getMessage());
+                        return null;
+                    }
+                })
+                .filter(java.util.Objects::nonNull)
                 .max(Instant::compareTo)
-                .ifPresent(latest -> {
-                    checkpoint.setLastProcessedTimestamp(latest);
-                    checkpointRepository.save(checkpoint);
-                    log.debug("Checkpoint actualizado a: {}", latest);
-                });
+                .ifPresentOrElse(
+                    latest -> {
+                        checkpoint.setLastProcessedTimestamp(latest);
+                        checkpointRepository.save(checkpoint);
+                        log.info("Checkpoint updated to: {}", latest);
+                    },
+                    () -> log.debug("No valid timestamps found to update checkpoint")
+                );
     }
 
     private UtmDataInputStatusCheckpoint getOrCreateCheckpoint() {