package APKPM::Store;

use base qw(Gearman::Driver::Worker);
use Moose;
use Time::HiRes;
use Data::Dump qw(dump);
use DBD::Pg;
use Redis;

with 'APKPM::Gearman';

sub prefix { 'Store_' }

sub process_name {
	my ( $self, $orig, $job_name ) = @_;
	warn "# process_name $orig $job_name\n";
	return "$orig ($job_name)";
}

sub dbh {
	DBI->connect_cached('DBI:Pg:dbname=apkpm','','', {
		RaiseError => 1,
		AutoCommit => 1,
	});
}

our $redis;
sub redis { $redis ||= Redis->new };

sub _create_index {
	my ( $self, $table, $timestamp ) = @_;

	my $sth = $self->dbh->prepare( qq{

select
    t.relname as table_name,
    i.relname as index_name,
    a.attname as column_name
from
    pg_class t,
    pg_class i,
    pg_index ix,
    pg_attribute a
where
    t.oid = ix.indrelid
    and i.oid = ix.indexrelid
    and a.attrelid = t.oid
    and a.attnum = ANY(ix.indkey)
    and t.relkind = 'r'
    and t.relname like ?
order by
    t.relname,
    i.relname;

	} );
	$sth->execute($table);
	my $sql = '';
	while( my $row = $sth->fetchrow_hashref ) {
		#warn "# _create_index $table ",dump($row);
		$sql .= qq|create index $row->{index_name}_$timestamp on ${table}_$timestamp($row->{column_name});\n|;
	}

	return $sql;
}

sub pg_insert {
	my ( $self, $base_table, $h ) = @_;

	my $table = $base_table;

	my @c;

	my $timestamp =
		exists $h->{timestamp} ? $h->{timestamp} :
		exists $h->{start}     ? $h->{start}     :
		warn "no timestamp partitioning on $table";

	my $create_table;
	if ( $timestamp ) {
		$timestamp =~ s/^(\d{4})-(\d{2})-(\d{2}).*$/$1_$2_$3/;
		my $part = $table . '_' . $timestamp;
		$create_table = qq{
			create table $part () inherits ($table) ;
		};
		$create_table .= $self->_create_index( $table, $timestamp );
		$table = $part;
		warn "# using partition $table";
	}

	if ( my $cols = $self->redis->get("pg.$table") ) {
		@c = split(/\s+/,$cols);
	} else {
		my $sth = $self->dbh->prepare( "select * from $table limit 1" );
		eval { $sth->execute; };
		if ( $@ ) {
			warn "ERROR $@\n# try $create_table\n";
			$self->dbh->do( $create_table );
			$sth->execute;
		}

		@c = @{ $sth->{NAME_lc} };
		$self->redis->set( "pg.$table" => join(' ',@c) );
		$self->redis->expire( "pg.$table" => 15 * 60 ); # refresh every 15 min
	}

	my $sql = "INSERT INTO $table (" . join(',',@c) . ') values (' . join(',', map { '?' } 0 .. $#c) . ')';
	warn $sql;
	my $sth = $self->dbh->prepare($sql);

	my $h_lc;
	$h_lc->{ lc $_ } = $h->{$_} foreach keys %$h;

	if ( my $username = $h->{username} ) {
		my $key = join('.', 'table', $base_table, $username);
		$self->redis->set( $key => $self->e_json($h) );
		$self->redis->expire( $key => 15 * 60 ); # 15 min timeout
	}

	$h_lc->{h} = $self->to_hstore( $h_lc->{h} ) if exists $h_lc->{h};

	$sth->execute( map { $h_lc->{$_} } @c );
}

sub insert : Job : Decode(d_json) : MinProcesses(0) {
	my ( $self, $job, $workload ) = @_;
	my $table = delete $workload->{_table} || die "no _table";
	$self->pg_insert($table => $workload);
}

sub sql : Job : Encode(e_json) : MinProcesses(1) {
	my ( $self, $job, $workload ) = @_;

	my $sth = $self->dbh->prepare($workload);
	my $rows = eval { $sth->execute };
	return { error => $@ } if $@;

	warn "# $rows rows get_username_table $workload\n";

	$rows = $sth->fetchall_arrayref;
	my @columns = @{ $sth->{NAME} };

	# decode hash column
	my $hash_col;
	foreach ( 0 .. $#columns ) {
		$hash_col = $_ if $columns[$_] eq 'h';
	}
	if ( defined $hash_col ) {
		map {
			my $hash = $_->[$hash_col];
			$hash =~ s/\@/\\\@/g;
			$_->[$hash_col] = eval "{ $hash }";
			$_->[$hash_col] = "ERROR: $@ for $hash" if $@;
		} @$rows
	}

	return {
		columns => \@columns,
		rows => $rows,
		hash_col => $hash_col,
	};
}

1;
