#!/usr/bin/perl

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

use Time::HiRes qw( time );

use lib '/app/tools/common/lib';
use Common::Session;
use Common::DB::Export;
use Common::DB::Export::Directory;

use lib '/app/tools/stats/lib';
use Stats::Redshift::Dimension;
use Stats::Redshift::UnloadDimension;

my ( $opt, $usage ) = clopt(
    [ 'help|?|h'     => "Print usage message and exit", { 'implies'   => { 'database' => 1 } } ],
    [ 'database|d=s' => 'Database to load',             { 'requiured' => 1 } ],
    [ 'force|f'      => 'Force stat refresh',           { 'default'   => 0 } ]
);
if ( $opt->help ) {
    logMessage( 'info', $usage->text, 'options' );
    exit 1;
}
my $begin_time = time();

my $exp    = Common::DB::Export->new( $opt->database );
my $type   = lc( $exp->database->client_type );
my $dim    = Stats::Redshift::Dimension->new($type);
my $schema = $exp->database->db_name;

logMessage( 'info', "Unloading all agg dimensions from redshift for $schema" );
my $ld = Stats::Redshift::UnloadDimension->new($type);
$ld->directory->schema($schema);

runlogcmd( "aws s3 rm --recursive " . $ld->directory->remote );
runlogcmd( "sudo rm -rf " . $ld->directory->local );

logMessage( 'info', "Checking if data has changed" );

my $dbhost = $exp->database->host->aws_host->hostname;

my ( $has_changed, $log_data, $what ) = $ld->has_changed( $exp->database, $opt->force );

if ( $has_changed or $opt->force ) {

    foreach my $t ( keys %{$what} ) {
        logMessage( 'info1', "$t mysql: $what->{$t}->{'old'} redshift: $what->{$t}->{'new'}" );
    }

    $log_data->{'date_started'} = DateTime->now('time_zone' => 'America/Los_Angeles')->datetime(' ');
    $ld->unload_client( $exp->database );

    logMessage( 'info1', "Importing data into $schema agg tables on $dbhost" );
    my $dbh = Common::Session::getdbh($dbhost);

    foreach my $table ( @{ $dim->listing } ) {
        my $sql = "TRUNCATE TABLE $schema.$table";
        logMessage( 'warn2', $sql );
        $dbh->do($sql);
    }
    $dbh->disconnect;

    my $dir = Common::DB::Export::Directory->new();
    $dir->mount('/mnt/mysql/stats');
    $dir->parent('schema-imports');
    $dir->schema($schema);

    my ($loginpath) = $dbhost =~ m/rps(.*)/;
    my $myexec = "mysqlimport --login-path=$loginpath --use-threads=4 --replace --local $schema ";

    my $files = $dir->localfiles();
    my $i     = 1;
    foreach my $file ( sort { $a <=> $b } @{$files} ) {

        my $path = $file->path;
        my $name = $file->name;
        my $idx  = sprintf( "(%03d/%03d)", $i++, scalar @{$files} );
        logMessage( 'info2', "$idx loading $schema $name" );
        my $out = runcmd("$myexec $path");
        if ( $out->[2] ) {
            logMessage( 'error3', $out->[1] );
        } else {
            $out->[0] =~ s/\n//g;
            logMessage( 'info3', $out->[0] );
        }

    }

    $log_data->{'tables_rebuilt'} = 1;

} else {
    logMessage( 'info1', $exp->database->db_name . " sale data unchanged use force (-f) to refresh stats" );
    $log_data->{'date_started'}   = DateTime->now('time_zone' => 'America/Los_Angeles')->datetime(' ');
    $log_data->{'tables_rebuilt'} = 0;
}

$log_data->{'date_finished'} = DateTime->now('time_zone' => 'America/Los_Angeles')->datetime(' ');

my $dbx = Common::Session::getdbx($dbhost);
$dbx->getdbh->do( 'USE ' . $schema );
my $row = $dbx->resultset('RPS::AggSaleLog')->create($log_data);
if ( defined $row ) {
    logMessage( 'info1', "AggSaleLog data created with ID: " . $row->log_id );
}

my $end_time = time();
my $elapsed  = sprintf( "%.2f", $end_time - $begin_time );

logMessage( 'info', "Complete - $elapsed secs elapsed" );