Composite keys as per HLD

Changed to SHA256-hash based mongo ids for composite keys due to unpredictable key data and need for any such key to be both stable and url-safe
Added record lineage reporting api

Author: kristan66

src/KeeperData.Bridge/Controllers/ImportController.cs

@@ -1,6 +1,7 @@
 using KeeperData.Bridge.Worker.Tasks;
 using KeeperData.Core.Database;
 using KeeperData.Core.ETL.Abstract;
+using KeeperData.Core.ETL.Utils;
 using KeeperData.Core.Reporting;
 using KeeperData.Core.Reporting.Dtos;
 using KeeperData.Infrastructure.Storage;
@@ -17,6 +18,8 @@ public class ImportController(
     ICollectionManagementService collectionManagementService,
     IReportingCollectionManagementService reportingCollectionManagementService) : ControllerBase
 {
+    private readonly RecordIdGenerator _recordIdGenerator = new();
+
     /// <summary>
     /// Starts a bulk file import process asynchronously.
     /// Returns immediately with an import ID once the lock has been acquired.
@@ -244,6 +247,93 @@ public async Task<IActionResult> GetFileReports(Guid importId, CancellationToken
         }
     }
 
+    /// <summary>
+    /// Gets the paginated lineage events for a specific record in chronological order.
+    /// Shows the complete history of changes to a record across all imports.
+    /// The recordId is a SHA256 hash generated by RecordIdGenerator from the composite key parts, making it URL-safe.
+    /// </summary>
+    /// <param name="collectionName">The collection name (e.g., "sam_cph_holdings")</param>
+    /// <param name="recordId">The URL-safe record ID (SHA256 hash generated from primary key parts)</param>
+    /// <param name="skip">Number of events to skip for pagination (default: 0)</param>
+    /// <param name="top">Number of events to return (default: 10, max: 100)</param>
+    /// <param name="cancellationToken">Cancellation token</param>
+    /// <returns>Paginated list of lineage events</returns>
+    [HttpGet("lineage/{collectionName}/{recordId}")]
+    [ProducesResponseType(typeof(PaginatedLineageEvents), StatusCodes.Status200OK)]
+    [ProducesResponseType(typeof(ErrorResponse), StatusCodes.Status400BadRequest)]
+    [ProducesResponseType(typeof(ErrorResponse), StatusCodes.Status404NotFound)]
+    [ProducesResponseType(typeof(ErrorResponse), StatusCodes.Status499ClientClosedRequest)]
+    public async Task<IActionResult> GetRecordLineageEvents(
+        string collectionName,
+        string recordId,
+        [FromQuery] int skip = 0,
+        [FromQuery] int top = 10,
+        CancellationToken cancellationToken = default)
+    {
+        logger.LogInformation("Received request to get lineage events for {CollectionName}/{RecordId} with skip={Skip}, top={Top}",
+            collectionName, recordId, skip, top);
+
+        // Validate parameters
+        if (skip < 0)
+        {
+            logger.LogWarning("Invalid skip parameter: {Skip}", skip);
+            return BadRequest(new ErrorResponse
+            {
+                Message = "Skip parameter must be greater than or equal to 0.",
+                Timestamp = DateTime.UtcNow
+            });
+        }
+
+        if (top <= 0 || top > 100)
+        {
+            logger.LogWarning("Invalid top parameter: {Top}", top);
+            return BadRequest(new ErrorResponse
+            {
+                Message = "Top parameter must be between 1 and 100.",
+                Timestamp = DateTime.UtcNow
+            });
+        }
+
+        // The recordId is a SHA256 hash generated by RecordIdGenerator from the composite key parts.
+        // ASP.NET Core URL-decodes the route parameter, but since our hash is already URL-safe,
+        // we can pass it directly to the reporting service.
+
+        try
+        {
+            var result = await importReportingService.GetRecordLineageEventsPaginatedAsync(
+                collectionName,
+                recordId,
+                skip,
+                top,
+                cancellationToken);
+
+            logger.LogInformation("Successfully retrieved {Count} of {Total} lineage events for {CollectionName}/{RecordId}",
+                result.Count, result.TotalEvents, collectionName, recordId);
+
+            return Ok(result);
+        }
+        catch (KeyNotFoundException ex)
+        {
+            logger.LogWarning("Lineage not found for {CollectionName}/{RecordId}: {Message}",
+                collectionName, recordId, ex.Message);
+            return NotFound(new ErrorResponse
+            {
+                Message = ex.Message,
+                Timestamp = DateTime.UtcNow
+            });
+        }
+        catch (OperationCanceledException)
+        {
+            logger.LogWarning("Get lineage events request was cancelled for {CollectionName}/{RecordId}",
+                collectionName, recordId);
+            return StatusCode(499, new ErrorResponse
+            {
+                Message = "Request was cancelled.",
+                Timestamp = DateTime.UtcNow
+            });
+        }
+    }
+
     /// <summary>
     /// Deletes a specific MongoDB collection by name.
     /// The collection name must be defined in DataSetDefinitions.
@@ -435,4 +525,102 @@ public async Task<IActionResult> DeleteAllReportingCollections(CancellationToken
             DeletedAtUtc = result.OperatedAtUtc
         });
     }
+
+    /// <summary>
+    /// Generates a URL-safe record ID from composite key parts using SHA256 hashing.
+    /// This endpoint helps clients construct the recordId needed for lineage queries.
+    /// </summary>
+    /// <param name="request">Request containing the key parts to hash</param>
+    /// <param name="cancellationToken">Cancellation token</param>
+    /// <returns>Generated record ID hash</returns>
+    [HttpPost("generate-record-id")]
+    [ProducesResponseType(typeof(GenerateRecordIdResponse), StatusCodes.Status200OK)]
+    [ProducesResponseType(typeof(ErrorResponse), StatusCodes.Status400BadRequest)]
+    public IActionResult GenerateRecordId([FromBody] GenerateRecordIdRequest request, CancellationToken cancellationToken = default)
+    {
+        logger.LogInformation("Received request to generate record ID from {count} key parts", request.KeyParts?.Length ?? 0);
+
+        // Validate request
+        if (request.KeyParts == null || request.KeyParts.Length == 0)
+        {
+            logger.LogWarning("Invalid request: KeyParts is null or empty");
+            return BadRequest(new ErrorResponse
+            {
+                Message = "KeyParts must contain at least one value.",
+                Timestamp = DateTime.UtcNow
+            });
+        }
+
+        // Check for null or empty values in key parts
+        for (int i = 0; i < request.KeyParts.Length; i++)
+        {
+            if (string.IsNullOrEmpty(request.KeyParts[i]))
+            {
+                logger.LogWarning("Invalid request: KeyPart at index {index} is null or empty", i);
+                return BadRequest(new ErrorResponse
+                {
+                    Message = $"KeyPart at index {i} cannot be null or empty.",
+                    Timestamp = DateTime.UtcNow
+                });
+            }
+        }
+
+        try
+        {
+            var recordId = _recordIdGenerator.GenerateId(request.KeyParts);
+
+            logger.LogInformation("Successfully generated record ID: {recordId}", recordId);
+
+            return Ok(new GenerateRecordIdResponse
+            {
+                RecordId = recordId,
+                KeyParts = request.KeyParts,
+                Timestamp = DateTime.UtcNow
+            });
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Failed to generate record ID");
+            return BadRequest(new ErrorResponse
+            {
+                Message = $"Failed to generate record ID: {ex.Message}",
+                Timestamp = DateTime.UtcNow
+            });
+        }
+    }
+}
+
+/// <summary>
+/// Request to generate a record ID from composite key parts.
+/// </summary>
+public record GenerateRecordIdRequest
+{
+    /// <summary>
+    /// The individual parts of the composite key (in order).
+    /// Each part will be joined and hashed to create the record ID.
+    /// </summary>
+    /// <example>["NORTH", "F001"]</example>
+    public required string[] KeyParts { get; init; }
+}
+
+/// <summary>
+/// Response containing the generated record ID.
+/// </summary>
+public record GenerateRecordIdResponse
+{
+    /// <summary>
+    /// The generated URL-safe record ID (SHA256 hash, 43 characters).
+    /// This ID can be used to query lineage events.
+    /// </summary>
+    public required string RecordId { get; init; }
+
+    /// <summary>
+    /// The key parts that were used to generate the record ID.
+    /// </summary>
+    public required string[] KeyParts { get; init; }
+
+    /// <summary>
+    /// The timestamp when the ID was generated.
+    /// </summary>
+    public DateTime Timestamp { get; init; }
 }

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

@@ -1,12 +1,11 @@
-using CsvHelper;
-using CsvHelper.Configuration;
 using KeeperData.Core.Database;
 using KeeperData.Core.ETL.Abstract;
 using KeeperData.Core.ETL.Utils;
 using KeeperData.Core.Reporting;
 using KeeperData.Core.Reporting.Dtos;
 using KeeperData.Core.Storage;
-using KeeperData.Core.Storage.Dtos;
+using CsvHelper;
+using CsvHelper.Configuration;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using MongoDB.Bson;
@@ -36,6 +35,7 @@ public class IngestionPipeline(
     private const int LineageEventBatchSize = DefaultInterval;
     private readonly IDatabaseConfig _databaseConfig = databaseConfig.Value;
     private readonly CsvRowCounter _rowCounter = csvRowCounter;
+    private readonly RecordIdGenerator _recordIdGenerator = new();
 
     // MongoDB field name constants
     private const string FieldId = "_id";
@@ -500,7 +500,6 @@ private async Task<FileIngestionMetrics> ProcessCsvRecordsAsync(Guid importId, I
                 Debug.WriteLine($"[keepetl] Processed batch of {BatchSize} records from {fileKey} in {batchStopwatch.ElapsedMilliseconds}ms. Total processed: {metrics.RecordsProcessed}, Created: {metrics.RecordsCreated}, Updated: {metrics.RecordsUpdated}, Deleted: {metrics.RecordsDeleted}");
                 Debug.WriteLine($"[keepetl] -- END BATCH ({batchStopwatch.Elapsed.TotalSeconds}s, {batchStopwatch.ElapsedMilliseconds}ms) --");
                 Debug.WriteLine($"[keepetl] ");
-                Debug.WriteLine($"[keepetl] ");
 
                 batch.Clear();
             }
@@ -724,9 +723,9 @@ private BsonDocument CreateDocumentFromCsvRecord(
         var now = DateTime.UtcNow;
         var accumulatorSet = new HashSet<string>(definition.Accumulators ?? []);
 
-        // Create composite primary key for _id
+        // Create URL-safe composite primary key for _id using RecordIdGenerator
         var compositeKeyParts = headers.PrimaryKeyHeaderNames.Select(pkHeader => csv.GetField(pkHeader) ?? string.Empty);
-        document[FieldId] = string.Join(EtlConstants.CompositeKeyDelimiter, compositeKeyParts);
+        document[FieldId] = _recordIdGenerator.GenerateId(compositeKeyParts);
 
         foreach (var header in headers.AllHeaders)
         {

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

@@ -5,22 +5,11 @@ public static class StandardDataSetDefinitionsBuilder
     public static DataSetDefinitions Build()
     {
         var list = new List<DataSetDefinition>();
-        var samCPHHolding = list.With(new DataSetDefinition("sam_cph_holdings", "LITP_SAMCPHHOLDING_{0}", ["CPH"], ChangeType.HeaderName,
-        [
-            "ADDRESS_PK",
-            "DISEASE_TYPE",
-            "INTERVAL",
-            "INTERVAL_UNIT_OF_TIME",
-            "CPH_RELATIONSHIP_TYPE",
-            "SECONDARY_CPH",
-            "ANIMAL_SPECIES_CODE",
-            "ANIMAL_PRODUCTION_USAGE_CODE",
-        ]));
-
+        var samCPHHolding = list.With(new DataSetDefinition("sam_cph_holdings", "LITP_SAMCPHHOLDING_{0}", ["CPH", "FEATURE_NAME", "SECONDARY_CPH", "ANIMAL_SPECIES_CODE"], ChangeType.HeaderName, []));
         var ctscphHolding = list.With(new DataSetDefinition("cts_cph_holding", "LITP_CTSCPHHOLDING_{0}", ["LID_FULL_IDENTIFIER"], ChangeType.HeaderName, []));
-        var ctsKeeper = list.With(new DataSetDefinition("cts_keeper", "LITP_CTSKEEPER_{0}", ["PAR_ID"], ChangeType.HeaderName, []));
+        var ctsKeeper = list.With(new DataSetDefinition("cts_keeper", "LITP_CTSKEEPER_{0}", ["PAR_ID", "LID_FULL_IDENTIFIER"], ChangeType.HeaderName, []));
         var samCPHHolder = list.With(new DataSetDefinition("sam_cph_holder", "LITP_SAMCPHHOLDER_{0}", ["PARTY_ID"], ChangeType.HeaderName, []));
-        var samHerd = list.With(new DataSetDefinition("sam_herd", "LITP_SAMHERD_{0}", ["CPHH", "HERDMARK"], ChangeType.HeaderName, []));
+        var samHerd = list.With(new DataSetDefinition("sam_herd", "LITP_SAMHERD_{0}", ["CPHH", "HERDMARK", "ANIMAL_PURPOSE_CODE"], ChangeType.HeaderName, []));
         var samParty = list.With(new DataSetDefinition("sam_party", "LITP_SAMPARTY_{0}", ["PARTY_ID"], ChangeType.HeaderName, []));
 
         return new DataSetDefinitions

src/KeeperData.Core/ETL/Utils/RecordIdGenerator.cs

@@ -0,0 +1,83 @@
+using System.Security.Cryptography;
+using System.Text;
+
+namespace KeeperData.Core.ETL.Utils;
+
+/// <summary>
+/// Generates URL-safe record identifiers from composite key parts using SHA256 hashing.
+/// Encapsulates all concerns around key generation, ensuring consistent and collision-resistant IDs.
+/// </summary>
+public class RecordIdGenerator
+{
+    /// <summary>
+    /// Generates a URL-safe record ID from one or more key parts.
+    /// Uses SHA256 hashing to create a consistent, collision-resistant identifier.
+    /// The hash is Base64URL-encoded for URL safety (43 characters).
+    /// </summary>
+    /// <param name="keyParts">The individual parts of the composite key</param>
+    /// <returns>A URL-safe record ID string (43 characters for SHA256)</returns>
+    /// <exception cref="ArgumentNullException">Thrown when keyParts is null</exception>
+    /// <exception cref="ArgumentException">Thrown when keyParts is empty or contains null/empty values</exception>
+    public string GenerateId(params string[] keyParts)
+    {
+        if (keyParts == null)
+        {
+            throw new ArgumentNullException(nameof(keyParts));
+        }
+
+        if (keyParts.Length == 0)
+        {
+            throw new ArgumentException("At least one key part is required", nameof(keyParts));
+        }
+
+        // Validate all parts are non-null and non-empty
+        for (int i = 0; i < keyParts.Length; i++)
+        {
+            if (string.IsNullOrEmpty(keyParts[i]))
+            {
+                throw new ArgumentException($"Key part at index {i} is null or empty", nameof(keyParts));
+            }
+        }
+
+        // Join parts with delimiter and hash
+        var composite = string.Join(EtlConstants.CompositeKeyDelimiter, keyParts);
+        return HashToBase64Url(composite);
+    }
+
+    /// <summary>
+    /// Generates a URL-safe record ID from an enumerable of key parts.
+    /// Uses SHA256 hashing to create a consistent, collision-resistant identifier.
+    /// </summary>
+    /// <param name="keyParts">The individual parts of the composite key</param>
+    /// <returns>A URL-safe record ID string</returns>
+    /// <exception cref="ArgumentNullException">Thrown when keyParts is null</exception>
+    /// <exception cref="ArgumentException">Thrown when keyParts is empty or contains null/empty values</exception>
+    public string GenerateId(IEnumerable<string> keyParts)
+    {
+        if (keyParts == null)
+        {
+            throw new ArgumentNullException(nameof(keyParts));
+        }
+
+        return GenerateId(keyParts.ToArray());
+    }
+
+    /// <summary>
+    /// Computes a SHA256 hash of the input string and returns it as a Base64URL-encoded string.
+    /// Base64URL encoding uses '-' instead of '+', '_' instead of '/', and removes padding '='.
+    /// </summary>
+    /// <param name="value">The string to hash</param>
+    /// <returns>Base64URL-encoded SHA256 hash (43 characters)</returns>
+    private static string HashToBase64Url(string value)
+    {
+        var bytes = Encoding.UTF8.GetBytes(value);
+        var hash = SHA256.HashData(bytes);
+
+        // Convert to Base64URL format
+        var base64 = Convert.ToBase64String(hash);
+        return base64
+            .Replace('+', '-')  // Replace + with -
+            .Replace('/', '_')  // Replace / with _
+            .TrimEnd('=');      // Remove padding
+    }
+}