#!/usr/bin/perl

use strict;
use warnings;
use open ':std', ':encoding(UTF-8)';

use Parallel::ForkManager;
use Term::ANSIColor qw(:constants);
use List::MoreUtils qw(first_index);
use Digest::MD5 qw(md5 md5_hex md5_base64);

use lib '/app/tools/common/lib';
use Common::Session;
use Common::DB::Diff::Table;
use Common::DB::Diff::Database;

## $SIG{CHLD} = sub { Parallel::ForkManager::wait_children($pm) };

my ( $opt, $usage ) = clopt(
    [ 'help|?|h' => "print usage message and exit", { 'implies' => { 'database' => 1 } } ],
    [
        'tabsopt' => 'hidden' => {
            'one_of' =>
              [ [ 'verify|v' => 'Verify all is_diff=Y', { 'implies' => { 'force' => 1 } } ], [ 'tables|t=s' => 'Table filter CSV' ] ],
        },
    ],
    [ 'chunksize|c=i' => 'Chunksize of rows to compare', { 'default'  => 100_000 } ],
    [ 'database|d=s'  => 'Database to compare',          { 'required' => 1 } ],
    [ 'force|f'       => 'Force compare if DB diff = Y', { 'default'  => 0 } ],
    [ 'min=i'         => 'Min rowid to start',           { 'default'  => 1 } ],
    [ 'max=i'         => 'Max rowid to finsh' ],
    [ 'primary|p=s'   => 'Primary DB default: db01', { 'default' => 'db01' } ],
    [ 'quiet|q'       => 'Quiet mode - just show exceptions' ],
    [ 'replica|r=s'   => 'Replicat DB default: db02',        { 'default' => 'db02' } ],
    [ 'threads|T=i'   => 'Parallel threads to compare with', { 'default' => 24 } ],
);

if ( $opt->help ) {
    logMessage( 'info', $usage->text, 'options' );
    exit 1;
}

logMessage( 'info', "Getting all tables for " . $opt->database );
my $dbdiff = Common::DB::Diff::Database->new( $opt->database );

my $tables;
if ( defined $opt->tabsopt ) {
    if ( $opt->tabsopt eq 'tables' ) {
        map { $tables->{$_}++ } split ',', $opt->tables;
    } elsif ( $opt->tabsopt eq 'verify' ) {
        my $dbx = Common::Session::getdbx('admindb');
        my $all = [ $dbx->resultset('Report::SysDbReplicaDiff')->search( { 'is_diff' => 'Y', 'dbname' => $opt->database } )->all ];
        if ( ( not defined $all or scalar @{$all} < 1 ) and not $opt->force ) {
            logMessage( 'fatal', $opt->database . " has no tables where is_diff=Y in admindb" );
            exit 1;
        } else {
            map { $tables->{ $_->tabname }++ } @{$all};
        }
        undef $dbx;
    }
}

## find all the rows we've already checked and cache them
my $cache;
my $search = {
    'primary_host' => $dbdiff->primary_host,
    'replica_host' => $dbdiff->replica_host,
    'dbname'       => $dbdiff->database,
};
my $dbx  = Common::Session::getdbx('admindb');
my $rset = $dbx->resultset('Report::SysDbReplicaDiff')->search($search);
while ( my $row = $rset->next ) {
    next if $opt->force and $row->is_diff eq 'Y';
    my $keys;
    foreach my $col (qw/primary_host replica_host dbname tabname minrow maxrow/) {
        push @{$keys}, $row->get_column($col);
    }
    my $key = join '-', @{$keys};
    $cache->{$key}++;
}
undef $dbx;

my $j = 1;
foreach my $tdiff ( @{ $dbdiff->tables } ) {
    my $idx = sprintf( "(%05d/%05d)", $j++, $dbdiff->count );

    next if defined $opt->tabsopt and not exists $tables->{ $tdiff->table };

    logMessage( 'info', "$idx " . $opt->database . " " . $tdiff->table );

    my $maxid = defined $opt->max ? $opt->max : $tdiff->primary->pkmax;

    my $pm = Parallel::ForkManager->new( $opt->threads );

    for ( my $i = $opt->min ; $i <= $maxid ; $i += $opt->chunksize ) {
        my $min = $i;
        my $max = $min + $opt->chunksize - 1;
        $max = $maxid if $max > $maxid;

        my $keys = [ ( $tdiff->primary_host, $tdiff->replica_host, $tdiff->primary->database, $tdiff->primary->table, $min ) ];
        my $key  = join '-', @{$keys};
        if ( exists $cache->{$key} and not $opt->force ) {
            my $object = $tdiff->primary->database . "." . $tdiff->primary->table;
            logMessage( 'warn1', "$object already in admin db for $min:$max" );
            next;
        }

        $pm->start and next;

        my $record = {
            'primary_host' => $tdiff->primary_host,
            'replica_host' => $tdiff->replica_host,
            'dbname'       => $tdiff->primary->database,
            'tabname'      => $tdiff->primary->table,
            'minrow'       => $min,
            'maxrow'       => $max
        };

        my $dbh1 = Common::Session::getdbh( $tdiff->primary_host );
        my $dbh2 = Common::Session::getdbh( $tdiff->replica_host );
        $record->{'keycol'} = $tdiff->primary->pkcol;

        my ( $err, $msg, $minrow, $maxrow, $pcnt, $rcnt ) = $tdiff->compare_signatures( $min, $max, $dbh1, $dbh2 );
        if ($err) {
            my $mismatched;
            if ( $pcnt == $rcnt ) {
                $mismatched = $tdiff->compare_columns( $minrow, $maxrow, $dbh1, $dbh2 );
                $record->{'diff_json'} = JSON::XS::encode_json($mismatched) if defined $mismatched;
                if ( not defined $mismatched ) {
                    $record->{'is_diff'} = 'N';
                    $err = 0;
                } else {
                    $record->{'is_diff'} = 'Y';
                }
            } elsif ( $pcnt != $rcnt ) {
                $record->{'is_diff'} = 'Y';
            }
        } else {
            $record->{'is_diff'} = 'N';
        }
        $record->{'primary_cnt'} = $pcnt;
        $record->{'replica_cnt'} = $rcnt;
        $record->{'message'}     = $msg;
        if ($err) {
            logMessage( 'error1', $msg );
        } else {
            logMessage( 'info1', $msg );
        }

        if ( $record->{'is_diff'} eq 'Y' ) {
            my $dbx2 = Common::Session::getdbx('admindb');
            $dbx2->resultset('Report::SysDbReplicaDiff')->update_or_create( $record, { 'key' => 'uidx_host_db_tab_row' } );
            undef $dbx2;
        }

        $pm->finish;

    }

    $pm->wait_all_children;

}