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 _logger; public MdpInboundSnapshotService( ISqlSugarClient db, MdpInboundAuthService auth, ILogger logger) { _db = db; _auth = auth; _logger = logger; } public async Task 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() .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 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() .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() .Where(r => r.SnapshotId == snapshotId && r.Status == "COMMITTED") .Select(r => r.SyncBatchId) .ToListAsync(ct); var snapshotKeys = new HashSet(StringComparer.Ordinal); if (batchIds.Count > 0) { var keys = await _db.Ado.SqlQueryAsync( $""" 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( $""" 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() .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() .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() .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() .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 LoadAsync( string snapshotId, string accessKey, long tenantId, string entityCode, CancellationToken ct) => _db.Queryable() .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 MarkStatusAsync(long id, string status, CancellationToken ct) => _db.Updateable() .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 } }; }