Bug fix: Prevent re-acquisition/re-ingestion of files if the files have not changed.

Author: kristan66

src/KeeperData.Core/ETL/Impl/AcquisitionPipeline.cs

@@ -31,7 +31,7 @@ public async Task StartAsync(Guid importId, string sourceType, CancellationToken
             report.CompletedAtUtc = DateTime.UtcNow;
             report.Error = ex.Message;
             await reportingService.UpsertImportReportAsync(report, ct);
-            
+
             throw;
         }
     }

src/KeeperData.Core/ETL/Impl/ImportOrchestrator.cs

@@ -118,7 +118,7 @@ public async Task StartAsync(ImportReport report, CancellationToken ct)
         return (fileSets, totalFiles);
     }
 
-    private async Task<IngestionTotals> ProcessAllFilesAsync(ImportReport report, ImmutableList<FileSet> fileSets, int totalFiles, 
+    private async Task<IngestionTotals> ProcessAllFilesAsync(ImportReport report, ImmutableList<FileSet> fileSets, int totalFiles,
         IBlobStorageService blobStorage, CancellationToken ct)
     {
         Debug.WriteLine($"[keepetl] Step 2: Processing and ingesting {totalFiles} files for ImportId: {report.ImportId}");
@@ -132,7 +132,9 @@ private async Task<IngestionTotals> ProcessAllFilesAsync(ImportReport report, Im
             Debug.WriteLine($"[keepetl] Processing file set: {fileSet.Definition.Name} with {fileSet.Files.Length} file(s)");
             logger.LogDebug("Processing file set for definition: {DefinitionName} with {FileCount} file(s) for ImportId: {ImportId}", fileSet.Definition.Name, fileSet.Files.Length, report.ImportId);
 
-            foreach (var file in fileSet.Files)
+            var orderedFiles = fileSet.Files.OrderBy(f => f.Timestamp).ToArray();
+
+            foreach (var file in orderedFiles)
             {
                 processedFileCount++;
 
@@ -159,7 +161,7 @@ private async Task<IngestionTotals> ProcessAllFilesAsync(ImportReport report, Im
         return totals with { FilesProcessed = processedFileCount };
     }
 
-    private async Task<IngestionTotals> ProcessSingleFileAsync(Guid importId, FileSet fileSet, EtlFile file, int currentFileNumber, 
+    private async Task<IngestionTotals> ProcessSingleFileAsync(Guid importId, FileSet fileSet, EtlFile file, int currentFileNumber,
         int totalFiles, IBlobStorageService blobStorage, ImportReport report, CancellationToken ct)
     {
         var fileStopwatch = Stopwatch.StartNew();
@@ -169,17 +171,17 @@ private async Task<IngestionTotals> ProcessSingleFileAsync(Guid importId, FileSe
 
         try
         {
-            // Check if file has already been ingested by retrieving MD5 from S3 metadata
+            // Check if file has already been ingested by retrieving ETag from S3 metadata
             var isAlreadyIngested = await IsFileAlreadyIngestedAsync(file.StorageObject.Key, blobStorage, ct);
             if (isAlreadyIngested)
             {
                 fileStopwatch.Stop();
                 Debug.WriteLine($"[keepetl] Skipping file {file.StorageObject.Key} - already ingested in a previous import");
                 logger.LogInformation("Skipping file {FileKey} - already ingested in a previous import for ImportId: {ImportId}", file.StorageObject.Key, importId);
 
-                return new IngestionTotals() 
-                { 
-                    FilesSkipped = 1 
+                return new IngestionTotals()
+                {
+                    FilesSkipped = 1
                 };
             }
 
@@ -189,7 +191,7 @@ private async Task<IngestionTotals> ProcessSingleFileAsync(Guid importId, FileSe
             await RecordSuccessfulIngestionAsync(importId, file, fileMetrics, fileStopwatch.ElapsedMilliseconds, ct);
 
             Debug.WriteLine($"[keepetl] Successfully ingested file: {file.StorageObject.Key} - Created: {fileMetrics.RecordsCreated}, Updated: {fileMetrics.RecordsUpdated}, Deleted: {fileMetrics.RecordsDeleted}, Duration: {fileStopwatch.ElapsedMilliseconds}ms");
-            logger.LogInformation("Successfully ingested file: {FileKey} - Created: {Created}, Updated: {Updated}, Deleted: {Deleted}, Duration: {Duration}ms", 
+            logger.LogInformation("Successfully ingested file: {FileKey} - Created: {Created}, Updated: {Updated}, Deleted: {Deleted}, Duration: {Duration}ms",
                 file.StorageObject.Key, fileMetrics.RecordsCreated, fileMetrics.RecordsUpdated, fileMetrics.RecordsDeleted, fileStopwatch.ElapsedMilliseconds);
 
             return new IngestionTotals
@@ -221,23 +223,23 @@ private async Task<bool> IsFileAlreadyIngestedAsync(
     {
         try
         {
-            // Retrieve file metadata from S3 to get the MD5 hash
+            // Retrieve file metadata from S3 to get the ETag
             var metadata = await blobStorage.GetMetadataAsync(fileKey, ct);
 
-            if (!metadata.UserMetadata.TryGetValue("MD5Hash", out var md5Hash) || string.IsNullOrEmpty(md5Hash))
+            if (string.IsNullOrEmpty(metadata.ETag))
             {
-                Debug.WriteLine($"[keepetl] No MD5 hash found in S3 metadata for file {fileKey} - will proceed with ingestion");
-                logger.LogDebug("No MD5 hash found in S3 metadata for file {FileKey} - will proceed with ingestion", fileKey);
+                Debug.WriteLine($"[keepetl] No ETag found for file {fileKey} - will proceed with ingestion");
+                logger.LogDebug("No ETag found for file {FileKey} - will proceed with ingestion", fileKey);
                 return false;
             }
 
-            // Check if a file with this key and MD5 has been previously ingested (not just acquired)
-            var isIngested = await reportingService.IsFileIngestedAsync(fileKey, md5Hash, ct);
+            // Check if a file with this key and ETag has been previously ingested (not just acquired)
+            var isIngested = await reportingService.IsFileIngestedAsync(fileKey, metadata.ETag, ct);
 
             if (isIngested)
             {
-                Debug.WriteLine($"[keepetl] File {fileKey} with MD5 {md5Hash} has already been ingested");
-                logger.LogWarning("File {FileKey} with MD5 {Md5Hash} has already been ingested in a previous import", fileKey, md5Hash);
+                Debug.WriteLine($"[keepetl] File {fileKey} with ETag {metadata.ETag} has already been ingested");
+                logger.LogWarning("File {FileKey} with ETag {ETag} has already been ingested in a previous import", fileKey, metadata.ETag);
             }
 
             return isIngested;
@@ -296,7 +298,7 @@ private async Task<FileIngestionMetrics> IngestFileAsync(Guid importId, IBlobSto
                 fileSet.Definition.PrimaryKeyHeaderNames,
                 fileSet.Definition.ChangeTypeHeaderName);
 
-            var metrics = await ProcessCsvRecordsAsync(importId, collection, csvContext.Csv, headers, file.StorageObject.Key, 
+            var metrics = await ProcessCsvRecordsAsync(importId, collection, csvContext.Csv, headers, file.StorageObject.Key,
                 collectionName, fileSet.Definition, progressTracker, report, ct);
 
             await csvContext.DisposeAsync();
@@ -448,8 +450,8 @@ private void ValidateChangeTypeHeader(string[] headers, string changeTypeHeaderN
     }
 
 
-    private async Task<FileIngestionMetrics> ProcessCsvRecordsAsync(Guid importId, IMongoCollection<BsonDocument> collection, CsvReader csv, 
-        CsvHeaders headers, string fileKey, string collectionName, DataSetDefinition definition, IngestionProgressTracker progressTracker, 
+    private async Task<FileIngestionMetrics> ProcessCsvRecordsAsync(Guid importId, IMongoCollection<BsonDocument> collection, CsvReader csv,
+        CsvHeaders headers, string fileKey, string collectionName, DataSetDefinition definition, IngestionProgressTracker progressTracker,
         ImportReport report, CancellationToken ct)
     {
         Debug.WriteLine($"[keepetl] Starting to process CSV records for file: {fileKey}");

src/KeeperData.Core/ETL/Impl/IngestionPipeline.cs

@@ -6,4 +6,9 @@ public static class EtlConstants
     public const string DateTimePattern = "yyyyMMddHHmmss";
     public const string CompositeKeyDelimiter = "@@";
     public const string LineageEventIdDelimiter = "||";
+
+    // S3-compliant metadata keys (lowercase with x-amz-meta- prefix)
+    // S3 automatically prefixes user metadata with "x-amz-meta-" and lowercases the keys
+    public const string MetadataKeySourceEncryptedLength = "x-amz-meta-sourceencryptedlength";
+    public const string MetadataKeySourceETag = "x-amz-meta-sourceetag";
 }

src/KeeperData.Core/EtlConstants.cs

@@ -17,7 +17,7 @@ public class ImportFileDocument
     public required string FileName { get; init; }
     public required string FileKey { get; init; }
     public required string DatasetName { get; init; }
-    public required string Md5Hash { get; init; }
+    public required string ETag { get; init; }
     public long FileSize { get; init; }
     public required string Status { get; init; }
 

src/KeeperData.Core/Reporting/Domain/ImportFileDocument.cs

@@ -28,6 +28,7 @@ public class AcquisitionPhaseDocument
     public int FilesDiscovered { get; set; }
     public int FilesProcessed { get; set; }
     public int FilesFailed { get; set; }
+    public int FilesSkipped { get; set; }
     public DateTime? StartedAtUtc { get; set; }
     public DateTime? CompletedAtUtc { get; set; }
 }
@@ -36,6 +37,7 @@ public class IngestionPhaseDocument
 {
     public required string Status { get; set; }
     public int FilesProcessed { get; set; }
+    public int FilesSkipped { get; set; }
     public int RecordsCreated { get; set; }
     public int RecordsUpdated { get; set; }
     public int RecordsDeleted { get; set; }

src/KeeperData.Core/Reporting/Domain/ImportReportDocument.cs

@@ -5,7 +5,7 @@ public record FileAcquisitionRecord
     public required string FileName { get; init; }
     public required string FileKey { get; init; }
     public required string DatasetName { get; init; }
-    public required string Md5Hash { get; init; }
+    public required string ETag { get; init; }
     public long FileSize { get; init; }
     public required string SourceKey { get; init; }
     public long DecryptionDurationMs { get; init; }

src/KeeperData.Core/Reporting/Dtos/FileAcquisitionRecord.cs

@@ -26,9 +26,9 @@ public record FileProcessingReport
     public required string DatasetName { get; init; }
 
     /// <summary>
-    /// Gets the MD5 hash of the file content for integrity verification.
+    /// Gets the ETag of the file content for integrity verification.
     /// </summary>
-    public required string Md5Hash { get; init; }
+    public required string ETag { get; init; }
 
     /// <summary>
     /// Gets the file size in bytes.

src/KeeperData.Core/Reporting/Dtos/FileProcessingReport.cs

@@ -128,7 +128,7 @@ public record IngestionPhaseReport
     public DateTime? CompletedAtUtc { get; set; }
 
     public IngestionCurrentFileStatus? CurrentFileStatus { get; set; }
-    
+
     public int FilesSkipped { get; set; }
 }
 

src/KeeperData.Core/Reporting/Dtos/ImportReport.cs

@@ -9,8 +9,8 @@ public interface IImportReportingService
     Task UpsertImportReportAsync(ImportReport report, CancellationToken ct);
 
     // File-level operations
-    Task<bool> IsFileProcessedAsync(string fileKey, string md5Hash, CancellationToken ct);
-    Task<bool> IsFileIngestedAsync(string fileKey, string md5Hash, CancellationToken ct);
+    Task<bool> IsFileProcessedAsync(string fileKey, string etag, CancellationToken ct);
+    Task<bool> IsFileIngestedAsync(string fileKey, string etag, CancellationToken ct);
     Task RecordFileAcquisitionAsync(Guid importId, FileAcquisitionRecord record, CancellationToken ct);
     Task RecordFileIngestionAsync(Guid importId, FileIngestionRecord record, CancellationToken ct);