feat(data-input): add checkpoint table and repository for data input status management

mjabascal10 · mjabascal10 · commit 06a0dd68c7ab · 2026-01-09T11:02:29.000-06:00
diff --git a/backend/src/main/java/com/park/utmstack/domain/datainput_ingestion/UtmDataInputStatus.java b/backend/src/main/java/com/park/utmstack/domain/datainput_ingestion/UtmDataInputStatus.java
@@ -1,17 +1,13 @@
-package com.park.utmstack.domain;
+package com.park.utmstack.domain.datainput_ingestion;
 
-import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.park.utmstack.domain.network_scan.UtmNetworkScan;
 import lombok.*;
 
 import javax.persistence.*;
-import javax.validation.constraints.NotEmpty;
 import javax.validation.constraints.NotNull;
 import javax.validation.constraints.Size;
 import java.io.Serializable;
 import java.time.Instant;
 import java.util.Objects;
-import java.util.concurrent.TimeUnit;
 
 /**
  * A UtmDataInputStatus.
diff --git a/backend/src/main/java/com/park/utmstack/domain/datainput_ingestion/UtmDataInputStatusCheckpoint.java b/backend/src/main/java/com/park/utmstack/domain/datainput_ingestion/UtmDataInputStatusCheckpoint.java
@@ -0,0 +1,23 @@
+package com.park.utmstack.domain.datainput_ingestion;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import javax.persistence.*;
+import java.time.Instant;
+
+@Entity
+@Getter
+@Setter
+@NoArgsConstructor
+@Table(name = "utm_data_input_status_checkpoint")
+public class UtmDataInputStatusCheckpoint {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(name = "last_processed_timestamp", nullable = false)
+    private Instant lastProcessedTimestamp;
+}
diff --git a/backend/src/main/java/com/park/utmstack/domain/network_scan/UtmNetworkScan.java b/backend/src/main/java/com/park/utmstack/domain/network_scan/UtmNetworkScan.java
@@ -3,7 +3,6 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.park.utmstack.domain.UtmAssetMetrics;
-import com.park.utmstack.domain.UtmDataInputStatus;
 import com.park.utmstack.domain.network_scan.enums.AssetRegisteredMode;
 import com.park.utmstack.domain.network_scan.enums.AssetStatus;
 import com.park.utmstack.domain.network_scan.enums.UpdateLevel;
diff --git a/backend/src/main/java/com/park/utmstack/repository/datainput_ingestion/UtmDataInputStatusCheckpointRepository.java b/backend/src/main/java/com/park/utmstack/repository/datainput_ingestion/UtmDataInputStatusCheckpointRepository.java
@@ -0,0 +1,8 @@
+package com.park.utmstack.repository.datainput_ingestion;
+
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatusCheckpoint;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface UtmDataInputStatusCheckpointRepository extends JpaRepository<UtmDataInputStatusCheckpoint, Long> {}
diff --git a/backend/src/main/java/com/park/utmstack/repository/datainput_ingestion/UtmDataInputStatusRepository.java b/backend/src/main/java/com/park/utmstack/repository/datainput_ingestion/UtmDataInputStatusRepository.java
@@ -1,6 +1,6 @@
-package com.park.utmstack.repository;
+package com.park.utmstack.repository.datainput_ingestion;
 
-import com.park.utmstack.domain.UtmDataInputStatus;
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatus;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
diff --git a/backend/src/main/java/com/park/utmstack/service/UtmDataInputStatusQueryService.java b/backend/src/main/java/com/park/utmstack/service/UtmDataInputStatusQueryService.java
@@ -1,8 +1,8 @@
 package com.park.utmstack.service;
 
-import com.park.utmstack.domain.UtmDataInputStatus;
-import com.park.utmstack.domain.UtmDataInputStatus_;
-import com.park.utmstack.repository.UtmDataInputStatusRepository;
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatus;
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatus_;
+import com.park.utmstack.repository.datainput_ingestion.UtmDataInputStatusRepository;
 import com.park.utmstack.service.dto.UtmDataInputStatusCriteria;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
diff --git a/backend/src/main/java/com/park/utmstack/service/UtmDataInputStatusService.java b/backend/src/main/java/com/park/utmstack/service/UtmDataInputStatusService.java
@@ -1,17 +1,19 @@
 package com.park.utmstack.service;
 
 import com.park.utmstack.config.Constants;
-import com.park.utmstack.domain.UtmDataInputStatus;
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatus;
 import com.park.utmstack.domain.UtmServerModule;
 import com.park.utmstack.domain.application_events.enums.ApplicationEventType;
 import com.park.utmstack.domain.chart_builder.types.query.FilterType;
 import com.park.utmstack.domain.chart_builder.types.query.OperatorType;
 import com.park.utmstack.domain.correlation.config.UtmDataTypes;
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatusCheckpoint;
 import com.park.utmstack.domain.network_scan.UtmNetworkScan;
 import com.park.utmstack.domain.network_scan.enums.AssetStatus;
 import com.park.utmstack.domain.network_scan.enums.UpdateLevel;
 import com.park.utmstack.domain.shared_types.alert.UtmAlert;
-import com.park.utmstack.repository.UtmDataInputStatusRepository;
+import com.park.utmstack.repository.datainput_ingestion.UtmDataInputStatusCheckpointRepository;
+import com.park.utmstack.repository.datainput_ingestion.UtmDataInputStatusRepository;
 import com.park.utmstack.repository.correlation.config.UtmDataTypesRepository;
 import com.park.utmstack.repository.network_scan.UtmNetworkScanRepository;
 import com.park.utmstack.service.application_events.ApplicationEventService;
@@ -22,13 +24,10 @@
 import com.park.utmstack.service.network_scan.UtmNetworkScanService;
 import com.park.utmstack.util.enums.AlertSeverityEnum;
 import com.park.utmstack.util.enums.AlertStatus;
-import com.utmstack.opensearch_connector.parsers.TermAggregateParser;
-import com.utmstack.opensearch_connector.types.BucketAggregation;
 import lombok.RequiredArgsConstructor;
 import org.apache.http.conn.util.InetAddressUtils;
 import org.opensearch.client.json.JsonData;
 import org.opensearch.client.opensearch._types.SortOrder;
-import org.opensearch.client.opensearch._types.aggregations.TopHitsAggregate;
 import org.opensearch.client.opensearch.core.SearchRequest;
 import org.opensearch.client.opensearch.core.SearchResponse;
 import org.slf4j.Logger;
@@ -38,13 +37,13 @@
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -68,6 +67,7 @@ public class UtmDataInputStatusService {
     private final ElasticsearchService elasticsearchService;
     private final UtmDataTypesRepository dataTypesRepository;
     private final UtmNetworkScanRepository networkScanRepository;
+    private final UtmDataInputStatusCheckpointRepository checkpointRepository;
 
 
     /**
@@ -441,9 +441,16 @@ private Map<String, Object> createAlertForDatasourceDown(UtmDataInputStatus inpu
 
     private Map<String, StatisticDocument> getLatestStatisticsByDataSource() {
 
+        UtmDataInputStatusCheckpoint checkpoint = this.checkpointRepository.findById(1L)
+                .orElseGet(() -> {
+                    UtmDataInputStatusCheckpoint newCheckpoint = new UtmDataInputStatusCheckpoint();
+                    newCheckpoint.setLastProcessedTimestamp(Instant.now().minus(1, ChronoUnit.HOURS));
+                    return newCheckpoint;
+                });
+
         ArrayList<FilterType> filters = new ArrayList<>();
         filters.add(new FilterType("type", OperatorType.IS, "enqueue_success"));
-        filters.add(new FilterType("@timestamp", OperatorType.IS_BETWEEN, List.of("now-24h", "now")));
+        filters.add(new FilterType("@timestamp", OperatorType.IS_GREATER_THAN, checkpoint.getLastProcessedTimestamp().toString()));
 
         SearchRequest sr = SearchRequest.of(s -> s
                 .query(SearchUtil.toQuery(filters))
@@ -457,7 +464,7 @@ private Map<String, StatisticDocument> getLatestStatisticsByDataSource() {
                         )
                 )
                 .sort(sort -> sort.field(f -> f.field("@timestamp").order(SortOrder.Desc)))
-                .size(10000) // máximo de dataSources esperados
+                .size(10000)
         );
 
         SearchResponse<StatisticDocument> response =
@@ -478,6 +485,15 @@ private Map<String, StatisticDocument> getLatestStatisticsByDataSource() {
             }
         });
 
+        Instant lastTimestamp = result.values().stream()
+                .map(doc -> Instant.parse(doc.getTimestamp()))
+                .max(Instant::compareTo)
+                .orElse(Instant.now());
+
+        checkpoint.setLastProcessedTimestamp(lastTimestamp);
+
+        this.checkpointRepository.save(checkpoint);
+
         return result;
     }
 
diff --git a/backend/src/main/java/com/park/utmstack/service/correlation/config/UtmDataTypesService.java b/backend/src/main/java/com/park/utmstack/service/correlation/config/UtmDataTypesService.java
@@ -2,7 +2,7 @@
 
 import com.park.utmstack.domain.application_events.enums.ApplicationEventType;
 import com.park.utmstack.domain.correlation.config.UtmDataTypes;
-import com.park.utmstack.repository.UtmDataInputStatusRepository;
+import com.park.utmstack.repository.datainput_ingestion.UtmDataInputStatusRepository;
 import com.park.utmstack.repository.correlation.config.UtmDataTypesRepository;
 import com.park.utmstack.repository.network_scan.UtmNetworkScanRepository;
 import com.park.utmstack.service.UtmDataInputStatusService;
diff --git a/backend/src/main/java/com/park/utmstack/service/dto/network_scan/NetworkScanDTO.java b/backend/src/main/java/com/park/utmstack/service/dto/network_scan/NetworkScanDTO.java
@@ -1,21 +1,20 @@
 package com.park.utmstack.service.dto.network_scan;
 
-import com.park.utmstack.domain.UtmDataInputStatus;
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatus;
 import com.park.utmstack.domain.network_scan.UtmAssetGroup;
 import com.park.utmstack.domain.network_scan.UtmAssetTypes;
 import com.park.utmstack.domain.network_scan.UtmNetworkScan;
 import com.park.utmstack.domain.network_scan.UtmPorts;
 import com.park.utmstack.domain.network_scan.enums.AssetRegisteredMode;
 import com.park.utmstack.domain.network_scan.enums.AssetStatus;
-import com.park.utmstack.repository.UtmDataInputStatusRepository;
+import com.park.utmstack.repository.datainput_ingestion.UtmDataInputStatusRepository;
 import lombok.Getter;
 import lombok.Setter;
 import org.springframework.util.CollectionUtils;
 
 import java.time.Instant;
 import java.util.*;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 @Getter
 @Setter
diff --git a/backend/src/main/java/com/park/utmstack/service/network_scan/UtmNetworkScanService.java b/backend/src/main/java/com/park/utmstack/service/network_scan/UtmNetworkScanService.java
@@ -8,7 +8,7 @@
 import com.park.utmstack.domain.network_scan.enums.PropertyFilter;
 import com.park.utmstack.domain.shared_types.enums.ImageShortName;
 import com.park.utmstack.repository.UtmAssetMetricsRepository;
-import com.park.utmstack.repository.UtmDataInputStatusRepository;
+import com.park.utmstack.repository.datainput_ingestion.UtmDataInputStatusRepository;
 import com.park.utmstack.repository.network_scan.UtmNetworkScanRepository;
 import com.park.utmstack.service.UtmImagesService;
 import com.park.utmstack.service.agent_manager.AgentGrpcService;
diff --git a/backend/src/main/java/com/park/utmstack/web/rest/UtmDataInputStatusResource.java b/backend/src/main/java/com/park/utmstack/web/rest/UtmDataInputStatusResource.java
@@ -1,6 +1,6 @@
 package com.park.utmstack.web.rest;
 
-import com.park.utmstack.domain.UtmDataInputStatus;
+import com.park.utmstack.domain.datainput_ingestion.UtmDataInputStatus;
 import com.park.utmstack.domain.application_events.enums.ApplicationEventType;
 import com.park.utmstack.service.UtmDataInputStatusQueryService;
 import com.park.utmstack.service.UtmDataInputStatusService;
diff --git a/backend/src/main/java/com/park/utmstack/web/rest/idp_provider/IdentityProviderResource.java b/backend/src/main/java/com/park/utmstack/web/rest/idp_provider/IdentityProviderResource.java
@@ -1,7 +1,6 @@
 package com.park.utmstack.web.rest.idp_provider;
 
 
-import com.park.utmstack.domain.UtmDataInputStatus;
 import com.park.utmstack.service.dto.idp_provider.dto.IdentityProviderConfigResponseDto;
 import com.park.utmstack.service.dto.idp_provider.dto.IdentityProviderCriteria;
 import com.park.utmstack.service.idp_provider.IdentityProviderService;
diff --git a/backend/src/main/resources/config/liquibase/changelog/20260109001_create_table_data_input_status_checkpoint.xml b/backend/src/main/resources/config/liquibase/changelog/20260109001_create_table_data_input_status_checkpoint.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="utf-8"?>
+<databaseChangeLog
+        xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.5.xsd">
+
+    <changeSet id="20260109001" author="manuel">
+        <createTable tableName="utm_data_input_status_checkpoint">
+
+            <column name="id" type="BIGINT" autoIncrement="true">
+                <constraints primaryKey="true" nullable="false"/>
+            </column>
+
+            <column name="last_processed_timestamp" type="TIMESTAMP WITH TIME ZONE">
+                <constraints nullable="false"/>
+            </column>
+
+        </createTable>
+
+    </changeSet>
+
+</databaseChangeLog>
diff --git a/backend/src/main/resources/config/liquibase/master.xml b/backend/src/main/resources/config/liquibase/master.xml
@@ -297,5 +297,7 @@
 
     <include file="/config/liquibase/changelog/20260105001_adding_crowdstrike_integration.xml" relativeToChangelogFile="false"/>
 
+    <include file="/config/liquibase/changelog/20260109001_create_table_data_input_status_checkpoint.xml" relativeToChangelogFile="false"/>
+
 
 </databaseChangeLog>
