#!/usr/bin/perl -w

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

my $limit = "limit 5";
my $riak_url = 'http://10.60.0.92:8098';
my $dbi = 'DBI:mysql:dbname=koha;host=10.60.0.10;port=3306';
my @tables = qw(
biblioitems
biblio
);

my $dbh = DBI->connect($dbi,"","") || die $DBI::errstr;
my $riak = RiakSearch->new( $riak_url );

warn $riak->request( 'GET' => '/koha.marcxml' );

sub riak_search_kv_hook {
	$riak->request( 'PUT' => shift, { props => { precommit => [ { mod => 'riak_search_kv_hook', fun => 'precommit' } ] } });
}

riak_search_kv_hook '/koha.marcxml';

foreach my $table ( @tables ) {

    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";

	riak_search_kv_hook "koha.$table";

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

        my $key = join('_', map { $row->{$_} } @pk);

        if ( my $marcxml = delete $row->{marcxml} ) {
			$riak->request( 'PUT' => "/koha.marcxml/$key", $marcxml, {
				'Content-Type' => 'text/xml',
			} );
        }

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

		my $headers;
		if ( exists $row->{biblionumber} && $key !~ m/biblionumber/ ) {
			$headers->{Links} = '/koha.biblio/' . $row->{biblionumber};
		}
		$riak->request( 'PUT' => "/koha.$table/$key", $row, $headers );
    }

}
