# Copyright (c) 2004  Joshua Hoblitt
#
# $Id: Server.pm,v 1.40 2007-05-05 02:59:31 jhoblitt Exp $

package Nebulous::Server;

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

our $VERSION = '0.05';

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

use DBI;
use Log::Log4perl;
use Nebulous::Server::Config;
use Nebulous::Server::Log;
use Nebulous::Server::SQL;
use Nebulous::Util qw( :standard );
use Params::Validate qw( validate_pos SCALAR SCALARREF );
use URI::file;

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

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 $db;
    eval {
        $db = DBI->connect(
            $config->dsn,
            $config->dbuser,
            $config->dbpasswd,
            {
                RaiseError => 1,
                PrintError => 0,
                AutoCommit => 0,
            },
        );

        $db->do( $sql->set_transaction_model );

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

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

    $log->debug( "leaving" );

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

    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 we're running under mod_perl & Apache::DBI is loaded we want to
    # reconnect to the database everytime the dbh is request.  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.
    if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
        my $db;
        eval {
            $db = DBI->connect(
                $config->dsn,
                $config->dbuser,
                $config->dbpasswd,
                {
                    RaiseError => 1,
                    PrintError => 0,
                    AutoCommit => 0,
                },
            );

            $db->do( $sql->set_transaction_model );

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

        $self->{db} = $db;

        return $db;
    }

    return $self->{db};
}

sub create_object
{
    my $self = shift;

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

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

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

    # Find the storage volume to use.  We are doing this first as there's no
    # point in creating a new storage object if we can't allocate a new
    # instance. ->_get_storage_volume() throws an exception on failure.
    my ($vol_id, $vol_path);
    eval {
        ($vol_id, $vol_path) = $self->_get_storage_volume($volume);
    };
    if ($@) {
        $log->logdie($@);
    }

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

        my $object_id;
        {
            # get object ID
            my $query = $db->prepare_cached( $sql->last_insert_id );
            $query->execute;
            ($object_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($object_id);
        }

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

        {
            # 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;
        }
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database error: $@");
    }

    # 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.
    my $filename = $key;
    # mange '/'s into ':'
    $filename =~ s|/|:|g;
    my $uri = URI::file->new("$vol_path/$ins_id.$filename");
    $log->debug("generated uri $uri");

    # TODO add some stuff here to retry if unsucessful
    eval {
        _create_empty_file($uri->file, $key);
    };
    if ($@) {
        $db->rollback;
        $log->logdie($@);
    }

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

        $db->commit;
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database 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 - @_");

    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, $key);

        # 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;
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database error: $@");
    }

    $log->debug("leaving");

    return $newkey;
}

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,
            callbacks   => {
                # check that the volume name requested is valid
                'is valid volume name' => sub { $self->_is_valid_volume_name($_[0]) },
            },
            optional    => 1,
        },
    );

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

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

    my ($vol_id, $vol_path);
    eval {
        ($vol_id, $vol_path) = $self->_get_storage_volume($vol_name);
    };
    if ($@) {
        $log->logdie($@);
    }

    my $ins_id;
    eval {
        my $so_id;
        {
            my $query = $db->prepare_cached( $sql->get_object_instances );
            my $rows = $query->execute($key, 1);

            unless ( $rows > 0 ) {
                $query->finish;
                $db->rollback;
                $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);
        }

        {
            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;
        }
    };
    if ($@) {
        $db->rollback;
        $log->logdie( "database error: $@" );
    }

    my $filename = $key;
    # mange '/'s into ':'
    $filename =~ s|/|:|g;
    my $uri = URI::file->new("$vol_path/$ins_id.$filename");
    $log->debug("generated uri $uri");

    # TODO add some stuff here to retry if unsucessful
    eval {
        _create_empty_file($uri->file, $key);
    };
    if ($@) {
        $db->rollback;
        $log->logdie($@);
    }

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

        $db->commit;
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database 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 - @_" );

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

    eval {
        # this will set update locks
        $query = $db->prepare_cached( $sql->get_object_locks );
        my $rows = $query->execute( $key );
        unless ( $rows == 1 ) {
            $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 ( $@ ) {
        $query->finish;
        $db->rollback;
        $log->logdie( "database error: $@" );
    }

    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 ) {
            eval { $db->rollback; };
            $log->logdie( "database error: $@" ) if $@;

            $log->logdie( "can not write lock twice" );
        }
            
        if ( $read_lock > 0 ) {
            eval { $db->rollback; };
            $log->logdie( "database error: $@" ) if $@;

            $log->logdie( "can not write lock after read lock" );
        }

        eval {
            my $query = $db->prepare_cached( $sql->set_write_lock );
            my $rows = $query->execute( $key );
        
            # 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;
        };
        if ( $@ ) {
            $db->rollback;
            $log->logdie( "database error: $@" );
        }

        $log->debug( "leaving" );

        return 1;
    }

    if ( $type eq 'read' ) {
        # can't set a read lock if there's a write lock
        if ( $write_lock ) {
            eval { $db->rollback; };
            $log->logdie( "database error: $@" ) if $@;

            $log->logdie( "can not read lock after write lock" );
        }

        eval {
            my $query = $db->prepare_cached( $sql->increment_read_lock );
            my $rows = $query->execute( $key );
        
            # 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;
        };
        if ( $@ ) {
            $db->rollback;
            $log->logdie( "database error: $@" );
        }

        $log->debug( "leaving" );

        return 1;
    }

    # never reached
    $log->logdie( "this should never happen" );
}

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 - @_" );

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

    eval {
        # this will set update locks
        $query = $db->prepare_cached( $sql->get_object_locks );
        my $rows = $query->execute( $key );
        unless ( $rows == 1 ) {
            $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 ( $@ ) {
        $query->finish;
        $db->rollback;
        $log->logdie( "database error: $@" );
    }

    if ( $type eq 'write' ) {
        # can't remove a write lock if it doesn't exist
        if ( $read_lock ) {
            eval { $db->rollback; };
            $log->logdie( "database error: $@" ) if $@;
           
            $log->logdie( "can not have a write lock under a read lock" );
        }
        unless ( $write_lock ) {
            eval { $db->rollback; };
            $log->logdie( "database error: $@" ) if $@;

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

        eval {
            my $query = $db->prepare_cached( $sql->delete_write_lock );
            my $rows = $query->execute( $key );
        
            # 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;
        };
        if ( $@ ) {
            $db->rollback;
            $log->logdie( "database error: $@" );
        }

        $log->debug( "leaving" );

        return 1;
    }

    if ( $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 ) {
            eval { $db->rollback; };
            $log->logdie( "database error: $@" ) if $@;
           
            $log->logdie( "can not have a read lock under a write lock" );
        }
           
        if ( $read_lock == 0 ) {
            eval { $db->rollback; };
            $log->logdie( "database error: $@" ) if $@;
           
            $log->logdie( "can not remove non-existant read lock" );
        }

        eval {
            my $query = $db->prepare_cached( $sql->decrement_read_lock );
            my $rows = $query->execute( $key );
        
            # 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;
        };
        if ( $@ ) {
            $db->rollback;
            $log->logdie( "database error: $@" );
        }

        $log->debug( "leaving" );

        return 1;
    }

    # never reached
    $log->logdie( "this should never happen" );
}


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 - @_");

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

        # 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 {
            unless ($rows == 2) {
                $log->logdie( "affected row count is $rows instead of 2" );
            }
        }
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database error: $@");
    }

    eval { 
        $db->commit;
    };
    if ($@) {
        $db->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 - @_");

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

        # 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" );
        }
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database error: $@");
    }

    my $row = $query->fetchrow_hashref;
    my $value = $row->{ 'value' };
    $query->finish;

    $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 - @_");

    my $query;
    eval {
        $query = $db->prepare_cached( $sql->list_object_xattr );
        # ext_id
        my $rows = $query->execute($key);
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database error: $@");
    }

    my @xattrs;
    while (my $row = $query->fetchrow_hashref) {
        push @xattrs, $row->{ 'name' };
    }

    $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 - @_");

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

        # 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" );
        }
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database error: $@");
    }

    eval { 
        $db->commit;
    };
    if ($@) {
        $db->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 - @_" );

    my $query;

    eval {
        $query = $db->prepare_cached( $sql->find_objects );
        $query->execute( $pattern );
    };
    $log->logdie( "database error: $@" ) if $@;

    my @keys;

    while ( my $row = $query->fetchrow_hashref ) {
        my $key = $row->{ 'ext_id' };
        push @keys, $key if $key;
    }

    $log->debug( "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,
            callbacks   => {
                # check that the volume name requested is valid
                'is valid volume name' => sub { $self->_is_valid_volume_name($_[0]) },
            },
            optional    => 1,
        },
    );

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

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

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

    my @locations;

    while (my $row = $query->fetchrow_hashref) {
        my $instance = $row->{ 'uri' };
        push @locations, $instance if $instance;
    }

    # 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 ) {
                $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 );
            
            # if we affected something other then one row 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 );

            # if we affected something other then two rows something very bad
            # has happened
            # this will have to be changed in order to support hardlinks
            unless ( $rows == 2 ) {
                $log->logdie( "affected row count is $rows instead of 2" );
            }
        }

        $db->commit;
    };
    if ( $@ ) {
        $db->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 - @_");

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

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

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

    $log->debug("leaving");

    return $stat;
}


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 = shift;

    my ($vol_id, $vol_path);
    eval {
        # TODO cache this?
        my $query;
        my $rows;
        if ( $name ) {
            $query = $db->prepare_cached( $sql->get_storage_volume_byname );
            # %free, name, avaiable, allocate
            $rows = $query->execute(0.95, $name, 1, 1);
            unless ($rows > 0) {
                $query->finish;
                $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_path, $free) = $query->fetchrow_array;
        $query->finish;
    };
    if ($@) {
        $db->rollback;
        $log->logdie("database error: $@");
    }

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

    $log->debug( "leaving" );

    return ($vol_id, $vol_path);
}


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

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

    my $ext_id;
    eval {
        my $query = $db->prepare_cached( $sql->get_object ); 
        $query->execute($key);
        ($ext_id) = $query->fetchrow_array;
        $query->finish;
    };
    if ($@) {
        $db->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 ($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;
    };
    if ($@) {
        $db->rollback;
        $log->logdie( "database error: $@" );
    }

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

    return;
}


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__
