# -*- cperl -*-
# -----------------------------------------------------------------------------
package Celaenito::JobQueueManager;
use strict;
use warnings;
use POE::Session;
use POE::Component::Client::UserAgent;
use Celaenito::DB;
use Celaenito::Logger;
use Celaenito::FetchAgent;
use Celaenito::LogPurger;
use UNIVERSAL;
use Carp;
use Hash::Util qw(lock_keys);
use SQLTime;
use Smart::Comments;
use constant CHECK_INTERVAL => 15; # 何秒毎にキューをチェックするか
use constant CONCURRENCY => 100; # 同時に幾つのジョブを實行するか
use constant CONC_PER_HOST => 2; # 一ホスト會たりの同時接續數
use constant ALIAS => 'jobqueue'; # このセッションの名前

sub run {
    my $class = shift;
    my $this = bless {} => $class;
    
    $this->{agents} = {}; # jobid => POE::Session<FetchAgent>
    
    lock_keys %$this;
    
    POE::Session->create(
	heap => $this,
	inline_states => {
	    _start     => \&_start_handler,
	    _stop      => \&_stop_handler,
	    _child     => \&_child_handler,
	    shutdown   => \&_shutdown_handler,
	    enqueue    => \&_enqueue_handler,
	    check      => \&_check_handler,
	    stop_agent => \&_stop_agent_handler,
	    count_agents => sub {
		scalar keys %{$this->{agents}};
	    },
	    get_agents => sub {
		map {
		    $_->get_heap;
		} values %{$this->{agents}};
	    },
	    _signal_stop => sub {
		$_[KERNEL]->sig_handled;
		$_[KERNEL]->yield('shutdown');
	    },
	},
    );

    # PoCo::Client::UserAgentも起動する
    POE::Component::Client::UserAgent->new(
	alias   => 'useragent',
	timeout => 10 * 60, # 10分
	agent   => 'Mozilla/1.4 (compatible; I am "Celaenito" the web archiver, sorry for my rudeness.)',
	# FIXME: max_size, etc
       );

    # LogPurgerも起動する
    Celaenito::LogPurger->run;
}

sub _start_handler {
    $_[KERNEL]->alias_set(ALIAS);
    $_[KERNEL]->delay('check', CHECK_INTERVAL);
    foreach (qw[HUP INT QUIT ABRT TERM]) {
	$_[KERNEL]->sig($_ => '_signal_stop');
    }

    logdebug sprintf(
	"started running...\n".
	"alias                   : %s\n".
	"checking interval       : %d sec\n".
	"concurrency             : %d jobs\n".
	"max connections per host: %d sockets\n",
	ALIAS, CHECK_INTERVAL, CONCURRENCY, CONC_PER_HOST);

    # この時点でstatusがrunであるjob_queue内の項目は、
    # 前回の起動時に正常終了しなかつた事を表す。
    # 警告を出して再實行(statusをwaitに)。
    my $sth = DB->prepare(q{SELECT * FROM job_queue WHERE status = "run"});
    $sth->execute;
    while (my $row = $sth->fetchrow_hashref) {
	logwarn sprintf(
	    "The job (jobid: %s) has status `run'. ".
	      "It means a FetchAgent for it has spawned, and suddenly died. ".
	      "Setting its status to `wait' so that it will be restarted...",
	    $row->{jobid});
	
	DB->do(
	    q{UPDATE job_queue SET status = "wait" WHERE jobid = ?}, undef,
	    $row->{jobid});
    }
}

sub _shutdown_handler {
    my $this = $_[HEAP];
    logdebug "Shutdown time has arrived. Telling agents to stop...";

    $_[KERNEL]->post(useragent => 'shutdown');
    $_[KERNEL]->post(logpurger => 'shutdown');
    
    $_[KERNEL]->alias_remove(ALIAS);
    $_[KERNEL]->delay('check');
    
    foreach (values %{$this->{agents}}) {
	$_[KERNEL]->post($_ => 'shutdown');
    }
}

sub _stop_handler {
    logdebug "STOPPED.";
}

sub _child_handler {
    my ($this, $event, $child_session) = @_[HEAP, ARG0, ARG1];
    my $child = $child_session->get_heap;
    
    if ($event eq 'lose') {
	if (UNIVERSAL::isa($child => 'Celaenito::FetchAgent')) {
	    logdebug sprintf(
		"FetchAgent (jobid: %s) stopped.",
		$child->jobid);
	    delete $this->{agents}{$child->jobid};
	}
    }
}

sub _check_handler {
    my $this = $_[HEAP];
    # FIXME: priorityを見てゐない。
	
    my $sth = DB->prepare(q{
SELECT * FROM job_queue
  LEFT JOIN target_url USING (urlid)
 WHERE status = "wait" AND begin_time <= NOW()
 ORDER BY begin_time ASC
});
    $sth->execute;
    while (my $row = $sth->fetchrow_hashref) {
	$this->{agents}{$row->{jobid}} and next;

	# concurrencyチェック。
	if (keys %{$this->{agents}} >= CONCURRENCY) {
	    #logdebug sprintf(
	#	"%d jobs are now running concurrently. I can't spawn more.",
	#	scalar(keys %{$this->{agents}}));
	    last;
	}

	# 一ホスト會たりの接續數制限
	my $host = URI->new($row->{url})->host;
	my $conc_per_host = grep {
	    $_->get_heap->host eq $host;
	} values %{$this->{agents}};
	
	if ($conc_per_host >= CONC_PER_HOST) {
 	    #logdebug sprintf(
	#	"%d jobs are now running concurrently against [$host]. I can't spawn more.",
	#	$conc_per_host);
	    last;
	}
	
	logdebug sprintf(
	    "It is the time to spawn FetchAgent for the urlid %d;\n".
	      "URL: %s\n".
		"Reserved Time: %s\n",
	    $row->{urlid}, $row->{url}, $row->{begin_time});
	
	$this->spawn($row->{jobid});
    }
    
    $_[KERNEL]->delay('check', CHECK_INTERVAL);
}

sub spawn {
    my ($this, $jobid) = @_;

    defined($jobid) or croak "jobid is undef";

    # invoked_by が NULL であれば、by と at を初期値に設定。
    my ($urlid, $by) = DB->selectrow_array(q{
        SELECT target_url.urlid, invoked_by
          FROM target_url
          LEFT JOIN job_queue USING (urlid)
         WHERE jobid = ?
    }, undef, $jobid);

    unless (defined $by) {
	DB->do(q{
            UPDATE target_url
               SET invoked_by = urlid,
                   invoked_at = NOW(),
                   last_modified = last_modified
             WHERE urlid = ?
        }, undef, $urlid);
    }

    my $agent_session = Celaenito::FetchAgent->run($jobid);
    $this->{agents}{$jobid} = $agent_session;
}

sub _enqueue_handler {
    # options: {
    #     invoked_by => urlid,  # この enqueue は他の URL から再歸的に行はれたものである。
    # }
    my ($urlid, $begin_time, $options) = @_[ARG0, ARG1, ARG2];

    # URLの存在チェック
    my $target = DB->selectrow_hashref(
	q{SELECT * FROM target_url WHERE urlid = ?}, undef, $urlid);
    if (!$target) {
	logerror sprintf(
	    "Failed to enqueue url (id: %s)\n".
	      "It is not in the target_url table, this must be a bug.\n".
		"Sender: %s",
	    $urlid, $_[SENDER]);
	return;
    }

    # invoked_by チェック
    if ($options and defined $options->{invoked_by}) {
	# リンク元の invoked_by と invoked_at をコピー
	my ($old_by, $old_at) = DB->selectrow_array(q{
            SELECT invoked_by,
                   UNIX_TIMESTAMP(invoked_at)
              FROM target_url
             WHERE urlid = ?
        }, undef, $options->{invoked_by});

	my ($by, $at) = DB->selectrow_array(q{
            SELECT invoked_by,
                   UNIX_TIMESTAMP(invoked_at)
              FROM target_url
             WHERE urlid = ?
        }, undef, $urlid);

	if (($by || '') eq ($old_by || '') and
	      ($at || 0) == ($old_at || 0)) {
	    # 等しい => リンクがループしてゐる
	    #logdebug sprintf(
	#	"Skipped to enqueue [%s] due to a duplicate recursion.",
	#	$target->{url});
	    return;
	}
	else {
	    # 等しくないので by と at を上書きして續行
	    DB->do(q{
		UPDATE target_url
                   SET invoked_by = ?,
                       invoked_at = FROM_UNIXTIME(?),
                       last_modified = last_modified
                 WHERE urlid = ?
            }, undef, $old_by, $old_at, $urlid);
	}
    }
    else {
	# by と at を NULL にする。
	DB->do(q{
            UPDATE target_url
               SET invoked_by = NULL,
                   invoked_at = NULL,
                   last_modified = last_modified
             WHERE urlid = ?
        }, undef, $urlid);
    }

    # invoked_by チェック(二つ目)
    if ($options and defined $options->{invoked_by}) {
	my $limit = 3600; # 30分
	my ($recent) = DB->selectrow_array(q{
            SELECT COUNT(*)
              FROM job_queue
             WHERE urlid = ? AND
                   UNIX_TIMESTAMP(begin_time) >= ? AND
                   status != "wait"
        }, undef, $urlid, time - $limit);
	if ($recent) {
	    # 最近この URL は fetch されたばかり。
	    #logdebug sprintf(
	#	"Skipped to enqueue [%s] due to the recent fetch.",
	#	$target->{url});
	    return;
	}
    }

    logdebug sprintf(
	"enqueue; urlid: %s\n".
	"begin_time: %s\n".
	"url       : %s",
	$urlid,
	scalar localtime($begin_time),
	$target->{url});

    # この同じurlidを持つジョブがキュー内に存在する時、
    # (1) status=waitなら、その實行時刻と$begin_timeのうち早い方に設定する
    # (2) status=runなら、このenqueueを無視する (取得に失敗したら自動でリトライする)
    # (3) status=doneの行は最初から無視。
    my $oldjob = DB->selectrow_hashref(q{
SELECT *
  FROM job_queue
 WHERE urlid = ? AND (status = "wait" OR status = "run")
}, undef, $urlid);
    if ($oldjob) {
	if ($oldjob->{status} eq 'wait') {
	    my ($changed, $earlier);
	    if (sqlparsetime($oldjob->{begin_time}) > $begin_time) {
		# begin_time書換へ
		DB->do(
		    q{UPDATE job_queue SET begin_time = ? WHERE jobid = ?},
		    undef, sqlmaketime($begin_time), $oldjob->{jobid});
		$changed = 1;
		$earlier = $begin_time;
	    }
	    else {
		$earlier = sqlparsetime($oldjob->{begin_time});
	    }
	    
	    #logdebug sprintf(
	#	"The job (urlid: %s) is already queued and waiting to be spawned. ".
	#	   ($changed ?
	#	      "begin-time is still %s." : "begin-time hastens to %s."),
	#	$urlid,
	#	scalar localtime($earlier));
	}
	elsif ($oldjob->{status} eq 'run') {
	    logdebug sprintf(
		"The job (urlid: %s) is currently running. ignore...",
		$urlid);
	}
    }
    else {
	DB->do(q{
INSERT INTO job_queue
       (urlid, status, begin_time)
VALUES (  ?  , "wait",     ?     )
}, undef, $urlid, sqlmaketime($begin_time));
    }
}

sub _stop_agent_handler {
    my ($this, $urlid) = @_[HEAP, ARG0];

    my $found;
    foreach my $session (values %{$this->{agents}}) {
	my $agent = $session->get_heap;
	
	if ($agent->urlid == $urlid) {
	    logdebug sprintf
	      'stop_agent: found an agent (jobid %d) whose urlid is %d. telling to abort...',
		$agent->jobid, $urlid;
	    
	    $_[KERNEL]->post($session => 'shutdown');
	    $found = 1;
	}
    }

    if (!$found) {
	logdebug sprintf
	  'stop_agent: no jobs whose urlid is %d are running.',
	    $urlid;
    }
}

1;
