61 lines
2.8 KiB
C#
61 lines
2.8 KiB
C#
using Dapper;
|
|
using Npgsql;
|
|
|
|
namespace Shared.Operations;
|
|
|
|
public sealed record UpsertReconcileItem(
|
|
string TenantId,
|
|
string ReconcileType,
|
|
string ReferenceNo,
|
|
string SourceLabel,
|
|
string TargetLabel,
|
|
string ExpectedValue,
|
|
string ActualValue,
|
|
string? DifferenceValue,
|
|
string? ReasonCode,
|
|
string? ReasonText,
|
|
string Status,
|
|
string? SourceId,
|
|
string? TargetId,
|
|
DateTimeOffset OccurredAt);
|
|
|
|
public sealed class ReconcileProjectionWriter(NpgsqlDataSource dataSource)
|
|
{
|
|
public async Task<Guid> UpsertAsync(UpsertReconcileItem item, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
var id = Guid.NewGuid();
|
|
const string sql = """
|
|
insert into kbx.reconcile_items(
|
|
id, tenant_id, reconcile_type, reference_no, source_label, target_label,
|
|
expected_value, actual_value, difference_value, reason_code, reason_text, status,
|
|
source_id, target_id, identity_key, occurred_at)
|
|
values(
|
|
@Id, @TenantId, @ReconcileType, @ReferenceNo, @SourceLabel, @TargetLabel,
|
|
@ExpectedValue, @ActualValue, @DifferenceValue, @ReasonCode, @ReasonText, @Status,
|
|
@SourceId, @TargetId, @IdentityKey, @OccurredAt)
|
|
on conflict(tenant_id, reconcile_type, reference_no, identity_key)
|
|
do update set
|
|
source_label=excluded.source_label, target_label=excluded.target_label,
|
|
expected_value=excluded.expected_value, actual_value=excluded.actual_value,
|
|
difference_value=excluded.difference_value, reason_code=excluded.reason_code,
|
|
reason_text=excluded.reason_text, status=excluded.status,
|
|
occurred_at=excluded.occurred_at, version=kbx.reconcile_items.version+1, updated_at=now()
|
|
where excluded.occurred_at >= kbx.reconcile_items.occurred_at
|
|
returning id;
|
|
""";
|
|
var args = new {
|
|
Id = id,
|
|
item.TenantId, item.ReconcileType, item.ReferenceNo, item.SourceLabel, item.TargetLabel,
|
|
item.ExpectedValue, item.ActualValue, item.DifferenceValue, item.ReasonCode, item.ReasonText,
|
|
item.Status, item.SourceId, item.TargetId, IdentityKey = $"{item.SourceId ?? "-"}|{item.TargetId ?? "-"}", item.OccurredAt,
|
|
};
|
|
var changedId = await connection.ExecuteScalarAsync<Guid?>(new CommandDefinition(sql, args, cancellationToken: ct));
|
|
if (changedId is not null) return changedId.Value;
|
|
return await connection.ExecuteScalarAsync<Guid>(new CommandDefinition("""
|
|
select id from kbx.reconcile_items
|
|
where tenant_id=@TenantId and reconcile_type=@ReconcileType and reference_no=@ReferenceNo and identity_key=@IdentityKey;
|
|
""", args, cancellationToken: ct));
|
|
}
|
|
}
|