package Stats::Redshift::UnloadDimension;

use constant 'ROLE' => 'arn:aws:iam::290365479392:role/service-role/AmazonRedshift-CommandsAccessRole-20220212T094750';

use strict;
use warnings;

use DateTime;

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

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

use base 'Stats::Redshift';

__PACKAGE__->mk_accessors(qw/dimensions directory database_list/);

sub _init {
    my $self = shift;
    $self->SUPER::_init();

    my $dim = Stats::Redshift::Dimension->new( $self->type );
    $self->dimensions( $dim->dimensions );

    $self->database_list( Stats::Redshift::DatabaseList->new() );

    $self->directory( Common::DB::Export::Directory->new() );
    $self->directory->parent('schema-imports');

    $self;
}

sub unload_dim {
    my $self     = shift;
    my $cid      = shift;
    my $dim      = shift;
    my $parallel = shift;

    my $schema = $self->redshift_schema;

    my $cols = $dim->columns;
    push @{$cols}, qw/s.units s.revenue/ if $dim->tablename !~ m/.*_summary$/;
    $cols = [ map { $_ eq 'bc.contributor_id' ? 's.author_id' : $_ } @{$cols} ];

    my $select = join ',', @{$cols};

    my $sql = "UNLOAD (\n'SELECT $select FROM ";
    $sql .= "${schema}." . $dim->tablename . " s\n";
    $sql .= "WHERE s.client_id=$cid'\n)\n";
    $sql .= "TO '" . $self->directory->remote . ".'\n";
    $sql .= "IAM_ROLE '" . ROLE . "'\n";
    $sql .= "ALLOWOVERWRITE\n";
    $sql .= "CSV DELIMITER '\t'\n";
    $sql .= "PARALLEL OFF\n" if not $parallel;
    $sql .= "MAXFILESIZE 12MB";
    ## $sql .= "GZIP\n";

    $self->dosql($sql);

}

sub unload_client {
    my $self  = shift;
    my $db    = shift;
    my $force = shift;

    my $dbh = Common::Session::getdbh('redshift');
    my $cid = $db->client_id;

    $self->directory->schema( $db->db_name );

    foreach my $dim ( @{ $self->dimensions } ) {
        $self->directory->table( $dim->tablename );

        my $table = $self->redshift_schema . "." . $dim->tablename;
        logMessage( 'info', "Unloading $table" );

        my $cntsql = "SELECT TO_CHAR(COUNT(*),'999,999,999'),COUNT(*) FROM $table WHERE client_id=?";
        my $count  = $dbh->selectall_arrayref( $cntsql, undef, $cid )->[0];
        if ( defined $count ) {
            my $format = $count->[0];
            $format =~ s/ *//g;
            my $ttlrows  = $count->[1];
            my $parallel = 1;
            if ( $ttlrows < 100_000 ) {
                logMessage( 'info2', "rowcount: $format unloading serially" );
                $parallel = 0;
            } else {
                logMessage( 'info2', "rowcount: $format unloading parallel" );
            }
            my $err = $self->unload_dim( $db->client_id, $dim, $parallel );
        } else {
            logMessage( 'error2', $dbh->errstr );
        }
    }
    $self->copy_local;
    $dbh->disconnect;

    logMessage( 'info', 'complete' );

    return 1;

}

sub unload_all {
    my $self = shift;
    my $type = $self->type;

    my $allclients = $self->database_list->{$type};
    foreach my $db ( @{$allclients} ) {
        logMessage( 'info', "Unloading client " . $db->db_name );
        $self->unload_client($db);
    }

}

sub copy_local {
    my $self = shift;

    $self->directory->table(undef);
    my $local  = $self->directory->local;
    my $remote = $self->directory->remote;

    $self->directory->localcreate;

    my $out = runlogcmd("aws s3 cp --recursive $remote/ $local/");
    if ( $out->[2] ) {
        logMessage( 'error2', $out->[1] );
        exit 1;
    }
}

sub has_changed {
    my $self  = shift;
    my $db    = shift;
    my $force = shift;

    ## get redshift agg_sale_log data
    my $rdbh   = Common::Session::getdbh('redshift');
    my $schema = $self->redshift_schema;
    my $rsth   = $rdbh->prepare("SELECT * FROM ${schema}.agg_sale_log WHERE client_id=?");
    $rsth->execute( $db->client_id );
    my $reds_data = $rsth->fetchrow_hashref;
    $rsth->finish;
    $rdbh->disconnect;

    my $has_changed = 0;
	my $what_changed;
    if ($force) {
        $has_changed = 1;
    } elsif ( not defined $reds_data ) {
        $has_changed = 0;
    } else {
        my $dbhost = $db->host->aws_host->hostname;

        my $dbh = Common::Session::getdbh($dbhost);

        my $dbname = $db->db_name;
        my $sth    = $dbh->prepare(
            qq{
        SELECT log_id, total_records, total_units, ROUND(total_revenue,2) total_revenue, max_date_modified
        FROM ${dbname}.agg_sale_log ORDER BY log_id DESC LIMIT 1
        }
        );
        $sth->execute;
        my $log_data = $sth->fetchrow_hashref;
        ## if we don't get any log_data back, we want to rebuild the agg tables
        $sth->finish;
        $dbh->disconnect;

        if ( not defined $log_data ) {
            $has_changed = 1;
        } else {
            foreach my $key (qw/total_records total_units total_revenue max_date_modified/) {
                if ( $reds_data->{$key} ne $log_data->{$key} ) {
                    $what_changed->{$key} = { 'new' => $reds_data->{$key}, 'old' => $log_data->{$key} };
                    $has_changed++;
                }
            }
        }
    }

    ## remove redshift specific columns
    map { delete $reds_data->{$_} } (qw/client_id date_loaded log_id/);

    ( $has_changed, $reds_data, $what_changed );
}

1;