|
|
@@ -8,30 +8,51 @@ using Microsoft.Data.SqlClient;
|
|
|
using Microsoft.Extensions.Options;
|
|
|
|
|
|
var builder = WebApplication.CreateBuilder(args);
|
|
|
-builder.Services.Configure<SyncOptions>(builder.Configuration.GetSection("XiaoshuSync"));
|
|
|
+builder.WebHost.ConfigureKestrel(options => options.Limits.MaxRequestBodySize = 1_048_576);
|
|
|
+builder.Services.AddOptions<SyncOptions>()
|
|
|
+ .Bind(builder.Configuration.GetSection("XiaoshuSync"))
|
|
|
+ .Validate(options => !string.IsNullOrWhiteSpace(options.KeyId), "XiaoshuSync KeyId 未配置")
|
|
|
+ .Validate(options => options.Secret.Length >= 32, "XiaoshuSync Secret 必须至少 32 位")
|
|
|
+ .Validate(options => options.MaxClockSkewSeconds is >= 30 and <= 900, "签名时间窗口必须在 30 至 900 秒之间")
|
|
|
+ .Validate(options => options.DefaultPageSize > 0 && options.MaxPageSize >= options.DefaultPageSize && options.MaxPageSize <= 5000, "同步分页配置无效")
|
|
|
+ .Validate(options => options.MaxCommandBodyBytes is >= 1024 and <= 1_048_576, "写入请求大小上限无效")
|
|
|
+ .Validate(options => options.PathBase.StartsWith('/') && !options.PathBase.EndsWith('/'), "PathBase 必须以 / 开头且不能以 / 结尾")
|
|
|
+ .ValidateOnStart();
|
|
|
builder.Services.AddSingleton<ReplayGuard>();
|
|
|
builder.Services.AddSingleton<LegacyRepository>();
|
|
|
var app = builder.Build();
|
|
|
+app.UsePathBase(app.Services.GetRequiredService<IOptions<SyncOptions>>().Value.PathBase);
|
|
|
|
|
|
+app.UseExceptionHandler(exceptionApp => exceptionApp.Run(async context =>
|
|
|
+{
|
|
|
+ context.Response.StatusCode = StatusCodes.Status500InternalServerError;
|
|
|
+ context.Response.ContentType = "application/json";
|
|
|
+ await context.Response.WriteAsJsonAsync(new { error = "sync_bridge_error", message = "同步桥执行失败,请查看服务器日志" });
|
|
|
+}));
|
|
|
app.Use(async (context, next) =>
|
|
|
{
|
|
|
- if (!context.Request.Path.StartsWithSegments("/xiaoshu-sync/v1")) { await next(); return; }
|
|
|
- var rejection = await RequestAuthenticator.ValidateAsync(context, context.RequestServices.GetRequiredService<IOptions<SyncOptions>>().Value, context.RequestServices.GetRequiredService<ReplayGuard>());
|
|
|
+ if (!context.Request.Path.StartsWithSegments("/v1") || context.Request.Path.Equals("/v1/health")) { await next(); return; }
|
|
|
+ context.Response.Headers.CacheControl = "no-store";
|
|
|
+ var configured = context.RequestServices.GetRequiredService<IOptions<SyncOptions>>().Value;
|
|
|
+ if (context.Request.Path.StartsWithSegments("/v1/commands") && context.Request.ContentLength > configured.MaxCommandBodyBytes) { context.Response.StatusCode = StatusCodes.Status413PayloadTooLarge; await context.Response.WriteAsJsonAsync(new { error = "payload_too_large" }); return; }
|
|
|
+ var rejection = await RequestAuthenticator.ValidateAsync(context, configured, context.RequestServices.GetRequiredService<ReplayGuard>());
|
|
|
if (rejection is not null) { context.Response.StatusCode = rejection.Value.Status; await context.Response.WriteAsJsonAsync(new { error = rejection.Value.Message }); return; }
|
|
|
await next();
|
|
|
});
|
|
|
|
|
|
-app.MapGet("/xiaoshu-sync/v1/manifest", async (LegacyRepository repository, CancellationToken cancellationToken) => Results.Ok(await repository.ManifestAsync(cancellationToken)));
|
|
|
-app.MapGet("/xiaoshu-sync/v1/changes", async (string dataset, string? cursor, int? limit, LegacyRepository repository, CancellationToken cancellationToken) =>
|
|
|
+app.MapGet("/v1/health", () => Results.Ok(new { service = "xiaoshu-legacy-sync", status = "ok", version = "v1", time = DateTimeOffset.UtcNow }));
|
|
|
+app.MapGet("/v1/manifest", async (LegacyRepository repository, CancellationToken cancellationToken) => Results.Ok(await repository.ManifestAsync(cancellationToken)));
|
|
|
+app.MapGet("/v1/changes", async (string dataset, string? cursor, int? limit, LegacyRepository repository, CancellationToken cancellationToken) =>
|
|
|
{
|
|
|
if (!DatasetCatalog.All.TryGetValue(dataset, out var definition)) return Results.BadRequest(new { error = "unsupported_dataset", message = $"不支持的数据集:{dataset}" });
|
|
|
return Results.Ok(await repository.ChangesAsync(definition, cursor, limit, cancellationToken));
|
|
|
});
|
|
|
-app.MapPost("/xiaoshu-sync/v1/commands/{operation}", async (string operation, HttpContext context, LegacyRepository repository, CancellationToken cancellationToken) =>
|
|
|
+app.MapPost("/v1/commands/{operation}", async (string operation, HttpContext context, LegacyRepository repository, IOptions<SyncOptions> configured, CancellationToken cancellationToken) =>
|
|
|
{
|
|
|
if (!CommandCatalog.Allowed.Contains(operation)) return Results.BadRequest(new { error = "unsupported_operation", message = $"不支持的写入操作:{operation}" });
|
|
|
+ if (context.Request.ContentLength is > 0 && context.Request.ContentLength > configured.Value.MaxCommandBodyBytes) return Results.StatusCode(StatusCodes.Status413PayloadTooLarge);
|
|
|
var command = await context.Request.ReadFromJsonAsync<LegacyCommand>(cancellationToken: cancellationToken);
|
|
|
- if (command is null || string.IsNullOrWhiteSpace(command.IdempotencyKey) || string.IsNullOrWhiteSpace(command.Reason) || string.IsNullOrWhiteSpace(command.Actor)) return Results.BadRequest(new { error = "invalid_command", message = "actor、reason 和 idempotencyKey 必填" });
|
|
|
+ if (command is null || string.IsNullOrWhiteSpace(command.IdempotencyKey) || command.IdempotencyKey.Length > 128 || string.IsNullOrWhiteSpace(command.Reason) || command.Reason.Length > 500 || string.IsNullOrWhiteSpace(command.Actor) || command.Actor.Length > 200 || command.Payload.ValueKind is not JsonValueKind.Object) return Results.BadRequest(new { error = "invalid_command", message = "actor、reason、idempotencyKey 和对象 payload 必填且长度必须有效" });
|
|
|
return Results.Ok(await repository.ExecuteCommandAsync(operation, command, cancellationToken));
|
|
|
});
|
|
|
app.MapGet("/health", () => Results.Ok(new { service = "xiaoshu-legacy-sync", status = "ok", time = DateTimeOffset.UtcNow }));
|
|
|
@@ -45,6 +66,9 @@ sealed class SyncOptions
|
|
|
public int MaxClockSkewSeconds { get; init; } = 300;
|
|
|
public int DefaultPageSize { get; init; } = 500;
|
|
|
public int MaxPageSize { get; init; } = 1000;
|
|
|
+ public int MaxCommandBodyBytes { get; init; } = 1_048_576;
|
|
|
+ public string LegacyTimeZoneId { get; init; } = "China Standard Time";
|
|
|
+ public string PathBase { get; init; } = "/xiaoshu-sync";
|
|
|
}
|
|
|
|
|
|
readonly record struct AuthRejection(int Status, string Message);
|
|
|
@@ -67,7 +91,7 @@ static class RequestAuthenticator
|
|
|
using var reader = new StreamReader(context.Request.Body, Encoding.UTF8, leaveOpen: true);
|
|
|
var body = await reader.ReadToEndAsync(); context.Request.Body.Position = 0;
|
|
|
var bodyHash = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(body))).ToLowerInvariant();
|
|
|
- var canonical = $"{context.Request.Method.ToUpperInvariant()}\n{context.Request.Path}{context.Request.QueryString}\n{timestampText}\n{nonce}\n{bodyHash}";
|
|
|
+ var canonical = $"{context.Request.Method.ToUpperInvariant()}\n{context.Request.PathBase}{context.Request.Path}{context.Request.QueryString}\n{timestampText}\n{nonce}\n{bodyHash}";
|
|
|
using var hmac = new HMACSHA256(Encoding.UTF8.GetBytes(options.Secret));
|
|
|
var expected = Convert.ToHexString(hmac.ComputeHash(Encoding.UTF8.GetBytes(canonical))).ToLowerInvariant();
|
|
|
return FixedText(signature.ToLowerInvariant(), expected) ? null : new(401, "请求签名无效");
|
|
|
@@ -89,9 +113,11 @@ sealed class ReplayGuard
|
|
|
sealed record LegacyCommand(string Actor, string Reason, string IdempotencyKey, string? ExpectedUpdatedAt, JsonElement Payload);
|
|
|
sealed record ProjectionRecord(string ClassName, string KeyField, string LegacyKey, Dictionary<string, object?> Fields);
|
|
|
sealed record ChangeItem(string Key, string Operation, DateTimeOffset? ChangedAt, IReadOnlyList<ProjectionRecord> Records);
|
|
|
-sealed record ChangePage(string Dataset, string Cursor, bool HasMore, int Count, DateTimeOffset ServerTime, IReadOnlyList<ChangeItem> Items);
|
|
|
-sealed record DatasetDefinition(string Key, string Label, string CountSql, string ChangesSql, string LegacyKeyColumn, IReadOnlyList<RecordMapping> Records);
|
|
|
+sealed record ChangePage(string Dataset, string Cursor, bool HasMore, int Count, DateTimeOffset ServerTime, string Phase, long Watermark, IReadOnlyList<ChangeItem> Items);
|
|
|
+sealed record DatasetDefinition(string Key, string Label, string CountSql, string KeysSql, string SnapshotSql, string LookupSql, string LegacyKeyColumn, IReadOnlyList<RecordMapping> Records);
|
|
|
sealed record RecordMapping(string ClassName, string KeyField, string Prefix, string SourcePrefix);
|
|
|
+sealed record CursorState(bool Incremental, long Position, long SnapshotWatermark);
|
|
|
+sealed record ChangeLogRow(long Id, string LegacyKey, string Operation, DateTimeOffset? ChangedAt, string? ItemKey);
|
|
|
|
|
|
static class DatasetCatalog
|
|
|
{
|
|
|
@@ -99,25 +125,40 @@ static class DatasetCatalog
|
|
|
public static readonly IReadOnlyDictionary<string, DatasetDefinition> All = new Dictionary<string, DatasetDefinition>(StringComparer.OrdinalIgnoreCase)
|
|
|
{
|
|
|
["users"] = new("users", "全部业务账号", "SELECT COUNT_BIG(*) FROM ZL_User",
|
|
|
- """SELECT TOP (@limit) CAST(u.UserID AS bigint) legacyKey,COALESCE(u.LastLoginTimes,u.RegTime) changedAt,u.UserID [user__legacyUserId],u.UserName [user__username],u.HoneyName [user__nickname],u.Email [user__email],u.Mobile [user__mobile],u.GroupID [user__legacyGroupId],u.ParentUserID [user__parentUserId],u.RegTime [user__registeredAt],u.LastLoginTimes [user__lastLoginAt],u.LoginTimes [user__loginCount],u.State [user__legacyState],u.Purse [user__purse],u.SilverCoin [user__silverCoin],u.UserExp [user__userExp],u.UserPoint [user__userPoint],u.DummyPurse [user__dummyPurse],u.UserCreit [user__credit] FROM ZL_User u WHERE u.UserID>@after ORDER BY u.UserID""", "legacyKey", [new("_User", "legacyUserId", "user__", "legacy-sync:user:")]),
|
|
|
+ "SELECT CAST(UserID AS bigint) legacyKey FROM ZL_User ORDER BY UserID",
|
|
|
+ """SELECT TOP (@limit) CAST(u.UserID AS bigint) legacyKey,COALESCE(u.LastLoginTimes,u.RegTime) changedAt,u.UserID [user__legacyUserId],u.UserName [user__username],u.HoneyName [user__nickname],u.Email [user__email],u.Mobile [user__mobile],u.GroupID [user__legacyGroupId],u.ParentUserID [user__parentUserId],u.RegTime [user__registeredAt],u.LastLoginTimes [user__lastLoginAt],u.LoginTimes [user__loginCount],u.State [user__legacyState],u.Purse [user__purse],u.SilverCoin [user__silverCoin],u.UserExp [user__userExp],u.UserPoint [user__userPoint],u.DummyPurse [user__dummyPurse],u.UserCreit [user__credit] FROM ZL_User u WHERE u.UserID>@after ORDER BY u.UserID""",
|
|
|
+ """SELECT TOP (1) CAST(u.UserID AS bigint) legacyKey,COALESCE(u.LastLoginTimes,u.RegTime) changedAt,u.UserID [user__legacyUserId],u.UserName [user__username],u.HoneyName [user__nickname],u.Email [user__email],u.Mobile [user__mobile],u.GroupID [user__legacyGroupId],u.ParentUserID [user__parentUserId],u.RegTime [user__registeredAt],u.LastLoginTimes [user__lastLoginAt],u.LoginTimes [user__loginCount],u.State [user__legacyState],u.Purse [user__purse],u.SilverCoin [user__silverCoin],u.UserExp [user__userExp],u.UserPoint [user__userPoint],u.DummyPurse [user__dummyPurse],u.UserCreit [user__credit] FROM ZL_User u WHERE u.UserID=@legacyKey""", "legacyKey", [new("_User", "legacyUserId", "user__", "legacy-sync:user:")]),
|
|
|
["nodes"] = new("nodes", "课程与词库目录", "SELECT COUNT_BIG(*) FROM ZL_Node",
|
|
|
- """SELECT TOP (@limit) CAST(n.NodeID AS bigint) legacyKey,NULL changedAt,n.NodeID [node__nodeId],n.ParentID [node__parentId],n.NodeName [node__nodeName],n.NodeType [node__nodeType],n.NodeDir [node__nodeDir],n.NodePic [node__nodePicUrl],n.Description [node__description],n.OrderID [node__orderId],n.CUser [node__cuser],n.CUName [node__cuname],n.EditDate [node__editDate] FROM ZL_Node n WHERE n.NodeID>@after ORDER BY n.NodeID""", "legacyKey", [new("Node", "nodeId", "node__", "legacy-sync:node:")]),
|
|
|
+ "SELECT CAST(NodeID AS bigint) legacyKey FROM ZL_Node ORDER BY NodeID",
|
|
|
+ """SELECT TOP (@limit) CAST(n.NodeID AS bigint) legacyKey,NULL changedAt,n.NodeID [node__nodeId],n.ParentID [node__parentId],n.NodeName [node__nodeName],n.NodeType [node__nodeType],n.NodeDir [node__nodeDir],n.NodePic [node__nodePicUrl],n.Description [node__description],n.OrderID [node__orderId],n.CUser [node__cuser],n.CUName [node__cuname],n.EditDate [node__editDate] FROM ZL_Node n WHERE n.NodeID>@after ORDER BY n.NodeID""",
|
|
|
+ """SELECT TOP (1) CAST(n.NodeID AS bigint) legacyKey,NULL changedAt,n.NodeID [node__nodeId],n.ParentID [node__parentId],n.NodeName [node__nodeName],n.NodeType [node__nodeType],n.NodeDir [node__nodeDir],n.NodePic [node__nodePicUrl],n.Description [node__description],n.OrderID [node__orderId],n.CUser [node__cuser],n.CUName [node__cuname],n.EditDate [node__editDate] FROM ZL_Node n WHERE n.NodeID=@legacyKey""", "legacyKey", [new("Node", "nodeId", "node__", "legacy-sync:node:")]),
|
|
|
["course-bindings"] = Content("course-bindings", "课程绑定", 58, "ZL_C_kcbd", "b", ["yhid","kcid","yxx","cksl","syjd"]),
|
|
|
["appointments"] = Content("appointments", "预约排课", 54, "ZL_C_order", "a", ["szyh","pl","fxpl","plxm","kcid","yykcid","dslx","jffs","bxrq","sdsd","yysj","dszt","kssj","jssj","scsj","szmd"]),
|
|
|
["lessons"] = Content("lessons", "上课记录", 59, "ZL_C_skjl", "l", ["xymz","jsmz","kcid","kclx","kcmc","pf","pjnr","pldp","kzsj","yyds","plpjsj","szmdid"]),
|
|
|
["learning-records"] = Content("learning-records", "每日学习记录", 56, "ZL_C_ss", "d", ["userId","pl","dqrq","dsid","learned","ygg","djq","xxqs","fxrl","szmdid"]),
|
|
|
["practice-records"] = Content("practice-records", "练习记录", 53, "ZL_C_lxjl", "p", ["yhid","kcid","scid","xxcs","jrscb"]),
|
|
|
["memory-records"] = Content("memory-records", "抗遗忘记录", 60, "ZL_C_gywjl", "m", ["yhid","plid","kcid","xxjlid","fxzt","wcsj","kywrq","kywsj","OrderID"]),
|
|
|
- ["assessments"] = Content("assessments", "测评档案", 61, "ZL_C_cespj", "e", ["userId","df","askid","wrong","answerid","dontKnow","prevScore","totalScore"]),
|
|
|
+ ["assessments"] = Content("assessments", "测评档案", 61, "ZL_C_cpda", "e", ["userId","df","askid","wrong","answerid","dontKnow","prev_score:prevScore","totalScore"]),
|
|
|
["vocabulary"] = Content("vocabulary", "词库", 52, "ZL_C_ck", "v", ["sy","yb","lj","yp"]),
|
|
|
["special-training"] = Content("special-training", "专项训练", 2, "ZL_C_Article", "r", ["ico","mp4","author","source","content","synopsis"], " AND c.NodeID=3")
|
|
|
};
|
|
|
|
|
|
private static DatasetDefinition Content(string key, string label, int modelId, string addonTable, string alias, string[] addonFields, string extraWhere = "")
|
|
|
{
|
|
|
- var addonSelect = string.Join(',', addonFields.Select(field => $"{alias}.[{field}] [addon__{ToCamel(field)}]"));
|
|
|
- var sql = $"SELECT TOP (@limit) CAST(c.GeneralID AS bigint) legacyKey,COALESCE(c.UpDateTime,c.CreateTime) changedAt,c.GeneralID [common__generalId],c.ItemID [common__itemId],c.ModelID [common__modelId],c.NodeID [common__nodeId],c.TableName [common__tableName],c.Title [common__title],c.Subtitle [common__subtitle],c.Inputer [common__inputer],c.CreateTime [common__createTime],c.UpDateTime [common__upDateTime],c.Status [common__status],c.OrderID [common__orderId],c.TopImg [common__topImg],{alias}.ID [addon__id],{addonSelect} FROM ZL_CommonModel c INNER JOIN {addonTable} {alias} ON {alias}.ID=c.ItemID WHERE c.ModelID={modelId}{extraWhere} AND c.GeneralID>@after ORDER BY c.GeneralID";
|
|
|
- return new(key, label, $"SELECT COUNT_BIG(*) FROM ZL_CommonModel c WHERE c.ModelID={modelId}{extraWhere} AND {ActiveCommon}", sql, "legacyKey", [new("CommonModel", "generalId", "common__", $"legacy-sync:{key}:common:"), new(ClassFor(modelId), "id", "addon__", $"legacy-sync:{key}:addon:")]);
|
|
|
+ var addonSelect = string.Join(',', addonFields.Select(field =>
|
|
|
+ {
|
|
|
+ var mapping = field.Split(':', 2);
|
|
|
+ var sourceField = mapping[0];
|
|
|
+ var targetField = mapping.Length == 2 ? mapping[1] : ToCamel(sourceField);
|
|
|
+ return $"{alias}.[{sourceField}] [addon__{targetField}]";
|
|
|
+ }));
|
|
|
+ var select = $"CAST(c.GeneralID AS bigint) legacyKey,COALESCE(c.UpDateTime,c.CreateTime) changedAt,c.GeneralID [common__generalId],c.ItemID [common__itemId],c.ModelID [common__modelId],c.NodeID [common__nodeId],c.TableName [common__tableName],c.Title [common__title],c.Subtitle [common__subtitle],c.Inputer [common__inputer],c.CreateTime [common__createTime],c.UpDateTime [common__upDateTime],c.Status [common__status],c.OrderID [common__orderId],c.TopImg [common__topImg],{alias}.ID [addon__id],{addonSelect}";
|
|
|
+ var from = $" FROM ZL_CommonModel c INNER JOIN {addonTable} {alias} ON {alias}.ID=c.ItemID WHERE c.ModelID={modelId}{extraWhere} AND {ActiveCommon}";
|
|
|
+ var snapshotSql = $"SELECT TOP (@limit) {select}{from} AND c.GeneralID>@after ORDER BY c.GeneralID";
|
|
|
+ var lookupSql = $"SELECT TOP (1) {select}{from} AND c.GeneralID=@legacyKey";
|
|
|
+ var countSql = $"SELECT COUNT_BIG(*){from}";
|
|
|
+ var keysSql = $"SELECT CAST(c.GeneralID AS bigint) legacyKey{from} ORDER BY c.GeneralID";
|
|
|
+ return new(key, label, countSql, keysSql, snapshotSql, lookupSql, "legacyKey", [new("CommonModel", "generalId", "common__", $"legacy-sync:{key}:common:"), new(ClassFor(modelId), "id", "addon__", $"legacy-sync:{key}:addon:")]);
|
|
|
}
|
|
|
private static string ClassFor(int modelId) => modelId switch { 2 => "ContentArticle", 52 => "VocabularyWord", 53 => "PracticeRecord", 54 => "CourseAppointment", 56 => "DailyStudyRecord", 58 => "CourseBinding", 59 => "LessonRecord", 60 => "MemoryPracticeRecord", 61 => "AssessmentProfile", _ => throw new InvalidOperationException() };
|
|
|
private static string ToCamel(string value) => value.Length == 0 ? value : char.ToLowerInvariant(value[0]) + value[1..];
|
|
|
@@ -135,21 +176,60 @@ sealed class LegacyRepository(IConfiguration configuration, IOptions<SyncOptions
|
|
|
public async Task<object> ManifestAsync(CancellationToken cancellationToken)
|
|
|
{
|
|
|
await using var connection = new SqlConnection(_connectionString); await connection.OpenAsync(cancellationToken);
|
|
|
+ var watermarkStart = await GlobalWatermarkAsync(connection, cancellationToken);
|
|
|
var datasets = new List<object>();
|
|
|
foreach (var definition in DatasetCatalog.All.Values)
|
|
|
{
|
|
|
await using var command = new SqlCommand(definition.CountSql, connection) { CommandTimeout = 120 };
|
|
|
var count = Convert.ToInt64(await command.ExecuteScalarAsync(cancellationToken), CultureInfo.InvariantCulture);
|
|
|
- datasets.Add(new { key = definition.Key, label = definition.Label, count });
|
|
|
+ var checksum = await KeyChecksumAsync(connection, definition, cancellationToken);
|
|
|
+ await using var watermarkCommand = new SqlCommand("SELECT COALESCE(MAX(Id),0) FROM dbo.XiaoshuSyncChangeLog WHERE Dataset=@dataset", connection);
|
|
|
+ watermarkCommand.Parameters.AddWithValue("@dataset", definition.Key);
|
|
|
+ var watermark = Convert.ToInt64(await watermarkCommand.ExecuteScalarAsync(cancellationToken), CultureInfo.InvariantCulture);
|
|
|
+ datasets.Add(new { key = definition.Key, label = definition.Label, count, watermark, checksum });
|
|
|
+ }
|
|
|
+ var watermarkEnd = await GlobalWatermarkAsync(connection, cancellationToken);
|
|
|
+ var now = DateTimeOffset.UtcNow;
|
|
|
+ return new { manifestTime = now, serverTime = now, consistent = watermarkStart == watermarkEnd, watermarkStart, watermarkEnd, datasets };
|
|
|
+ }
|
|
|
+ private static async Task<long> GlobalWatermarkAsync(SqlConnection connection, CancellationToken cancellationToken)
|
|
|
+ {
|
|
|
+ await using var command = new SqlCommand("SELECT COALESCE(MAX(Id),0) FROM dbo.XiaoshuSyncChangeLog", connection);
|
|
|
+ return Convert.ToInt64(await command.ExecuteScalarAsync(cancellationToken), CultureInfo.InvariantCulture);
|
|
|
+ }
|
|
|
+ private static async Task<string> KeyChecksumAsync(SqlConnection connection, DatasetDefinition definition, CancellationToken cancellationToken)
|
|
|
+ {
|
|
|
+ using var hash = IncrementalHash.CreateHash(HashAlgorithmName.SHA256);
|
|
|
+ await using var command = new SqlCommand(definition.KeysSql, connection) { CommandTimeout = 180 };
|
|
|
+ await using var reader = await command.ExecuteReaderAsync(CommandBehavior.SequentialAccess, cancellationToken);
|
|
|
+ while (await reader.ReadAsync(cancellationToken))
|
|
|
+ {
|
|
|
+ var key = Convert.ToString(reader[0], CultureInfo.InvariantCulture) ?? "";
|
|
|
+ hash.AppendData(Encoding.UTF8.GetBytes(key)); hash.AppendData("\n"u8);
|
|
|
}
|
|
|
- return new { manifestTime = DateTimeOffset.UtcNow, serverTime = DateTimeOffset.UtcNow, datasets };
|
|
|
+ return Convert.ToHexString(hash.GetHashAndReset()).ToLowerInvariant();
|
|
|
}
|
|
|
public async Task<ChangePage> ChangesAsync(DatasetDefinition definition, string? cursor, int? requestedLimit, CancellationToken cancellationToken)
|
|
|
{
|
|
|
- var after = CursorCodec.Decode(cursor); var limit = Math.Clamp(requestedLimit ?? _options.DefaultPageSize, 1, _options.MaxPageSize);
|
|
|
+ var state = CursorCodec.Decode(cursor); var limit = Math.Clamp(requestedLimit ?? _options.DefaultPageSize, 1, _options.MaxPageSize);
|
|
|
await using var connection = new SqlConnection(_connectionString); await connection.OpenAsync(cancellationToken);
|
|
|
- await using var command = new SqlCommand(definition.ChangesSql, connection) { CommandTimeout = 180 };
|
|
|
- command.Parameters.Add(new SqlParameter("@after", SqlDbType.BigInt) { Value = after }); command.Parameters.Add(new SqlParameter("@limit", SqlDbType.Int) { Value = limit + 1 });
|
|
|
+ if (state.Incremental) return await IncrementalChangesAsync(connection, definition, state, limit, cancellationToken);
|
|
|
+ var snapshotWatermark = state.SnapshotWatermark;
|
|
|
+ if (snapshotWatermark == 0)
|
|
|
+ {
|
|
|
+ await using var watermarkCommand = new SqlCommand("SELECT COALESCE(MAX(Id),0) FROM dbo.XiaoshuSyncChangeLog WHERE Dataset=@dataset", connection);
|
|
|
+ watermarkCommand.Parameters.AddWithValue("@dataset", definition.Key);
|
|
|
+ snapshotWatermark = Convert.ToInt64(await watermarkCommand.ExecuteScalarAsync(cancellationToken), CultureInfo.InvariantCulture);
|
|
|
+ }
|
|
|
+ await using var command = new SqlCommand(definition.SnapshotSql, connection) { CommandTimeout = 180 };
|
|
|
+ command.Parameters.Add(new SqlParameter("@after", SqlDbType.BigInt) { Value = state.Position }); command.Parameters.Add(new SqlParameter("@limit", SqlDbType.Int) { Value = limit + 1 });
|
|
|
+ var items = await ReadProjectionItemsAsync(command, definition, cancellationToken);
|
|
|
+ var hasMore = items.Count > limit; if (hasMore) items.RemoveAt(items.Count - 1); var next = items.Count == 0 ? state.Position : long.Parse(items[^1].Key, CultureInfo.InvariantCulture);
|
|
|
+ var nextCursor = hasMore ? CursorCodec.EncodeSnapshot(next, snapshotWatermark) : CursorCodec.EncodeIncremental(snapshotWatermark);
|
|
|
+ return new ChangePage(definition.Key, nextCursor, hasMore, items.Count, DateTimeOffset.UtcNow, "snapshot", snapshotWatermark, items);
|
|
|
+ }
|
|
|
+ private async Task<List<ChangeItem>> ReadProjectionItemsAsync(SqlCommand command, DatasetDefinition definition, CancellationToken cancellationToken)
|
|
|
+ {
|
|
|
var items = new List<ChangeItem>(); await using var reader = await command.ExecuteReaderAsync(cancellationToken);
|
|
|
while (await reader.ReadAsync(cancellationToken))
|
|
|
{
|
|
|
@@ -161,16 +241,46 @@ sealed class LegacyRepository(IConfiguration configuration, IOptions<SyncOptions
|
|
|
{
|
|
|
var name = reader.GetName(index); if (!name.StartsWith(mapping.Prefix, StringComparison.Ordinal)) continue;
|
|
|
var field = name[mapping.Prefix.Length..]; var value = reader.IsDBNull(index) ? null : reader.GetValue(index);
|
|
|
- if (value is DateTime date) value = new DateTimeOffset(DateTime.SpecifyKind(date, DateTimeKind.Local)).ToUniversalTime(); fields[field] = value;
|
|
|
+ if (value is DateTime date) value = LegacyDateTime.ToUtc(date, _options.LegacyTimeZoneId); fields[field] = value;
|
|
|
}
|
|
|
var recordKey = Convert.ToString(fields.GetValueOrDefault(mapping.KeyField), CultureInfo.InvariantCulture) ?? key; fields["sourceKey"] = mapping.SourcePrefix + recordKey;
|
|
|
records.Add(new ProjectionRecord(mapping.ClassName, mapping.KeyField, recordKey, fields));
|
|
|
}
|
|
|
- DateTimeOffset? changedAt = reader["changedAt"] is DateTime changed ? new DateTimeOffset(DateTime.SpecifyKind(changed, DateTimeKind.Local)).ToUniversalTime() : null;
|
|
|
+ DateTimeOffset? changedAt = reader["changedAt"] is DateTime changed ? LegacyDateTime.ToUtc(changed, _options.LegacyTimeZoneId) : null;
|
|
|
items.Add(new ChangeItem(key, "upsert", changedAt, records));
|
|
|
}
|
|
|
- var hasMore = items.Count > limit; if (hasMore) items.RemoveAt(items.Count - 1); var next = items.Count == 0 ? after : long.Parse(items[^1].Key, CultureInfo.InvariantCulture);
|
|
|
- return new ChangePage(definition.Key, CursorCodec.Encode(next), hasMore, items.Count, DateTimeOffset.UtcNow, items);
|
|
|
+ return items;
|
|
|
+ }
|
|
|
+ private async Task<ChangePage> IncrementalChangesAsync(SqlConnection connection, DatasetDefinition definition, CursorState state, int limit, CancellationToken cancellationToken)
|
|
|
+ {
|
|
|
+ const string logSql = "SELECT TOP (@limit) Id,LegacyKey,Operation,ChangedAt,ItemKey FROM dbo.XiaoshuSyncChangeLog WHERE Dataset=@dataset AND Id>@after ORDER BY Id";
|
|
|
+ await using var logCommand = new SqlCommand(logSql, connection) { CommandTimeout = 120 };
|
|
|
+ logCommand.Parameters.AddWithValue("@dataset", definition.Key); logCommand.Parameters.AddWithValue("@after", state.Position); logCommand.Parameters.AddWithValue("@limit", limit + 1);
|
|
|
+ var changes = new List<ChangeLogRow>(); await using (var reader = await logCommand.ExecuteReaderAsync(cancellationToken))
|
|
|
+ {
|
|
|
+ while (await reader.ReadAsync(cancellationToken))
|
|
|
+ {
|
|
|
+ DateTimeOffset? changedAt = reader["ChangedAt"] is DateTime changed ? new DateTimeOffset(DateTime.SpecifyKind(changed, DateTimeKind.Utc)) : null;
|
|
|
+ changes.Add(new(Convert.ToInt64(reader["Id"], CultureInfo.InvariantCulture), Convert.ToString(reader["LegacyKey"], CultureInfo.InvariantCulture) ?? "", Convert.ToString(reader["Operation"], CultureInfo.InvariantCulture) ?? "U", changedAt, reader["ItemKey"] is DBNull ? null : Convert.ToString(reader["ItemKey"], CultureInfo.InvariantCulture)));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ var hasMore = changes.Count > limit; if (hasMore) changes.RemoveAt(changes.Count - 1);
|
|
|
+ var items = new List<ChangeItem>();
|
|
|
+ foreach (var change in changes)
|
|
|
+ {
|
|
|
+ if (change.Operation.Equals("D", StringComparison.OrdinalIgnoreCase)) { items.Add(DeleteItem(definition, change)); continue; }
|
|
|
+ await using var lookup = new SqlCommand(definition.LookupSql, connection) { CommandTimeout = 120 };
|
|
|
+ lookup.Parameters.Add(new SqlParameter("@legacyKey", SqlDbType.BigInt) { Value = long.Parse(change.LegacyKey, CultureInfo.InvariantCulture) });
|
|
|
+ var projected = await ReadProjectionItemsAsync(lookup, definition, cancellationToken);
|
|
|
+ items.Add(projected.Count == 0 ? DeleteItem(definition, change) : projected[0] with { ChangedAt = change.ChangedAt ?? projected[0].ChangedAt });
|
|
|
+ }
|
|
|
+ var next = changes.Count == 0 ? state.Position : changes[^1].Id;
|
|
|
+ return new ChangePage(definition.Key, CursorCodec.EncodeIncremental(next), hasMore, items.Count, DateTimeOffset.UtcNow, "incremental", next, items);
|
|
|
+ }
|
|
|
+ private static ChangeItem DeleteItem(DatasetDefinition definition, ChangeLogRow change)
|
|
|
+ {
|
|
|
+ var records = definition.Records.Select((mapping, index) => new ProjectionRecord(mapping.ClassName, mapping.KeyField, index == 0 ? change.LegacyKey : change.ItemKey ?? change.LegacyKey, new Dictionary<string, object?>())).ToList();
|
|
|
+ return new ChangeItem(change.LegacyKey, "delete", change.ChangedAt, records);
|
|
|
}
|
|
|
public async Task<object> ExecuteCommandAsync(string operation, LegacyCommand command, CancellationToken cancellationToken)
|
|
|
{
|
|
|
@@ -184,6 +294,33 @@ sealed class LegacyRepository(IConfiguration configuration, IOptions<SyncOptions
|
|
|
|
|
|
static class CursorCodec
|
|
|
{
|
|
|
- public static string Encode(long value) => Convert.ToBase64String(Encoding.UTF8.GetBytes($"v1:{value}"));
|
|
|
- public static long Decode(string? value) { if (string.IsNullOrWhiteSpace(value)) return 0; try { var decoded = Encoding.UTF8.GetString(Convert.FromBase64String(value)); return decoded.StartsWith("v1:") && long.TryParse(decoded[3..], out var parsed) ? Math.Max(0, parsed) : 0; } catch { return 0; } }
|
|
|
+ public static string EncodeSnapshot(long position, long watermark) => Encode($"v2:s:{Math.Max(0, position)}:{Math.Max(0, watermark)}");
|
|
|
+ public static string EncodeIncremental(long position) => Encode($"v2:c:{Math.Max(0, position)}");
|
|
|
+ public static CursorState Decode(string? value)
|
|
|
+ {
|
|
|
+ if (string.IsNullOrWhiteSpace(value)) return new(false, 0, 0);
|
|
|
+ try
|
|
|
+ {
|
|
|
+ var decoded = Encoding.UTF8.GetString(Convert.FromBase64String(value)); var parts = decoded.Split(':');
|
|
|
+ if (parts.Length == 2 && parts[0] == "v1" && long.TryParse(parts[1], out var legacy)) return new(false, Math.Max(0, legacy), 0);
|
|
|
+ if (parts.Length >= 3 && parts[0] == "v2" && long.TryParse(parts[2], out var position))
|
|
|
+ {
|
|
|
+ if (parts[1] == "c") return new(true, Math.Max(0, position), 0);
|
|
|
+ if (parts[1] == "s") return new(false, Math.Max(0, position), parts.Length > 3 && long.TryParse(parts[3], out var watermark) ? Math.Max(0, watermark) : 0);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ catch { }
|
|
|
+ return new(false, 0, 0);
|
|
|
+ }
|
|
|
+ private static string Encode(string value) => Convert.ToBase64String(Encoding.UTF8.GetBytes(value));
|
|
|
+}
|
|
|
+
|
|
|
+static class LegacyDateTime
|
|
|
+{
|
|
|
+ public static DateTimeOffset ToUtc(DateTime value, string timeZoneId)
|
|
|
+ {
|
|
|
+ var unspecified = DateTime.SpecifyKind(value, DateTimeKind.Unspecified);
|
|
|
+ var zone = TimeZoneInfo.FindSystemTimeZoneById(timeZoneId);
|
|
|
+ return new DateTimeOffset(TimeZoneInfo.ConvertTimeToUtc(unspecified, zone), TimeSpan.Zero);
|
|
|
+ }
|
|
|
}
|