#!/usr/bin/perl -w

use strict;
use DBI;
use RiakSearch;
use Data::Dump qw/dump/;


my ( $riak_url, $table, $dbi ) = @ARGV;

$riak_url ||= 'http://10.60.0.92:8098';
$table    ||= 'biblioitems';
$dbi      ||= 'DBI:mysql:dbname=koha;host=10.60.0.10;port=3306';

my $batch_size = 1000;
my $offset = $ENV{OFFSET} || 0;

my $dbh = DBI->connect($dbi,"","") || die $DBI::errstr;
my $riak = RiakSearch->new( $riak_url );
#$riak->{args} = 'w=1';


sub riak_search_kv_hook {
	my $bucket = shift;
	$riak->request( 'PUT' => $bucket, { props => {
		precommit => [ { mod => 'riak_search_kv_hook', fun => 'precommit' } ],
#		precommit => [],
#		last_write_wins => 'true',
		}
	});
	warn "riak_search_kv_hook $bucket ", $riak->request( 'GET' => '/koha.marcxml' );
}

riak_search_kv_hook 'koha.marcxml';
riak_search_kv_hook "koha.$table";

while(1) {

	my $limit = "LIMIT $batch_size OFFSET $offset";
	warn "SELECT * FROM $table $limit\n";

    my $sth = $dbh->prepare(qq{ select * from $table $limit}) || die $dbh->errstr();
    $sth->execute || die $sth->errstr();
    my @pk = $dbh->primary_key( undef, undef, $table );

    print "import ", $sth->rows, " rows from $table pk:",dump( @pk ),"...\n";

    while (my $row = $sth->fetchrow_hashref() ) {

        my $key = join('_', map { $row->{$_} } @pk);
		$key ||= "o.$offset";
		my $biblionumber = $row->{biblionumber};

        if ( my $marcxml = delete $row->{marcxml} ) {
			$riak->request( 'PUT' => "/koha.marcxml/$key", $marcxml, {
				'Content-Type' => 'text/xml',
				'Link' => qq|</riak/koha.$table/$biblionumber>; riaktag="biblio"|,
			} );
        }

#		warn "## $key ",dump($row);

		my $headers;
		foreach my $reference ( qw(biblio biblioitems borrowers items) ) {
			my $singular = $1 if $reference =~ m/^(\w+?)s?$/;
			my $col = $singular . 'number';
			next if $key =~ m/$col/;
			my $number = $row->{$col} || next;
			push @{ $headers->{Link} }, qq|</riak/koha.$reference/$number>; riaktag="$reference"|;
		}

		$riak->request( 'PUT' => "/koha.$table/$key", $row, $headers );

		$offset++;
    }

	last if $sth->rows < $batch_size;

}

warn "END total_rows $offset\n";
