# Copyright (c) 2004-2008  Joshua Hoblitt
#
# $Id: Server.pm,v 1.93 2008-10-13 20:41:17 jhoblitt Exp $

package Nebulous::Server;

use strict;
use warnings FATAL => qw( all );
no warnings qw( uninitialized );

our $VERSION = '0.15';

use base qw( Class::Accessor::Fast );

use DBI;
use Digest::SHA1 qw( sha1_hex );
use File::ExtAttr qw( setfattr );
use File::Path;
use File::Spec;
use Log::Log4perl;
use Nebulous::Server::Config;
use Nebulous::Server::Log;
use Nebulous::Server::SQL;
use Nebulous::Keys qw( parse_neb_key parse_neb_volume );
use Params::Validate qw( validate_pos SCALAR SCALARREF UNDEF );
use URI::file;

__PACKAGE__->mk_accessors(qw( log sql config ));

use constant SUBPATH_DEPTH  => 2;
use constant NFS_RETRIES    => 100;

sub new
{
    my $class = shift;

    # let Nebulous::Server::Config validate our params
    my $config = Nebulous::Server::Config->init( @_ );

    # log4perl is not avaliable until we call init()
    Nebulous::Server::Log->init($config);
    my $log = Log::Log4perl::get_logger( "Nebulous::Server" );

    my $sql = Nebulous::Server::SQL->new;

    $log->debug( "entered - @_" );

    my $self = bless {}, ref $class || $class;
    $self->log($log);
    $self->sql($sql);
    $self->config($config);

    # ask for the db handle as a means of validating the database parameters
    $self->db;

    $log->debug( "leaving" );
    
    return $self;
}


sub db
{
    my $self = shift;

    if (@_) {
        $self->{db} = $_[0];
        return $self;
    }

    my $log     = $self->log;
    my $sql     = $self->sql;
    my $config  = $self->config;

    # if the dbh is still alive, return it
    if (defined $self->{db} and $self->{db}->ping) {
        $log->debug("db handle is still alive");
        return $self->{db};
    }
    # otherwise create a new connection
    $log->debug("db handle is dead/unopened");

    # if we're running under mod_perl & Apache::DBI is loaded we want to
    # reconnect to the database everytime the dbh is requested.  The rational is
    # that if we're running under mod_perl this is probably a log running
    # processes and the database might have gone away on us.  Apache::DBI will
    # take care of getting a valid dbh back.
    my $db;
    eval {
        $db = DBI->connect_cached(
            $config->dsn,
            $config->dbuser,
            $config->dbpasswd,
            {
                RaiseError => 1,
                PrintError => 0,
                AutoCommit => 0,
            },
        );

        $db->do( $sql->set_transaction_model );
        $log->debug( "connected to database: ", sub { $db->data_sources; } );
        $db->commit;
        $log->debug("commit");
    };
    if ( $@ ) {
        $db->rollback if $db;
        $log->debug("rollback");
        $log->logdie( "database error: $@" );
    }

    $self->{db} = $db;

    return $db;
}


sub create_object
{
    my $self = shift;

    my ($key, $vol_name) = validate_pos(@_,
        {
            type        => SCALAR,
        },
        {
            type        => SCALAR|UNDEF,
#            callbacks   => {
#                # check that the volume requested is valid
#                'is valid volume name' => sub {
#                    return 1 if not defined $_[0];
#                    $self->_is_valid_volume_name($_[0])
#                },
#            },
            optional    => 1,
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug( "entered - @_" );

    # vol_name overrides the key implied volume
    $key = parse_neb_key($key, $vol_name);
    $vol_name = $key->volume;

    # the key's volume can't be validiated on input for this method so we have
    # to check it after parsing the key
    if (defined $vol_name
        and not $self->_is_valid_volume_name($key->volume)) {
        if ($key->soft_volume) {
            $log->warn( "$vol_name is not a known volume name" );
            $vol_name = undef;
        } else {
            die "$vol_name is not a valid volume name"
        }
    }
        
    my ($vol_id, $vol_host, $vol_path, $vol_xattr)
        = $self->_get_storage_volume($vol_name, $key->soft_volume);

    my $uri;
    eval {
        {
            # create storage_object
            my $query = $db->prepare_cached( $sql->new_object ); 
            $query->execute('NULL', $key->path);
        }

        my $so_id;
        {
            # get object ID
            my $query = $db->prepare_cached( $sql->last_insert_id );
            $query->execute;
            ($so_id) = $query->fetchrow_array;
            # XXX finish seems to be required when using LAST_INSERT_ID() or we
            # get a warning about the stmt handling still be active the next
            # time LAST_INSERT_ID() is invoked
            $query->finish;
        }

        {
            # create storage_object_attr
            my $query = $db->prepare_cached( $sql->new_object_attr ); 
            $query->execute($so_id);
        }

        {
            
            # create instance with no URI
#            my $query = $db->prepare_cached( $sql->new_instance );
            my $query = $db->prepare_cached( $sql->new_object_instance );
            $query->execute($vol_id);
        }

        my $ins_id;
        {
            # get instance ID
            my $query = $db->prepare_cached( $sql->last_insert_id );
            $query->execute;
            ($ins_id) = $query->fetchrow_array;
            # XXX finish seems to be required when using LAST_INSERT_ID() or we
            # get a warning about the stmt handling still be active the next
            # time LAST_INSERT_ID() is invoked
            $query->finish;
        }

        # Unfortunately, since we want to use the instance row's ID as part of the
        # actual on disk file name we can't try to create the file until after
        # we've create both a new storage_storage object and instance.

        # TODO add some stuff here to retry if unsucessful
        $uri = $self->_create_empty_instance_file($key->path, $so_id, $ins_id, $vol_path, $vol_xattr);
        $log->debug("created $uri on volume ID: $vol_id");

        {
            # update the instance with URI & vol_id that the file is on
            my $query = $db->prepare_cached( $sql->update_instance_uri );
            # vol_id, uri, ins_id
            $query->execute($vol_id, "$uri", $ins_id);
        }

        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
#        and not $key->soft_volume
        $db->rollback;
        $log->debug("rollback");
        $log->logdie("error: $@");
    }

    $log->debug("leaving");

    return "$uri";
}


sub rename_object
{
    my $self = shift;

    my ($key, $newkey) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        # make sure the "newkey" doesn't already exist
        {
            type        => SCALAR,
            callbacks   => {
                'is not valid object key' => sub { not $self->_is_valid_object_key($_[0]) },
            },
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # ignore volumes
    $key    = parse_neb_key($key);
    $newkey = parse_neb_key($newkey);

    eval {
        # rename storage_object
        my $query = $db->prepare_cached($sql->rename_object); 
        # this SQL statment takes the new key name as the first param
        my $rows = $query->execute($newkey->path, $key->path);

        # if we affected more then one row something very bad has happened.
        unless ($rows == 1) {
            $query->finish;
            $log->logdie("affected row count is $rows instead of 1");
        }

        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie("database error: $@");
    }

    $log->debug("leaving");

    return $newkey;
}


sub swap_objects
{
    my $self = shift;

    my ($key1, $key2) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # ignore volumes
    $key1 = parse_neb_key($key1);
    $key2 = parse_neb_key($key2);

    # order of operations for the swap:
    # key1 -> key1.swap
    # key2 -> key1
    # key1.swap -> key2

    eval {
        {
            # key1 -> key1.swap
            my $query = $db->prepare_cached($sql->rename_object); 
            # this SQL statment takes the new key name as the first param
            my $rows = $query->execute($key1->path . ".swap", $key1->path);

            # if we affected more then one row something very bad has happened.
            unless ($rows == 1) {
                $query->finish;
                $log->logdie("affected row count is $rows instead of 1");
            }
        }

        {
            # key2 -> key1
            my $query = $db->prepare_cached($sql->rename_object); 
            # this SQL statment takes the new key name as the first param
            my $rows = $query->execute($key1->path, $key2->path);

            # if we affected more then one row something very bad has happened.
            unless ($rows == 1) {
                $query->finish;
                $log->logdie("affected row count is $rows instead of 1");
            }
        }

        {
            # key1.swap -> key2
            my $query = $db->prepare_cached($sql->rename_object); 
            # this SQL statment takes the new key name as the first param
            my $rows = $query->execute($key2->path, $key1->path . ".swap");

            # if we affected more then one row something very bad has happened.
            unless ($rows == 1) {
                $query->finish;
                $log->logdie("affected row count is $rows instead of 1");
            }
        }

        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie("database error: $@");
    }

    $log->debug("leaving");

    return 1;
}


sub replicate_object
{
    my $self = shift;

    my ($key, $vol_name) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key'
                    => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR|UNDEF,
#            callbacks   => {
#                # check that the volume name requested is valid
#                'is valid volume name' => sub {
#                    return 1 if not defined $_[0];
#                    $self->_is_valid_volume_name($_[0])
#                },
#            },
            optional    => 1,
        },
    );

    # if a volume name is explicity specified then we should make the
    # replication onto that volume (even if there is alread an instance on that
    # volume) if at all possible and throw an error if we can not.
    # if a volume name IS NOT specified then we should make the replication
    # onto any (hopefully the best) volume that DOES NOT already have an
    # instance on it.  If all avilable volume already have an instance on them
    # then we should throw an error 

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # vol_name overrides the key implied volume
    $key = parse_neb_key($key, $vol_name);
    $vol_name = $key->volume;

    if (defined $vol_name
        and not $self->_is_valid_volume_name($key->volume)) {
        if ($key->soft_volume) {
            $log->warn( "$vol_name is not a known volume name" );
            $vol_name = undef;
        } else {
            die "$vol_name is not a valid volume name"
        }
    }
        
    my ($vol_id, $vol_host, $vol_path, $vol_xattr);
    if (defined $vol_name) {
        ($vol_id, $vol_host, $vol_path, $vol_xattr)
            = $self->_get_storage_volume($vol_name);
    } else {
        ($vol_id, $vol_host, $vol_path, $vol_xattr)
            = $self->_get_replication_volume($key);
    }

    my $uri;
    eval {
        my $so_id;
        {
            # verify that at least one instance is currently available
            my $query = $db->prepare_cached( $sql->get_object_instances );
            my $rows = $query->execute($key->path, 1);

            unless ( $rows > 0 ) {
                $query->finish;
                $log->logdie( "storage object does not exist" );
            }

            $so_id = $query->fetchrow_hashref->{ 'so_id' };
            $query->finish;
        }

        {
            my $query = $db->prepare_cached( $sql->new_instance );
            $query->execute($so_id, $vol_id);
        }

        my $ins_id;
        {
            my $query = $db->prepare_cached( $sql->last_insert_id );
            $query->execute();
            ($ins_id) = $query->fetchrow_array;
            # XXX finish seems to be required when using LAST_INSERT_ID() or we
            # get a warning about the stmt handling still being active the next
            # time LAST_INSERT_ID() is invoked
            $query->finish;
        }

        # Unfortunately, since we want to use the instance row's ID as part of
        # the actual on disk file name we can't try to create the file until
        # after we've create both a new storage_storage object and instance.

        # TODO add some stuff here to retry if unsucessful
        $uri = $self->_create_empty_instance_file($key->path, $so_id, $ins_id, $vol_path, $vol_xattr);

        {
            my $query = $db->prepare_cached( $sql->update_instance_uri );
            # vol_id, uri, ins_id
            $query->execute($vol_id, $uri, $ins_id);
        }

        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
        $db->rollback;
        # handle soft volumes
        if (defined $vol_name and defined $key->soft_volume) {
            $log->debug("retrying with 'any' volume");
            return $self->replicate_object($key->path, 'any');
        }
        $log->debug("rollback");
        $log->logdie("error: $@");
    }

    $log->debug("leaving");

    return "$uri";
}


sub lock_object
{
    my $self = shift;

    my ( $key, $type ) = validate_pos( @_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR,
            callbacks   => {
                'is read or write' => sub { $_[0] =~ /^(?:read|write)$/ },
            },
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug( "entered - @_" );

    # ignore volume
    $key = parse_neb_key($key);

    my $so_id;
    my $read_lock;
    my $write_lock;

    eval {
        {
            # this will set update locks
            my $query = $db->prepare_cached( $sql->get_object_locks );
            my $rows = $query->execute( $key->path );
            unless ( $rows == 1 ) {
                $query->finish;
                $log->logdie( "storage object does not exist" );
            }

            my $row = $query->fetchrow_hashref;
            $query->finish;

            $so_id      = $row->{ 'so_id' };
            $read_lock  = $row->{ 'read_lock' };
            $write_lock = $row->{ 'write_lock' };
        }

        if ($type eq 'write') {
            # can't set a write lock twice and
            # can't set a write lock if there are read locks
            if ($write_lock) {
                $log->logdie("can not write lock twice -- retry");
            }
            
            if ($read_lock > 0) {
                $log->logdie("can not write lock after read lock -- retry");
            }

            {
                my $query = $db->prepare_cached( $sql->set_write_lock );
                my $rows = $query->execute($key->path);
            
                # if we affected more then one row something very bad has happened.
                unless ($rows == 1) {
                    $log->logdie("affected row count is $rows instead of 1");
                }

            }
        } elsif ($type eq 'read') {
            # can't set a read lock if there's a write lock
            if ($write_lock) {
                $log->logdie("can not read lock after write lock -- retry");
            }

            {
                my $query = $db->prepare_cached( $sql->increment_read_lock );
                my $rows = $query->execute($key->path);
            
                # if we affected more then one row something very bad has happened.
                unless ($rows == 1) {
                    $log->logdie("affected row count is $rows instead of 1");
                }
            }
        }

        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie("error: $@");
    }

    $log->debug("leaving");

    return 1;
}


sub unlock_object
{
    my $self = shift;

    my ( $key, $type ) = validate_pos( @_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR,
            callbacks   => {
                'is read or write' => sub { $_[0] =~ /^(?:read|write)$/ },
            },
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug( "entered - @_" );

    # ignore volume
    $key = parse_neb_key($key);

    my $so_id;
    my $read_lock;
    my $write_lock;

    eval {
        {
            # this will set update locks
            my $query = $db->prepare_cached( $sql->get_object_locks );
            my $rows = $query->execute($key->path);
            unless ($rows == 1) {
                $query->finish;
                $log->logdie("storage object does not exist");
            }

            my $row = $query->fetchrow_hashref;
            $query->finish;

            $so_id      = $row->{ 'so_id' };
            $read_lock  = $row->{ 'read_lock' };
            $write_lock = $row->{ 'write_lock' };
        }

        if ($type eq 'write') {
            # can't remove a write lock if it doesn't exist
            if ($read_lock) {
                $log->logdie("can not have a write lock under a read lock");
            }

            unless ($write_lock) {
                $log->logdie("can not remove non-existant write lock");
            }

            {
                my $query = $db->prepare_cached( $sql->delete_write_lock );
                my $rows = $query->execute($key->path);
            
                # if we affected more then one row something very bad has happened.
                unless ($rows == 1) {
                    $log->logdie("affected row count is $rows instead of 1");
                }
            }
        } elsif ($type eq 'read') {
            # can't remove a read lock if there's a write lock and
            # can't remove a read lock if there aren't any
            if ($write_lock) {
                $log->logdie("can not have a read lock under a write lock");
            }
               
            if ($read_lock == 0) {
                $log->logdie("can not remove non-existant read lock");
            }

            {
                my $query = $db->prepare_cached( $sql->decrement_read_lock );
                my $rows = $query->execute($key->path);
            
                # if we affected more then one row something very bad has happened.
                unless ($rows == 1) {
                    $log->logdie("affected row count is $rows instead of 1");
                }

            }
        }
        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie("error: $@");
    }

    $log->debug( "leaving" );

    return 1;
}


sub setxattr_object
{
    my $self = shift;

    my ($key, $name, $value, $flags) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR,
        },
        {
            type        => SCALAR,
        },
        {
            type        => SCALAR,
            callbacks   => {
                'is read or write' => sub { $_[0] =~ /^(?:create|replace)$/i },
            },
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # ignore volume
    $key = parse_neb_key($key);

    eval {
        my $query;

        if ($flags eq 'create') {
            $query = $db->prepare_cached( $sql->new_object_xattr );
        } else {
            # replace
            $query = $db->prepare_cached( $sql->replace_object_xattr );
        }

        # name, value, ext_id
        my $rows = $query->execute($name, $value, $key->path);
        $query->finish;

        # if we affected more then one row something very bad has happened.
        if ($flags eq 'create') {
            unless ($rows == 1) {
                $log->logdie( "affected row count is $rows instead of 1" );
            }
        } else {
            # replace_object_xattr can effect either 1 or 2 rows.  2 rows in
            # the case of a replace and 1 if the xattr didn't already exist.
            unless ($rows == 1 or $rows == 2) {
                $log->logdie( "affected row count is $rows instead of 2" );
            }
        }

        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie("database error: $@");
    }

    $log->debug("leaving");

    return 1;
}


sub getxattr_object
{
    my $self = shift;

    my ($key, $name) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR,
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # ignore volume
    $key = parse_neb_key($key);

    my $value;
    eval {
        my $query = $db->prepare_cached( $sql->get_object_xattr );
        # ext_id, name
        my $rows = $query->execute($key->path, $name);

        # no rows returned means that the xattr does not exist
        if ($rows == 0) {
            $query->finish;
            $log->logdie( "xattr $key:$name does not exist" );
        }
        # if we go more then one row bad something very bad has happened.
        unless ($rows == 1) {
            $query->finish;
            $log->logdie( "affected row count is $rows instead of 1" );
        }

        my $row = $query->fetchrow_hashref;
        # XXX: DBI bug? ->finish is needed here even though $query is going out
        # of scope
        $query->finish;
        $value = $row->{ 'value' };
    };
    $log->logdie("database error: $@") if $@;

    $log->debug("leaving");

    return $value;
}


sub listxattr_object
{
    my $self = shift;

    my ($key) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # ignore volume
    $key = parse_neb_key($key);

    my @xattrs;
    eval {
        my $query = $db->prepare_cached( $sql->list_object_xattr );
        # ext_id
        my $rows = $query->execute($key->path);

        while (my $row = $query->fetchrow_hashref) {
            push @xattrs, $row->{ 'name' };
        }
    };
    $log->logdie("database error: $@") if $@;

    $log->debug("leaving");

    return \@xattrs;
}


sub removexattr_object
{
    my $self = shift;

    my ($key, $name) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR,
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # ignore volume
    $key = parse_neb_key($key);

    eval {
        my $query = $db->prepare_cached( $sql->remove_object_xattr );
        # ext_id, name
        my $rows = $query->execute($key->path, $name);
        $query->finish;

        # if we affected more then one row something very bad has happened.
        unless ($rows == 1) {
            $log->logdie( "affected row count is $rows instead of 1" );
        }

        $db->commit;
        $log->debug("commit");
    };
    if ($@) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie("database error: $@");
    }

    $log->debug("leaving");

    return 1;
}


sub find_objects
{
    my $self = shift;

    my ( $pattern ) = validate_pos( @_,
        {
            type        => SCALAR,
            optional    => 1,
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug( "entered - @_" );

    unless ($pattern) {
        $log->debug( "leaving" );
        $log->logdie("no keys found");
    }

    # attempt to strip off neb:// if it exists
    $pattern =~ s|^neb:||;

    my @keys;
    eval {
        my $query = $db->prepare_cached( $sql->find_objects );
        $query->execute( $pattern );

        while ( my $row = $query->fetchrow_hashref ) {
            my $key = $row->{ 'ext_id' };
            push @keys, $key if $key;
        }
    };
    $log->logdie("database error: $@") if $@;

    $log->logdie("no keys found") unless ( scalar @keys );

    $log->debug( "leaving" );

    return \@keys;
}


sub find_instances
{
    my $self = shift;

    my ($key, $vol_name) = validate_pos(@_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
        {
            type        => SCALAR|UNDEF,
#            callbacks   => {
#                # check that the volume name requested is valid
#                'is valid volume name' => sub {
#                    return 1 if not defined $_[0];
#                    $self->_is_valid_volume_name($_[0])
#                },
#            },
            optional    => 1,
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # vol_name overrides the key implied volume
    $key = parse_neb_key($key, $vol_name);
    $vol_name = $key->volume;

    # the key's volume can't be validiated on input for this method so we have
    # to check it after parsing the key
    if (defined $vol_name
        and not $self->_is_valid_volume_name($key->volume)) {
        if ($key->soft_volume) {
            $log->warn( "$vol_name is not a known volume name" );
            $vol_name = undef;
        } else {
            die "$vol_name is not a valid volume name"
        }
    }

    my @locations;
    eval {
        my $query;
        if ($vol_name) {
            $query = $db->prepare_cached( $sql->get_object_instances_by_vol_name );
            # ext_id, name, available
            my $rows = $query->execute($key->path, $vol_name, 1);
            unless ($rows > 0) {
                $query->finish;
                $log->logdie("no instances on storage volume or volume is not avaiable for key: $key volume: $vol_name");
            }
        } else {
            $query = $db->prepare_cached( $sql->get_object_instances );
            # ext_id, available
            my $rows = $query->execute($key->path, 1);
            unless ($rows > 0) {
                $query->finish;
                $log->logdie("no instances available for key: $key");
            }
        }

        while (my $row = $query->fetchrow_hashref) {
            my $instance = $row->{ 'uri' };
            push @locations, $instance if $instance;
        }
    };
    if ($@) {
        $db->rollback;
        # handle soft volumes
        if (defined $vol_name and defined $key->soft_volume) {
            $log->debug("retrying with 'any' volume");
            return $self->find_instances($key->path, 'any');
        }
        $log->logdie("database error: $@");
    }

    # XXX remove this?
    $log->logdie("no instances found") unless (scalar @locations);

    $log->debug("found: @locations");

    $log->debug("leaving");

    return \@locations;
}


sub delete_instance
{
    my $self = shift;

    my ( $uri ) = validate_pos( @_,
        {
            type => SCALAR|SCALARREF,
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug( "entered - @_" );

    eval {
        my $so_id;
        my $instances;
        # get so_id
        {
            my $query = $db->prepare_cached( $sql->get_object_from_uri );
            my $rows = $query->execute( $uri );

            unless ( $rows > 0 ) {
                $query->finish;
                $log->logdie( "no instance is associated with uri" );
            }

            $so_id = $query->fetchrow_hashref->{ 'so_id' };
            $query->finish;

        }

        {
            my $query = $db->prepare_cached( $sql->get_instance_count );
            $query->execute( $so_id );

            $instances = $query->fetchrow_hashref->{ 'count(ins_id)' };
            $query->finish;
        }

        # remove instance
        {
            my $query = $db->prepare_cached( $sql->delete_instance );
            my $rows = $query->execute( $uri );
            $query->finish;
            
            # if we affected something other then two rows something very bad
            # has happened
            unless ( $rows == 1 ) {
                $log->logdie( "affected row count is $rows instead of 1" );
            }
        }

        # if we just deleted the last instance associated with a storage object
        # remove it too
        if ( $instances == 1 ) {
            # we just removed the last instance
            my $query = $db->prepare_cached( $sql->delete_object );
            my $rows = $query->execute( $so_id );
            $query->finish;

            # TODO: this will have to be changed in order to support hardlinks
            unless ( $rows == 1 ) {
                $log->logdie( "affected row count is $rows instead of 2" );
            }
        }

        $db->commit;
        $log->debug("commit");
    };
    if ( $@ ) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie( "database error: $@" );
    }

    $log->debug( "leaving" );

    return 1;
}


sub stat_object
{
    my $self = shift;

    my ( $key ) = validate_pos( @_,
        {
            type        => SCALAR,
            callbacks   => {
                'is valid object key' => sub { $self->_is_valid_object_key($_[0]) },
            },
        },
    );

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    # ignore volume
    $key = parse_neb_key($key);

    my $stat;
    eval {
        my $query = $db->prepare_cached( $sql->stat_object );
        my $rows = $query->execute($key->path);

        unless ($rows == 1) {
            $log->logdie("no storage object found");
        }

        $stat = $query->fetchrow_arrayref;
        $query->finish;
    };
    $log->logdie("database error: $@") if $@;

    $log->debug("leaving");

    return $stat;
}


sub mounts
{
    my $self = shift;

    validate_pos(@_); 

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug("entered - @_");

    my $stats;
    my $query;
    eval {
        $query = $db->prepare_cached( $sql->get_mounted_volumes );
        $query->execute();

        # suck that table into an AoA
        $stats = $query->fetchall_arrayref;

        $query->finish;
    };
    $log->logdie("database error: $@") if $@;

    $log->logdie("no mounted volumes found") unless (scalar @$stats);

    $log->debug("leaving");

    return $stats;
}


sub _get_storage_volume
{
    my $self = shift;

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  = $self->db;

    no warnings qw( uninitialized );
    $log->debug( "entered - @_" );
    use warnings;

    my ($name, $soft_volume) = @_;

    my ($vol_id, $vol_host, $vol_path, $xattr);
    eval {
        my $query;
        my $rows;
        if ( $name ) {
            $query = $db->prepare_cached( $sql->get_storage_volume_by_name );
            # %free, name, avaiable, allocate
            $rows = $query->execute(0.95, $name, 1, 1);
            # XXX destinguish between non-existant and unaviable
            unless ($rows > 0) {
                $query->finish;

                # if a volume name was specified, and is soft, and we failed to
                # find it, fall back to any volume
                if ($soft_volume) {
                    ($vol_id, $vol_host, $vol_path, $xattr) = $self->_get_storage_volume;
                    return; # this just returns out of the eval not from the subroutine
                }
                $log->logdie("storage volume: $name is not available");
            }
            # when matching by name we shouldn't ever match more than once
            if ($rows > 1) {
                $query->finish;
                $log->logdie("affected row count is $rows instead of 1");
            }
        } else {
            $query = $db->prepare_cached( $sql->get_storage_volume );
            # %free, avaiable, allocate
            $rows = $query->execute(0.95, 1, 1);
            # there has to be atleast one storage volume
            unless ($rows > 0) {
                $query->finish;
                $log->logdie("no storage volume is available");
            }
        }

        my $free;
        ($vol_id, $vol_host, $vol_path, $xattr, $free) = $query->fetchrow_array;
        $query->finish;
    };
    $log->logdie("database error: $@") if $@;

    $log->logdie("failed to find a suitable volume" )
        unless defined $vol_id and defined $vol_path;

    $log->debug( "leaving" );

    return ($vol_id, $vol_host, $vol_path, $xattr);
}


sub _get_replication_volume
{
    my $self = shift;

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    no warnings qw( uninitialized );
    $log->debug( "entered - @_" );
    use warnings;

    my $key = shift;

    $key = parse_neb_key($key);

    my ($vol_id, $vol_host, $vol_path, $xattr);
    eval {
        my $rows;
        my $query = $db->prepare_cached( $sql->get_replication_volume_for_ext_id );
        # ext_id, %free, avaiable, allocate
        $rows = $query->execute($key->path, 0.95, 1, 1);
        # XXX destinguish between non-existant and unaviable
        unless ($rows > 0) {
            $query->finish;
            $log->logdie("can't find a suitable storage volume to replicate $key to");
        }
        # when matching by name we shouldn't ever match more than once
        if ($rows > 1) {
            $query->finish;
            $log->logdie("affected row count is $rows instead of 1");
        }

        my $free;
        ($vol_id, $vol_host, $vol_path, $xattr, $free) = $query->fetchrow_array;
        $query->finish;
    };
    $log->logdie("database error: $@") if $@;

    $log->logdie("failed to find a suitable volume" )
        unless defined $vol_id and defined $vol_path;

    $log->debug( "leaving" );

    return ($vol_id, $vol_host, $vol_path, $xattr);
}


sub _is_valid_object_key
{
    my ($self, $key) = @_;

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $key = parse_neb_key($key);

    my $ext_id;
    eval {
        my $query = $db->prepare_cached( $sql->get_object ); 
        $query->execute($key->path);
        ($ext_id) = $query->fetchrow_array;
        $query->finish;
    };
    if ($@) {
        $db->rollback;
        $log->debug("rollback");
        $log->logdie( "database error: $@" );
    }

    if (defined $ext_id) {
        return 1;
    } 

    return;
}


sub _is_valid_volume_name
{
    my ($self, $vol_name) = @_;

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    my $volume_info = parse_neb_volume($vol_name);

    $vol_name = $volume_info->{volume};

    # handle "any" volume
    if ($vol_name eq 'any') {
        return 1;
    }

    my ($vol_id, $vol_path);
    eval {
        my $query = $db->prepare_cached( $sql->get_volume_by_name ); 
        $query->execute( $vol_name );
        my $free;
        ($vol_id, $vol_path, $free) = $query->fetchrow_array;
        $query->finish;
    };
    $log->logdie("database error: $@") if $@;

    if (defined $vol_id and defined $vol_path) {
        return 1;
    } 

    return;
}


sub _create_empty_instance_file
{
    my $self = shift;

    my ($key, $so_id, $ins_id, $vol_path, $xattr) =  @_;

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  = $self->db;

    my $uri;
    eval {
        my $storage_path = $self->_generate_storage_path($key, $vol_path);
        my $storage_filename = $self->_generate_storage_filename($key, $ins_id);
        unless (-d $storage_path) {
            _retry(sub { mkpath(@_) }, $storage_path, 0, 0775)
                or die "can't create storage path: $storage_path";
        }
        # check to make sure at least the parent directory has the proper
        # permissions
        my $mode = [_retry(sub { stat($storage_path) } )]->[2] & 07777;
        unless ($mode == 0775) {
            $log->error("$storage_path has the wrong permissions of: %04o", $mode);
            _retry(sub { chmod(0775, $storage_path) }) or die "can't chmod $storage_path: $!";
        }

        my $fqpn = File::Spec->catfile($storage_path, $storage_filename);
        $uri = URI::file->new($fqpn);
        $log->debug("generated uri $uri");
        $self->_create_empty_file($uri->file);
    };
    if ($@) {
        $log->logdie($@);
    }

    if ($xattr) {
        my $path = $uri->file;
        die "can not set xattr on $path: $!"
            unless (setfattr($path, 'user.nebulous_key', $key));
    }

    return $uri;
}


sub _create_empty_file
{
    my $self = shift;

    my ($path) = @_;

    # perl's open() can't do an O_CREAT | O_EXCL
    die "file $path already exists" if (-e $path);

    my $fh;
    die "can not open $path: $!"
        unless (_retry(sub { open($fh, '>', $path) }));

    # chmod before fsync() to make sure the changed perms hit the disk too
    die "can not chmod $path: $!"
        unless (_retry(sub { chmod(0664, $path) }));

    die "can not flush $path: $!"
        unless ($fh->flush);

    die "can not sync $path: $!"
        unless ($fh->sync);

    die "can not close $path: $!"
        unless (_retry(sub { close($fh) }));

    return $path;
}


sub _generate_storage_filename
{
    my $self = shift;

    my ($key, $ins_id) = @_;

    my $filename = $key;
    # mangle '/'s into ':'
    $filename =~ s|/|:|g;

    return "$ins_id.$filename"
}


sub _generate_storage_path
{
    my $self = shift;

    my ($key, $vol_path) = @_;

    # taken and modified from Cache::File::cache_file_path()
    # Copyright (C) 2003-2006 Chris Leishman.  All Rights Reserved.
    my $shakey = sha1_hex($key);
    my (@path) = unpack('A2' x SUBPATH_DEPTH, $shakey);

    return File::Spec->catdir($vol_path, @path);
}


sub _retry
{
    my $func = shift;

    my @ret;
    for (my $i = 0; $i < NFS_RETRIES; $i++) {
        eval {
            @ret = $func->(@_);
        };
        if ($@) {
            die $@;
            sleep 1;
            next;
        }

        last;
    }

    # if the loop ended and $@ is set, rethrow the error
    if ($@) {
        die $@;
    }

    return @ret;
}


sub DESTROY
{
    my $self = shift;

    my $log = $self->log;
    my $sql = $self->sql;
    my $db  =$self->db;

    $log->debug( "entered" );

    $self->db->disconnect;        

    $log->debug( "disconnected from database: ", sub { $db->data_sources; } );

    $log->debug( "leaving" );
}


1;

__END__
