-- usp_SyncApply: set-based apply of pending SyncQueue rows.
-- Per distinct (TargetSchema, TargetTable) it builds column lists from sys.columns
-- (excluding the ID key, computed, identity and rowversion columns) and runs a
-- dynamic-SQL MERGE (last-write-wins by SourceLogID DESC) for Insert/Update and a
-- DELETE for the last op = Delete. SET IDENTITY_INSERT ON preserves Access PKs.
--
-- CORRECTNESS NOTE: a SINGLE ranked CTE (rn=1 per RecordID, over ALL pending ops
-- regardless of OperateType) is read by BOTH the upsert and delete branches.
-- Earlier this was two independent ranked CTEs (one over Insert/Update, one over
-- Delete), which let a stale Delete outrank a newer Insert for the same RecordID
-- and silently drop a row whose true last op was an Insert. Routing off the one
-- winning row's OperateType guarantees only the genuine last op wins.

CREATE OR ALTER PROCEDURE ProductionDataBaseSync.usp_SyncApply
    @MaxRetries INT = 5
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @sch NVARCHAR(128), @tbl NVARCHAR(255), @FullName NVARCHAR(514);
    DECLARE @sql NVARCHAR(MAX), @cols NVARCHAR(MAX), @upd NVARCHAR(MAX), @ins NVARCHAR(MAX);

    -- Re-queue rows still under the retry budget.
    UPDATE ProductionDataBaseSync.SyncQueue
       SET Status = 'pending'
     WHERE Status = 'error' AND RetryCount < @MaxRetries;

    DECLARE cur CURSOR LOCAL FAST_FORWARD FOR
        SELECT DISTINCT TargetSchema, TargetTable
          FROM ProductionDataBaseSync.SyncQueue
         WHERE Status = 'pending';

    OPEN cur;
    FETCH NEXT FROM cur INTO @sch, @tbl;

    WHILE @@FETCH_STATUS = 0
    BEGIN
        SET @FullName = QUOTENAME(@sch) + N'.' + QUOTENAME(@tbl);

        BEGIN TRY
            BEGIN TRAN;
            -- @cols: comma-quoted column names (for INSERT target list)
            -- @upd:  tgt.col = JSON_VALUE(src.RowData,'$.col')
            -- @ins:  JSON_VALUE(src.RowData,'$.col')  (for INSERT VALUES)
            -- NOTE: every JSON path key is wrapped in double quotes ('$."col"').
            -- This is mandatory for non-ASCII column names (e.g. Chinese names
            -- like 名字) and harmless for ASCII names, so we quote unconditionally.
            SELECT
                @cols = STRING_AGG(QUOTENAME(c.name), N',')
                            WITHIN GROUP (ORDER BY c.column_id),
                @upd  = STRING_AGG(QUOTENAME(c.name) + N'=JSON_VALUE(src.RowData,''$."' + c.name + N'"'')', N',')
                            WITHIN GROUP (ORDER BY c.column_id),
                @ins  = STRING_AGG(N'JSON_VALUE(src.RowData,''$."' + c.name + N'"'')', N',')
                            WITHIN GROUP (ORDER BY c.column_id)
              FROM sys.columns c
             WHERE c.object_id = OBJECT_ID(@FullName)
               AND c.is_computed = 0
               AND c.is_identity = 0
               AND TYPE_NAME(c.system_type_id) <> 'timestamp'
               AND c.name <> 'ID';

            IF @cols IS NOT NULL
            BEGIN
                -- Upsert: winners (rn=1) whose winning OperateType is
                -- Insert/Update and carry RowData. The ranked CTE covers ALL
                -- pending ops so the rn=1 row is the true last op for that
                -- RecordID; an earlier Delete can no longer outrank a newer
                -- Insert for the same RecordID.
                SET @sql = N'SET IDENTITY_INSERT ' + @FullName + N' ON;'
                         + N';WITH ranked AS ('
                         + N'  SELECT RecordID, OperateType, RowData,'
                         + N'         ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn'
                         + N'    FROM ProductionDataBaseSync.SyncQueue'
                         + N'   WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending'''
                         + N')'
                         + N'MERGE ' + @FullName + N' WITH (HOLDLOCK) AS tgt'
                         + N' USING ('
                         + N'  SELECT RecordID, RowData FROM ranked'
                         + N'   WHERE rn=1 AND OperateType IN (''Insert'',''Update'') AND RowData IS NOT NULL'
                         + N') AS src ON tgt.ID = TRY_CAST(src.RecordID AS int)'
                         + N' WHEN MATCHED THEN UPDATE SET ' + @upd
                         + N' WHEN NOT MATCHED THEN INSERT (ID,' + @cols + N')'
                         + N'  VALUES (TRY_CAST(src.RecordID AS int),' + @ins + N');'
                         + N'SET IDENTITY_INSERT ' + @FullName + N' OFF;';
                EXEC sp_executesql @sql,
                     N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl;

                -- Delete: winners (rn=1) whose winning OperateType is Delete.
                -- Same ranked CTE — only the genuine last op can be a delete.
                SET @sql = N';WITH ranked AS ('
                         + N'  SELECT RecordID, OperateType,'
                         + N'         ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn'
                         + N'    FROM ProductionDataBaseSync.SyncQueue'
                         + N'   WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending'''
                         + N')'
                         + N'DELETE t FROM ' + @FullName + N' t'
                         + N' JOIN (SELECT RecordID FROM ranked WHERE rn=1 AND OperateType=''Delete'') d'
                         + N'   ON t.ID = TRY_CAST(d.RecordID AS int);';
                EXEC sp_executesql @sql,
                     N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl;
            END

            UPDATE ProductionDataBaseSync.SyncQueue
               SET Status = 'applied', AppliedAt = SYSDATETIME()
             WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending';

            COMMIT;
        END TRY
        BEGIN CATCH
            -- ROLLBACK unwinds the per-table transaction (and, if the caller
            -- started one, the caller's too). A proc that errors under an outer
            -- transaction is expected to leave the txn uncommittable; the caller
            -- issues its own rollback. We roll back here so the partial per-table
            -- work is discarded before marking rows as 'error'/'dead'.
            IF @@TRANCOUNT > 0 ROLLBACK;
            UPDATE ProductionDataBaseSync.SyncQueue
               SET Status      = CASE WHEN RetryCount + 1 >= @MaxRetries THEN 'dead' ELSE 'error' END,
                   RetryCount   = RetryCount + 1,
                   ErrorMsg     = ERROR_MESSAGE()
             WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending';
        END CATCH

        FETCH NEXT FROM cur INTO @sch, @tbl;
    END

    CLOSE cur;
    DEALLOCATE cur;
END
GO