package Stats::Redshift::ExportSchema;

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

use JSON::XS;
use FileHandle;
use File::Basename;
use Parallel::ForkManager;
use Text::CSV::Easy(qw/csv_build csv_parse/);

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

use lib '/app/tools/stat/lib';
use Stats::Redshift;
use Stats::Redshift::DatabaseList;

use base 'Class::Accessor';

__PACKAGE__->mk_accessors(qw/database_list redshift_rps redshift_dtmv refresh/);

sub new {
    my $class = shift;
    my $self  = {};
    bless $self, $class;
    $self->_init();
    $self;
}

sub _init {
    my $self = shift;
    $self->database_list( Stats::Redshift::DatabaseList->new );

    my $rpstabs  = $self->_redshift_tables('rps');
    my $dtmvtabs = $self->_redshift_tables('dtmv');

    $self->redshift_rps($rpstabs);
    $self->redshift_dtmv($dtmvtabs);

    $self->refresh(0);

}

sub export_database {
    my $self = shift;
    my $name = shift;

    my $red = Common::DB::Export::Redshift->new($name);
    my $exp = $red->exporter;
    $exp->parallel(8);
    $exp->compress(1);

    logMessage( 'info', "Exporting $name to tables to redshift" );

    my $cid  = $red->database->client_id;
    my $type = lc( $red->database->client_type );

    my $tablist;
    if ( $type eq 'rps' ) {
        $tablist = $self->redshift_rps;
    } else {
        $tablist = $self->redshift_dtmv;
    }

    if ( $self->refresh ) {
        my $rs     = Stats::Redshift->new($type);
        my $schema = $red->redshift_schema;
        foreach my $tab ( sort @{$tablist} ) {
            my $sql = "delete from $schema.$tab where client_id=$cid";
            logMessage( 'warn2', $sql );
            $rs->dosql($sql);
        }
    }
    $red->directory->localremove;
    $red->directory->remoteremove;

    foreach my $tab ( sort @{$tablist} ) {
        my $sql = $self->outfile_sql($cid);

        logMessage( 'info2', "Exporting " . $red->schema . ".$tab" );
        ## set table directory
        $red->directory->table($tab);
        $exp->directory( $red->directory );

        my $files = $exp->export_table( $tab, $sql );
        if ( ref $files eq 'ARRAY' ) {
            my $manifest = $red->create_manifest($files);
            $red->send_to_s3;
            $red->load_redshift($tab);
        }
    }

}

sub truncate_all {
    my $self = shift;
    foreach my $type (qw/rps dtmv/) {
        my $rs = Stats::Redshift->new($type);

        my $tables;
        if ( $type eq 'rps' ) {
            $tables = $self->redshift_rps;
        } else {
            $tables = $self->redshift_dtmv;
        }

        ## truncate all tables on redshift
        logMessage( 'warn', "Removing all data for " . $rs->redshift_schema . " tables" );
        foreach my $tab ( @{$tables} ) {
            $rs->truncate($tab);
        }
    }
    logMessage( 'info', 'complete' );
}

sub outfile_sql {
    my $self = shift;
    my $cid  = shift;
    my $sql  = (
        qq{
SELECT /* :comment: */ $cid, NOW(), t.* 
FROM :database:.:table: t
WHERE 1
AND :primary: >= :min:
AND :primary:  < :max:
INTO OUTFILE ":file:" 
FIELDS TERMINATED BY '\t' OPTIONALLY ENCLOSED BY '"'
}
    );
}

sub _redshift_tables {
    my $self = shift;
    my $type = shift;

    ## connect to redshift
    my $rsdbh  = Common::Session::getdbh('redshift');
    my $tables = $rsdbh->selectcol_arrayref(
        qq{
select tablename from pg_tables 
where schemaname=?
and tablename != 'sale_stage'
and tablename != 'clientlist'
and tablename not like 'agg%'
order by tablename
}, undef, $type
    );
    $rsdbh->disconnect;
    $tables;
}

1;