# -*- cperl -*-
# -----------------------------------------------------------------------------
package Celaenito::FetchAgent;
use strict;
use warnings;
use POE::Session;
use POE::Kernel;
use Celaenito::DB;
use Celaenito::Logger;
use Celaenito::FileStorage;
use Hash::Util qw(lock_keys);
use HTTP::Request;
use HTTP::Status;
use HTTPAntiAbuse;
use LinkExtractor;
use Data::Dumper;
use Time::HiRes;
use SQLTime;
use Kibibyte;
use Smart::Comments;

# 接續失敗時に於るリトライ囘數。
# 最大リトライ囘數に達すると、そのエントリは削除される。
our %RETRY_COUNT; # canonical url => num

# 最大リトライ囘數。
our $RETRY_COUNT_LIMIT = 5;

# リトライ間隔の最大價。單位は分。
our $RETRY_MAX_INTERVAL = 60;

# 再歸取得を許可するプロトコル
our %ALLOWED_PROTO = map {$_ => 1} qw(http https ftp);

sub is_allowed_proto ($) {
    my $proto = shift;
    
    defined $proto and
      $ALLOWED_PROTO{lc $proto};
}

sub min (@) {
    @_ or return undef;
    my $n = $_[0];
    foreach (@_) {
	$n > $_ and $n = $_;
    }
    $n;
}

sub max (@) {
    @_ or return undef;
    my $n = $_[0];
    foreach (@_) {
	$n < $_ and $n = $_;
    }
    $n;
}

sub ordinal ($) {
    my $num = shift;
    if ($num == 1) {
	"1st";
    }
    elsif ($num == 2) {
	"2nd";
    }
    elsif ($num == 3) {
	"3rd";
    }
    else {
	"${num}th";
    }
}

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

    $this->{jobid} = $jobid;
    $this->{urlid} = undef;
    $this->{url} = undef;
    $this->{options} = {};
    $this->{dl_starttime} = undef;
    $this->{size} = undef;
    $this->{progress} = undef;
    $this->{data} = '';
    $this->{shutdown_flag} = undef;

    lock_keys %$this;

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

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

    my ($rows) = DB->do(
	q{UPDATE job_queue SET status = "run", begin_time = NOW() WHERE jobid = ?},
	undef, $this->{jobid});
    
    if ($rows == 0) {
	logwarn sprintf(
	    "The job (jobid: %s) is being duplicated. exit...",
	    $this->{jobid});
	return;
    }
    else {
	logdebug sprintf(
	    "Job (jobid: %s) started running...",
	    $this->{jobid});
    }

    $this->_begin;
}

sub _begin {
    my $this = shift;
    
    # UserAgentにURLを送る。古いファイルがあればIf-Modified-Since付き。
    my ($url, $options_dumped, $urlid) = DB->selectrow_array(q{
SELECT url, options, target_url.urlid
  FROM target_url
  LEFT JOIN job_queue USING (urlid)
 WHERE jobid = ?
}, undef, $this->{jobid});

    unless (defined $urlid) {
	logerror sprintf(
	    "The job (jobid: %s) is an orphan. exit...",
	    $this->{jobid});
	return;
    }

    my $last_modified = do {
	my $ver = FS->get_latest_version($urlid);
	if (defined $ver) {
	    my $info = FS->get_info($urlid, $ver);
	    sqlparsetime $info->{timestamp};
	}
	else {
	    undef;
	}
    };
    
    $this->{url} = $url;
    $this->{urlid} = $urlid;
    $options_dumped and
      %{$this->{options}} = %{eval $options_dumped};

    my $req = HTTP::Request->new(
	GET => $url);

    if ($last_modified) {
	# If-Modified-Sinceを付ける。
	$req->headers->if_modified_since($last_modified);
    }

    if (my $offset = length $this->{data}) {
	# 途中まで落としたデータが在る。
	$req->header(Range => sprintf('bytes=%d-', $offset));
    }
    
    $poe_kernel->post(
	useragent => 'request',
	request  => $req,
	response => $poe_kernel->get_active_session->postback(_downloaded => $this),
	callback => sub {
	    my ($chunk, $res, $proto) = @_;

	    if ($this->{shutdown_flag}) {
		logdebug sprintf
		  'is terminating transfer due to the shutdown request [%s]',
		    $res->request->uri->as_string;
		die "flagged shutdown";
	    }
	    
	    $res->content($res->content . $chunk);
	    my $length_current = length($res->content);
	    my $length_said = $res->header('Content-Length');

	    $this->{size} = $length_said;
	    
	    if ($length_said) {
		$this->{progress} = $length_current / $length_said;
		#logdebug sprintf
		#  'got % 3d%% (%d/%d) [%s]',
		#    $this->{progress} * 100, $length_current, $length_said, $res->request->uri->as_string;
	    }
	    else {
		#logdebug sprintf
		#  'got %d bytes (length unknown) [%s]',
		#    $length_current, $res->request->uri->as_string;
	    }

	    return undef; # keep downloading
	},
	chunksize => 100 * 1024,
       );
    
    unless ($this->{dl_starttime}) {
	$this->{dl_starttime} = Time::HiRes::time;
    }
}

sub _is_allowed_for_recurse {
    my $tag = lc shift;

    if ($tag eq 'a' or $tag eq 'frame' or $tag eq 'iframe') {
	return 1;
    }
    else {
	return undef;
    }
}

sub _downloaded_postback {
    my ($this) = @{$_[ARG0]};
    my ($req, $res, $entry) = @{$_[ARG1]};

    if ($this->{shutdown_flag}) {
	return;
    }

    # 今囘がリダイレクトであれば、何もせず終了するだけで
    # LWPが勝手にリクエストを續けてくれる。
    if ($res->is_redirect and $res->code ne RC_NOT_MODIFIED) {
	lognotice sprintf
	  "is forwarded.\nURL: %s\nStatus: %s",
	    $req->uri->as_string,
	      $res->status_line;
	return;
    }

    # 落としたサイズの方が Content-Length より小さければ転送が途中で切
    # れてゐるのでリトライ。
    if ($res->is_success) {
	if ($res->code == 200) {
	    $this->{data} = $res->content;
	}
	elsif ($res->code == 206) {
	    $this->{data} .= $res->content;
	}

	my $len_said = $res->header('Content-Length');
	my $len_read = length $res->content;
	if ($len_said and $len_read < $len_said) {
	    lognotice sprintf
	      "Stream for [%s] got broken after reading %s bytes. ".
		"Retrying to read the remaining %s bytes...",
		  $req->uri->as_string, $len_read, $len_said - $len_read;

	    $this->_begin;
	    return;
	}
	else {
	    # 殘りの部分では $res->content を使ふので、ここで
	    # $res->content に $this->{data} を設定。
	    $res->content($this->{data});
	}
    }

    # HTTPプロトコル違反に對處
    unabuse $res;
    
    # リダイレクトが起こつた時、$reqはリダイレクト後のURIになつてゐる。
    my $req_uri = $req->uri->canonical;

    lognotice sprintf
      "ended.\nURL: %s\nStatus: %s",
	$req->uri->as_string,
	  $res->status_line;

    # job_logにREPLACEで入れる。
    my $log_message = do {
	if ($res->is_success) {
	    my $elapsed = Time::HiRes::time - $this->{dl_starttime};
	    my $size = length($res->content);
	    
	    sprintf
	      '%s, got %s in %.1f sec (%s)',
		$res->status_line,
		Kibibyte::format($size => '%.2f %s'),
		$elapsed,
		$elapsed == 0 ? '-' : Kibibyte::format($size / $elapsed => '%.2f %s/s');
	}
	else {
	    $res->status_line;
	}
    };
    
    DB->do(q{
REPLACE INTO job_log
       (jobid, urlid, message, url)
VALUES (  ?  ,   ?  ,    ?   ,  ? )
}, undef, $this->{jobid}, $this->{urlid}, $log_message, $this->{url});

    my $enqueued;
    my $enqueue_after;
    if ($res->is_success) {
	# リトライ履歴があれば、それを削除。
	if (exists $RETRY_COUNT{$req_uri->as_string}) {
	    logdebug sprintf(
		"[%s] has been retried for %d times. now this url is retrievable.",
		$req_uri->as_string,
		$RETRY_COUNT{$req_uri->as_string} || 0);

	    delete $RETRY_COUNT{$req_uri->as_string};
	}
	
	# このファイルをDBに保存。
	FS->store(
	    urlid    => $this->{urlid},
	    response => $res,
	   );

	if ($this->{options}{overwrite}) {
	    FS->delete_old_versions($this->{urlid});
	}

	# auto_updateが有效なら、これを再びenqueue
	if (my $interval = $this->{options}{auto_update_interval}) {
	    $enqueue_after = sub {
		$poe_kernel->post(
		    jobqueue => enqueue => $this->{urlid}, time + $interval * 60); # intervalの單位は分
		logdebug "re-enqueued [$this->{url}] after $interval minutes";
	    };
	}
    }
    elsif ($res->code == 500 and defined($res->header('Client-Warning'))) {
	# 接續失敗等の場合。
	# 自動更新間隔が指定されてゐれば、その値と60分のうち
	# 短い方を間隔としてenqueueする。無ければ60分でenqueueする。
	#
	# 但し同じURLを5回以上リトライしようとした場合は、
	# 一度そのリトライをスキップする事により、リトライを終了する。
	# (enqueueさへしなければリトライは止まるので。)
	my $url = $req_uri->as_string;
	my $count = $RETRY_COUNT{$url};
	if ($count and $count >= $RETRY_COUNT_LIMIT) {
	    logdebug "[$url] has been retried too many times ($count). stop...";
	    delete $RETRY_COUNT{$url};
	}
	else {
	    $RETRY_COUNT{$url} = ($count || 0) + 1;
	    my $interval = do {
		if ($_ = $this->{options}{auto_update_interval}) {
		    min $_, $RETRY_MAX_INTERVAL;
		}
		else {
		    $RETRY_MAX_INTERVAL;
		}
	    };
	    $enqueue_after = sub {
		$poe_kernel->post(
		    jobqueue => enqueue => $this->{urlid}, time + $interval * 60);
		logdebug sprintf(
		    're-enqueued [%s] after %d minutes due to transporting failure. '.
		      'this is %s retry; remaining %d times.',
		    $this->{url},
		    $interval,
		    ordinal($RETRY_COUNT{$url}),
		    $RETRY_COUNT_LIMIT - $RETRY_COUNT{$url});
	    };
	}
    }
    else {
	if ($this->{options}{auto_update_interval}) {
	    if ($res->code eq '304' or
		  $this->{options}{keep_update_on_disappear}) {
		# 失敗はしたが再びキューに入れる。
		# 304 Not Modifiedの場合も同樣。

		if (my $interval = $this->{options}{auto_update_interval}) {
		    $enqueue_after = sub {
			$poe_kernel->post(
			    jobqueue => enqueue => $this->{urlid}, time + $interval * 60); # intervalの單位は分
			logdebug "re-enqueued [$this->{url}] after $interval minutes.";
		    };
		}
	    }
	}
    }

    # リンク先のenqueue處理。長い…
    my $links = do {
	my $ver = FS->get_latest_version($this->{urlid});
	if (defined $ver) {
	    FS->get_links($this->{urlid}, $ver);
	}
	else {
	    [];
	}
    };
    foreach my $link (@$links) {
	# recursive_fetchがoffであつても、a以外のタグでリンクされた
	# 他のファイルは必ずキューに入れる。但し、後に述べる例外規
	# 則により、a 以外のタグによるリンクへは再歸取得フラグが傳
	# 播しない。
	my $baseurl = defined $link->[3] ? $link->[3] : $req_uri;
	my $linkuri = URI->new_abs($link->[2], $baseurl)->canonical;

	# リンク先のfragmentは削除。
	$linkuri->fragment(undef);

	# aタグのリンクであり、strip_query_stringが指定されてゐるなら
	# リンク先のquery stringも削除。
	if ($link->[1] and _is_allowed_for_recurse($link->[1]) and
	      $this->{options}{strip_query_string}) {

	    if (defined $linkuri->query) {
		lognotice sprintf '[%s] has a link to [%s]; its query string will be stripped due to configuration.',
		  $req_uri->as_string, $linkuri->as_string;
		    
		$linkuri->query(undef);
	    }
	}
	    
	# リンク先のプロトコルをチェック
	unless (is_allowed_proto($linkuri->scheme)) {
	    #lognotice sprintf '[%s] has a link to [%s], but this scheme is not allowed. skip...',
	    #  $req_uri->as_string, $linkuri->as_string;
	    next;
	}

	# リンク先が自分自身であればスキップ
	if ($req_uri->as_string eq $linkuri->as_string) {
	    lognotice sprintf '[%s] has a link to [%s], but this is itself. skip...',
	      $req_uri->as_string, $linkuri->as_string;
	    next;
	}
	    
	if ($link->[1] and _is_allowed_for_recurse($link->[1])) {
	    # 再歸取得を行ふ場合のみenqueue。
	    if (!$this->{options}{recursive_fetch}) {
		next;
	    }
		
	    if ($this->{options}{no_parent} and
		  $req_uri->host_port eq $linkuri->host_port) {
		# no_parentチェック
		# リンク先とリンク元のホスト:ポートが同一である時、
		# リンク元が /foo/bar/baz.html ならばリンク先は /foo/bar/ から始まらなければならない。
		$req_uri->path =~ m|^(.*?/)[^/]*$|;
		my $prefix = $1;
		    
		if ($linkuri->path !~ m/^\Q$prefix\E/) {
		    lognotice sprintf 'is skipping link from [%s] to [%s] due to no_parent flag being true.',
		      $req_uri->as_string, $linkuri->as_string;
		    next;
		}
	    }
		
	    if (!$this->{options}{span_hosts} and
		  $req_uri->host ne $linkuri->host) {
		# span_hostsチェック失敗
		lognotice sprintf 'is skipping link from [%s] to [%s] due to span_hosts flag being false.',
		  $req_uri->as_string, $linkuri->as_string;
		next;
	    }
	}

	# ここでは賢い動作が必要である。
	# (1) 絶對パスに變換されたリンクがtarget_urlに存在しない場合、
	#     元のtarget_urlから以下の値を繼承した別のエントリを插入。
	#   a. priority
	#   b. overwrite
	#   c. options.auto_update_interval (但しinherit_auto_updateが眞の場合のみ)
	#   d. options.keep_update_on_disappear
	#   e. options.inherit_auto_update
	#   f. options.recursive_fetch
	#      (但しrecursion_levelが1以上の場合はrecursion_level - 1が1以上の場合のみ)
	#   g. options.recursion_level (但し元の値から1を引く)
	#   h. options.no_parent
	#   i. options.span_hosts
	#   j. options.strip_query_string
	# (2) 既に存在してゐる場合、元のtarget_urlの以下の値を置き換へる。
	#   a. priority: 上書き
	#   b. overwrite: しない方を優先
	#   c. options.keep_update_interval: 上書き
	#   d. options.keep_update_on_disappear: 上書き
	#   e. options.inherit_auto_update: 上書き
	#   f. options.recursive_fetch: 上書き
	#   g. options.recursion_level: 今囘のrecursion_level-1
	#   h. options.no_parent: 上書き
	#   i. options.span_hosts: 上書き
	#   j. options.strip_query_string: 上書き
	# (3) 但し例外として、recursive_fetchフラグはa, frame, iframe等のタグで
	#     リンクされたURLにしか傳播されない。auto_update_intervalも同樣である。
	my $self_ent = DB->selectrow_hashref(
	    q{SELECT * FROM target_url WHERE urlid = ?}, undef, $this->{urlid});
	my $old_ent = DB->selectrow_hashref(
	    q{SELECT * FROM target_url WHERE url = ?}, undef, $linkuri->as_string);
	if (!$old_ent) {
	    # optionsを作る
	    my $new_options = {};
	    if ($this->{options}{overwrite}) {
		$new_options->{overwrite} = 1;
	    }
	    if ($this->{options}{inherit_auto_update} and
		  _is_allowed_for_recurse($link->[1])) {
		    
		$new_options->{inherit_auto_update} = 1;
		    
		if ($_ = $this->{options}{auto_update_interval}) {
		    $new_options->{auto_update_interval} = $_;
			
		    if ($this->{options}{keep_update_on_disappear}) {
			$new_options->{keep_update_on_disappear} = 1;
		    }
		}
	    }
	    if ($this->{options}{recursive_fetch} and
		  _is_allowed_for_recurse($link->[1])) {
		    
		my $level = $this->{options}{recursion_level} || 0;
		if ($level > 0) {
		    if ($level == 1) {
			# ここで再歸取得終了。HTLが尽きた爲。
		    }
		    else {
			$new_options->{recursive_fetch} = 1;
			$new_options->{recursion_level} = $level - 1;
		    }
		}
		else {
		    $new_options->{recursive_fetch} = 1;
		}
		    
		if ($this->{options}{no_parent}) {
		    $new_options->{no_parent} = 1;
		}
		    
		if ($this->{options}{span_hosts}) {
		    $new_options->{span_hosts} = 1;
		}

		if ($this->{options}{strip_query_string}) {
		    $new_options->{strip_query_string} = 1;
		}
	    }
	    my $options_dumped;
	    if (%$new_options) {
		$options_dumped =
		  Data::Dumper->new([$new_options])
		      ->Indent(0)
			->Purity(1)
			  ->Useqq(1)
			    ->Terse(1)
			      ->Dump;
	    }
		
	    # target_urlにINSERT
	    DB->do(q{
INSERT INTO target_url
       (url, priority, label, added_date, options)
VALUES ( ? ,    ?    , NULL ,    NOW()  ,    ?   )
},
		   undef,
		   $linkuri->as_string,
		   $self_ent->{priority},
		   $options_dumped,
		  );
	    my ($urlid) = DB->selectrow_array(q{SELECT LAST_INSERT_ID()});
		
	    #lognotice "Added into the target_url recursively:\n".
	    #  "URL ID  : ".$urlid."\n".
	    #  "URL     : ".$linkuri->as_string."\n".
	    #  "Priority: ".$self_ent->{priority}."\n".
	    #  "Options : ".($options_dumped || '');
		
	    # INSERTに成功したら、これをjob_queueに入れる。
	    $poe_kernel->post(
		'jobqueue', enqueue => $urlid, time, {invoked_by => $this->{urlid}});
	    $enqueued = 1;
	}
	else {
	    # priorityを撰擇
	    my $priority = $self_ent->{priority};
		
	    # 古いoptionsを解凍
	    my $old_options = {};
	    if ($_ = $old_ent->{options}) {
		%$old_options = %{ eval $_ };
	    }
		    
	    # optionsを作る
	    my $new_options = {};
	    if ($old_options->{overwrite} and $this->{options}{overwrite}) {
		$new_options->{overwrite} = 1;
	    }
	    if ($this->{options}{inherit_auto_update} and
		  _is_allowed_for_recurse($link->[1])) {
		    
		$new_options->{inherit_auto_update} = 1;

		if ($_ = $this->{options}{auto_update_interval}) {
		    $new_options->{auto_update_interval} = $_;

		    if ($this->{options}{keep_update_on_disappear}) {
			$new_options->{keep_update_on_disappear} = 1;
		    }
		}
	    }
	    if ($this->{options}{recursive_fetch} and
		  _is_allowed_for_recurse($link->[1])) {
		    
		my $level;
		my $enable;
		if ($level = $this->{options}{recursion_level}) {
		    # 少なくとも1以上なので、これが0になれば消滅する。
		    $level--;
		    if ($level >= 1) {
			# まだ殘つてゐる。
			$enable = 1;
		    }
		}
		else {
			# 無制限である。
		    $enable = 1;
		}
		    
		if ($enable) {
		    $new_options->{recursive_fetch} = 1;
		    $level and
		      $new_options->{recursion_level} = $level;
			
		    $this->{options}{no_parent} and
		      $new_options->{no_parent} = $this->{options}{no_parent};
		    $this->{options}{span_hosts} and
		      $new_options->{span_hosts} = $this->{options}{span_hosts};
		    $this->{options}{strip_query_string} and
		      $new_options->{strip_query_string} = $this->{options}{strip_query_string};
		}
	    }

	    # 新舊オプションまたは優先度が異つてゐる場合のみ、
	    # これをシリアライズしてtarget_urlをUPDATEする。
	    my $update_options = do {
		if ($old_ent->{priority} ne $priority) {
		    1;
		}
		else {
		    if (keys %$old_options != keys %$new_options) {
			1;
		    }
		    else {
			my $differ;
			while (my ($key, $oldval) = each %$old_options) {
			    if ($oldval ne $new_options->{$key}) {
				$differ = 1;
				last;
			    }
			}
			$differ;
		    }
		}
	    };

	    if ($update_options) {
		my $options_dumped;
		if (%$new_options) {
		    $options_dumped =
		      Data::Dumper->new([$new_options])
			  ->Indent(0)
			    ->Purity(1)
			      ->Useqq(1)
				->Terse(1)
				  ->Dump;
		}
		
		# target_urlをUPDATE
		DB->do(q{
UPDATE target_url
   SET priority = ?,
       options  = ?,
       last_modified = last_modified
 WHERE urlid = ?
}, undef, $priority, $options_dumped, $old_ent->{urlid});

		lognotice "Modified the target_url recursively:\n".
		  "URL ID  : ".$old_ent->{urlid}."\n".
		    "URL     : ".$old_ent->{url}."\n".
		      "Priority: ".$priority."\n".
			"Options : ".($options_dumped || '');
	    }
	    
	    # これをjob_queueに入れる。
	    $poe_kernel->post(
		'jobqueue', enqueue => $old_ent->{urlid}, time, {invoked_by => $this->{urlid}});
	    $enqueued = 1;
	}
    }

    if ($enqueue_after) {
	$enqueue_after->();
	$enqueued = 1;
    }
    
    if ($enqueued) {
	# JobQueueManagerにcheckをpostし、直ぐにキューのチェックを開始させる。
	#$poe_kernel->post(jobqueue => 'check');
    }

    # job_queue.statusをdoneにする。
    DB->do(q{UPDATE job_queue SET status = "done" WHERE jobid = ?}, undef,
	   $this->{jobid});
}

sub _stop_handler {

}

sub _shutdown_handler {
    logdebug 'will shutdown as soon as possible...';
    shift->{shutdown_flag} = 1;
}

sub jobid {
    shift->{jobid};
}

sub urlid {
    shift->{urlid};
}

sub size {
    # may be undef
    shift->{size};
}

sub host {
    my $this = shift;

    $this->{url} ? URI->new($this->{url})->host : undef;
}

sub progress {
    # 0 <= x <= 1 || undef
    shift->{progress}
}

1;
