xiaoshu-app/legacy-sync-bridge/Program.cs

using System.Collections.Concurrent;
using System.Data;
using System.Globalization;
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using Microsoft.Data.SqlClient;
using Microsoft.Extensions.Options;

var builder = WebApplication.CreateBuilder(args);
builder.Services.Configure<SyncOptions>(builder.Configuration.GetSection("XiaoshuSync"));
builder.Services.AddSingleton<ReplayGuard>();
builder.Services.AddSingleton<LegacyRepository>();
var app = builder.Build();

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 (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) =>
{
    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) =>
{
    if (!CommandCatalog.Allowed.Contains(operation)) return Results.BadRequest(new { error = "unsupported_operation", message = $"不支持的写入操作：{operation}" });
    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 必填" });
    return Results.Ok(await repository.ExecuteCommandAsync(operation, command, cancellationToken));
});
app.MapGet("/health", () => Results.Ok(new { service = "xiaoshu-legacy-sync", status = "ok", time = DateTimeOffset.UtcNow }));
app.Run();

sealed class SyncOptions
{
    public string KeyId { get; init; } = "";
    public string Secret { get; init; } = "";
    public string[] AllowedIps { get; init; } = [];
    public int MaxClockSkewSeconds { get; init; } = 300;
    public int DefaultPageSize { get; init; } = 500;
    public int MaxPageSize { get; init; } = 1000;
}

readonly record struct AuthRejection(int Status, string Message);

static class RequestAuthenticator
{
    public static async Task<AuthRejection?> ValidateAsync(HttpContext context, SyncOptions options, ReplayGuard replayGuard)
    {
        if (string.IsNullOrWhiteSpace(options.KeyId) || string.IsNullOrWhiteSpace(options.Secret)) return new(503, "同步桥密钥尚未配置");
        var remoteIp = context.Connection.RemoteIpAddress?.ToString() ?? "";
        if (options.AllowedIps.Length > 0 && !options.AllowedIps.Contains(remoteIp, StringComparer.OrdinalIgnoreCase)) return new(403, "来源 IP 不在白名单");
        var keyId = context.Request.Headers["X-Xiaoshu-Key-Id"].ToString();
        var timestampText = context.Request.Headers["X-Xiaoshu-Timestamp"].ToString();
        var nonce = context.Request.Headers["X-Xiaoshu-Nonce"].ToString();
        var signature = context.Request.Headers["X-Xiaoshu-Signature"].ToString();
        if (!long.TryParse(timestampText, out var timestamp) || string.IsNullOrWhiteSpace(nonce) || string.IsNullOrWhiteSpace(signature) || !FixedText(keyId, options.KeyId)) return new(401, "签名头缺失或无效");
        if (Math.Abs(DateTimeOffset.UtcNow.ToUnixTimeSeconds() - timestamp) > options.MaxClockSkewSeconds) return new(401, "请求时间戳已过期");
        if (!replayGuard.TryUse(nonce, timestamp, options.MaxClockSkewSeconds)) return new(409, "nonce 已使用，拒绝重放");
        context.Request.EnableBuffering();
        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}";
        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, "请求签名无效");
    }
    private static bool FixedText(string left, string right) => left.Length == right.Length && CryptographicOperations.FixedTimeEquals(Encoding.UTF8.GetBytes(left), Encoding.UTF8.GetBytes(right));
}

sealed class ReplayGuard
{
    private readonly ConcurrentDictionary<string, long> _nonces = new();
    public bool TryUse(string nonce, long timestamp, int lifetimeSeconds)
    {
        var floor = DateTimeOffset.UtcNow.ToUnixTimeSeconds() - lifetimeSeconds;
        foreach (var stale in _nonces.Where(item => item.Value < floor).Select(item => item.Key)) _nonces.TryRemove(stale, out _);
        return nonce.Length is >= 16 and <= 128 && _nonces.TryAdd(nonce, timestamp);
    }
}

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 RecordMapping(string ClassName, string KeyField, string Prefix, string SourcePrefix);

static class DatasetCatalog
{
    private const string ActiveCommon = "COALESCE(c.Status,99)<>-2";
    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:")]),
        ["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:")]),
        ["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"]),
        ["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:")]);
    }
    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..];
}

static class CommandCatalog
{
    public static readonly HashSet<string> Allowed = new(StringComparer.OrdinalIgnoreCase) { "member.create", "member.update-profile", "member.reset-password", "member.change-status", "member.change-group", "member.adjust-balance", "member.bind-agent", "coach.create", "coach.update", "coach.delete", "appointment.create", "appointment.update", "appointment.cancel", "appointment.restore", "appointment.complete", "appointment.transition", "course-binding.save", "course-binding.recycle", "course-binding.restore", "vocabulary.save", "vocabulary.move", "vocabulary.recycle", "vocabulary.restore", "vocabulary.import", "special-training.save", "special-training.publish", "special-training.unpublish", "special-training.recycle", "special-training.restore" };
}

sealed class LegacyRepository(IConfiguration configuration, IOptions<SyncOptions> options)
{
    private readonly string _connectionString = configuration.GetConnectionString("LegacySqlServer") ?? throw new InvalidOperationException("LegacySqlServer connection string is missing");
    private readonly SyncOptions _options = options.Value;
    public async Task<object> ManifestAsync(CancellationToken cancellationToken)
    {
        await using var connection = new SqlConnection(_connectionString); await connection.OpenAsync(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 });
        }
        return new { manifestTime = DateTimeOffset.UtcNow, serverTime = DateTimeOffset.UtcNow, datasets };
    }
    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);
        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 });
        var items = new List<ChangeItem>(); await using var reader = await command.ExecuteReaderAsync(cancellationToken);
        while (await reader.ReadAsync(cancellationToken))
        {
            var key = Convert.ToString(reader[definition.LegacyKeyColumn], CultureInfo.InvariantCulture) ?? ""; var records = new List<ProjectionRecord>();
            foreach (var mapping in definition.Records)
            {
                var fields = new Dictionary<string, object?>(StringComparer.Ordinal);
                for (var index = 0; index < reader.FieldCount; index++)
                {
                    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;
                }
                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;
            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);
    }
    public async Task<object> ExecuteCommandAsync(string operation, LegacyCommand command, CancellationToken cancellationToken)
    {
        await using var connection = new SqlConnection(_connectionString); await connection.OpenAsync(cancellationToken);
        await using var sql = new SqlCommand("dbo.XiaoshuSync_ExecuteCommand", connection) { CommandType = CommandType.StoredProcedure, CommandTimeout = 120 };
        sql.Parameters.AddWithValue("@Operation", operation); sql.Parameters.AddWithValue("@Actor", command.Actor); sql.Parameters.AddWithValue("@Reason", command.Reason); sql.Parameters.AddWithValue("@IdempotencyKey", command.IdempotencyKey); sql.Parameters.AddWithValue("@ExpectedUpdatedAt", (object?)command.ExpectedUpdatedAt ?? DBNull.Value); sql.Parameters.AddWithValue("@PayloadJson", command.Payload.GetRawText());
        var output = sql.Parameters.Add("@ResultJson", SqlDbType.NVarChar, -1); output.Direction = ParameterDirection.Output; await sql.ExecuteNonQueryAsync(cancellationToken);
        using var document = JsonDocument.Parse(Convert.ToString(output.Value, CultureInfo.InvariantCulture) ?? "{}"); return new { operation, idempotencyKey = command.IdempotencyKey, result = document.RootElement.Clone(), serverTime = DateTimeOffset.UtcNow };
    }
}

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; } }
}