223 lines
13 KiB
C#
223 lines
13 KiB
C#
using System.Security.Cryptography;
|
|
using System.Text.Json;
|
|
using Dapper;
|
|
using Npgsql;
|
|
|
|
namespace Shared.Excel;
|
|
|
|
public sealed class ImportRepository(NpgsqlDataSource dataSource)
|
|
{
|
|
private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web);
|
|
|
|
public async Task<Guid> CreateSessionAsync(
|
|
string tenantId,
|
|
string userId,
|
|
ImportDefinition definition,
|
|
string fileName,
|
|
string contentType,
|
|
byte[] data,
|
|
IReadOnlyList<string> sourceColumns,
|
|
IReadOnlyList<ImportMapping> mapping,
|
|
CancellationToken ct)
|
|
{
|
|
var id = Guid.NewGuid();
|
|
var hash = Convert.ToHexString(SHA256.HashData(data)).ToLowerInvariant();
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
await using var tx = await connection.BeginTransactionAsync(ct);
|
|
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
insert into kbx.import_sessions(
|
|
id, tenant_id, user_id, screen_id, import_type, original_file_name, status,
|
|
source_columns, mapping, created_at, updated_at)
|
|
values(
|
|
@Id, @TenantId, @UserId, @ScreenId, @ImportType, @FileName, @Status,
|
|
cast(@SourceColumns as jsonb), cast(@Mapping as jsonb), now(), now());
|
|
""", new {
|
|
Id = id, TenantId = tenantId, UserId = userId,
|
|
definition.ScreenId, ImportType = definition.Id, FileName = fileName,
|
|
Status = ImportStatuses.MappingRequired,
|
|
SourceColumns = JsonSerializer.Serialize(sourceColumns, Json),
|
|
Mapping = JsonSerializer.Serialize(mapping, Json),
|
|
}, tx, cancellationToken: ct));
|
|
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
insert into kbx.import_files(session_id, content_type, file_size, sha256, data)
|
|
values(@SessionId, @ContentType, @FileSize, @Sha256, @Data);
|
|
""", new { SessionId = id, ContentType = contentType, FileSize = data.LongLength, Sha256 = hash, Data = data }, tx, cancellationToken: ct));
|
|
|
|
await tx.CommitAsync(ct);
|
|
return id;
|
|
}
|
|
|
|
public async Task<(string ContentType, byte[] Data)> GetFileAsync(Guid sessionId, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
return await connection.QuerySingleAsync<(string, byte[])>(new CommandDefinition(
|
|
"select content_type, data from kbx.import_files where session_id=@SessionId",
|
|
new { SessionId = sessionId }, cancellationToken: ct));
|
|
}
|
|
|
|
public async Task<IReadOnlyList<ImportMapping>?> GetSavedMappingAsync(
|
|
string tenantId, string userId, string importType, IReadOnlyList<string> sourceColumns, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
var signature = ImportSourceSignature.Create(sourceColumns);
|
|
var json = await connection.QueryFirstOrDefaultAsync<string>(new CommandDefinition("""
|
|
select mapping::text from kbx.saved_import_mappings
|
|
where tenant_id=@TenantId and user_id=@UserId and import_type=@ImportType and source_signature=@Signature
|
|
order by updated_at desc limit 1;
|
|
""", new { TenantId = tenantId, UserId = userId, ImportType = importType, Signature = signature }, cancellationToken: ct));
|
|
return json is null ? null : JsonSerializer.Deserialize<ImportMapping[]>(json, Json);
|
|
}
|
|
|
|
public async Task SaveNamedMappingAsync(
|
|
string tenantId, string userId, string importType, string name, IReadOnlyList<string> sourceColumns, IReadOnlyList<ImportMapping> mapping, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
insert into kbx.saved_import_mappings(id,tenant_id,user_id,import_type,name,source_signature,mapping,created_at,updated_at)
|
|
values(@Id,@TenantId,@UserId,@ImportType,@Name,@Signature,cast(@Mapping as jsonb),now(),now())
|
|
on conflict(tenant_id,user_id,import_type,name) do update
|
|
set source_signature=excluded.source_signature,mapping=excluded.mapping,updated_at=now();
|
|
""", new {
|
|
Id = Guid.NewGuid(), TenantId=tenantId, UserId=userId, ImportType=importType, Name=name,
|
|
Signature=ImportSourceSignature.Create(sourceColumns), Mapping=JsonSerializer.Serialize(mapping, Json)
|
|
}, cancellationToken: ct));
|
|
}
|
|
|
|
public async Task SaveMappingAsync(Guid sessionId, IReadOnlyList<ImportMapping> mapping, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
update kbx.import_sessions set mapping=cast(@Mapping as jsonb), status=@Status, updated_at=now()
|
|
where id=@SessionId;
|
|
""", new { SessionId = sessionId, Mapping = JsonSerializer.Serialize(mapping, Json), Status = ImportStatuses.MappingRequired }, cancellationToken: ct));
|
|
}
|
|
|
|
public async Task<ImportSessionDto?> GetAsync(Guid sessionId, string tenantId, string userId, bool includeErrors, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
var row = await connection.QuerySingleOrDefaultAsync<SessionRow>(new CommandDefinition("""
|
|
select id, import_type as "ImportType", screen_id as "ScreenId", original_file_name as "FileName", status,
|
|
total_rows as "TotalRows", valid_rows as "ValidRows", invalid_rows as "InvalidRows", warning_rows as "WarningRows", created_rows as "CreatedRows", updated_rows as "UpdatedRows",
|
|
progress_percent as "ProgressPercent", source_columns::text as "SourceColumnsJson",
|
|
mapping::text as "MappingJson", created_at as "CreatedAt", completed_at as "CompletedAt"
|
|
from kbx.import_sessions
|
|
where id=@SessionId and tenant_id=@TenantId and user_id=@UserId;
|
|
""", new { SessionId = sessionId, TenantId = tenantId, UserId = userId }, cancellationToken: ct));
|
|
if (row is null) return null;
|
|
|
|
IReadOnlyList<ImportIssue>? errors = null;
|
|
if (includeErrors)
|
|
{
|
|
var issueJson = await connection.QueryAsync<string>(new CommandDefinition("""
|
|
select errors::text from kbx.import_rows
|
|
where session_id=@SessionId and jsonb_array_length(errors) > 0
|
|
order by row_number limit 100;
|
|
""", new { SessionId = sessionId }, cancellationToken: ct));
|
|
errors = issueJson.SelectMany(x => JsonSerializer.Deserialize<ImportIssue[]>(x, Json) ?? []).ToArray();
|
|
}
|
|
|
|
return new ImportSessionDto(
|
|
row.Id, row.ImportType, row.ScreenId, row.FileName, row.Status,
|
|
row.TotalRows, row.ValidRows, row.InvalidRows, row.WarningRows,
|
|
row.CreatedRows, row.UpdatedRows, row.ProgressPercent,
|
|
JsonSerializer.Deserialize<string[]>(row.SourceColumnsJson, Json) ?? [],
|
|
JsonSerializer.Deserialize<ImportMapping[]>(row.MappingJson, Json) ?? [],
|
|
errors, row.CreatedAt, row.CompletedAt);
|
|
}
|
|
|
|
public async Task<(string ImportType, IReadOnlyList<ImportMapping> Mapping)> GetWorkDefinitionAsync(Guid sessionId, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
var row = await connection.QuerySingleAsync<(string ImportType, string MappingJson)>(new CommandDefinition(
|
|
"select import_type as "ImportType", mapping::text as "MappingJson" from kbx.import_sessions where id=@SessionId",
|
|
new { SessionId = sessionId }, cancellationToken: ct));
|
|
return (row.ImportType, JsonSerializer.Deserialize<ImportMapping[]>(row.MappingJson, Json) ?? []);
|
|
}
|
|
|
|
|
|
public async Task<bool> TryTransitionAsync(Guid sessionId, IReadOnlyCollection<string> from, string to, int progress, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
var affected = await connection.ExecuteAsync(new CommandDefinition("""
|
|
update kbx.import_sessions
|
|
set status=@To, progress_percent=@Progress, updated_at=now()
|
|
where id=@SessionId and status = any(@From);
|
|
""", new { SessionId = sessionId, From = from.ToArray(), To = to, Progress = progress }, cancellationToken: ct));
|
|
return affected == 1;
|
|
}
|
|
|
|
public async Task SetStatusAsync(Guid sessionId, string status, int progress, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
await connection.ExecuteAsync(new CommandDefinition(
|
|
"update kbx.import_sessions set status=@Status, progress_percent=@Progress, updated_at=now() where id=@SessionId",
|
|
new { SessionId = sessionId, Status = status, Progress = progress }, cancellationToken: ct));
|
|
}
|
|
|
|
public async Task ReplaceRowsAsync(Guid sessionId, IReadOnlyList<ImportRowValidation> rows, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
await using var tx = await connection.BeginTransactionAsync(ct);
|
|
await connection.ExecuteAsync(new CommandDefinition("delete from kbx.import_rows where session_id=@SessionId", new { SessionId = sessionId }, tx, cancellationToken: ct));
|
|
foreach (var row in rows)
|
|
{
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
insert into kbx.import_rows(session_id,row_number,raw_data,normalized_data,status,errors,warnings,domain_key)
|
|
values(@SessionId,@RowNumber,cast(@RawData as jsonb),cast(@NormalizedData as jsonb),@Status,cast(@Errors as jsonb),cast(@Warnings as jsonb),@DomainKey);
|
|
""", new {
|
|
SessionId = sessionId, row.RowNumber,
|
|
RawData = JsonSerializer.Serialize(row.RawData, Json),
|
|
NormalizedData = row.NormalizedData?.RootElement.GetRawText(),
|
|
Status = row.IsValid ? "VALID" : "INVALID",
|
|
Errors = JsonSerializer.Serialize(row.Errors, Json),
|
|
Warnings = JsonSerializer.Serialize(row.Warnings, Json),
|
|
row.DomainKey,
|
|
}, tx, cancellationToken: ct));
|
|
}
|
|
|
|
var total = rows.Count;
|
|
var valid = rows.Count(x => x.IsValid);
|
|
var invalid = total - valid;
|
|
var warningRows = rows.Count(x => x.Warnings.Count > 0);
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
update kbx.import_sessions
|
|
set status=@Status,total_rows=@Total,valid_rows=@Valid,invalid_rows=@Invalid,
|
|
warning_rows=@Warnings,progress_percent=100,updated_at=now()
|
|
where id=@SessionId;
|
|
""", new { SessionId = sessionId, Status = ImportStatuses.Validated, Total = total, Valid = valid, Invalid = invalid, Warnings = warningRows }, tx, cancellationToken: ct));
|
|
await tx.CommitAsync(ct);
|
|
}
|
|
|
|
public async Task CompleteAsync(Guid sessionId, ImportCommitResult result, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
var status = result.Failed > 0 ? ImportStatuses.PartiallyCompleted : ImportStatuses.Completed;
|
|
await connection.ExecuteAsync(new CommandDefinition("""
|
|
update kbx.import_sessions
|
|
set status=@Status, created_rows=@Created, updated_rows=@Updated,
|
|
invalid_rows=invalid_rows + @Failed, progress_percent=100,
|
|
completed_at=now(), updated_at=now()
|
|
where id=@SessionId;
|
|
""", new { SessionId = sessionId, Status = status, result.Created, result.Updated, result.Failed }, cancellationToken: ct));
|
|
}
|
|
|
|
public async Task<IReadOnlyDictionary<int, IReadOnlyList<ImportIssue>>> GetErrorsAsync(Guid sessionId, CancellationToken ct)
|
|
{
|
|
await using var connection = await dataSource.OpenConnectionAsync(ct);
|
|
var rows = await connection.QueryAsync<(int RowNumber, string ErrorsJson)>(new CommandDefinition("""
|
|
select row_number as "RowNumber", errors::text as "ErrorsJson" from kbx.import_rows
|
|
where session_id=@SessionId and jsonb_array_length(errors) > 0 order by row_number;
|
|
""", new { SessionId = sessionId }, cancellationToken: ct));
|
|
return rows.ToDictionary(x => x.RowNumber, x => (IReadOnlyList<ImportIssue>)(JsonSerializer.Deserialize<ImportIssue[]>(x.ErrorsJson, Json) ?? []));
|
|
}
|
|
|
|
private sealed record SessionRow(
|
|
Guid Id, string ImportType, string ScreenId, string FileName, string Status,
|
|
int TotalRows, int ValidRows, int InvalidRows, int WarningRows,
|
|
int CreatedRows, int UpdatedRows, int ProgressPercent,
|
|
string SourceColumnsJson, string MappingJson,
|
|
DateTimeOffset CreatedAt, DateTimeOffset? CompletedAt);
|
|
}
|