File Coverage

blib/lib/FlatFile/DataStore.pm

Criterion	Covered	Total	%
statement	728	777	93.6
branch	304	378	80.4
condition	23	40	57.5
subroutine	87	88	98.8
pod	36	72	50.0
total	1178	1355	86.9

#---------------------------------------------------------------------
  package FlatFile::DataStore;
#---------------------------------------------------------------------

=head1 NAME

FlatFile::DataStore - Perl module that implements a flatfile
datastore.

=head1 SYNOPSYS

 use FlatFile::DataStore;

 # new datastore object

 my $dir  = "/my/datastore/directory";
 my $name = "dsname";
 my $ds   = FlatFile::DataStore->new( { dir => $dir, name => $name } );

 # create a record

 my $record_data = "This is a test record.";
 my $user_data   = "Test1";
 my $record = $ds->create( {
     data => \$record_data,
     user => $user_data,
     } );
 my $record_number = $record->keynum;

 # retrieve it

 $record = $ds->retrieve( $record_number );

 # update it

 $record->data( "Updating the test record." );
 $record = $ds->update( $record );

 # delete it

 $record = $ds->delete( $record );

 # get its history

 my @records = $ds->history( $record_number );

=head1 DESCRIPTION

FlatFile::DataStore implements a simple flatfile datastore.  When you
create (store) a new record, it is appended to the flatfile.  When you
update an existing record, the existing entry in the flatfile is
flagged as updated, and the updated record is appended to the
flatfile.  When you delete a record, the existing entry is flagged as
deleted, and a "delete record" is I to the flatfile.

The result is that all versions of a record are retained in the
datastore, and running a history will return all of them.  Another
result is that each record in the datastore represents a transaction:
create, update, or delete.

Methods support the following actions:

 - create
 - retrieve
 - update
 - delete
 - history

Additionally, FlatFile::DataStore::Utils provides the
methods

 - validate
 - migrate

and others.

See FlatFile::DataStore::Tiehash for a tied interface.

=head1 VERSION

FlatFile::DataStore version 1.03

=cut

our $VERSION = '1.03';

use 5.008003;
use strict;
use warnings;

use URI::Escape;
use File::Path;
use Fcntl qw(:DEFAULT :flock);
use Digest::MD5 qw(md5_hex);
use Carp;

use FlatFile::DataStore::Preamble;
use FlatFile::DataStore::Record;
use FlatFile::DataStore::Toc;
use Math::Int2Base qw( base_chars int2base base2int );

use Data::Omap qw( :ALL );
sub untaint;

#---------------------------------------------------------------------
# globals:

my %Preamble = qw(
    indicator   1
    transind    1
    date        1
    transnum    1
    keynum      1
    reclen      1
    thisfnum    1
    thisseek    1
    prevfnum    1
    prevseek    1
    nextfnum    1
    nextseek    1
    user        1
    );

my %Optional = qw(
    dirmax      1
    dirlev      1
    tocmax      1
    keymax      1
    prevfnum    1
    prevseek    1
    nextfnum    1
    nextseek    1
    userdata    1
    );

# attributes that we generate (vs. user-supplied)
my %Generated = qw(
    uri         1
    crud        1
    userlen     1
    dateformat  1
    specs       1
    regx        1
    preamblelen 1
    fnumlen     1
    fnumbase    1
    translen    1
    transbase   1
    keylen      1
    keybase     1
    toclen      1
    datamax     1
    tocs        1
    );

# all attributes, including some more user-supplied ones
my %Attrs = ( %Preamble, %Optional, %Generated, qw(
    name        1
    dir         1
    desc        1
    recsep      1
    ) );

my $Ascii_chars = qr/^[ -~]+$/;  # i.e., printables

#---------------------------------------------------------------------

=head1 CLASS METHODS

=head2 FlatFile::DataStore->new();

Constructs a new FlatFile::DataStore object.

Accepts hash ref giving values for C and C.

 my $ds = FlatFile::DataStore->new(
     { dir  => $dir,
       name => $name,
     } );

To initialize a new datastore, edit the "$dir/$name.uri" file
and enter a configuration URI (as the only line in the file),
or pass the URI as the value of the C parameter, e.g.,

 my $ds = FlatFile::DataStore->new(
     { dir  => $dir,
       name => $name,
       uri  => join( ";" =>
           "http://example.com?name=$name",
           "desc=My%20Data%20Store",
           "defaults=medium",
           "user=8-%20-%7E",
           "recsep=%0A",
           ),
     } );

(See URI Configuration below.)

Also accepts a C parameter, which sets the default user
data for this instance, e.g.,

 my $ds = FlatFile::DataStore->new(
     { dir  => $dir,
       name => $name,
       userdata => ':',
     } );

Returns a reference to the FlatFile::DataStore object.

=cut

sub new {
    my( $class, $parms ) = @_;

    my $self = bless {}, $class;

    $self = $self->init( $parms ) if $parms;  # $self could change ...
    return $self;
}

#---------------------------------------------------------------------
#
# =head2 init(), called by new() to initialize a datastore object
#
# Parms (from hash ref):
#
#     dir  ... the directory where the datastore lives
#     name ... the name of the datastore
#     uri  ... a uri to be used to configure the datastore
#
# If dir/name.uri exists, init() will load its values.
# If uri is passed in, it will be used to initialize the datastore
# only if:
#
#     1) there isn't a .uri file, or
#     2) the .uri file is one line long, or
#     3) the .uri file has more lines (4) but no data files exist yet
#
# Private method.
#
# =cut
#

sub init {
    my( $self, $parms ) = @_;

    my $dir  = $parms->{'dir'};
    my $name = $parms->{'name'};

    croak qq/Need "dir" and "name"/
        unless defined $dir and defined $name;

    croak qq/Directory doesn't exist: $dir/
        unless -d $dir;

    $self->dir( $dir );
    $self->name( $name );

    # uri file may be
    # - one line: just the uri, or
    # - four lines: uri, object, uri_md5, object_md5
    #
    # if new_uri and uri file has
    # - one line ... new_uri can replace old one
    # - four lines (and new_uri is different) ...
    #   new_uri can replace the old uri (and object)
    #   but only if there aren't any data files yet

    my $new_uri = $parms->{'uri'};

    my $uri_file = "$dir/$name.uri";
    my( $uri, $obj, $uri_md5, $obj_md5 );

    if( -e $uri_file ) {
        my @lines = $self->read_file( $uri_file ); chomp @lines;

        if( @lines == 4 ) {
            ( $uri, $obj, $uri_md5, $obj_md5 ) = @lines;

            croak qq/URI MD5 check failed/    unless $uri_md5 eq md5_hex( $uri );
            croak qq/Object MD5 check failed/ unless $obj_md5 eq md5_hex( $obj );

            # new uri ok only if no data has been added yet
            if( $new_uri         and
                $new_uri ne $uri and
                not -e $self->which_datafile( 1 ) ) {
                    $uri = $new_uri;
            }
            else {
                untaint trusted => $obj;
                $self = eval $obj;  # note: *new* $self

                croak qq/Problem with URI file, $uri_file: $@/ if $@;

                $self->dir( $dir );  # dir not in object
            }
        }
        elsif( @lines == 1 ) {
            $uri = $new_uri || shift @lines;
        }
        else {
            croak qq/Invalid URI file: $uri_file/;
        }
    }
    else {
        $uri = $new_uri;
    }

    # if there isn't an object, the datastore hasn't been
    # initialized yet, so if we have a uri (either passed in
    # or read from the uri file, let's initialize it
    # (we could have an instance that only contains name and dir)

    if( !$obj and $uri ) {

        $self->uri( $uri );

        # Note: 'require', not 'use'.  This isn't
        # a "true" module -- we're just bringing in
        # some more FlatFile::DataStore methods.

        require FlatFile::DataStore::Initialize;

        my $uri_parms = $self->burst_query( \%Preamble );
        for my $attr ( keys %$uri_parms ) {

            croak qq/Unrecognized parameter: $attr/ unless $Attrs{ $attr };

            # (note: using $attr as method name here)
            $self->$attr( $uri_parms->{ $attr } );
        }

        # check that all fnums and seeks are the same ...
        #
        # (note: prevfnum, prevseek, nextfnum, and nextseek are
        # optional, but if you have one of them, you must have
        # all four, so checking for one of them here, i.e.,
        # prevfnum, is enough)

        if( $self->prevfnum ) {

            croak qq/fnum parameters differ/
                unless $self->thisfnum eq $self->prevfnum and
                       $self->thisfnum eq $self->nextfnum;

            croak qq/seek parameters differ/
                unless $self->thisseek eq $self->prevseek and
                       $self->thisseek eq $self->nextseek;

        }

        # now for some generated attributes ...
        my( $len, $base );

        # (we can use thisfnum because all fnums are the same)
        ( $len, $base ) = split /-/, $self->thisfnum;
        $self->fnumlen(    $len                        );
        $self->fnumbase(   $base                       );

        ( $len, $base ) = split /-/, $self->transnum;
        $self->translen(   $len                        );
        $self->transbase(  $base                       );

        ( $len, $base ) = split /-/, $self->keynum;
        $self->keylen(     $len                        );
        $self->keybase(    $base                       );

        $self->dateformat( (split /-/, $self->date)[1] );
        $self->regx(       $self->make_preamble_regx   );
        $self->crud(       $self->make_crud            );
        $self->tocs(       {}                          );
        $self->dir(        $dir                        );  # dir not in uri

        $self->toclen( 10          +  # blanks between parts
            3 *    $self->fnumlen  +  # datafnum, tocfnum, keyfnum
            2 *    $self->keylen   +  # numrecs keynum
            6 *    $self->translen +  # transnum and cruds
            length $self->recsep );

        # (note: we can use thisseek because all seeks are the same)
        ( $len, $base ) = split /-/, $self->thisseek;
        my $maxnum = substr( base_chars( $base ), -1) x $len;
        my $maxint = base2int $maxnum, $base;

        # if we give a datamax, it can't be larger than maxint
        if( my $max = $self->datamax ) {
            $self->datamax( convert_max( $max ) );
            if( $self->datamax > $maxint ) {

                croak join '' =>
                    "datamax too large: (", $self->datamax, ") ",
                    "thisseek is ", $self->thisseek,
                    " so maximum datamax is $maxnum base-$base ",
                    "(decimal: $maxint)";
            }
        }
        else {
            $self->datamax( $maxint );
        }

        if( my $max = $self->dirmax ) {
            $self->dirmax( convert_max( $max ) );
            $self->dirlev( 1 ) unless $self->dirlev;
        }

        if( my $max = $self->keymax ) {
            $self->keymax( convert_max( $max ) );
        }

        if( my $max = $self->tocmax ) {
            $self->tocmax( convert_max( $max ) );
        }

        if( my $user = $self->user ) {
            $self->userlen( (split /-/, $user)[0] );
        }

        for my $attr ( keys %Attrs ) {
            croak qq/Uninitialized attribute: $attr/
                if not $Optional{ $attr } and not defined $self->$attr;
        }

        $self->initialize;
    }

    for( $parms->{'userdata'} ) {
        $self->userdata( $_ ) if defined;
    }

    return $self;  # this is either the same self or a new self
}

#---------------------------------------------------------------------

=head1 OBJECT METHODS, Record Processing (CRUD)

=head2 create( $record )

 or create( { data => \$record_data, user => $user_data } )
 or create( { record => $record[, data => \$record_data][, user => $user_data] } )

Creates a record. If the parameter is a record object,
the record data and user data will be gotten from it.
Otherwise, if the parameter is a hash reference, the
expected keys are:

 - record => FlatFile::DataStore::Record object
 - data => string or scalar reference
 - user => string

If no record is passed, both 'data' and 'user' are required.
Otherwise, if a record is passed, the record data and user
data will be gotten from it unless one or both are explicitly
provided.

Returns a Flatfile::DataStore::Record object.

Note: the record data (but not the user data) is stored in the
FF::DS::Record object as a scalar reference.  This is done for
efficiency in the cases where the record data may be very large.
Likewise, the data parm passed to create() may be a scalar
reference.

=cut

sub create {
    my $self = shift;
    my( $data_ref, $user_data ) = $self->normalize_parms( @_ );

    # get next keynum
    #   (we don't call nextkeynum(), because we need the
    #   $top_toc object for other things, too)

    my $top_toc = $self->new_toc( { int => 0 } );
    my $keyint  = $top_toc->keynum + 1;
    my $keylen  = $self->keylen;
    my $keybase = $self->keybase;
    my $keynum  = int2base $keyint, $keybase, $keylen;

    croak qq/Database exceeds configured size, keynum too long: $keynum/
        if length $keynum > $keylen;

    # get keyfile
    #   need to lock files before getting seek positions
    #   want to lock keyfile before datafile

    my( $keyfile, $keyfint ) = $self->keyfile( $keyint );
    my $keyfh                = $self->locked_for_write( $keyfile );
    my $keyseek              = -s $keyfile;  # seekpos into keyfile

    # get datafile ($datafnum may increment)
    my $datafnum = $top_toc->datafnum || 1;  # (||1 only in create)
    $datafnum    = int2base $datafnum, $self->fnumbase, $self->fnumlen;
    my $reclen   = length $$data_ref;

    my $datafile;
    ( $datafile, $datafnum ) = $self->datafile( $datafnum, $reclen );
    my $datafh               = $self->locked_for_write( $datafile );
    my $dataseek             = -s $datafile;  # seekpos into datafile

    # get next transaction number
    my $transint = $self->nexttransnum( $top_toc );

    # make new record
    my $record = $self->new_record( {
        data     => $data_ref,
        preamble => {
            indicator => $self->crud->{'create'},
            transind  => $self->crud->{'create'},
            date      => now( $self->dateformat ),
            transnum  => $transint,
            keynum    => $keyint,
            reclen    => $reclen,
            thisfnum  => $datafnum,
            thisseek  => $dataseek,
            user      => $user_data,
            } } );

    # write record to datafile
    my $preamble = $record->preamble_string;
    my $dataline = $preamble . $$data_ref . $self->recsep;
    $self->write_bytes( $datafh, $dataseek, \$dataline );

    # write preamble to keyfile
    $self->write_bytes( $keyfh, $keyseek, \($preamble . $self->recsep) );

    # update table of contents (toc) file
    my $toc = $self->new_toc( { num => $datafnum } );

    # (note: datafnum and tocfnum are set in toc->new)
    $toc->keyfnum(   $keyfint          );
    $toc->keynum(    $keyint           );
    $toc->transnum(  $transint         );
    $toc->create(    $toc->create  + 1 );
    $toc->numrecs(   $toc->numrecs + 1 );
    $toc->write_toc( $toc->datafnum    );

    # update top toc
    $top_toc->datafnum( $toc->datafnum        );
    $top_toc->keyfnum(  $toc->keyfnum         );
    $top_toc->tocfnum(  $toc->tocfnum         );
    $top_toc->keynum(   $toc->keynum          );
    $top_toc->transnum( $toc->transnum        );
    $top_toc->create(   $top_toc->create  + 1 );
    $top_toc->numrecs(  $top_toc->numrecs + 1 );

    $top_toc->write_toc( 0 );

    close $datafh or die "Can't close $datafile: $!";
    close $keyfh  or die "Can't close $keyfile: $!";

    return $record;
}

#---------------------------------------------------------------------

=head2 retrieve( $num[, $pos] )

Retrieves a record.  The parm C<$num> may be one of

 - a key number, i.e., record sequence number
 - a file number

The parm C<$pos> is required if C<$num> is a file number.

Here's why: When $num is a record key sequence number (key number), a
preamble is retrieved from the datastore key file.  In that preamble
is the file number and seek position where the record data may be
gotten.  Otherwise, when $num is a file number, the application (you)
must supply the seek position into that file.  Working from an array
of record history is the most likely time you would do this.

Returns a Flatfile::DataStore::Record object.

=cut

sub retrieve {
    my( $self, $num, $pos ) = @_;

    for( $num ) {
        croak qq/Not a number: '$_'/ unless m{^ [0-9]+ $}x;
    }

    my $fnum;
    my $seekpos;
    my $keystring;

    if( defined $pos ) {
        for( $pos ) {
            croak qq/Not a number: '$_'/ unless m{^ [0-9]+ $}x;
        }
        $fnum    = $num;
        $seekpos = $pos;
    }
    else {
        my $keynum  = $num;
        my $recsep  = $self->recsep;
        my $keyseek = $self->keyseek( $keynum );

        my $keyfile = $self->keyfile( $keynum );
        my $keyfh   = $self->locked_for_read( $keyfile );

        my $trynum  = $self->lastkeynum;

        croak qq/Record doesn't exist: $keynum/ if $keynum > $trynum;

        $keystring = $self->read_preamble( $keyfh, $keyseek );
        close $keyfh or die "Can't close $keyfile: $!";

        my $parms  = $self->burst_preamble( $keystring );

        $fnum    = $parms->{'thisfnum'};
        $seekpos = $parms->{'thisseek'};
    }

    my $datafile = $self->which_datafile( $fnum );
    my $datafh   = $self->locked_for_read( $datafile );
    my $record   = $self->read_record( $datafh, $seekpos );
    close $datafh or die "Can't close $datafile: $!";

    # if we got the record via key file, check that preambles match
    if( $keystring ) {
        my $string = $record->preamble_string;

        croak qq/Mismatch: "$string" ne "$keystring"/ if $string ne $keystring;
    }

    return $record;
}

#---------------------------------------------------------------------

=head2 retrieve_preamble( $keynum )

Retrieves a preamble.  The parm C<$keynum> is a key number, i.e.,
record sequence number

Returns a Flatfile::DataStore::Preamble object.

This method allows getting information about the record, e.g., if
it's deleted, what's in the user data, etc., without the overhead of
retrieving the full record data.

=cut

sub retrieve_preamble {
    my( $self, $keynum ) = @_;

    for( $keynum ) {
        croak qq/Not a number: '$_'/ unless m{^ [0-9]+ $}x;
    }

    my $keyseek = $self->keyseek( $keynum );
    my $keyfile = $self->keyfile( $keynum );
    my $keyfh   = $self->locked_for_read( $keyfile );

    my $trynum  = $self->lastkeynum;

    croak qq/Record doesn't exist: $keynum/ if $keynum > $trynum;

    my $keystring = $self->read_preamble( $keyfh, $keyseek );
    close $keyfh or die "Can't close $keyfile: $!";

    my $preamble  = $self->new_preamble( { string => $keystring } );

    return $preamble;
}

#---------------------------------------------------------------------

=head2 locate_record_data( $num[, $pos] )

Rather than retrieving a record, this subroutine positions you at the
record data in the data file.  This might be handy if, for example,
the record data is text, and you just want part of it.  You can scan
the data and get what you want without having to read the entire
record.  Or the data might be XML and you could parse it using SAX
without reading it all into memory.

The parm C<$num> may be one of

 - a key number, i.e., record sequence number
 - a file number

The parm C<$pos> is required if C<$num> is a file number.  See
retrieve() above for why.

Returns a list containing the file handle (which is already locked
for reading in binmode), the seek position, and the record length.

You will be positioned at the seek position, so you could begin
reading data, e.g., via C<< <$fh> >>:

    my( $fh, $pos, $len ) = $ds->locate_record_data( $keynum );
    my $got;
    while( <$fh> ) {
        last if ($got += length) > $len;  # in case we read the recsep
        # [do something with $_ ...]
        last if $got == $len;
    }
    close $fh;

The above loop assumes you know each line of the data ends in a
newline.  Also keep in mind that the file is opened in binmode,
so you will be reading bytes (octets), not necessarily characters.
Decoding these octets is up to you.

XXX ("opened in binmode"?) does that make the example wrong
    wrt non-unix OS's

=cut

sub locate_record_data {
    my( $self, $num, $pos ) = @_;

    for( $num ) {
        croak qq/Not a number: '$_'/ unless m{^ [0-9]+ $}x;
    }

    my $fnum;
    my $seekpos;
    my $keystring;
    my $reclen;

    if( defined $pos ) {
        for( $pos ) {
            croak qq/Not a number: '$_'/ unless m{^ [0-9]+ $}x;
        }
        $fnum    = $num;
        $seekpos = $pos;
    }
    else {
        my $keynum  = $num;
        my $recsep  = $self->recsep;
        my $keyseek = $self->keyseek( $keynum );

        my $keyfile = $self->keyfile( $keynum );
        my $keyfh   = $self->locked_for_read( $keyfile );

        my $trynum  = $self->lastkeynum;

        croak qq/Record doesn't exist: $keynum/ if $keynum > $trynum;

        $keystring = $self->read_preamble( $keyfh, $keyseek );
        close $keyfh or die "Can't close $keyfile: $!";

        my $parms  = $self->burst_preamble( $keystring );

        $fnum    = $parms->{'thisfnum'};
        $seekpos = $parms->{'thisseek'};
        $reclen  = $parms->{'reclen'};
    }

    my $datafile = $self->which_datafile( $fnum );
    my $datafh   = $self->locked_for_read( $datafile );
    my $preamble = $self->read_preamble( $datafh, $seekpos );

    # if we got the record via key file, check that preambles match
    if( $keystring ) {

        croak qq/Mismatch: "$preamble" ne "$keystring"/
            if $preamble ne $keystring;
    }

    # if not via key file, we still need the record length
    else {
        my $parms  = $self->burst_preamble( $preamble );
        $reclen  = $parms->{'reclen'};
    }

    $seekpos += $self->preamblelen;  # skip to record data

    sysseek $datafh, $seekpos, 0 or

        croak qq/Can't seek to $seekpos in $datafile: $!/;

    return $datafh, $seekpos, $reclen;
}

#---------------------------------------------------------------------

=head2 update( $record )

 or update( { string => $preamble_string, data => \$record_data, user => $user_data } )
 or update( { preamble => $preamble_obj, data => \$record_data, user => $user_data } )
 or update( { record => $record_obj
    [, preamble => $preamble_obj]
    [, string   => $preamble_string]
    [, data     => \$record_data]
    [, user     => $user_data] } )

Updates a record. If the parameter is a record object,
the preamble, record data, and user data will be gotten
from it.  Otherwise, if the parameter is a hash reference,
the expected keys are:

 - record   => FlatFile::DataStore::Record object
 - preamble => FlatFile::DataStore::Preamble object
 - string   => a preamble string (the string attribute of a preamble object)
 - data     => string or scalar reference
 - user     => string

If no record is passed, 'preamble' (or 'string'), 'data', and
'user' are required.  Otherwise, if a record is passed, the
preamble, record data and user data will be gotten from it
unless any of them are explicitly provided.

Returns a Flatfile::DataStore::Record object.

=cut

sub update {
    my $self = shift;
    my( $data_ref, $user_data, $pr_obj ) = $self->normalize_parms( @_ );

    croak qq/Must have at least a previous preamble for update/
        unless $pr_obj;

    my $prevnext = $self->prevfnum;  # boolean

    my $prevpreamble = $pr_obj->string;
    my $keyint       = $pr_obj->keynum;
    my $prevind      = $pr_obj->indicator;
    my $prevfnum     = $pr_obj->thisfnum;
    my $prevseek     = $pr_obj->thisseek;

    # update is okay for these:
    my $create = $self->crud->{'create'};
    my $update = $self->crud->{'update'};
    my $delete = $self->crud->{'delete'};

    croak qq/update not allowed: $prevind/
        unless $prevind =~ /[\Q$create$update$delete\E]/;

    # get keyfile
    #   need to lock files before getting seek positions
    #   want to lock keyfile before datafile

    my( $keyfile, $keyfint ) = $self->keyfile( $keyint );
    my $keyfh                = $self->locked_for_write( $keyfile );
    my $keyseek              = $self->keyseek( $keyint );

    my $try = $self->read_preamble( $keyfh, $keyseek );

    croak qq/Mismatch: "$try" ne "$prevpreamble"/ unless $try eq $prevpreamble;

    # get datafile ($datafnum may increment)
    my $top_toc  = $self->new_toc( { int => 0 } );
    my $datafnum = int2base $top_toc->datafnum, $self->fnumbase, $self->fnumlen;
    my $reclen   = length $$data_ref;

    my $datafile;
    ( $datafile, $datafnum ) = $self->datafile( $datafnum, $reclen );
    my $datafh               = $self->locked_for_write( $datafile );
    my $dataseek             = -s $datafile;  # seekpos into datafile

    # get next transaction number
    my $transint = $self->nexttransnum( $top_toc );

    # make new record
    my $preamble_hash = {
        indicator => $update,
        transind  => $update,
        date      => now( $self->dateformat ),
        transnum  => $transint,
        keynum    => $keyint,
        reclen    => $reclen,
        thisfnum  => $datafnum,
        thisseek  => $dataseek,
        user      => $user_data,
        };
    if( $prevnext ) {
        $preamble_hash->{'prevfnum'} = $prevfnum;
        $preamble_hash->{'prevseek'} = $prevseek;
    }
    my $record = $self->new_record( {
        data     => $data_ref,
        preamble => $preamble_hash,
        } );

    # write record to datafile
    my $preamble = $record->preamble_string;
    my $dataline = $preamble . $$data_ref . $self->recsep;
    $self->write_bytes( $datafh, $dataseek, \$dataline );

    # write preamble to keyfile (recsep there already)
    $self->write_bytes( $keyfh, $keyseek, \$preamble );

    # update the old preamble
    if( $prevnext ) {
        $prevpreamble = $self->update_preamble( $prevpreamble, {
            indicator => $self->crud->{ 'oldupd' },
            nextfnum  => $datafnum,
            nextseek  => $dataseek,
            } );
        my $prevdatafile = $self->which_datafile( $prevfnum );
        if( $prevdatafile eq $datafile ) {
            $self->write_bytes( $datafh, $prevseek, \$prevpreamble );
        }
        else {
            my $prevdatafh = $self->locked_for_write( $prevdatafile );
            $self->write_bytes( $prevdatafh, $prevseek, \$prevpreamble );
            close $prevdatafh or die "Can't close $prevdatafile: $!";
        }
    }

    # update table of contents (toc) file
    my $toc = $self->new_toc( { num => $datafnum } );

    # note: datafnum and tocfnum are set in toc->new
    $toc->keyfnum(  $top_toc->keyfnum );  # keep last nums going
    $toc->keynum(   $top_toc->keynum  );
    $toc->transnum( $transint         );
    $toc->update(   $toc->update  + 1 );
    $toc->numrecs(  $toc->numrecs + 1 );

    # was the previous record in another data file?
    if( $prevnext ) {
        if( $prevfnum ne $datafnum ) {
            my $prevtoc = $self->new_toc( { num => $prevfnum } );
            $prevtoc->oldupd(    $prevtoc->oldupd  + 1 );
            $prevtoc->numrecs(   $prevtoc->numrecs - 1 ) if $prevind ne $delete;
            $prevtoc->write_toc( $prevtoc->datafnum    );
        }
        else {
            $toc->oldupd(  $toc->oldupd  + 1 );
            $toc->numrecs( $toc->numrecs - 1 ) if $prevind ne $delete;
        }
    }
    else {
        $toc->numrecs( $toc->numrecs - 1 ) if $prevind ne $delete;
    }

    $toc->write_toc( $toc->datafnum );

    # update top toc
    $top_toc->datafnum( $toc->datafnum        );
    $top_toc->tocfnum(  $toc->tocfnum         );
    $top_toc->transnum( $toc->transnum        );
    $top_toc->update(   $top_toc->update  + 1 );
    $top_toc->oldupd(   $top_toc->oldupd  + 1 ) if $prevnext;
    $top_toc->numrecs(  $top_toc->numrecs + 1 ) if $prevind eq $delete;

    $top_toc->write_toc( 0 );

    close $datafh or die "Can't close $datafile: $!";
    close $keyfh  or die "Can't close $keyfile: $!";

    return $record;
}

#---------------------------------------------------------------------

=head2 delete( $record )

 or delete( { string => $preamble_string, data => \$record_data, user => $user_data } )
 or delete( { preamble => $preamble_obj, data => \$record_data, user => $user_data } )
 or delete( { record => $record_obj
    [, preamble => $preamble_obj]
    [, string   => $preamble_string]
    [, data     => \$record_data]
    [, user     => $user_data] } )

Deletes a record.  The parameters are the same as for update().

Returns a Flatfile::DataStore::Record object.

=cut

sub delete {
    my $self = shift;
    my( $data_ref, $user_data, $pr_obj ) = $self->normalize_parms( @_ );

    croak qq/Must have at least a previous preamble for delete/
        unless $pr_obj;

    my $prevnext = $self->prevfnum;  # boolean

    my $prevpreamble = $pr_obj->string;
    my $keyint       = $pr_obj->keynum;
    my $prevind      = $pr_obj->indicator;
    my $prevfnum     = $pr_obj->thisfnum;
    my $prevseek     = $pr_obj->thisseek;

    # delete is okay for these:
    my $create = $self->crud->{'create'};
    my $update = $self->crud->{'update'};

    croak qq/delete not allowed: $prevind/
        unless $prevind =~ /[\Q$create$update\E]/;

    # get keyfile
    # need to lock files before getting seek positions
    # want to lock keyfile before datafile

    my( $keyfile, $keyfint ) = $self->keyfile( $keyint );
    my $keyfh                = $self->locked_for_write( $keyfile );
    my $keyseek              = $self->keyseek( $keyint );

    my $try = $self->read_preamble( $keyfh, $keyseek );

    croak qq/Mismatch: "$try" ne "$prevpreamble"/ unless $try eq $prevpreamble;

    # get datafile ($datafnum may increment)
    my $top_toc  = $self->new_toc( { int => 0 } );
    my $datafnum = int2base $top_toc->datafnum, $self->fnumbase, $self->fnumlen;
    my $reclen   = length $$data_ref;

    my $datafile;
    ( $datafile, $datafnum ) = $self->datafile( $datafnum, $reclen );
    my $datafh               = $self->locked_for_write( $datafile );
    my $dataseek             = -s $datafile;  # seekpos into datafile

    # get next transaction number
    my $transint = $self->nexttransnum( $top_toc );

    # make new record
    my $delete = $self->crud->{'delete'};
    my $preamble_hash = {
        indicator => $delete,
        transind  => $delete,
        date      => now( $self->dateformat ),
        transnum  => $transint,
        keynum    => $keyint,
        reclen    => $reclen,
        thisfnum  => $datafnum,
        thisseek  => $dataseek,
        user      => $user_data,
        };
    if( $prevnext ) {
        $preamble_hash->{'prevfnum'} = $prevfnum;
        $preamble_hash->{'prevseek'} = $prevseek;
    }
    my $record = $self->new_record( {
        data     => $data_ref,
        preamble => $preamble_hash,
        } );

    # write record to datafile
    my $preamble = $record->preamble_string;
    my $dataline = $preamble . $$data_ref . $self->recsep;
    $self->write_bytes( $datafh, $dataseek, \$dataline );

    # write preamble to keyfile (recsep there already)
    $self->write_bytes( $keyfh, $keyseek, \$preamble );

    # update the old preamble
    if( $prevnext ) {
        $prevpreamble = $self->update_preamble( $prevpreamble, {
            indicator => $self->crud->{ 'olddel' },
            nextfnum  => $datafnum,
            nextseek  => $dataseek,
            } );
        my $prevdatafile = $self->which_datafile( $prevfnum );
        if( $prevdatafile eq $datafile ) {
            $self->write_bytes( $datafh, $prevseek, \$prevpreamble );
        }
        else {
            my $prevdatafh = $self->locked_for_write( $prevdatafile );
            $self->write_bytes( $prevdatafh, $prevseek, \$prevpreamble );
            close $prevdatafh or die "Can't close $prevdatafile: $!";
        }
    }

    # update table of contents (toc) file
    my $toc = $self->new_toc( { num => $datafnum } );

    # note: datafnum and tocfnum are set in toc->new
    $toc->keyfnum(  $top_toc->keyfnum );  # keep last nums going
    $toc->keynum(   $top_toc->keynum  );
    $toc->transnum( $transint         );
    $toc->delete(   $toc->delete + 1  );

    # was the previous record in another data file?
    if( $prevnext ) {
        if( $prevfnum ne $datafnum ) {
            my $prevtoc = $self->new_toc( { num => $prevfnum } );
            $prevtoc->olddel(    $prevtoc->olddel  + 1 );
            $prevtoc->numrecs(   $prevtoc->numrecs - 1 );
            $prevtoc->write_toc( $prevtoc->datafnum    );
        }
        else {
            $toc->olddel(  $toc->olddel  + 1 );
            $toc->numrecs( $toc->numrecs - 1 );
        }
    }
    else {
        $toc->numrecs( $toc->numrecs - 1 );
    }

    $toc->write_toc( $toc->datafnum );

    # update top toc
    $top_toc->datafnum( $toc->datafnum        );
    $top_toc->tocfnum(  $toc->tocfnum         );
    $top_toc->transnum( $toc->transnum        );
    $top_toc->delete(   $top_toc->delete  + 1 );
    $top_toc->olddel(   $top_toc->olddel  + 1 ) if $prevnext;
    $top_toc->numrecs(  $top_toc->numrecs - 1 );

    $top_toc->write_toc( 0 );

    close $datafh or die "Can't close $datafile: $!";
    close $keyfh  or die "Can't close $keyfile: $!";

    return $record;
}

#---------------------------------------------------------------------
#
# =head2 normalize_parms( $parms )
#
# Parses parameters for create(), update(), and delete()
#
# If the parameter is a record object, then the preamble, record data,
# and user data will be gotten from it.
#
# Otherwise, if the parameter is a hash reference, the expected keys
# are:
#
#     - record   => FlatFile::DataStore::Record object
#     - preamble => FlatFile::DataStore::Preamble object
#     - string   => a preamble string (the string attribute of a preamble object)
#     - data     => string or scalar reference
#     - user     => string
#
# Returns record data (scalar ref), user data, preamble object
#
# Note that create() ignores the returned preamble, but update() and
# delete() do not.
#
# Private method.
#
# =cut
#

sub normalize_parms {
    my( $self, $parms ) = @_;

    croak qq/Bad call/ unless $parms;

    my( $data_ref, $user_data, $preamble );

    my $reftype = ref $parms;
    if( $reftype =~ /Record/ ) {
        $data_ref  = $parms->dataref;
        $user_data = $parms->user;
        $preamble  = $parms->preamble;
    }
    elsif( $reftype eq "HASH" ) {
        for( $parms->{'data'} ) {
            if( ref ) { $data_ref = $_ }
            else      { $data_ref = \$_ if defined }
        }
        for( $parms->{'user'} )  {
            $user_data = $_ if defined;
        }
        for( $parms->{'string'} ) {
            $preamble = $self->new_preamble( { string => $_ } )
                if defined;
        }
        for( $parms->{'preamble'} ) {
            $preamble = $_ if defined;
        }
        for( $parms->{'record'} ) {
            last unless defined;
            $data_ref  = $_->dataref  unless $data_ref;
            $user_data = $_->user     unless defined $user_data;
            $preamble  = $_->preamble unless $preamble;
        }
    }
    else {
        croak qq/Parameter must be a hashref or a record object/;
    }
    croak qq/No record data/ unless $data_ref;

    return $data_ref, $user_data, $preamble;
}

#---------------------------------------------------------------------

=head2 exists()

Tests if a datastore exists.  Currently, a datastore "exists" if there
is a .uri file -- whether the file is valid or not.

May be called on a datastore object, e.g.,

    $ds->exists()

Or may be called as a class method, e.g.,

    FlatFile::DataStore->exists({
        name => 'example',
        dir  => '/dbs/example',
        })

If called as a class method, you must pass a hashref that provides
values for 'name' and 'dir'.

=cut

sub exists {
    my( $self, $parms ) = @_;

    my( $dir, $name );

    if( ref $self ) {  # object method
        $dir  = $self->dir;
        $name = $self->name;

        # empty object, so datastore doesn't exist
        return unless $dir and $name;
    }

    else {  # class method

        if( $parms ) {
            $dir  = $parms->{'dir'};
            $name = $parms->{'name'};
        }

        # required for class method
        croak qq/Need dir and name/ unless $dir and $name;
    }

    -e "$dir/$name.uri";  # returned
}

#---------------------------------------------------------------------

=head2 history( $keynum )

Retrieves a record's history.  The parm C<$keynum> is always a key
number, i.e., a record sequence number.

Returns an array of FlatFile::DataStore::Record objects.

The first element of this array is the current record.  The last
element is the original record.  That is, the array is in reverse
chronological order.

=cut

sub history {
    my( $self, $keynum ) = @_;

    for( $keynum ) {
        croak qq/Not a number: '$_'/ unless m{^ [0-9]+ $}x;
    }

    my @history;

    my $rec = $self->retrieve( $keynum );
    push @history, $rec;

    my $prevfnum = $rec->prevfnum;
    my $prevseek = $rec->prevseek;

    while( $prevfnum ) {

        my $rec = $self->retrieve( $prevfnum, $prevseek );
        push @history, $rec;

        $prevfnum = $rec->prevfnum;
        $prevseek = $rec->prevseek;
    }

    return @history;
}

#---------------------------------------------------------------------

=head1 OBJECT METHODS, Accessors

In the specifications below, square braces ([]) denote optional
parameters, not anonymous arrays, e.g., C<[$omap]> indicates that
C<$omap> is optional, instead of implying that you need to pass it
inside an array.

=head2 $ds->specs( [$omap] )

Sets and returns the C attribute value if C<$omap> is given,
otherwise just returns the value.

An 'omap' is an ordered hash as defined in

    http://yaml.org/type/omap.html

and implemented here using Data::Omap.  That is, it's an array of
single-key hashes.  This ordered hash contains the specifications for
constructing and parsing a record preamble as defined in the name.uri
file.

In list context, the value returned is a list of hashrefs.  In scalar
context, the value returned is an arrayref containing the list of
hashrefs.

=cut

sub specs {
    my( $self, $omap ) = @_;
    for( $self->{specs} ) {
        if( $omap ) {

            croak qq/Invalid omap: /.omap_errstr()
                unless omap_is_valid( $omap );

            $_ = $omap;
        }
        return unless defined;
        return @$_ if wantarray;
        return $_;
    }
}

#---------------------------------------------------------------------

=head2 $ds->dir( [$dir] )

Sets and returns the C attribute value if C<$dir> is given,
otherwise just returns the value.

If C<$dir> is given and is a null string, the C object attribute
is removed from the object.  If C<$dir> is not null, the directory
must already exist.  In other words, this module will not create the
directory where the database is to be stored.

=cut

sub dir {
    my( $self, $dir ) = @_;
    if( defined $dir and $dir eq "" ) { delete $self->{dir} }
    else {
        for( $self->{dir} ) {
            if( defined $dir ) {

                croak qq/Directory doesn't exist: $dir/ unless -d $dir;

                $_ = $dir
            }
            return $_;
        }
    }
}

#---------------------------------------------------------------------

=head2 Preamble accessors (from the uri)

The following methods set and return their respective attribute values
if C<$value> is given.  Otherwise, they just return the value.

 $ds->indicator( [$value] );  # length-characters
 $ds->transind(  [$value] );  # length-characters
 $ds->date(      [$value] );  # length-format
 $ds->transnum(  [$value] );  # length-base
 $ds->keynum(    [$value] );  # length-base
 $ds->reclen(    [$value] );  # length-base
 $ds->thisfnum(  [$value] );  # length-base
 $ds->thisseek(  [$value] );  # length-base
 $ds->prevfnum(  [$value] );  # length-base
 $ds->prevseek(  [$value] );  # length-base
 $ds->nextfnum(  [$value] );  # length-base
 $ds->nextseek(  [$value] );  # length-base
 $ds->user(      [$value] );  # length-characters

=head2 Other accessors

 $ds->name(        [$value] ); # from uri, name of datastore
 $ds->desc(        [$value] ); # from uri, description of datastore
 $ds->recsep(      [$value] ); # from uri, character(s)
 $ds->uri(         [$value] ); # full uri as is
 $ds->preamblelen( [$value] ); # length of preamble string
 $ds->toclen(      [$value] ); # length of toc entry
 $ds->keylen(      [$value] ); # length of stored keynum
 $ds->keybase(     [$value] ); # base   of stored keynum
 $ds->translen(    [$value] ); # length of stored transaction number
 $ds->transbase(   [$value] ); # base   of stored transaction number
 $ds->fnumlen(     [$value] ); # length of stored file number
 $ds->fnumbase(    [$value] ); # base   of stored file number
 $ds->userlen(     [$value] ); # format from uri
 $ds->dateformat(  [$value] ); # format from uri
 $ds->regx(        [$value] ); # capturing regx for preamble string
 $ds->datamax(     [$value] ); # maximum bytes in a data file
 $ds->crud(        [$value] ); # hash ref, e.g.,

     {
        create => '+',
        oldupd => '#',
        update => '=',
        olddel => '*',
        delete => '-',
        '+' => 'create',
        '#' => 'oldupd',
        '=' => 'update',
        '*' => 'olddel',
        '-' => 'delete',
     }

 (logical actions <=> symbolic indicators)

=head2 Accessors for optional attributes

 $ds->dirmax(   [$value] );  # maximum files in a directory
 $ds->dirlev(   [$value] );  # number of directory levels
 $ds->tocmax(   [$value] );  # maximum toc entries
 $ds->keymax(   [$value] );  # maximum key entries
 $ds->userdata( [$value] );  # default user data

If no C, directories will keep being added to.

If no C, toc, key, and data files will reside in top-level
directory.  If C is given, C defaults to 1.

If no C, there will be only one toc file, which will grow
indefinitely.

If no C, there will be only one key file, which will grow
indefinitely.

If no C, will default to a null string (padded with spaces)
unless supplied another way.

=cut

sub indicator {for($_[0]->{indicator} ){$_=$_[1]if@_>1;return$_}}
sub transind  {for($_[0]->{transind}  ){$_=$_[1]if@_>1;return$_}}
sub date      {for($_[0]->{date}      ){$_=$_[1]if@_>1;return$_}}
sub transnum  {for($_[0]->{transnum}  ){$_=$_[1]if@_>1;return$_}}
sub keynum    {for($_[0]->{keynum}    ){$_=$_[1]if@_>1;return$_}}
sub reclen    {for($_[0]->{reclen}    ){$_=$_[1]if@_>1;return$_}}
sub thisfnum  {for($_[0]->{thisfnum}  ){$_=$_[1]if@_>1;return$_}}
sub thisseek  {for($_[0]->{thisseek}  ){$_=$_[1]if@_>1;return$_}}

# prevfnum, prevseek, nextfnum, nextseek are optional attributes;
# prevfnum() is set up to avoid autovivification, because it is
# the accessor used to test if these optional attributes are set

sub prevfnum {
    my $self = shift;
    return $self->{prevfnum} = $_[0] if @_;
    return $self->{prevfnum} if exists $self->{prevfnum};
}

sub prevseek  {for($_[0]->{prevseek}  ){$_=$_[1]if@_>1;return$_}}
sub nextfnum  {for($_[0]->{nextfnum}  ){$_=$_[1]if@_>1;return$_}}
sub nextseek  {for($_[0]->{nextseek}  ){$_=$_[1]if@_>1;return$_}}
sub user      {for($_[0]->{user}      ){$_=$_[1]if@_>1;return$_}}

sub name        {for($_[0]->{name}        ){$_=$_[1]if@_>1;return$_}}
sub desc        {for($_[0]->{desc}        ){$_=$_[1]if@_>1;return$_}}
sub recsep      {for($_[0]->{recsep}      ){$_=$_[1]if@_>1;return$_}}
sub uri         {for($_[0]->{uri}         ){$_=$_[1]if@_>1;return$_}}
sub userlen     {for($_[0]->{userlen}     ){$_=$_[1]if@_>1;return$_}}
sub dateformat  {for($_[0]->{dateformat}  ){$_=$_[1]if@_>1;return$_}}
sub regx        {for($_[0]->{regx}        ){$_=$_[1]if@_>1;return$_}}
sub crud        {for($_[0]->{crud}        ){$_=$_[1]if@_>1;return$_}}
sub tocs        {for($_[0]->{tocs}        ){$_=$_[1]if@_>1;return$_}}
sub datamax     {for($_[0]->{datamax}     ){$_=$_[1]if@_>1;return$_}}

sub preamblelen {for($_[0]->{preamblelen} ){$_=0+$_[1]if@_>1;return$_}}
sub toclen      {for($_[0]->{toclen}      ){$_=0+$_[1]if@_>1;return$_}}
sub keylen      {for($_[0]->{keylen}      ){$_=0+$_[1]if@_>1;return$_}}
sub keybase     {for($_[0]->{keybase}     ){$_=0+$_[1]if@_>1;return$_}}
sub translen    {for($_[0]->{translen}    ){$_=0+$_[1]if@_>1;return$_}}
sub transbase   {for($_[0]->{transbase}   ){$_=0+$_[1]if@_>1;return$_}}
sub fnumlen     {for($_[0]->{fnumlen}     ){$_=0+$_[1]if@_>1;return$_}}
sub fnumbase    {for($_[0]->{fnumbase}    ){$_=0+$_[1]if@_>1;return$_}}

# optional (set up to avoid autovivification):

sub dirmax {
    my $self = shift;
    return $self->{dirmax} = $_[0] if @_;
    return $self->{dirmax} if exists $self->{dirmax};
}
sub dirlev {
    my $self = shift;
    return $self->{dirlev} = 0+$_[0] if @_;
    return $self->{dirlev} if exists $self->{dirlev};
}
sub tocmax {
    my $self = shift;
    return $self->{tocmax} = $_[0] if @_;
    return $self->{tocmax} if exists $self->{tocmax};
}
sub keymax {
    my $self = shift;
    return $self->{keymax} = $_[0] if @_;
    return $self->{keymax} if exists $self->{keymax};
}

# default to null string (will be space-padded)
sub userdata {
    my $self = shift;
    return $self->{userdata} = $_[0] if @_;
    return '' unless exists $self->{userdata};
    return $self->{userdata};
}

#---------------------------------------------------------------------
#
# =head2 new_toc( \%parms )
#
# This method is a wrapper for FlatFile::DataStore::Toc->new().
#
# Private method.
#
# =cut
#

sub new_toc {
    my( $self, $parms ) = @_;
    $parms->{'datastore'} = $self;
    FlatFile::DataStore::Toc->new( $parms );
}

#---------------------------------------------------------------------
#
# =head2 new_preamble( \%parms )
#
# This method is a wrapper for FlatFile::DataStore::Preamble->new().
#
# Private method.
#
# =cut
#

sub new_preamble {
    my( $self, $parms ) = @_;
    $parms->{'datastore'} = $self;
    FlatFile::DataStore::Preamble->new( $parms );
}

#---------------------------------------------------------------------
#
# =head2 new_record( \%parms )
#
# This method is a wrapper for FlatFile::DataStore::Record->new().
#
# Private method.
#
# =cut
#

sub new_record {
    my( $self, $parms ) = @_;
    my $preamble = $parms->{'preamble'};
    if( ref $preamble eq 'HASH' ) {  # not an object
        $parms->{'preamble'} = $self->new_preamble( $preamble );
    }
    FlatFile::DataStore::Record->new( $parms );
}

#---------------------------------------------------------------------
#
# =head2 keyfile( $keyint )
#
# Takes an integer that is the record sequence number and returns the
# path to the keyfile where that record's preamble is.
#
# Private method.
#
# =cut
#

sub keyfile {
    my( $self, $keyint ) = @_;

    my $name     = $self->name;
    my $fnumlen  = $self->fnumlen;
    my $fnumbase = $self->fnumbase;

    my $keyfint = 1;
    my $keyfile = $name;

    # get key file number (if any) based on keymax and keyint
    if( my $keymax = $self->keymax ) {
        $keyfint = int( $keyint / $keymax ) + 1;
        my $keyfnum = int2base $keyfint, $fnumbase, $fnumlen;

        croak qq/Database exceeds configured size, keyfnum too long: $keyfnum/
            if length $keyfnum > $fnumlen;

        $keyfile .= ".$keyfnum";
    }

    $keyfile .= ".key";

    # get path based on dirlev (if any), dirmax, and key file number
    if( my $dirlev = $self->dirlev ) {
        my $dirmax = $self->dirmax;
        my $path   = "";
        my $this   = $keyfint;
        for( 1 .. $dirlev ) {
            my $dirint = $dirmax? (int( ( $this - 1 ) / $dirmax ) + 1): 1;
            my $dirnum = int2base $dirint, $fnumbase, $fnumlen;
            $path = $path? "$dirnum/$path": $dirnum;
            $this = $dirint;
        }
        $path = $self->dir . "/$name/key$path";
        mkpath( $path ) unless -d $path;
        $keyfile = "$path/$keyfile";
    }
    else {
        $keyfile = $self->dir . "/$keyfile";
    }

    return ( $keyfile, $keyfint ) if wantarray;
    return $keyfile;

}

#---------------------------------------------------------------------
#
# =head2 datafile(), called by create(), update(), and delete()
#
# Similar to which_datafile(), this method takes a file number
# and returns the path to that datafile.  Unlike which_datafile(),
# this method also takes a record length to check for overflow.
#
# That is, if the record about to be written would make a datafile
# become too large (> datamax), the file number is incremented,
# and the path to that new datafile is returned -- along with the
# new file number.  Calls to datafile() should always take this
# new file number into account.
#
# Will croak if the record is way too big (> datamax) or if the new
# file number is longer than the max length for file numbers.  In
# either case, a new datastore must be configured to handle the
# extra data, and the old datastore must be migrated to it.
#
# Private method.
#
# =cut
#

sub datafile {
    my( $self, $fnum, $reclen ) = @_;

    my $datafile = $self->which_datafile( $fnum );

    # check if we're about to overfill the data file
    # and if so, increment fnum for a new data file

    my $datamax   = $self->datamax;
    my $checksize = $self->preamblelen + $reclen + length $self->recsep;
    my $datasize  = -s $datafile || 0;

    if( $datasize + $checksize > $datamax ) {

        croak qq/Record too long: $checksize > $datamax/
            if $checksize > $datamax;

        my $fnumlen  = $self->fnumlen;
        my $fnumbase = $self->fnumbase;
        $fnum = int2base( 1 + base2int( $fnum, $fnumbase ), $fnumbase, $fnumlen );

        croak qq/Database exceeds configured size, fnum too long: $fnum/
            if length $fnum > $fnumlen;

        $datafile = $self->which_datafile( $fnum );
    }

    return $datafile, $fnum;
}

#---------------------------------------------------------------------
#
# =head2 which_datafile()
#
# Takes a file number and returns the path to that datafile.
#
# Takes into account dirlev and dirmax, if set, and will create
# new directories as needed.
#
# Private method.
#
# =cut
#

sub which_datafile {
    my( $self, $datafnum ) = @_;

    my $name     = $self->name;
    my $datafile = "$name.$datafnum.data";

    # get path based on dirlev, dirmax, and data file number
    if( my $dirlev   = $self->dirlev ) {
        my $fnumlen  = $self->fnumlen;
        my $fnumbase = $self->fnumbase;
        my $dirmax   = $self->dirmax;
        my $path     = "";
        my $this     = base2int $datafnum, $fnumbase;
        for( 1 .. $dirlev ) {
            my $dirint = $dirmax? (int( ( $this - 1 ) / $dirmax ) + 1): 1;
            my $dirnum = int2base $dirint, $fnumbase, $fnumlen;
            $path = $path? "$dirnum/$path": $dirnum;
            $this = $dirint;
        }
        $path = $self->dir . "/$name/data$path";
        mkpath( $path ) unless -d $path;
        $datafile = "$path/$datafile";
    }
    else {
        $datafile = $self->dir . "/$datafile";
    }

    return $datafile;
}

#---------------------------------------------------------------------
#
# =head2 sub all_datafiles(), called by validate utility
#
# Returns an array of paths for all of the data files in the
# datastore.
#
# Private method.
#
# =cut
#

sub all_datafiles {
    my( $self ) = @_;

    my $fnumlen  = $self->fnumlen;
    my $fnumbase = $self->fnumbase;
    my $top_toc  = $self->new_toc( { int => 0 } );
    my $datafint = $top_toc->datafnum;
    my @files;
    for( 1 .. $datafint ) {
        my $datafnum = int2base $_, $fnumbase, $fnumlen;
        push @files, $self->which_datafile( $datafnum );
    }
    return @files;
}

#---------------------------------------------------------------------

=head1 OBJECT METHODS, Other

=head2 howmany( [$regx] )

Returns count of records whose indicators match regx, e.g.,

    $self->howmany( qr/create|update/ );
    $self->howmany( qr/delete/ );
    $self->howmany( qr/oldupd|olddel/ );

If no regx, howmany() returns numrecs from the toc file, which
should give the same number as qr/create|update/.

=cut

sub howmany {
    my( $self, $regx ) = @_;

    my $top_toc = $self->new_toc( { int => 0 } );

    return $top_toc->numrecs unless $regx;

    my $howmany = 0;
    for( qw( create update delete oldupd olddel ) ) {
        $howmany += $top_toc->$_() if /$regx/ }
    return $howmany;
}

#---------------------------------------------------------------------

=head2 lastkeynum()

Returns the last key number used, i.e., the sequence number of the
last record added to the datastore, as an integer.

=cut

sub lastkeynum {
    my( $self ) = @_;

    my $top_toc = $self->new_toc( { int => 0 } );
    my $keyint  = $top_toc->keynum;

    return $keyint;
}

#---------------------------------------------------------------------

=head2 nextkeynum()

Returns lastkeynum()+1 (a convenience method).  This could be useful
for adding a new record to a hash tied to a datastore, e.g.,

    $h{ $ds->nextkeynum } = "New record data.";

(but also note that there is a "null key" convention for this -- see
FlatFile::DataStore::Tiehash)

=cut

sub nextkeynum {
   for( $_[0]->lastkeynum ) {
       return 0 unless defined;
       return $_ + 1;
   }
}

#---------------------------------------------------------------------
#
# =head2 keyseek( $keyint )
#
# Gets seekpos of a particular line in the key file.
#
# Takes the record sequence number as an integer and returns
# the seek position needed to retrieve the record's preamble from
# the pertinent keyfile.
#
# Interestingly, this seek position is only a function of the keyint
# and keymax values, so this routine doesn't need to know (and doesn't
# return) which keyfile we're seeking into.
#
# Private method.
#
# =cut
#

sub keyseek {
    my( $self, $keyint ) = @_;

    my $keylen = $self->preamblelen + length( $self->recsep );

    my $keyseek;
    if( my $keymax = $self->keymax ) {
        my $skip = int( $keyint / $keymax );
        $keyseek = $keylen * ( $keyint - ( $skip * $keymax ) ); }
    else {
        $keyseek = $keylen * $keyint; }

    return $keyseek;
}

#---------------------------------------------------------------------
#
# =head2 nexttransnum(), get next transaction number
#
# Takes a FF::DS::Toc (table of contents) object, which should be
# the "top" Toc that has many of the key values for the datastore.
#
# Returns the next transaction number as an integer.
# Note: transaction numbers begin with 1 (not 0).
#
# Will croak if this number is longer than allowed by the current
# configuration.  In that case, a new datastore that allows for
# more transactions must be configured and the old datastore
# migrated to it.
#
# Private method.
#
# =cut
#

sub nexttransnum {
    my( $self, $top_toc ) = @_;

    $top_toc ||= $self->new_toc( { int => 0 } );

    my $transint  = $top_toc->transnum + 1;
    my $translen  = $self->translen;
    my $transbase = $self->transbase;
    my $transnum  = int2base $transint, $transbase, $translen;

    croak qq/Database exceeds configured size, transnum too long: $transnum/
        if length $transnum > $translen;

    return $transint;
}

#---------------------------------------------------------------------
#
# =head2 burst_preamble()
#
# Takes a preamble string (as stored on disk) and parses out all
# of the values, based on regx and specs.
#
# Returns a hash ref of these values.
#
# Called by FF::DS::Preamble->new() to create an object from a string,
# and by retrieve() and locate_record_data() to get the file number
# and seek pos for reading a record.
#
# Private method.
#
# =cut
#

sub burst_preamble {
    my( $self, $string ) = @_;

    croak qq/No preamble to burst/ unless $string;

    my @fields = $string =~ $self->regx;

    croak qq/Something is wrong with preamble: $string/ unless @fields;

    my %parms;
    my $i;
    for( $self->specs ) {  # specs() returns an array of hashrefs
        my( $key, $aref )       = %$_;
        my( $pos, $len, $parm ) = @$aref;
        my $field = $fields[ $i++ ];
        for( $key ) {
            if( /indicator|transind|date/ ) {
                $parms{ $key } = $field;
            }
            elsif( /user/ ) {
                my $try = $field;
                $try =~ s/\s+$//;
                $parms{ $key } = $try;
            }
            elsif( /fnum/ ) {
                next if $field =~ /^-+$/;
                $parms{ $key } = $field;
            }
            else {
                next if $field =~ /^-+$/;
                $parms{ $key } = base2int( $field, $parm );
            }
        }
    }
    return \%parms;
}

#---------------------------------------------------------------------
#
# =head2 update_preamble()
#
# Called by update() and delete() to flag old recs.
#
# Takes a preamble string and a hash ref of values to change, and
# returns a new preamble string with those values changed.
#
# Will croak if the new preamble does not match the regx attribute.
#
# Private method.
#
# =cut
#

sub update_preamble {
    my( $self, $preamble, $parms ) = @_;

    my $omap = $self->specs;

    for( keys %$parms ) {

        my $value = $parms->{ $_ };

        my $specs = omap_get_values( $omap, $_ );
        croak qq/Unrecognized field: $_/ unless $specs;

        my( $pos, $len, $parm ) = @{$specs};

        my $try;
        if( /indicator|transind|date|user/ ) {
            $try = sprintf "%-${len}s", $value;

            croak qq/Invalid value for $_: $try/
                unless $try =~ $Ascii_chars;
        }
        # the fnums should be in their base form already
        elsif( /fnum/ ) {
            $try = sprintf "%0${len}s", $value;
        }
        else {
            $try = int2base $value, $parm, $len;
        }

        croak qq/Value of $_ too long: $try/ if length $try > $len;

        substr $preamble, $pos, $len, $try;  # update the field
    }

    croak qq/Something is wrong with preamble: $preamble/
        unless $preamble =~ $self->regx;

    return $preamble;
}

#---------------------------------------------------------------------
# file read/write:
#---------------------------------------------------------------------

#---------------------------------------------------------------------
#
# =head2 locked_for_read()
#
# Takes a file name, opens it for input, locks it, sets binmode, and
# returns the open file handle.
#
# Private method.
#
# =cut
#

sub locked_for_read {
    my( $self, $file ) = @_;
    untaint path => $file;

    my $fh;
    sysopen( $fh, $file, O_RDONLY|O_CREAT )
                         or croak qq/Can't open $file for read: $!/;
    flock $fh, LOCK_SH   or croak qq/Can't lock $file shared: $!/;
    binmode $fh;

    return $fh;
}

#---------------------------------------------------------------------
#
# =head2 locked_for_write()
#
# Takes a file name, opens it for read/write, locks it, sets binmode,
# and returns the open file handle.
#
# Private method.
#
# =cut
#

sub locked_for_write {
    my( $self, $file ) = @_;
    untaint path => $file;

    my $fh;
    sysopen( $fh, $file, O_RDWR|O_CREAT ) or croak qq/Can't open $file for read-write: $!/;
    my $ofh = select( $fh ); $| = 1; select ( $ofh );  # flush buffers
    flock $fh, LOCK_EX                    or croak qq/Can't lock $file exclusive: $!/;
    binmode $fh;

    return $fh;
}

#---------------------------------------------------------------------
#
# =head2 read_record()
#
# Takes an open file handle and a seek position and
#
#     - seeks there to read the preamble
#     - seeks to the record data and reads that
#     - returns a record object created from the preamble and data
#
# Private method.
#
# =cut
#

sub read_record {
    my( $self, $fh, $seekpos ) = @_;

    # we don't call read_preamble() because we need len anyway
    my $len  = $self->preamblelen;
    my $sref = $self->read_bytes( $fh, $seekpos, $len );
    my $preamble = $self->new_preamble( { string => $$sref } );

    $seekpos    += $len;
    $len         = $preamble->reclen;
    my $recdata  = $self->read_bytes( $fh, $seekpos, $len );

    my $record = $self->new_record( {
        preamble => $preamble,
        data     => $recdata,  # scalar ref
        } );

    return $record;
}

#---------------------------------------------------------------------
#
# =head2 read_preamble()
#
# Takes an open file handle (probably the key file) and a seek
# position and
#
#     - seeks there to read the preamble
#     - returns the preamble string (not an object)
#
# Private method.
#
# =cut
#

sub read_preamble {
    my( $self, $fh, $seekpos ) = @_;

    my $len  = $self->preamblelen;
    my $sref = $self->read_bytes( $fh, $seekpos, $len );

    return $$sref;  # want the string, not the ref
}

#---------------------------------------------------------------------
#
# =head2 read_bytes()
#
# Takes an open file handle, a seek position and a length, reads
# that many bytes from that position, and returns a scalar
# reference to that data.  It is expected that the file is set
# to binmode.
#
# Private method.
#
# =cut
#

sub read_bytes {
    my( $self, $fh, $seekpos, $len ) = @_;

    my $string;
    sysseek $fh, $seekpos, 0 or croak qq/Can't seek: $!/;
    my $rc = sysread $fh, $string, $len;
    croak qq/Can't read: $!/ unless defined $rc;

    return \$string;
}

#---------------------------------------------------------------------
#
# =head2 write_bytes()
#
# Takes an open file handle, a seek position, and a scalar
# reference and writes that data to the file at that position.
# It is expected that the file is set to binmode.
#
# Private method.
#
# =cut
#

sub write_bytes {
    my( $self, $fh, $seekpos, $sref ) = @_;

    sysseek  $fh, $seekpos, 0 or croak qq/Can't seek: $!/;
    syswrite $fh, $$sref      or croak qq/Can't write: $!/;

}

#---------------------------------------------------------------------
#
# =head2 read_file(), used by init() to read the .uri file
#
# Takes a file name, locks it for reading, and returns the
# contents as an array of lines
#
# Private method.
#
# =cut
#

sub read_file {
    my( $self, $file ) = @_;
    untaint path => $file;

    my $fh;
    open  $fh, '<', $file or croak qq/Can't open $file for read: $!/;
    flock $fh, LOCK_SH    or croak qq/Can't lock $file shared: $!/;
    # binmode $fh;  # NO binmode here, please

    return <$fh>;
}


#---------------------------------------------------------------------
#
# =head2 now(), expects a string that contains
#
#     'yyyy', 'mm', 'da', 'tttttt' (hhmmss) in some order, or
#     'yy',   'm',  'd',  'ttt'    (hms)    in some order
#
# ('yyyy' is a magic string that denotes decimal vs. base62)
#
# Returns current date formatted as requested.
#
# Private method.
#
# =cut
#

sub now {
    my( $format ) = @_;
    my( $yr, $mo, $da, $hr, $mn, $sc ) =
        sub{($_[5]+1900,$_[4]+1,$_[3],$_[2],$_[1],$_[0])}->(localtime);
    for( $format ) {
        if( /yyyy/ ) {  # decimal
            s/ yyyy   / sprintf "%04d", $yr                   /ex;  # Y10K bug
            s/ mm     / sprintf "%02d", $mo                   /ex;
            s/ dd     / sprintf "%02d", $da                   /ex;
            s/ tttttt / sprintf "%02d%02d%02d", $hr, $mn, $sc /ex;
        }
        else {          # base62
            s/ yy  / int2base( $yr, 62 )  /ex;  # Y3844 bug
            s/ m   / int2base( $mo, 62 )  /ex;
            s/ d   / int2base( $da, 62 )  /ex;
            s/ ttt / int2base( $hr, 62 ).
                     int2base( $mn, 62 ).
                     int2base( $sc, 62 )  /ex;
        }
    }
    return $format;
}

#---------------------------------------------------------------------
#
# =head2 TIEHASH() supports tied hash access
#
# Returns datastore object.
#
# Note: because of how new_toc and new_record are implemented, I
# couldn't make Tiehash a subclass, so I'm requiring it into this
# class.  This may change in the future -- or not.
#
# Somewhat private method.
#
# =cut
#

sub TIEHASH {

    # Note: 'require', not 'use'.  This isn't
    # a "true" module -- we're just bringing in
    # some more FlatFile::DataStore methods.

    require FlatFile::DataStore::Tiehash;

    my $class = shift;
    $class->new( @_ );
}

#---------------------------------------------------------------------
BEGIN {
my %allow = (
    trusted  =>  qr{^       (.*) $}x,  # i.e., anything
    path     =>  qr{^ ([-.\w/]*) $}x,  # e.g., /tmp/sess/s.1.data
);

sub untaint {
    my( $key, $var ) = @_;
    for( $var ) {
        return unless defined;
        return if /^$/;
    }
    for( $key ) {
        die "Not defined: $_" unless $allow{ $_ };            # programmer error
        if( $var =~ /$allow{ $_ }/ ) { $_[1] = $1          }  # must set the alias
        else                         { die "Invalid $_.\n" }  # intentionally coy
    }
}}

1;  # returned

__END__