lib/APKPM/Store.pm

   1 package APKPM::Store;
   2
   3 use base qw(Gearman::Driver::Worker);
   4 use Moose;
   5 use Time::HiRes;
   6 use Data::Dump qw(dump);
   7 use DBD::Pg;
   8 use Redis;
   9
  10 with 'APKPM::Gearman';
  11
  12 sub prefix { 'Store_' }
  13
  14 sub process_name {
  15         my ( $self, $orig, $job_name ) = @_;
  16         warn "# process_name $orig $job_name\n";
  17         return "$orig ($job_name)";
  18 }
  19
  20 sub dbh {
  21         DBI->connect_cached('DBI:Pg:dbname=apkpm','','', {
  22                 RaiseError => 1,
  23                 AutoCommit => 1,
  24         });
  25 }
  26
  27 our $redis;
  28 sub redis { $redis ||= Redis->new };
  29
  30 sub pg_insert {
  31         my ( $self, $table, $h ) = @_;
  32
  33         my @c;
  34
  35         my $timestamp =
  36                 exists $h->{timestamp} ? $h->{timestamp} :
  37                 exists $h->{start}     ? $h->{start}     :
  38                 warn "no timestamp partitioning on $table";
  39
  40         my $create_table;
  41         if ( $timestamp ) {
  42                 $timestamp =~ s/^(\d{4})-(\d{2})-(\d{2}).*$/$1_$2_$3/;
  43                 my $part = $table . '_' . $timestamp;
  44                 $create_table = qq{
  45                         create table $part () inherits ($table) ;
  46                 };
  47                 $table = $part;
  48                 warn "# using partition $table";
  49         }
  50
  51         if ( my $cols = $self->redis->get("pg.$table") ) {
  52                 @c = split(/\s+/,$cols);
  53         } else {
  54                 my $sth = $self->dbh->prepare( "select * from $table limit 1" );
  55                 eval { $sth->execute; };
  56                 if ( $@ ) {
  57                         warn "ERROR $@\n# try $create_table\n";
  58                         $self->dbh->do( $create_table );
  59                         $sth->execute;
  60                 }
  61
  62                 @c = @{ $sth->{NAME_lc} };
  63                 $self->redis->set( "pg.$table" => join(' ',@c) );
  64                 $self->redis->expire( "pg.$table" => 15 * 60 ); # refresh every 15 min
  65         }
  66
  67         my $sql = "INSERT INTO $table (" . join(',',@c) . ') values (' . join(',', map { '?' } 0 .. $#c) . ')';
  68         warn $sql;
  69         my $sth = $self->dbh->prepare($sql);
  70
  71         my $h_lc;
  72         $h_lc->{ lc $_ } = $h->{$_} foreach keys %$h;
  73
  74         if ( my $username = $h->{username} ) {
  75                 my $key = join('.', 'table', $table, $username);
  76                 $self->redis->set( $key => $self->e_json($h) );
  77                 $self->redis->expire( $key => 15 * 60 ); # 15 min timeout
  78         }
  79
  80         $h_lc->{h} = $self->to_hstore( $h_lc->{h} ) if exists $h_lc->{h};
  81
  82         $sth->execute( map { $h_lc->{$_} } @c );
  83 }
  84
  85 sub insert : Job : Decode(d_json) : MinProcesses(0) {
  86         my ( $self, $job, $workload ) = @_;
  87         my $table = delete $workload->{_table} || die "no _table";
  88         $self->pg_insert($table => $workload);
  89 }
  90
  91 sub sql : Job : Encode(e_json) : MinProcesses(1) {
  92         my ( $self, $job, $workload ) = @_;
  93
  94         my $sth = $self->dbh->prepare($workload);
  95         my $rows = eval { $sth->execute };
  96         return { error => $@ } if $@;
  97
  98         warn "# $rows rows get_username_table $workload\n";
  99
 100         $rows = $sth->fetchall_arrayref;
 101         my @columns = @{ $sth->{NAME} };
 102
 103         # decode hash column
 104         my $hash_col;
 105         foreach ( 0 .. $#columns ) {
 106                 $hash_col = $_ if $columns[$_] eq 'h';
 107         }
 108         if ( defined $hash_col ) {
 109                 map {
 110                         my $hash = $_->[$hash_col];
 111                         $hash =~ s/\@/\\\@/g;
 112                         $_->[$hash_col] = eval "{ $hash }";
 113                         $_->[$hash_col] = "ERROR: $@ for $hash" if $@;
 114                 } @$rows
 115         }
 116
 117         return {
 118                 columns => \@columns,
 119                 rows => $rows,
 120                 hash_col => $hash_col,
 121         };
 122 }
 123
 124 1;