Auto Fix SQL Server Transactional Replication Error 20598 for UPDATEs

<#Assumption: 1. the schema of all the publishing tables remains the same, and by default, it is 'dbo'
            2. The computer on which this script runs has sql server 2012 PowerShell module installed
            3. The computer on which this script runs has PowerShell V3 or above and .net 3.5 or above installed
Usage:   1. modify $schema / $PubSvr / $PubDB / $SubSvr / $SubDB / $Distributor to be your environment settings
         2. Test in the test environment by initially set $debug = $true, recommended to run in the ISE first to get familiar with the script
         3. When there are potential many rows missing (for update), please set $duration to a proper time (like 10 min), 
            and it will loop to fix all the missing rows one by one.
#>
#Requires -Version 3.0
import-module sqlps -DisableNameChecking;

[int] $duration = 3 # in minutes, how long this script will run, this is to handle multiple missing rows
[boolean]$debug = $true; # when $true, it will print out a query statement that can be executed on the subscriber server to insert the missing row
                         # it will not do any real fix work as such it will only print out the first fixing query.
[string]$schema='dbo';# the schema of the all publishing tables, 
[string]$PubSvr = 'tp_w520\sql2k8r2'; #publisher, in my example, this is sql server 2008R2
[string]$PubDB = 'testdb'; #publication database
[string]$SubSvr = 'tp_w520';#subscriber, in my example, this is sql server 2012
[string]$SubDB = 'testdb'; #subscription database
[string]$Distributor = 'tp_w520';#replication distributor

[datetime]$end_time = (get-date).AddMinutes($duration);

#first let's see what's the latest error 20598
[string]$maxid = '0'; 
[string]$xact_seq='0x00';
[string]$command_id='0'

$qry = "select id from dbo.MSPublisher_Databases where publisher_db='$pubdb';"
$result2 = invoke-sqlcmd -ServerInstance $Distributor -database Distribution -Query $qry;

while ((get-date) -le $end_time) {
    $qry = @"
    declare @id int;
    select @id = max(id) from dbo.MSRepl_Errors where id > $maxid and error_code=20598 and time > dateadd(hour, -1, getdate())
    and  (xact_seqno<>$xact_seq or command_id <> $command_id);
    select top 1 id=cast(id as varchar(8)), xact_seqno=convert(varchar(42), xact_seqno,1), command_id=cast(Command_id as varchar(8)) from dbo.MSRepl_Errors where id = @id
"@
    $result = $null;
    $result=invoke-sqlcmd -ServerInstance $Distributor -Database Distribution -Query $qry;

    if ($result -eq $null) {
        start-sleep -Seconds 15;
        continue;
    }
    else {
        $maxid = $result.ID;
        $xact_seq=$result.xact_seqno;
        $command_id = $result.command_id;
    }

    $qry = @"
    exec sp_browsereplcmds @xact_seqno_start='$($result.xact_seqno)', @xact_seqno_end='$($result.xact_seqno)'
    , @command_id=$($result.command_id), @publisher_database_id=$($result2.id);
"@;

    $result3 = invoke-sqlcmd -ServerInstance $Distributor -database Distribution -Query $qry; 
    [string]$odbc_qry = $result3.command;

    [string]$reg_pattern ="sp_msupd_$($schema)(.+?)\]"
    if ($odbc_qry -match $reg_pattern) {
        $TableName ="$($schema).$($matches[1])"
    }
    else {
       write-error 'The current error cannot be solved by this solution, which aims only at missing rows when updating.'}

    #need to get the parameter values and put into $pm
    if ($odbc_qry -match "(\{CALL \[sp_MSUpd_.+)\]") {
       if ($odbc_qry -match "\((.+?)\)") {
        [string]$pm = $matches[1];
       }
       else {
        write-error "cannot find parameters";
        return;
       }
    }
    else {
        write-error 'The current error cannot be solved by this solution, which aims only at missing rows when updating.'
        return;
    }

    #next we need to script out the columns of the Subscription Table
    $svr = new-object "microsoft.sqlserver.management.smo.server" $SubSvr
    $db = $svr.databases[$subdb];

    $tbl=$db.Tables[$TableName.split('.')[1], $TableName.split('.')[0]];

    [string]$col_list = "";

    $tbl.Columns |  % {$col_list += $_.Name + ',' };
    $col_list = $col_list.remove($col_list.length-1);
 
    #we need to grab the where clause in the replication update stored proc of the subscription table
    $p = $db.StoredProcedures["sp_MSUpd_$($TableName.replace(""."", """"))"];

    if ($p -eq $null) {
        write-error "cannot find the stored proce of sp_MS$($Fix_Type)_$($TableName.replace(""."", """"))"
        return;
    }

    $ar = $pm.split(",")
    [int]$i = 0;
    $where_clause = @{};
    while ($i -lt $p.parameters.count ) {
        if ($p.parameters[$i].name -match '@pkc[0-9]{1,2}') {
            switch ($p.parameters[$i].DataType.Name) {
             'datetime' { $ar[$i] = "'" + $ar[$i] + "'"; break;}
             'uniqueidentifier' {$ar[$i]=$ar[$i].replace("{", "'");$ar[$i]=$ar[$i].replace("}", "'"); break;}
            }
             $where_clause.add($p.parameters[$i].name, $ar[$i]);
            #$where_clause. += New-Object -TypeName PSObject -Property @{ $($p.parameters[$i].name)=$ar[$i]; }
            
        }
        $i += 1;
    }

    $re = new-object "System.Text.RegularExpressions.regex" ("where.+?(?=if)", [System.Text.RegularExpressions.RegexOptions]::Singleline)
    $where=$re.match($p.TextBody) 
    [string]$str = $where.Value;

    foreach ($k in $where_clause.keys) {
        $str=$str.replace($k, $where_clause[$k]);
    }

    [string]$qry = "select $col_list from $TableName $str";

    if ($debug) {
        "insert into $subsvr.$subdb.$tablename ($col_list) `
         select $col_list from $pubsvr.$pubdb.$TableName $str;
        "
    }
    else { # start to auto insert the missing row
        [System.Data.datarow]$r=$null;

        $r = invoke-sqlcmd -ServerInstance $SubSvr -Database $SubDB -Query $qry;

        if ($r -eq $null) {#if subsvr.subdb does not have the row
        #start to insert the missing row from the pubsvr.pubdb.tablename
            [string]$conn =  "Data Source=$SubSvr; trusted_connection=true; initial catalog=$SubDB;"
            $sqlbcp = new-object "system.data.sqlclient.sqlbulkcopy" ($conn, [System.Data.SqlClient.SqlBulkCopyOptions]::KeepIdentity)
            $r=$null
            $r = invoke-sqlcmd -ServerInstance $PubSvr -Database $PubDB -Query $qry;

            $sqlbcp.DestinationTableName = $TableName;

            $tbl.Columns | % {$sqlbcp.ColumnMappings.Add($_.name, $_.name) } | out-null;
            $sqlbcp.WriteToServer($r);
       }#if subsvr.subdb does not have the row
    } #end to auto insert the missing row
    start-sleep -Seconds 15; # repl error will take some time before being written to distribution.dbo.msrepl_errors table.
}