# -*- cperl -*-
# -----------------------------------------------------------------------------
# $Id: FileStorage.pm,v 1.3 2004/10/12 10:26:43 admin Exp $
# -----------------------------------------------------------------------------
# 既知の問題:
#   bigintで収まり切らないやうな大量のバージョンを扱へない。
#   時系列順に番號を付けてゐる今の方法では、古いバージョン
#   を消しても解決しない。
# -----------------------------------------------------------------------------
package Celaenito::FileStorage;
use strict;
use warnings;
use Celaenito::DB;
use Celaenito::Logger;
use LinkExtractor;
use Data::Dumper;
use Carp;
use Digest::MD5 qw(md5_hex);
use Hash::Util qw(lock_keys);
use SQLTime;
use Smart::Comments;
require Exporter;
our @ISA = qw(Exporter);
our @EXPORT = qw(FS);
our $FS;

sub FS () {
    if (!$FS) {
	$FS = bless {} => __PACKAGE__;

	lock_keys %$FS;
    }
    $FS;
}

sub store {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;

    # 後で失敗すると困るので先に分割サイズを求めて置く。
    my $splitsize = Celaenito::DB->sizelimit - 1024; # 1Kibの余裕を持たせる
    if ($splitsize <= 0) {
	die "Illegal splitsize: $splitsize";
    }
    
    my %args = @_;
    my $urlid = $args{urlid};
    my $response = $args{response};

    # Content-Typeヘッダは二つある可能性がある。
    # HTTPプロトコル上で與へられた實際のヘッダと、
    # HTMLの<head>タグ内で與へられたヘッダが共に在る場合である。
    # 最後の項目を撰擇する。
    # (LWP::UserAgentのparse_headを切る、と云ふ方法もあるが…)
    my $content_type = do {
	my @array = $response->header('Content-Type');
	$array[-1];
    };

    # MD5を計算
    my $md5 = md5_hex($response->content);

    # このurlidの直前のバージョンが(あれば)確認し、
    # もし全くの同一ファイルであれば何もせず終了
    my $latest_ver = get_latest_version($urlid);
    if (defined $latest_ver) {
	if ($md5 eq get_md5($urlid, $latest_ver)) {
	    my $latest_content = get_content($urlid, $latest_ver);
	    if ($response->content eq $latest_content) {
		logdebug "store: the data is exactly same as its latest version. ".
		  "no need to save this.";

		# content_type だけは更新
		DB->do(q{
                    UPDATE versions
                       SET content_type = ?
                     WHERE urlid = ? AND version = ?
                }, undef, $content_type, $urlid, $latest_ver);
		return;
	    }
	}
    }

    my $new_version = defined $latest_ver ?
      $latest_ver + 1 : 0;

    # target_url の last_modified を更新
    # サーバーからの應答に Last-Modified があればそれを、無ければ現在時刻を使ふ。
    my $timestamp =
      $response->headers->last_modified || time;
    
    DB->do(q{
	UPDATE target_url
	   SET last_modified = NULL
	 WHERE urlid = ?
    }, undef, $urlid);

    my $type;
    if (0) {
	# FIXME: xdelta
	$type = 'xdelta';
	# $response->content(...);
    }
    else {
	$type = 'checkpoint';
    }

    # MD5が一致するファイルを探し、もしあれば完全一致で同一性を確認し、
    # 全くの同一ファイルであればrefcountをインクリメントするだけ。
    foreach my $pair (find_with_md5($md5)) {
	if ($response->content eq get_content(@$pair)) {
	    logdebug "store: the data is exactly same as (refid, version) = ($pair->[0], $pair->[1]). ".
	      "incrementing refcount instead of inserting newly...";

	    my $pieceid = get_piece_id(@$pair);
	    DB->do(q{
		UPDATE pieces_info
		   SET refcount = refcount + 1
		 WHERE pieceid = ?
	    }, undef, $pieceid);

	    DB->do(q{
		INSERT INTO versions
		       (urlid, version, type, timestamp,
			content_type, pieceid)
		VALUES (  ?  ,    ?   ,  ?  ,     ?    ,
			  ?     ,    ?   )
	       }, undef,
		   $urlid, $new_version, $type, sqlmaketime($timestamp),
		   $content_type, $pieceid);
	    return;
	}
    }

    my $size = length($response->content);
    my $n_splits = int($size / $splitsize);
    $size / $splitsize != $n_splits and $n_splits++;

    DB->do(q{
LOCK TABLES versions WRITE,
            pieces_info WRITE,
            pieces_storage WRITE
});

    DB->do(q{
	INSERT INTO pieces_info
	       (splits, data_size, md5sum)
	VALUES (  ?   ,     ?    ,   ?   )
    }, undef, $n_splits, $size, $md5);

    my ($pieceid) = DB->selectrow_array(
	q{SELECT LAST_INSERT_ID()});

    DB->do(q{
	INSERT INTO versions
	       (urlid, version, type, timestamp,
		content_type, pieceid)
	VALUES (  ?  ,    ?   ,  ?  ,     ?    ,
		      ?     ,    ?   )
    }, undef,
	   $urlid, $new_version, $type, sqlmaketime($timestamp),
	   $content_type, $pieceid);
    
    insert_data_with_split(
	$pieceid, $splitsize, $response->content);

    DB->do(q{UNLOCK TABLES});
    undef;
}

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

sub insert_data_with_split {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my ($pieceid, $splitsize, $content) = @_;
    my $len = length $content;

    if ($splitsize <= 0) {
	croak "Splitsize is not positive! Want me to insert infinite rows into pieces_storage!?";
    }

    my $sth = DB->prepare(q{
INSERT INTO pieces_storage
       (pieceid, split, content)
VALUES (   ?   ,   ?  ,    ?   )
});
    for (my $offset = 0, my $i = 0;
	 $offset < $len;
	 $offset += $splitsize, $i++) {
	my $blocksize = min $len - $offset, $splitsize;
	my $block = substr $content, $offset, $blocksize;
	
	$sth->execute($pieceid, $i, $block);
    }
}

sub get_latest_version  {
    # $urlid: URL ID
    # $older_than: エポック(省略可)
    # 該當するバージョン番號があればそれを、無ければundefを返す。
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my ($urlid, $older_than) = @_;

    my @where = "urlid = $urlid";
    if (defined $older_than) {
	push @where, sprintf "UNIX_TIMESTAMP(timestamp) <= ".DB->quote($older_than);
    }
    my $where = 'WHERE '.join(' AND ', @where);

    my ($ver) = DB->selectrow_array(qq{
        SELECT version
          FROM versions
        $where
         ORDER BY version DESC
         LIMIT 1
    });
    $ver;
}

sub versions_list {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my $urlid = shift;

    my $sth = DB->prepare(q{
	SELECT version
	  FROM versions
	 WHERE urlid = ?
	 ORDER BY version ASC
    });
    $sth->execute($urlid);
    
    my @result;
    while (my $row = $sth->fetch) {
	push @result, $row->[0];
    }

    @result;
}

sub get_info  {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my ($urlid, $version) = @_;

    my $info = DB->selectrow_hashref(q{
	SELECT *
	  FROM versions
	  LEFT JOIN pieces_info USING (pieceid)
	 WHERE urlid = ? AND version = ?
    }, undef, $urlid, $version);
    $info;
}

sub get_size  {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;

    my $info = get_info($_[0], $_[1]);
    $info ? $info->{data_size} : undef;
}

sub get_md5  {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;

    my $info = get_info($_[0], $_[1]);
    $info ? $info->{md5sum} : undef;
}

sub get_piece_id  {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;

    my $info = get_info($_[0], $_[1]);
    $info ? $info->{pieceid} : undef;
}

sub get_content  {
    # [scalar context]
    #   戻り値: ファイルの中身そのもの
    #
    # [array context]
    #   戻り値: (ファイル中身, [リンク, ...])
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my ($urlid, $version) = @_;

    defined $urlid or croak "get_content: urlid is undef";
    defined $version or croak "get_content: version is undef";

    my $checkpoint;
    my @patches;

    my $type;
    ($type, $checkpoint) = get_content_without_patching($urlid, $version);
    while ($type eq 'xdelta') {
	unshift @patches, $checkpoint;
	$version--;
	($type, $checkpoint) = get_content_without_patching($urlid, $version);
    }

    if (@patches) {
	# FIXME: パッチを當てる
	croak "Patches are not implemented. sorry";
    }

    if (wantarray) {
	my $info = get_info($urlid, $version);
	my $links = extract_links($checkpoint, $info->{content_type});

	@$links = grep { $_->[2] !~ m/^#/ } @$links;

	($checkpoint, $links);
    }
    else {
	$checkpoint;
    }
}

sub get_links {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my ($urlid, $version) = @_;
    
    defined $urlid or croak "get_links: urlid is undef";
    defined $version or croak "get_links: version is undef";

    my $info = get_info($urlid, $version);
    if (can_extract_links($info->{content_type})) {
	(undef, $_) = get_content($urlid, $version);
	return $_;
    }
    else {
	# サポートしてゐない MIME タイプなので、ファイルの中身を取って
	# 來ると無駄になる。
	return;
    }
}

sub get_content_without_patching {
    # 戻り値: (type, data)
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my ($urlid, $version) = @_;

    my $info = get_info($urlid, $version);
    if (!defined $info) {
	croak "version not found: urlid($urlid), version($version)";
    }

    my $sth = DB->prepare(q{
SELECT content
  FROM pieces_storage
 WHERE pieceid = ?
 ORDER BY split ASC
});
    $sth->execute($info->{pieceid});
    my $data = '';
    while (my $row = $sth->fetch) {
	$data .= $row->[0];
    }
    if ($sth->rows != $info->{splits}) {
	croak sprintf "Number of split mismatches (in pieces_info: %d, actual: %d) (urlid: %d, version: %d)",
	  $info->{splits}, $sth->rows + 0, $urlid, $version;
    }

    return ($info->{type}, $data);
}

sub find_with_md5  {
    # 戻り値: ([urlid, version], ...)
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my $md5 = shift;

    my $sth = DB->prepare(q{
	SELECT urlid, version
	  FROM versions
	  LEFT JOIN pieces_info USING (pieceid)
	 WHERE md5sum = ?
    });
    $sth->execute($md5);
    my @result;
    while (my $row = $sth->fetch) {
	push @result, [ @$row ];
    }

    @result;
}

sub delete_url {
    # 指定されたURL IDの全バージョンを消す。
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my $urlid = shift;

    my $sth = DB->prepare(q{
	SELECT version
	  FROM versions
	 WHERE urlid = ?
    });
    $sth->execute($urlid);
    while (my $ver = $sth->fetch) {
	delete_version($urlid, $ver->[0]);
    }
    
    undef;
}

sub delete_old_versions {
    # 最新でないバージョンを全て消す
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my $urlid = shift;
    
    my $latest = get_latest_version($urlid);
    if (!defined $latest) {
	croak "no such urlid: $urlid";
    }
    
    logdebug "delete_old_versions: urlid $urlid; latest: $latest";

    my $sth = DB->prepare(q{
	SELECT version
	  FROM versions
	 WHERE urlid = ? AND version != ?
    });
    $sth->execute($urlid, $latest);
    while (my $ver = $sth->fetch) {
	logdebug "delete_old_versions: urlid $urlid; $ver->[0] is not latest.";
	delete_version($urlid, $ver->[0]);
    }
}

sub delete_version {
    $_[0] and ref($_[0]) eq __PACKAGE__ and shift;
    my ($urlid, $version) = @_;

    my ($pieceid) = DB->selectrow_array(
	q{SELECT pieceid FROM versions WHERE urlid = ? AND version = ?},
	undef, $urlid, $version);
    if (!defined $pieceid) {
	croak "version not found: urlid($urlid), version($version)";
    }

    DB->do(q{
	LOCK TABLES
	  versions WRITE,
	  pieces_info WRITE,
	  pieces_storage WRITE,
	  target_url WRITE
    });

    my ($refcount) = DB->selectrow_array(
	q{SELECT refcount FROM pieces_info where pieceid = ?},
	undef, $pieceid);
    if ($refcount == 1) {
	# 本當にこれを參照してゐるバージョンが一つだけである事を確認
	my ($count) = DB->selectrow_array(
	    q{SELECT COUNT(*) FROM versions WHERE pieceid = ?}, undef, $pieceid);
	if ($count > 1) {
	    logwarn "refcount of pieceid $pieceid is being zero, but it actually is referenced $count times.";
	    DB->do(q{
		UPDATE pieces_info
		   SET refcount = ?
		 WHERE pieceid = ?
	    }, undef, $count, $pieceid);
	}
	else {
	    logdebug "refcount of pieceid $pieceid became zero. deleting...";
	    DB->do(q{DELETE FROM pieces_storage WHERE pieceid = ?}, undef, $pieceid);
	    DB->do(q{DELETE FROM pieces_info WHERE pieceid = ?}, undef, $pieceid);
	}
    }
    else {
	DB->do(q{
	    UPDATE pieces_info
	       SET refcount = refcount - 1
	     WHERE pieceid = ?
	}, undef, $pieceid);

	$refcount--;
	logdebug "refcount of pieceid $pieceid became $refcount. it keeps alive...";
    }

    DB->do(q{
	DELETE FROM versions
	 WHERE urlid = ? AND version = ?
    }, undef, $urlid, $version);

    DB->do(q{UNLOCK TABLES});
}

sub total_files {
    ($_) = DB->selectrow_array(
	q{SELECT COUNT(*) FROM pieces_info});
    $_;
}

sub total_size {
    ($_) = DB->selectrow_array(
	q{SELECT SUM(data_size) FROM pieces_info});
    $_;
}

1;
