# -*- cperl -*-
package Celaenito::Indexer;
use strict;
use warnings;
use POE::Session;
use POE::Kernel;
use Celaenito::DB;
use Celaenito::HyperEstraier;
use Celaenito::Logger;
use Smart::Comments;
use Time::HiRes qw(gettimeofday tv_interval);
use Hash::Util qw(lock_keys);
use constant INTERVAL => 3; # index 状態になってから次に index 状態になるまでの時間
use constant BULK     => 3; # 一囘の index 状態で幾つまでの URL を處理するか

sub run {
    my ($class) = @_;
    my $this = bless {} => $class;

    $this->{sth} = undef;
    $this->{total} = 0;         # 索引生成が必要な URL の數
    $this->{done} = 0;          # 生成した索引の數

    lock_keys %$this;

    POE::Session->create(
        heap => $this,
        inline_states => {
            _start   => \&_start_handler,
            _stop    => \&_stop_handler,
            index    => \&_index_handler,
            shutdown => \&_shutdown_handler,
        },
       );
}

sub _reset_state () {
    DB->do(q{
        DELETE FROM est_state
    });
}

sub _state ($;$) {
    my ($key, $newval) = @_;

    if (@_ == 1) {
        ($_) = DB->selectrow_array(q{
            SELECT state_value
              FROM est_state
             WHERE state_key = ?
          }, undef, $key);
        $_;
    }
    else {
        if (defined $newval) {
            DB->do(q{
                REPLACE INTO est_state
                        (state_key, state_value)
                 VALUES (?        , ?          )
            }, undef, $key, $newval);
        }
        else {
            DB->do(q{
                DELETE FROM est_state
                 WHERE state_key = ?
            }, undef, $key);
        }
        $newval;
    }
}

sub _start_handler {
    my $this = $_[HEAP];

    # casket が空だった場合は最初から。
    if (EST->doc_num == 0) {
        logdebug
          'The HyperEst DB is empty.';
        
        _reset_state();
    }

    my $last_lastmod = _state('last_lastmod') || 0;
    my $last_urlid   = _state('last_urlid')   || -1;

    ($this->{total}) = DB->selectrow_array(q{
        SELECT COUNT(*)
          FROM target_url
         WHERE last_modified > FROM_UNIXTIME(?) OR
               (last_modified = FROM_UNIXTIME(?) AND urlid > ?)
      }, undef, $last_lastmod, $last_lastmod, $last_urlid);

    $this->{sth} = DB->prepare(q{
        SELECT urlid,
               url,
               UNIX_TIMESTAMP(last_modified) AS last_modified
          FROM target_url
         WHERE last_modified > FROM_UNIXTIME(?) OR
               (last_modified = FROM_UNIXTIME(?) AND urlid > ?)
         ORDER BY last_modified ASC, urlid ASC
      });
    
    #my $before = [gettimeofday];
    
    $this->{sth}->execute(
    $last_lastmod, $last_lastmod, $last_urlid);

    #my $duration = tv_interval($before);
    #logdebug sprintf(
    #    'took %f seconds to examine which URLs should be indexed.',
    #    $duration);

    $_[KERNEL]->delay('index', INTERVAL);
}

sub _stop_handler {
    
}

sub _index_handler {
    my $this = $_[HEAP];
    my $continue = 1;

    my @rows;
    foreach (1 .. BULK) {
        if (my $row = $this->{sth}->fetchrow_hashref) {
            # もしこの URL にバージョンが一つも無く、且つジョブが豫約濟
            # なら、今囘はそれを無視する。
            my ($n_vers) = DB->selectrow_array(q{
                SELECT COUNT(*)
                  FROM versions
                 WHERE urlid = ?
              }, undef, $row->{urlid});

            if ($n_vers == 0) {
                my ($reserved) = DB->selectrow_array(q{
                    SELECT COUNT(*)
                      FROM job_queue
                     WHERE urlid = ? AND status != 'done'
                  }, undef, $row->{urlid});

                $reserved and next;
            }
        
            push @rows, { %$row };
        }
        else {
            push @rows, undef;
            last;
        }
    }

    grep {defined} @rows
      or return;

    EST->write(
        sub {
            my $E = shift;
        
            foreach my $row (@rows) {
                unless (defined $row) {
            
                    if ($this->{sth}->rows) {
                        logdebug sprintf(
                            'indexed %d URLs and we have no more to index.',
                            $this->{sth}->rows);
                    }
            
                    $continue = undef;
                    last;
                }

                ### require: defined $row->{urlid}
                ### require: defined $row->{url}
                ### require: defined $row->{last_modified}

                my $act;
                if (my $doc = EST->draft($row->{urlid})) {
                    $E->put_doc($doc, 0);

                    $act = 'indexed';
                }
                else {
                    my $deleted = EST->delete($row->{urlid});
            
                    $act = ($deleted ? 'deleted' : 'skipped');
                }
        
                $this->{done}++;
                logdebug sprintf(
                    '%s [%d/%d] (%d %%) [%s]',
                    $act,
                    $this->{done}, $this->{total},
                    $this->{done} / $this->{total} * 100, $row->{url});

                _state last_lastmod => $row->{last_modified};
                _state last_urlid => $row->{urlid};
            }
        });

    $continue and
      $_[KERNEL]->delay('index', INTERVAL);
}

sub _shutdown_handler {
    logdebug 'is going to shutdown...';

    $_[KERNEL]->delay('index');
}

1;
