AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/Inbound/MdpInboundSnapshotService.cs

using System.Security.Cryptography;
using System.Text.RegularExpressions;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.DataPlatform.Inbound;

public sealed class MdpInboundSnapshotService : ITransient
{
    private static readonly Regex TableNameRe = new(@"^[A-Za-z0-9_]+$", RegexOptions.Compiled);
    private const decimal DefaultDiffThreshold = 0.20m;

    private readonly ISqlSugarClient _db;
    private readonly MdpInboundAuthService _auth;
    private readonly ILogger<MdpInboundSnapshotService> _logger;

    public MdpInboundSnapshotService(
        ISqlSugarClient db,
        MdpInboundAuthService auth,
        ILogger<MdpInboundSnapshotService> logger)
    {
        _db = db;
        _auth = auth;
        _logger = logger;
    }

    public async Task<MdpInboundOutcome> OpenAsync(
        string entityCode, string accessKey, long tenantId, string clientIp, CancellationToken ct)
    {
        var code = (entityCode ?? string.Empty).Trim().ToUpperInvariant();
        var denied = await _auth.CheckAsync(accessKey, tenantId, code, clientIp, ct);
        if (denied != null)
            return Fail(denied.Value.Status, denied.Value.Message);

        await ExpireOpenAsync(tenantId, accessKey, code, ct);

        var existing = await _db.Queryable<MdpInboundSnapshot>()
            .Where(s => s.TenantId == tenantId && s.AccessKey == accessKey && s.Status == "OPEN")
            .Where("UPPER(entity_code) = @code", new SugarParameter("@code", code))
            .FirstAsync(ct);
        if (existing != null)
            return Fail(409, "snapshot already open");

        var now = DateTime.Now;
        var snapshotId = $"SNAP_{now:yyyyMMddHHmmss}_{RandomToken(8)}";
        await _db.Insertable(new MdpInboundSnapshot
        {
            SnapshotId = snapshotId,
            TenantId = tenantId,
            AccessKey = accessKey,
            EntityCode = code,
            Status = "OPEN",
            LastSeq = 0,
            RowCount = 0,
            OpenedAt = now,
            ExpireAt = now.AddHours(24),
            CreateTime = now,
            UpdateTime = now
        }).ExecuteCommandAsync(ct);

        return Ok(new { snapshotId, expireAt = now.AddHours(24), status = "OPEN" }, 201);
    }

    public async Task EnsureAcceptAsync(
        string snapshotId, string accessKey, long tenantId, string entityCode, int? seq, CancellationToken ct)
    {
        if (string.IsNullOrWhiteSpace(snapshotId))
            return;
        if (seq is null or < 1)
            throw new MdpInboundBatchException(400, "seq required");

        var snap = await LoadAsync(snapshotId, accessKey, tenantId, entityCode, ct)
                   ?? throw new MdpInboundBatchException(400, "snapshot not found");

        if (snap.ExpireAt < DateTime.Now)
        {
            await MarkStatusAsync(snap.Id, "EXPIRED", ct);
            throw new MdpInboundBatchException(400, "snapshot expired");
        }
        if (!string.Equals(snap.Status, "OPEN", StringComparison.OrdinalIgnoreCase))
            throw new MdpInboundBatchException(409, "snapshot not open");
        if (seq.Value != snap.LastSeq + 1)
        {
            _logger.LogWarning("inbound snapshot seq gap snapshot={Snap} expected={Exp} got={Got}",
                snapshotId, snap.LastSeq + 1, seq);
            throw new MdpInboundBatchException(400, "seq gap");
        }
    }

    public async Task AdvanceAsync(string snapshotId, int seq, int rowCount, CancellationToken ct)
    {
        if (string.IsNullOrWhiteSpace(snapshotId))
            return;
        var n = await _db.Ado.ExecuteCommandAsync(
            """
            UPDATE mdp_inbound_snapshot
               SET last_seq=@seq, row_count=row_count+@rows, update_time=@now
             WHERE snapshot_id=@id AND status='OPEN' AND last_seq=@prev
            """,
            new SugarParameter("@seq", seq),
            new SugarParameter("@rows", rowCount),
            new SugarParameter("@now", DateTime.Now),
            new SugarParameter("@id", snapshotId),
            new SugarParameter("@prev", seq - 1));
        if (n != 1)
            throw new InvalidOperationException("snapshot seq advance failed");
    }

    public async Task<MdpInboundOutcome> CommitAsync(
        string entityCode, string snapshotId, string accessKey, long tenantId, string clientIp,
        bool force, CancellationToken ct)
    {
        var code = (entityCode ?? string.Empty).Trim().ToUpperInvariant();
        var denied = await _auth.CheckAsync(accessKey, tenantId, code, clientIp, ct);
        if (denied != null)
            return Fail(denied.Value.Status, denied.Value.Message);

        var grant = await _auth.GetGrantAsync(accessKey, code, ct);
        if (grant == null)
            return Fail(403, "entity not authorized");
        var entity = await _db.Queryable<MdpEntity>()
            .Where(e => e.InboundEnabled == 1)
            .Where("UPPER(entity_code) = @code", new SugarParameter("@code", code))
            .FirstAsync(ct);
        if (entity == null)
            return Fail(403, "entity not authorized");

        var snap = await LoadAsync(snapshotId, accessKey, tenantId, code, ct)
                   ?? throw new MdpInboundBatchException(400, "snapshot not found");

        if (snap.ExpireAt < DateTime.Now)
        {
            await MarkStatusAsync(snap.Id, "EXPIRED", ct);
            return Fail(400, "snapshot expired");
        }

        if (force)
        {
            if (!string.Equals(snap.Status, "DIFF_BLOCKED", StringComparison.OrdinalIgnoreCase))
                return Fail(409, "force only when DIFF_BLOCKED");
        }
        else if (!string.Equals(snap.Status, "OPEN", StringComparison.OrdinalIgnoreCase))
        {
            return Fail(409, "snapshot not open");
        }

        if (string.IsNullOrWhiteSpace(entity.TargetTableName) || !TableNameRe.IsMatch(entity.TargetTableName))
            return Fail(500, "illegal target_table_name");

        var sourceTable = string.IsNullOrWhiteSpace(entity.SourceTableName) ? entity.EntityCode : entity.SourceTableName;
        var batchIds = await _db.Queryable<MdpInboundRequest>()
            .Where(r => r.SnapshotId == snapshotId && r.Status == "COMMITTED")
            .Select(r => r.SyncBatchId)
            .ToListAsync(ct);

        var snapshotKeys = new HashSet<string>(StringComparer.Ordinal);
        if (batchIds.Count > 0)
        {
            var keys = await _db.Ado.SqlQueryAsync<string>(
                $"""
                SELECT source_biz_key FROM `{entity.TargetTableName}`
                 WHERE tenant_id=@t AND source_system=@sys AND source_table=@st
                   AND sync_batch_id IN ({string.Join(",", batchIds.Select((_, i) => "@b" + i))})
                """,
                new[]
                {
                    new SugarParameter("@t", tenantId),
                    new SugarParameter("@sys", grant.SourceCode),
                    new SugarParameter("@st", sourceTable)
                }.Concat(batchIds.Select((b, i) => new SugarParameter("@b" + i, b))).ToArray());
            foreach (var k in keys.Where(x => !string.IsNullOrWhiteSpace(x)))
                snapshotKeys.Add(k);
        }

        var allKeys = await _db.Ado.SqlQueryAsync<string>(
            $"""
            SELECT source_biz_key FROM `{entity.TargetTableName}`
             WHERE tenant_id=@t AND source_system=@sys AND source_table=@st
            """,
            new SugarParameter("@t", tenantId),
            new SugarParameter("@sys", grant.SourceCode),
            new SugarParameter("@st", sourceTable));
        var missing = allKeys.Where(k => !string.IsNullOrWhiteSpace(k) && !snapshotKeys.Contains(k)).Distinct().ToList();
        var total = Math.Max(allKeys.Count, 1);
        var ratio = (decimal)missing.Count / total;

        if (!force && ratio > DefaultDiffThreshold)
        {
            await _db.Updateable<MdpInboundSnapshot>()
                .SetColumns(s => new MdpInboundSnapshot
                {
                    Status = "DIFF_BLOCKED",
                    DiffMissing = missing.Count,
                    DiffRatio = ratio,
                    UpdateTime = DateTime.Now
                })
                .Where(s => s.Id == snap.Id)
                .ExecuteCommandAsync(ct);
            _logger.LogWarning("inbound snapshot DIFF_BLOCKED snap={Snap} missing={N} ratio={R}",
                snapshotId, missing.Count, ratio);
            return Fail(409, "diff threshold exceeded");
        }

        var tran = await _db.Ado.UseTranAsync(async () =>
        {
            if (missing.Count > 0)
            {
                var inParams = missing.Select((k, i) => new SugarParameter("@k" + i, k)).ToArray();
                var emptyObj = "{}";
                await _db.Ado.ExecuteCommandAsync(
                    $"""
                    UPDATE `{entity.TargetTableName}`
                       SET raw_data = JSON_SET(IFNULL(raw_data, '{emptyObj}'), '$.is_deleted', 1),
                           process_status='PENDING',
                           update_time=@now
                     WHERE tenant_id=@t AND source_system=@sys AND source_table=@st
                       AND source_biz_key IN ({string.Join(",", missing.Select((_, i) => "@k" + i))})
                    """,
                    new[]
                    {
                        new SugarParameter("@now", DateTime.Now),
                        new SugarParameter("@t", tenantId),
                        new SugarParameter("@sys", grant.SourceCode),
                        new SugarParameter("@st", sourceTable)
                    }.Concat(inParams).ToArray());
            }

            await _db.Updateable<MdpInboundSnapshot>()
                .SetColumns(s => new MdpInboundSnapshot
                {
                    Status = "COMMITTED",
                    DiffMissing = missing.Count,
                    DiffRatio = ratio,
                    CommittedAt = DateTime.Now,
                    UpdateTime = DateTime.Now
                })
                .Where(s => s.Id == snap.Id)
                .ExecuteCommandAsync(ct);
        });

        if (!tran.IsSuccess)
        {
            _logger.LogError(tran.ErrorException, "inbound snapshot commit failed snap={Snap}", snapshotId);
            return Fail(500, "snapshot commit failed");
        }

        return Ok(new
        {
            snapshotId,
            status = "COMMITTED",
            diffMissing = missing.Count,
            diffRatio = ratio,
            force
        });
    }

    public async Task ExpireStaleOpenAsync(CancellationToken ct)
    {
        await _db.Updateable<MdpInboundSnapshot>()
            .SetColumns(s => new MdpInboundSnapshot { Status = "EXPIRED", UpdateTime = DateTime.Now })
            .Where(s => s.Status == "OPEN" && s.ExpireAt < DateTime.Now)
            .ExecuteCommandAsync(ct);
    }

    private async Task ExpireOpenAsync(long tenantId, string accessKey, string entityCode, CancellationToken ct)
    {
        await _db.Updateable<MdpInboundSnapshot>()
            .SetColumns(s => new MdpInboundSnapshot { Status = "EXPIRED", UpdateTime = DateTime.Now })
            .Where(s => s.TenantId == tenantId && s.AccessKey == accessKey && s.Status == "OPEN" && s.ExpireAt < DateTime.Now)
            .Where("UPPER(entity_code) = @code", new SugarParameter("@code", entityCode))
            .ExecuteCommandAsync(ct);
    }

    private Task<MdpInboundSnapshot> LoadAsync(
        string snapshotId, string accessKey, long tenantId, string entityCode, CancellationToken ct) =>
        _db.Queryable<MdpInboundSnapshot>()
            .Where(s => s.SnapshotId == snapshotId && s.AccessKey == accessKey && s.TenantId == tenantId)
            .Where("UPPER(entity_code) = @code", new SugarParameter("@code", (entityCode ?? "").ToUpperInvariant()))
            .FirstAsync(ct);

    private Task<int> MarkStatusAsync(long id, string status, CancellationToken ct) =>
        _db.Updateable<MdpInboundSnapshot>()
            .SetColumns(s => new MdpInboundSnapshot { Status = status, UpdateTime = DateTime.Now })
            .Where(s => s.Id == id)
            .ExecuteCommandAsync(ct);

    private static string RandomToken(int len)
    {
        var bytes = RandomNumberGenerator.GetBytes(len);
        return Convert.ToHexString(bytes)[..len];
    }

    private static MdpInboundOutcome Fail(int status, string message) =>
        new() { HttpStatus = status, Body = new { code = status, message, data = (object)null } };

    private static MdpInboundOutcome Ok(object data, int http = 200) =>
        new() { HttpStatus = http, Body = new { code = 0, message = "ok", data } };
}