CREATE TABLE dbo.Products (
    Sku       VARCHAR(40)   NOT NULL PRIMARY KEY,
    Price     DECIMAL(18,2) NOT NULL,
    Currency  CHAR(3)       NOT NULL,
    UpdatedAt DATETIME2     NOT NULL CONSTRAINT DF_Prod_Upd DEFAULT SYSUTCDATETIME()
);

CREATE OR ALTER PROCEDURE dbo.MergePriceList
    @PriceListJson NVARCHAR(MAX)
AS
BEGIN
    SET NOCOUNT ON;
    MERGE dbo.Products AS tgt
    USING (
        SELECT sku, price, currency
        FROM OPENJSON(@PriceListJson, '$.items')
        WITH (
            sku      VARCHAR(40)   '$.sku',
            price    DECIMAL(18,2) '$.price',
            currency CHAR(3)       '$.currency'
        )
    ) AS src
        ON tgt.Sku = src.sku
    WHEN MATCHED THEN
        UPDATE SET Price = src.price, Currency = src.currency,
                   UpdatedAt = SYSUTCDATETIME()
    WHEN NOT MATCHED THEN
        INSERT (Sku, Price, Currency)
        VALUES (src.sku, src.price, src.currency);
END;

using System.ComponentModel.DataAnnotations;
using System.Text.Json;
using CC.Art2link.Pipelines.Domain.Models.PipelineComponents;

public sealed class PriceListDisassemblerConfig
{
    [Range(1, int.MaxValue)] public int MinRows { get; set; }
    [Required] public string ExpectedHeader { get; set; } = string.Empty;
    [Required] public string Supplier { get; set; } = string.Empty;
}

public sealed class PriceListDisassembler
    : PipelineComponentBase<PriceListDisassemblerConfig>
{
    public override string Name => "PriceListDisassembler";

    protected override Task<PipelineComponentOutput> ExecuteAsync(
        PipelineComponentInput input,
        PriceListDisassemblerConfig config,
        CancellationToken cancellationToken)
    {
        // 1. Read the raw CSV body the SFTP download handed in.
        var lines = input.Body
            .Split('\n')
            .Select(l => l.TrimEnd('\r'))
            .Where(l => l.Length > 0)
            .ToList();

        // 2. Validate before parsing.
        if (lines.Count == 0 || lines[0] != config.ExpectedHeader)
            return Fail("price list header missing or unexpected");

        if (lines.Count - 1 < config.MinRows)
            return Fail($"price list has {lines.Count - 1} rows, "
                        + $"expected at least {config.MinRows}");

        // 3. Disassemble the CSV rows into the canonical PriceList shape.
        var items = new List<object>();
        string? validFrom = null;
        foreach (var row in lines.Skip(1))
        {
            var cols = row.Split(',');
            if (cols.Length < 5
                || !decimal.TryParse(cols[2], out var price)
                || price <= 0m)
                return Fail($"invalid price in row: {row}");

            validFrom ??= cols[4];
            items.Add(new { sku = cols[0], price, currency = cols[3] });
        }

        var json = JsonSerializer.Serialize(new
        {
            supplier  = config.Supplier,
            validFrom,
            items
        });

        // 4. Emit one JSON message and classify it as the canonical type,
        //    so the bus carries PriceList JSON, never the raw CSV.
        return Task.FromResult(new PipelineComponentOutput
        {
            Success  = true,
            Messages = [new PipelineMessage
            {
                Body        = json,
                MessageType = "PriceList"
            }]
        });

        static Task<PipelineComponentOutput> Fail(string msg) =>
            Task.FromResult(new PipelineComponentOutput
            {
                Success = false, ErrorMessage = msg
            });
    }
}

sku,description,unit_price,currency,valid_from
TP-4501,Trail Pack 45L,61.20,EUR,2026-06-05
HX-0090,Hex Stove,18.75,EUR,2026-06-05
GL-1240,Glacier Lantern,9.40,EUR,2026-06-05

{
  "supplier": "NordicGear",
  "validFrom": "2026-06-05",
  "items": [
    { "sku": "TP-4501", "price": 61.20, "currency": "EUR" },
    { "sku": "HX-0090", "price": 18.75, "currency": "EUR" },
    { "sku": "GL-1240", "price": 9.40,  "currency": "EUR" }
  ]
}