bucardo 4.99.5-1 / usr / bin / bucardo

#!/usr/bin/env perl
# -*-mode:cperl; indent-tabs-mode: nil; cperl-indent-level: 4-*-

## Script to control Bucardo
##
## Copyright 2006-2012 Greg Sabino Mullane <greg@endpoint.com>
##
## Please see http://bucardo.org/ for full documentation
##
## Run as bucardo --help for some basic instructions

package bucardo;
use strict;
use warnings;
use 5.008003;
use DBI;
use IO::Handle;
use Getopt::Long;
Getopt::Long::Configure(qw/no_ignore_case/);
use Time::HiRes 'sleep';
use POSIX qw/ceil setsid localeconv/;
use Data::Dumper 'Dumper';
$Data::Dumper::Indent = 1;

our $VERSION = '4.99.5';

## For the tests, we want to check that it compiles without actually doing anything
return 1 if $ENV{BUCARDO_TEST};

## No buffering on the standard output streams
*STDOUT->autoflush(1);
*STDERR->autoflush(1);

## All the variables we use often and want to declare here without 'my'
use vars qw/$dbh $SQL $sth %sth $count $info %global $SYNC $GOAT $TABLE $SEQUENCE $DB $DBGROUP $HERD
            $CUSTOMCODE $CUSTOMNAME $CUSTOMCOLS /;

## How to show dates from the database, e.g. start time of a sync
my $DATEFORMAT       = q{Mon DD, YYYY HH24:MI:SS};
my $SHORTDATEFORMAT  = q{HH24:MI:SS};

## How long we hang out between checks after a kick or when waiting for notices
my $WAITSLEEP        = 1;

## Determine how we were called
## If we were called from a different directory, and the base directory is in our path,
## we strip out the directory part
my $progname = $0;
if (exists $ENV{PATH} and $progname =~ m{(.+)/(.+)}) {
    my ($base, $name) = ($1,$2);
    for my $seg (split /\:/ => $ENV{PATH}) {
        if ($seg eq $base) {
            $progname = $name;
            last;
        }
    }
}

## We must have at least one argument to do anything
help() unless @ARGV;

## Default arguments - most are for the bc constructor
my $bcargs = {
              quiet        => 0,
              verbose      => 0,
              bcverbose    => 1,
              dbname       => 'bucardo',
              dbuser       => 'bucardo',
              dbpass       => '',
              sendmail     => 0,
              extraname    => '',
              debugfilesep => 0,
              debugdir     => '/var/log/bucardo',
              debugname    => '',
              debugsyslog  => 1,
              debugfile    => 1,
              cleandebugs  => 0,
              batch        => 0,
          };

## These options must come before the GetOptions call
for my $arg (@ARGV) {
    if ($arg eq '--no-bucardorc') {
        $bcargs->{'no-bucardorc'} = 1;
    }
    if ($arg =~ /--bucardorc=(.+)/) {
        $bcargs->{'bucardorc'} = $1;
    }
    if ($arg =~ /^-+\?$/) {
        help();
    }
}

## Values are first read from a .bucardorc, either in the current dir, or the home dir.
## If those do not exist, check for a global rc file
## These will be overwritten by command-line args.
my $file;
if (! $bcargs->{'no-bucardorc'}) {
    if ($bcargs->{bucardorc}) {
        -e $bcargs->{bucardorc} or die qq{Could not find the file "$bcargs->{bucardorc}"\n};
        $file = $bcargs->{bucardorc};
    }
    elsif (-e '.bucardorc') {
        $file = '.bucardorc';
    }
    elsif (-e "$ENV{HOME}/.bucardorc") {
        $file = "$ENV{HOME}/.bucardorc";
    }
    elsif (-e '/etc/bucardorc') {
        $file = '/etc/bucardorc';
    }
}
if (defined $file) {
    open my $rc, '<', $file or die qq{Could not open "$file": $!\n};
    while (<$rc>) {

        ## Skip any lines starting with a hash
        next if /^\s*#/;

        ## Format is foo=bar, with whitespace allowed
        next unless /^\s*(\w+)\s*=\s*(.+?)\s*$/o;
        my ($name,$value) = ($1,$2); ## no critic (ProhibitCaptureWithoutTest)
        $bcargs->{$name} = $value;
    }
    close $rc or die;
}

GetOptions ## no critic (ProhibitCallsToUndeclaredSubs)
    ($bcargs,
     'verbose+',
     'vv',
     'quiet+',
     'notimer',
     'help',
     'debug',
     'version',
     'sort=i',
     'showdays',
     'compress',
     'retry=i',
     'retrysleep=i',
     'batch',
     'dryrun|dry-run',
     'confirm',
     'tsep=s',

     ## These are sent to the constructor:
     'bcverbose',
     'dbport=i',
     'dbhost=s',
     'dbname=s',
     'dbuser=s',
     'dbpass=s',
     'sendmail=i',
     'extraname=s',
     'debugfilesep',
     'debugname=s',
     'debugsyslog=i',
     'debugdir=s',
     'debugfile=i',
     'cleandebugs=i',

     ## Used internally
     'force',
     'schema|n=s@',
     'exclude-schema|N=s@',
     'table|t=s@',
     'exclude-table|T=s@',
     'db=s',
     'herd=s',
     'piddir=s',
     ## These two already handled above, but need to be here so GetOptions is happy:
     'bucardorc=s',
     'no-bucardorc',

) or die "\n";

## If --help is set, ignore everything else, show help, then exit
help() if $bcargs->{help};

## If --version is set, ignore everything else, show the version, and exit
if ($bcargs->{version}) {
    print "$progname version $VERSION\n";
    exit 0;
}

## Sometimes we want to be as quiet as possible
my $QUIET = delete $bcargs->{quiet};

## Quick shortcut for lots of verbosity
$bcargs->{vv} and $bcargs->{verbose} = 2;

## Set some global arguments
my $VERBOSE = delete $bcargs->{verbose};
my $DEBUG   = delete $bcargs->{debug} || $ENV{BUCARDO_DEBUG} || 0;

## Do we compress time outputs by stripping out whitespace?
my $COMPRESS = delete $bcargs->{compress} || 0;

## Do we retry after a sleep period on failed kicks?
my $RETRY      = delete $bcargs->{retry} || 0;
my $RETRYSLEEP = delete $bcargs->{retrysleep} || 0;

## Allow people to turn off the cool timer when kicking syncs
my $NOTIMER = delete $bcargs->{notimer} || 0;

## Anything left over is the verb and noun(s)
my $verb = shift || '';

## No verb? Show a help message and exit
help() unless $verb;

## Standardize the verb as lowercase, and grab the rest of the args as the "nouns"
$verb = lc $verb;
my @nouns = @ARGV;

## Allow alternate underscore format
if ($verb =~ /^(\w+)_(\w+)$/) {
    $verb = $1;
    unshift @nouns => $2;
}

## Make a single string version, mostly for output in logs
my $nouns = join ' ' => @nouns;
## The verb may have a helper, usually a number
my $adverb;

## Installation must happen before we try to connect!
install() if $verb =~ /instal/i;

## For everything else, we need to connect to a previously installed Bucardo database

## Create a quick data source name
my $DSN = "dbi:Pg:dbname=$bcargs->{dbname}";
$bcargs->{dbhost} and length $bcargs->{dbhost} and $DSN .= ";host=$bcargs->{dbhost}";
$bcargs->{dbport} and length $bcargs->{dbport} and $DSN .= ";port=$bcargs->{dbport}";

## Connect to the database
$dbh = DBI->connect($DSN, $bcargs->{dbuser}, $bcargs->{dbpass}, {AutoCommit=>0,RaiseError=>1,PrintError=>0});

## We only want to concern ourselves with things in the bucardo schema
$dbh->do('SET search_path = bucardo');

## Make sure we find a valid Postgres version
## Why do we check this after a successful install?
## In case they get pg_dumped to a different (older) database. It has happened! :)
check_version($dbh); ## dies on invalid version

## Listen for the MCP. Not needed for old-school non-payload LISTEN/NOTIFY, but does no harm
$dbh->do('LISTEN bucardo');
$dbh->commit();

## Set some global variables based on information from the bucardo_config table

## The reason file records startup and shutdown messages
my $REASONFILE = get_config('reason_file') or die "Invalid reason_file!\n";
my $REASONFILE_LOG = "$REASONFILE.log";

## The directory Bucardo.pm writes PID and other information to
my $PIDDIR = get_config('piddir') or die "Invalid piddir!\n";

## The PID file of the master control file (MCP)
## If this exists, it is a good bet that Bucardo is currently running
my $PIDFILE = "$PIDDIR/bucardo.mcp.pid";

## The stop file whose existence tells all Bucardo processes to exit immediately
my $stopfile = get_config('stopfile') or die "Invalid stopfile!\n";
my $STOPFILE = "$PIDDIR/$stopfile";

## Aliases for terms people may shorten, misspell, etc.
## Mostly used for database columns when doing an 'update'
our %alias = (
    'ssp'                 => 'server_side_prepares',
    'server_side_prepare' => 'server_side_prepares',
    'port'                => 'dbport',
    'host'                => 'dbhost',
    'name'                => 'dbname',
    'user'                => 'dbuser',
    'pass'                => 'dbpass',
    'password'            => 'dbpass',
    'service'             => 'dbservice',
);

## Columns that cannot be changed: used in the update_* subroutines
my %column_no_change = (
    'id'    => 1,
    'cdate' => 1,
);

## Regular expression for a valid database group name
my $re_dbgroupname = qr{\w[\w\d]*};

## Regular expression for a valid database name
my $re_dbname = qr{\w[\w\d]*};

## Send a ping to the MCP to make sure it is alive and responding
ping() if $verb eq 'ping';

## Display more detailed help than --help
superhelp() if $verb eq 'help';

## Make sure the Bucardo database has the latest schema
upgrade() if $verb eq 'upgrade' or $verb eq 'uprgade';

## All the rest of the verbs require use of global information
## Thus, we load everything right now
load_bucardo_info();

## View the status of one or more syncs
status_all()    if $verb eq 'status' and ! @nouns;
status_detail() if $verb eq 'status';

## Stop, start, or restart the main Bucardo daemon
stop()          if $verb eq 'stop';
start()         if $verb eq 'start' or $verb eq 'strt';
restart()       if $verb eq 'restart';

## Reload the configuration file
reload_config() if $verb eq 'reload_config';

## Reload the whole daemon
reload()        if $verb eq 'reload';

## Show information about something: database, table, sync, etc.
list_item()     if $verb eq 'list' or $verb eq 'l' or $verb eq 'lsit' or $verb eq 'liast'
    or $verb eq 'lisy' or $verb eq 'lit';

## Add something
add_item()      if $verb eq 'add';

## Remove something
remove_item()   if $verb eq 'remove' or $verb eq 'delete' or $verb eq 'del';

## Update something
update_item()   if $verb eq 'update' or $verb eq 'upd';

## Inspect something
inspect()       if $verb eq 'inspect';

## Inject a message into the Bucardo logs
message()       if $verb eq 'message';

## Show or set an item from the bucardo.config table
config()        if $verb eq 'set' or $verb eq 'show' or $verb eq 'config';

## Validate a sync
validate()      if $verb eq 'validate';

## Purge the delta/track tables
purge()         if $verb eq 'purge';

## There are only a few valid verbs left, so we check for them now
if ($verb ne 'kick' and $verb ne 'activate' and $verb ne 'deactivate') {
    ## Show help and exit
    help();
}

## For all remaining verbs, we expect a list of syncs with an optional decimal "timeout"

## If there are no syncs, no sense in going on!
if (! keys %$SYNC) {
    die qq{No syncs have been created yet!\n};
}

## The final list of syncs we are going to do something to
my @syncs;

## The fail msg on a non-match
my $msg;

## Loop through each noun and handle it
SYNCMATCH: for my $sync (@nouns) {

    ## Quick skipping of noise word 'sync'
    next if $sync eq 'sync';

    ## If this is a number, it's a timeout, so set it and skip to the next noun
    if ($sync =~ /^\d+$/) {
        $adverb = $sync;
        next SYNCMATCH;
    }

    ## If they want all syncs, grab them all and stop reading any more nouns
    if ($sync eq 'all') {
        undef @syncs;
        for my $name (sort keys %$SYNC) {
            push @syncs => $name;
        }
        last SYNCMATCH;
    }

    ## The rest are all ways of finding the sync they want
    ## Change the name to a Perl-regex friendly form
    (my $term = $sync) =~ s/%/\*/g;
    $term =~ s/([^\.])\*/$1.*/g;
    $term =~ s/^\*/.*/;

    if ($term =~ /\*/) {
        for my $name (sort keys %$SYNC) {
            push @syncs => $name if $name =~ /^$term$/;
        }
        next SYNCMATCH;
    }

    ## Now that wildcards are out, we must have an absolute match
    if (! exists $SYNC->{$sync}) {
        $msg = qq{Sync "$sync" does not appear to exist\n};
        ## No sense in going on
        last SYNCMATCH;
    }

    ## Got a direct match, so store it away
    push @syncs => $sync;

}

## If syncs is empty, a regular expression search failed
if (!@syncs) {
    $msg = qq{No matching syncs were found\n};
}

## If we have a message, something is wrong
if (defined $msg) {
    ## Be nice and print a list of active syncs
    my @goodsyncs;
    for my $s (sort keys %$SYNC) {
        push @goodsyncs => $s if $SYNC->{$s}{status} eq 'active';
    }
    if (@goodsyncs) {
        $msg .= "Active syncs:\n";
        $msg .= join "\n" => map { " $_" } @goodsyncs;
    }
    die "$msg\n";
}

## Activate or deactivate one or more syncs
vate_sync() if $verb eq 'activate' or $verb eq 'deactivate';

## Kick one or more syncs
kick() if $verb eq 'kick';

## If we reach here (and we should not), display help and exit
help();

exit;

## Everything from here on out is subroutines


sub get_config {

    ## Given a name, return the matching value from the bucardo_config table
    ## Arguments: one
    ## 1. setting name
    ## Returns: bucardo_config.value string

    my $name = shift;

    $SQL = 'SELECT setting FROM bucardo.bucardo_config WHERE LOWER(name) = ?';
    $sth = $dbh->prepare_cached($SQL);
    $count = $sth->execute(lc $name);
    if ($count < 1) {
        $sth->finish();
        die "Invalid bucardo_config setting: $name\n";
    }
    return $sth->fetchall_arrayref()->[0][0];

} ## end of get_config


sub numbered_relations {

    ## Sorting function
    ## Arguments: none (implicit $a / $b via Perl sorting)
    ## Returns: winning value
    ## Sorts relations of the form schema.table
    ## in which we do alphabetical first, but switch to numeric order
    ## for any numbers at the end of the schema or the table
    ## Thus, public.foobar1 will come before public.foobar10

    ## Pull in the names to be sorted, dereference as needed
    my $uno = ref $a ? "$a->{schemaname}.$a->{tablename}" : $a;
    my $dos = ref $b ? "$b->{schemaname}.$b->{tablename}" : $b;

    ## Break apart the first item into schema and table
    die if $uno !~ /(.+)\.(.+)/;
    my ($schema1,$sbase1,$table1,$tbase1) = ($1,$1,$2,$2);
    ## Store ending numbers if available: if not, use 0
    my ($snum1, $tnum1) = (0,0);
    $sbase1 =~ s/(\d+)$// and $snum1 = $1;
    $tbase1 =~ s/(\d+)$// and $tnum1 = $1;

    ## Break apart the second item into schema and table
    die if $dos !~ /(.+)\.(.+)/;
    my ($schema2,$sbase2,$table2,$tbase2) = ($1,$1,$2,$2);
    my ($snum2, $tnum2) = (0,0);
    $sbase2 =~ s/(\d+)$// and $snum2 = $1;
    $tbase2 =~ s/(\d+)$// and $tnum2 = $1;

    return (
        $sbase1 cmp $sbase2
     or $snum1 <=> $snum2
     or $tbase1 cmp $tbase2
     or $tnum1 <=> $tnum2);

} ## end of numbered_relations


sub check_version {

    ## Quick check that we have the minumum supported version
    ## This is for the bucardo database itself
    ## Arguments: one
    ## 1. Database handle
    ## Returns: undef (may die if the version is not good)

    my $dbh = shift;
    my $res = $dbh->selectall_arrayref('SELECT version()')->[0][0];
    if ($res !~ /(\d+)\.(\d+)/) {
        die "Sorry, unable to determine the database version\n";
    }
    my ($maj,$min) = ($1,$2);
    if ($maj < 8 or (8 == $maj and $min < 1)) {
        die "Sorry, Bucardo requires Postgres version 8.1 or higher. This is only $maj.$min\n";
    }

    return;

} ## end of check_version


sub help {

    ## Give detailed help about usage of this program
    ## Arguments: none
    ## Returns: never, always exits

    ## Nothing to do if we are being quiet
    exit 0 if $QUIET;

    warn qq{Usage: $progname args
  install         ** Installs the Bucardo database

  start <reason>  ** Force any existing sync(s) to quit, then starts Bucardo

  stop <reason>   ** Tell all Bucardo processes to stop permanently

  list <type> [name]   ** View information about dbs, dbgroups, herds, syncs, tables, sequences, customnames, customcols, customcode or "all" of them
  add <type> <name>    ** Add a db, dbgroup, herd, sync, table, sequence, customcode, customcols customname or customcode
  remove <type> <name> ** Remove a db, dbgroup, herd, sync, table, sequence, customname, customcol or customcode

  For detailed help on the above, prefix with a help like this:
    $progname help add db

  kick <syncname(s)> [timeout]
                  ** Kick off one or more syncs, optionally wait for result (0 = wait until done)

  ping [timeout]  ** Ping the MCP process for a response, return a Nagios-friendly string

  status [--sort=col#]
                  ** List information about all syncs

  status syncname[s]
                  ** List detailed information about one or more syncs

  reload syncname[s]
                 ** Reload a sync

  validate syncname[s]
                 ** Validate a sync

  purge [database]
                 ** Purge the delta and track tables

  upgrade
                 ** Upgrade Bucardo to the current version

  message
                 ** Write a message to the Bucardo logs

  show [all|item]
  set [item=value]
                 ** View or set configuration parameters

  reload_config   ** Force a running Bucardo to reload the bucardo_config table

For more details, try 'man bucardo'
};

    exit 0;

} ## end of help


sub superhelp {

    ## Show detailed help by examining the verb and nouns
    ## Arguments: none
    ## Returns: never, always exits

    ## If there are no nouns, we can only show the generic help
    help() if ! @nouns;

    ## Grab the first two words and standardize them
    my $word1 = lc $nouns[0];
    my $word2 = lc $nouns[1];
    my $word3 = lc $nouns[2];

    ## Swap them out if the second one is 'all' and the third exists
    if (defined $word2 and $word2 eq 'all' and defined $word3) {
        $word2 = $word3;
    }

    ## Because usage() sometimes relies on the verb
    $verb = $word1;

    ## Standardize the 'thing' argument
    $word2 = standardize_name($word2);

    ## TODO: Write a test to ensure full coverage below

    if ('add' eq $word1) {
        if ('customcode' eq $word2) {
            warn usage('add_customcode') . "\n";
            exit 0;
        }
        if ('customname' eq $word2) {
            warn usage('add_customname') . "\n";
            exit 0;
        }
        if ('customcols' eq $word2) {
            warn usage('add_customcols') . "\n";
            exit 0;
        }
        if ('database' eq $word2) {
            warn usage('add_database') . "\n";
            exit 0;
        }
        if ('dbgroup' eq $word2) {
            warn usage('add_dbgroup') . "\n";
            exit 0;
        }
        if ('herd' eq $word2) {
            warn usage('add_herd') . "\n";
            exit 0;
        }
        if ('sync' eq $word2) {
            warn usage('add_sync') . "\n";
            exit 0;
        }
        if ('table' eq $word2) {
            warn usage('add_table') . "\n";
            exit 0;
        }
        if ('sequence' eq $word2) {
            warn usage('add_sequence') . "\n";
            exit 0;
        }
        warn usage('add') . "\n";
        exit 0;
    }
    if ('config' eq $word1) {
        warn usage('config') . "\n";
        exit 0;
    }
    if ('inspect' eq $word1) {
        if ('table' eq $word2) {
            warn usage('inspect_table') . "\n";
            exit 0;
        }
        if ('herd' eq $word2) {
            warn usage('inspect_herd') . "\n";
            exit 0;
        }
        if ('sync' eq $word2) {
            warn usage('inspect_sync') . "\n";
            exit 0;
        }

        warn usage('inspect') . "\n";
        exit 0;
    }
    if ('kick' eq $word1) {
        warn usage('kick') . "\n";
        exit 0;
    }
    if ('list' eq $word1 or 'l' eq $word1 or 'liast' eq $word1) {
        if ('customcode' eq $word2 or 'custom_code' eq $word2) {
            warn usage('list_customcode') . "\n";
            exit 0;
        }
        ## Keep dbg before db
        if ('dbg' eq $word2 or 'dbgroup' eq $word2) {
            warn usage('list_dbgroups') . "\n";
            exit 0;
        }
        if ($word2 =~ /^(?:db|database)/) {
            warn usage('list_databases') . "\n";
            exit 0;
        }
        if ('herd' eq $word2) {
            warn usage('list_herds') . "\n";
            exit 0;
        }
        if ('sync' eq $word2) {
            warn usage('list_syncs') . "\n";
            exit 0;
        }
        if ('table' eq $word2 or 'tab' eq $word2) {
            warn usage('list_tables') . "\n";
            exit 0;
        }
        if ('sequence' eq $word2 or 'seq' eq $word2) {
            warn usage('list_sequences') . "\n";
            exit 0;
        }
        warn usage('list') . "\n";
        exit 0;
    }
    if ('message' eq $word1) {
        warn usage('message') . "\n";
        exit 0;
    }
    if ('ping' eq $word1) {
        warn usage('ping') . "\n";
        exit 0;
    }
    if ('reload' eq $word1) {
        warn usage('reload') . "\n";
        exit 0;
    }
    if ('reload_config' eq $word1) {
        warn usage('reload_config') . "\n";
        exit 0;
    }
    if ('remove' eq $word1 or 'delete' eq $word1) {
        if ('customcode' eq $word2 or 'custom_code' eq $word2) {
            warn usage('remove_customcode') . "\n";
            exit 0;
        }
        if ('db' eq $word2 or 'database' eq $word2) {
            warn usage('remove_database') . "\n";
            exit 0;
        }
        if ('dbg' eq $word2 or 'dbgroup' eq $word2) {
            warn usage('remove_dbgroup') . "\n";
            exit 0;
        }
        if ('herd' eq $word2) {
            warn usage('remove_herd') . "\n";
            exit 0;
        }
        if ('sync' eq $word2) {
            warn usage('remove_sync') . "\n";
            exit 0;
        }
        if ('table' eq $word2 or 'tab' eq $word2) {
            warn usage('remove_table') . "\n";
            exit 0;
        }
        if ('sequence' eq $word2 or 'seq' eq $word2) {
            warn usage('remove_sequence') . "\n";
            exit 0;
        }
        warn usage('remove') . "\n";
        exit 0;
    }
    if ('restart' eq $word1) {
        warn usage('restart') . "\n";
        exit 0;
    }
    if ('start' eq $word1) {
        warn usage('start') . "\n";
        exit 0;
    }
    if ('status' eq $word1) {
        warn usage('status') . "\n";
        exit 0;
    }
    if ('stop' eq $word1) {
        warn usage('stop') . "\n";
        exit 0;
    }
    if ('update' eq $word1) {
        warn usage('update') . "\n";
        exit 0;
    }
    if ('upgrade' eq $word1) {
        warn usage('upgrade') . "\n";
        exit 0;
    }
    if ('validate' eq $word1) {
        warn usage('validate') . "\n";
        exit 0;
    }
    if ('purge' eq $word1) {
        warn usage('purge') . "\n";
        exit 0;
    }

    ## Generic fallthrough
    help();

    exit 0;

} ## end of superhelp


sub ping {

    ## See if the MCP is alive and responds to pings
    ## Default is to wait 15 seconds
    ## Arguments: none, but looks in @nouns for a timeout
    ## Returns: never, exits

    ## Set the default timeout, but override if any remaining args start with a number
    my $timeout = 15;
    for (@nouns) {
        if (/^(\d+)/) {
            $timeout = $1;
            last;
        }
    }

    $VERBOSE and print "Pinging MCP, timeout = $timeout\n";
    $dbh->do('LISTEN bucardo_mcp_pong');
    $dbh->do('NOTIFY bucardo_mcp_ping');
    $dbh->commit();
    my $starttime = time;
    sleep 0.1;

    ## Loop until we timeout or get a confirmation from the MCP
  P:{
        ## Grab any notices that have come in
        my $notify = $dbh->func('pg_notifies');
        if (defined $notify) {
            ## Extract the PID that sent this notice
            my ($name, $pid, $payload) = @$notify;
            ## We are done: ping successful
            $QUIET or print "OK: Got response from PID $pid\n";
            exit 0;
        }

        ## Rollback, sleep, and check for a timeout
        $dbh->rollback();
        sleep 0.5;
        my $totaltime = time - $starttime;
        if ($timeout and $totaltime >= $timeout) {
            ## We are done: ping failed
            $QUIET or print "CRITICAL: Timed out ($totaltime s), no ping response from MCP\n";
            exit 1;
        }
        redo;
    }

    return;

} ## end of ping


sub start {

    ## Attempt to start the Bucardo daemon
    ## Arguments: none
    ## Returns: undef

    ## Write a note to the 'reason' log file
    ## This will automatically write any nouns in as well
    append_reason_file('start');

    ## Refuse to go on if we get a ping response within 5 seconds
    $QUIET or print "Checking for existing processes\n";

    ## We refuse to start if the MCP PID file exists
    my $oldpid = 0;
    if (-e $PIDFILE) {
        $msg = "Cannot start, PID file $PIDFILE exists\n";
        open my $fh, '<', $PIDFILE or die qq{Could not open "$PIDFILE": $!\n};
        if (<$fh> !~ /(\d+)/) { ## no critic
            warn qq{File "$PIDFILE" does not start with a PID!\n};
        }
        else {
            $msg = "Cannot start, process $1 is already running (file=$PIDFILE)\n";
        }
        close $fh or warn qq{Could not close $PIDFILE: $!\n};

        $QUIET or print $msg;

        append_reason_file('fail');

        exit 1;
    }

    ## Verify that the version in the database matches our version
    my $dbversion = get_config('bucardo_current_version')
        or die "Could not find Bucardo version!\n";
    if ($dbversion ne $VERSION) {
        my $message = "Version mismatch: bucardo is $VERSION, but bucardo database is $dbversion\n";
        append_reason_file('fail');
        warn $message;
        warn "Perhaps you need to run 'bucardo upgrade' ?\n";
        exit 1;
    }

    ## Create a new Bucardo daemon
    require Bucardo;
    my $bc = Bucardo->new($bcargs);

    ## Verify that the version of Bucardo.pm matches our version
    my $pm_version = $bc->{version} || 'unknown';
    if ($VERSION ne $pm_version) {
        my $message = "Version mismatch: bucardo is $VERSION, but Bucardo.pm is $pm_version\n";
        append_reason_file('fail');
        die $message;
    }

    ## Just in case, stop it
    stop_bucardo();

    if (-e $STOPFILE) {
        print "Removing $STOPFILE\n" unless $QUIET;
        unlink $STOPFILE;
    }

    $QUIET or print qq{Starting Bucardo\n};

    ## Disconnect from our local connection before we fork
    $dbh->disconnect();

    ## Fork and setsid to disassociate ourselves from the daemon
    if (fork) {
        ## We are the kid, do nothing
    }
    else {
        close STDERR or warn "Could not close STDERR\n";
        close STDOUT or warn "Could not close STDOUT\n";
        setsid() or die;
        ## Here we go!
        $bc->start_mcp();
    }

    exit 0;

} ## end of start


sub stop {

    ## Attempt to stop the Bucardo daemon
    ## Arguments: none
    ## Returns: undef

    ## Write a note to the 'reason' log file
    append_reason_file('stop');

    print "Creating $STOPFILE ... " unless $QUIET;
    stop_bucardo();
    print "Done\n" unless $QUIET;

    ## If this was called directly, just exit now
    exit 0 if $verb eq 'stop';

    return;

} ## end of stop


sub stop_bucardo {

    ## Create the semaphore that tells all Bucardo processes to exit
    ## Arguments: none
    ## Returns: undef

    ## Create the file, and write some quick debug information into it
    ## The only thing the processe care about is if the file exists
    open my $stop, '>', $STOPFILE or die qq{Could not create "$STOPFILE": $!\n};
    print {$stop} "Stopped by $progname on " . (scalar localtime) . "\n";
    close $stop or warn qq{Could not close "$STOPFILE": $!\n};

    return;

} ## end of stop_bucardo


sub restart {

    ## Simple, really: stop, wait, start!
    ## Arguments: none
    ## Returns: undef

    stop();
    sleep 1;
    start();

    return;

} ## end of restart


sub reload_config {

    ## Reload configuration settings from the bucardo database,
    ## then restart all controllers and kids
    ## Arguments: none directly (but processes the nouns to check for numeric arg)
    ## Returns: never, exits

    ## Scan the nouns for a numeric argument.
    ## If found, set as the adverb.
    ## This will cause us to wait for confirmation or reload before exiting
    for (@nouns) {
        if (/^\d+$/) {
            $adverb = $1;
            last;
        }
    }

    $QUIET or print q{Forcing Bucardo to reload the bucardo_config table};

    ## We may want to wait to hear from the MCP that it is done
    my $done = 'bucardo_reload_config_finished';
    $dbh->do('NOTIFY bucardo_reload_config');
    if (defined $adverb) {
        print '...';
        $dbh->do("LISTEN $done");
    }
    $dbh->commit();

    ## If we are not waiting around for a confirmation, just exit now
    if (!defined $adverb) {
        print "\n";
        exit 0;
    }

    ## Wait a little bit, then scan for the confirmation message
    sleep 0.1;
    wait_for_notice($dbh, $done);
    print "DONE!\n";

    exit 0;

} ## end of reload_config


sub wait_for_notice {

    ## Keep hanging out until we get the notice we are waiting for
    ## Arguments: two
    ## 1. Database handle
    ## 2. String to listen for
    ## Returns: undef

    my ($ldbh, $string) = @_;

  WAITIN: {
        for my $notice (@{ db_get_notices($ldbh) }) {
            my ($name) = @$notice;
            last WAITIN if $name eq $string;
        }
        $dbh->commit();
        sleep($WAITSLEEP);
        redo;
    }

    return;

} ## end of wait_for_notice


sub reload {

    ## Ask for one or more syncs to be reloaded
    ## Arguments: none directly (but processes the nouns for a list of syncs)
    ## Returns: never, exits

    my $usage = usage('reload');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    for my $syncname (@nouns) {

        ## Be nice and allow things like $0 reload sync foobar
        next if $syncname eq 'sync';

        ## Make sure this sync exists, and grab its status
        $SQL = 'SELECT status FROM bucardo.sync WHERE name = ?';
        $sth = $dbh->prepare($SQL);
        $count = $sth->execute($syncname);
        if ($count != 1) {
            warn "Invalid sync: $syncname\n";
            $sth->finish();
            next;
        }
        my $status = $sth->fetch()->[0];

        ## Skip any syncs that are not active
        if ($status ne 'active') {
            warn qq{Cannot reload: status of sync "$syncname" is $status\n};
            next;
        }

        ## We wait for the MCP to tell us that each sync is done reloading
        my $done = "bucardo_reloaded_sync_$syncname";
        print "Reloading sync $syncname...";
        $dbh->do(qq{LISTEN "$done"});
        $dbh->do(qq{NOTIFY "bucardo_reload_sync_$syncname"});
        $dbh->commit();

        ## Sleep a little, then wait until we hear a confirmation from the MCP
        sleep 0.1;
        wait_for_notice($dbh, $done);
        print "\n";

    } ## end each sync to be reloaded

    exit 0;

} ## end of reload


sub validate {

    ## Attempt to validate one or more syncs
    ## Arguments: none directly (but processes the nouns for a list of syncs)
    ## Returns: never, exits

    my $usage = usage('validate');

    if (!@nouns) {
        warn "$usage\n";
        exit 0;
    }

    ## Build the list of syncs to validate
    my @synclist;

    ## Nothing specific is the same as 'all'
    if ($nouns[0] eq 'all' and ! defined $nouns[1]) {
        @synclist = sort keys %$SYNC;
        if (! @synclist) {
            print "Sorry, there are no syncs to validate!\n";
            exit 0;
        }
    }
    else {
        for my $name (@nouns) {

            ## Be nice and allow things like $0 validate sync foobar
            next if $name eq 'sync';

            if (! exists $SYNC->{$name}) {
                die qq{Sorry, there is no sync named "$name"\n};
            }
            push @synclist => $name;
        }
    }

    ## Get the largest sync name so we can line up the dots all pretty
    my $maxsize = 1;
    for my $name (@synclist) {
        $maxsize = length $name if length $name > $maxsize;
    }
    $maxsize += 3;

    ## Loop through and validate each in turn,
    ## waiting for a positive response from the MCP
    for my $name (@synclist) {

        printf "Validating sync $name%s",
            '.' x ($maxsize - length $name);

        my $done = "bucardo_validated_sync_$name";
        $dbh->do(qq{LISTEN "$done"});

        $SQL = 'SELECT validate_sync(?)';
        $sth = $dbh->prepare($SQL);
        $sth->execute($name);
        $dbh->commit();

        ## Wait a bit, then hang out until the MCP confirms the validation
        sleep 0.1;
        wait_for_notice($dbh, $done);
        print "DONE!\n";
    }

    exit 0;

} ## end of validate


sub purge {

    ## Purge the delta and track tables for one or more tables, for one or more databases
    ## Arguments: variable
    ## Returns: never, exits

    ## TODO: databases, tables, timeslices

    my $usage = usage('purge');

    ## Nothing specific is the same as 'all'
    my $doall = 0;
    if (!@nouns or ($nouns[0] eq 'all' and ! defined $nouns[1])) {
        $doall = 1;
        for my $dbname (sort keys %$DB) {
            my $db = $DB->{$dbname};
            ## Do not purge inactive databases
            next if $db->{status} ne 'active';

            ## Do not purge unless they are a source
            next if ! $db->{issource};

            print "Checking db $dbname\n";

            ## Make sure it has a bucardo schema.
            ## May not if validate_sync has never been run!
            my $dbh = connect_database($dbname);

            if (! schema_exists('bucardo')) {
                warn "Cannot purge database $dbname: no bucardo schema!\n";
                next;
            }

            ## Run the purge_delta on this database
            $SQL = 'SELECT bucardo.bucardo_purge_delta(?)';
            $sth = $dbh->prepare($SQL);
            $sth->execute('1 second');
            my $results = $sth->fetchall_arrayref()->[0][0];
            ## Dump the resulting message back to the user
            ## Should be like this: Tables processed: 3
            print "$dbname: $results\n";

            $dbh->commit();

        }
    }
    if (! $doall) {
        for my $name (@nouns) {
            die "Purging name $name\n";
        }
    }

    exit 0;

} ## end of purge


sub add_item {

    ## Add an item to the internal bucardo database
    ## Arguments: none directly (but processes the nouns)
    ## Returns: never, exits

    my $usage = usage('add');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## First word is the type of thing we are adding
    my $thing = shift @nouns;

    ## Account for variations and abbreviations
    $thing = standardize_name($thing);

    ## All of these will exit and do not return
    add_customcode() if $thing eq 'customcode';
    add_customname() if $thing eq 'customname';
    add_customcols() if $thing eq 'customcols';
    add_database()   if $thing eq 'database';
    add_dbgroup()    if $thing eq 'dbgroup';
    add_herd()       if $thing eq 'herd';
    add_sync()       if $thing eq 'sync';

    ## The rest is tables and sequences
    ## We need to support 'add table all' as well as 'add all tables'

    my $second_arg = $nouns[0] || '';

    ## Rearrange the args as needed, and determine if we want 'all'
    my $do_all = 0;

    if ($thing eq 'all') {
        $do_all = 1;
        $thing = shift @nouns;
        $thing = standardize_name($thing);
    }
    elsif (lc $second_arg eq 'all') {
        $do_all = 1;
        shift @nouns;
    }

    ## Quick check in case someone thinks they should add a goat
    if ($thing =~ /^goat/i) {
        warn qq{Cannot add a goat: use add table or add sequence instead\n};
        exit 1;
    }

    ## Add a table
    if ($thing eq 'table') {
        if ($do_all) {
            ## Add all the tables, and return the output
            print add_all_tables();
            ## The above does not commit, so make sure we do it here
            $dbh->commit();
            exit 0;
        }
        else {
            add_table('table');
        }
    }

    ## Add a sequence
    if ($thing eq 'sequence') {
        if ($do_all) {
            ## Add all the sequences, and return the output
            print add_all_sequences();
            ## The above does not commit, so make sure we do it here
            $dbh->commit();
            exit 0;
        }
        else {
            add_table('sequence');
        }
    }

    ## Anything past this point is an error
    if ($do_all) {
        warn qq{The 'all' option can only be used with 'table' and 'sequence'\n};
        exit 1;
    }

    warn "$usage\n";
    exit 1;

} ## end of add_item


sub update_item {

    ## Update some object in the database
    ## This merely passes control on to the more specific update_ functions
    ## Arguments: none (but parses nouns)
    ## Returns: undef

    my $usage = usage('update');

    ## Must have at least two nouns
    if ($#nouns < 2) {
        warn "$usage\n";
        exit 1;
    }

    ## What type of thing are we updating?
    my $thing = shift @nouns;

    ## Account for variations and abbreviations
    $thing = standardize_name($thing);

    ## These do return due to recursion, so we must exit afterwards
    (update_customcode() or exit 1) if $thing eq 'customcode';
    ## We do not update customname or customcols
    ## The dbgroup must be checked before the database (dbg vs db)
    (update_database(@nouns) or exit 1) if $thing eq 'database';
    (update_dbgroup(@nouns)  or exit 1) if $thing eq 'dbgroup';
    (update_herd(@nouns)     or exit 1) if $thing eq 'herd';
    (update_sync(@nouns)     or exit 1) if $thing eq 'sync';
    (update_table(@nouns)    or exit 1) if $thing eq 'table' or $thing eq 'sequence';

    ## If we got this far, we have an unknown thing
    warn "$usage\n";
    exit 1;

} ## end of update_item


sub list_item {

    ## Show information about one or more items in the bucardo database
    ## Arguments: none, but parses nouns
    ## Returns: 0 on success, -1 on error

    my $usage = usage('list');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## First word is the type if thing we are listing
    my $thing = shift @nouns;

    ## Account for variations and abbreviations
    $thing = standardize_name($thing);

    SWITCH: {
        $thing eq 'customcode' and do {
                list_customcodes();
                last SWITCH;
        };
        $thing eq 'customname' and do {
                list_customnames();
                last SWITCH;
        };
        $thing eq 'customcols' and do {
                list_customcols();
                last SWITCH;
        };
        ## The dbgroup must be checked before the database (dbg vs db)
        $thing eq 'dbgroup' and do {
                list_dbgroups();
                last SWITCH;
        };
        $thing eq 'database' and do {
                list_databases();
                last SWITCH;
        };
        $thing eq 'herd' and do {
                list_herds();
                last SWITCH;
        };
        $thing eq 'sync' and do {
                list_syncs();
                last SWITCH;
        };
        $thing eq 'table' and do {
                list_tables();
                last SWITCH;
        };
        $thing eq 'sequence' and do {
                list_sequences();
                last SWITCH;
        };
        $thing eq 'all' and do {
                print "-- customcodes:\n";  list_customcodes();
                print "-- customnames:\n";  list_customnames();
                print "-- customcols:\n";   list_customcols();
                print "-- dbgroups:\n";     list_dbgroups();
                print "-- databases:\n";    list_databases();
                print "-- herds:\n";        list_herds();
                print "-- syncs:\n";        list_syncs();
                print "-- tables:\n";       list_tables();
                print "-- sequences:\n";    list_sequences();
                print "\n";
                last SWITCH;
        };

        ## catch all
        ## Cannot list anything else
        warn "$usage\n";
        exit 1;

    } # SWITCH

    exit 0;
} ## end of list_item


sub remove_item {

    ## Delete from the bucardo database
    ## Arguments: none, but parses nouns
    ## Returns: never, exits

    my $usage = usage('remove');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## First word is the type if thing we are removing
    my $thing = shift @nouns;

    ## Account for variations and abbreviations
    $thing = standardize_name($thing);

    ## All of these will exit and do not return
    remove_customcode() if $thing eq 'customcode';
    remove_customname() if $thing eq 'customname';
    remove_customcols() if $thing eq 'customcols';
    ## The dbgroup must be checked before the database (dbg vs db)
    remove_database()   if $thing eq 'database';
    remove_dbgroup()    if $thing eq 'dbgroup';
    remove_herd()       if $thing eq 'herd';
    remove_sync()       if $thing eq 'sync';
    remove_table()      if $thing eq 'table';
    remove_sequence()   if $thing eq 'sequence';

    ## Do not know how to remove anything else
    warn "$usage\n";
    exit 1;

} ## end of remove_item


##
## Database-related subroutines: add, remove, update, list
##

sub add_database {

    ## Add one or more databases. Inserts to the bucardo.db table
    ## By default, we do a test connection as well (turn off with the --force argument)
    ## Arguments: two or more
    ## 1. The internal name Bucardo uses to refer to this database
    ## 2+ name=value parameters, dash-dash arguments
    ## Returns: undef
    ## Example: bucardo add db nyc1 dbname=nyc1 dbhost=nyc1.example.com dbgroup=sales

    ## Grab our generic usage message
    my $usage = usage('add_database');

    ## The first word is the internal name (bucardo.db.name)
    my $item_name = shift @nouns || '';

    ## No name is a problem
    if (! length $item_name) {
        warn "$usage\n";
        exit 1;
    }

    ## Inputs and aliases, database column name, flags, default value
    my $validcols = qq{
        null                     name                 0                $item_name
        type|dbtype              dbtype               0                postgres
        user|username|dbuser     dbuser               0                bucardo
        pass|password|dbpass     dbpass               0                null
        db|dbname                dbname               0                null
        port|dbport|pgport       dbport               numeric          ENV:PGPORT
        host|dbhost|pghost       dbhost               0                ENV:PGHOSTADDR|PGHOST
        conn|dbconn|pgconn       dbconn               0                null
        stat|status              status               =active|inactive null
        group|dbgroup            none                 0                skip
        addalltables             none                 0                skip
        addallsequences          none                 0                skip
        server_side_prepares|ssp server_side_prepares TF               null
        service|dbservice        dbservice            0                null
        makedelta                makedelta            =on|off          null
    };


    ## TODO: See if makedelta still needed
    ## If so, add to wiki page: http://bucardo.org/wiki/Bucardo/add_database#Optional_arguments

    my ($dbcols,$cols,$phs,$vals,$extra)
        = process_simple_args({cols => $validcols, list => \@nouns, usage => $usage});

    ## Must have a database name
    if (! exists $vals->{dbname}) {
        print qq{Cannot add database: must supply a database name to connect to\n};
        exit 1;
    }

    ## Cannot add if already there
    if (exists $DB->{$item_name}) {
        print qq{Cannot add database: the name "$item_name" already exists\n};
        exit 1;
    }

    ## Clean up and standardize the type name
    my $dbtype = $vals->{dbtype} = standardize_rdbms_name($vals->{dbtype});

    ## Attempt to insert this into the database
    $SQL = "INSERT INTO bucardo.db ($cols) VALUES ($phs)";
    debug("SQL: $SQL");
    debug(Dumper $vals);
    $sth = $dbh->prepare($SQL);
    my $evalok = 0;
    eval {
        $sth->execute(map { $vals->{$_} } sort keys %$vals);
        $evalok = 1;
    };
    if (! $evalok) {
        if ($@ =~ /"db_dsn_unique"/) {
            die qq{Cannot add database: already have a connection with the same parameters\n};
        }
        if ($@ =~ /"db_name_sane"/) {
            die qq{Invalid name: you cannot refer to this database as "$item_name"\n};
        }
        die "Failed to add database: $@\n";
    }

    ## Store certain messages so we can output them in a desired order
    my $finalmsg = '';

    ## Test database handle
    my $testdbh;

    ## May want to do a test connection to the database
  TESTCONN: {

        ## Nothing else to do for flatfiles
        last TESTCONN if 'flatfile' eq $dbtype;

        ## Get the module namem the way to refer to its database
        ## This also makes sure we have a valid type
        my %dbtypeinfo = (
            drizzle  => ['DBD::drizzle',  'Drizzle database'],
            mongo    => ['MongoDB',       'MongoDB'],
            mysql    => ['DBD::mysql',    'MySQL database'],
            mariadb  => ['DBD::mysql',    'MariaDB database'],
            oracle   => ['DBD::Oracle',   'Oracle database'],
            postgres => ['DBD::Pg',       'PostgreSQL database'],
            redis    => ['Redis',         'Redis database'],
            sqlite   => ['DBD::SQLite',   'SQLite database'],
        );
        if (! exists $dbtypeinfo{$dbtype}) {
            die qq{Unknown database type: $dbtype\n};
        }
        my ($module,$fullname) = @{ $dbtypeinfo{$dbtype} };

        ## Gather connection information from the database via db_getconn
        $SQL = 'SELECT bucardo.db_getconn(?)';
        $sth = $dbh->prepare($SQL);
        $sth->execute($item_name);
        my $dbconn = $sth->fetchall_arrayref()->[0][0];

        ## Must be able to load the Perl driver
        $evalok = 0;
        eval {
            eval "require $module";
            $evalok = 1;
        };
        if (! $evalok) {
            die "Cannot add unless the Perl module '$module' is available: $@\n";
        }

        ## Reset for the evals below
        $evalok = 0;

        ## Standard args for the DBI databases
        ## We put it here as we may move around with the Postgres bucardo user trick
        my ($type,$dsn,$user,$pass) = split /\n/ => $dbconn;

        ## Sometimes we need to adjust the username on the fly and retry
        my $realuser = $user;

        ## Handle all of the ones that do not use standard DBI first

        if ('mongo' eq $dbtype) {

            my $dsn = {};
            for my $line (split /\n/ => $dbconn) {
                next if $line !~ /(\w+):\s+(.+)/;
                $dsn->{$1} = $2;
            }

            eval {
                $testdbh = MongoDB::Connection->new($dsn); ## no critic
                $evalok = 1;
            };
        }

        elsif ('redis' eq $dbtype) {

            my $tempdsn = {};
            for my $line (split /\n/ => $dbconn) {
                next if $line !~ /(\w+):\s+(.+)/;
                $tempdsn->{$1} = $2;
            }
            my $server;
            if (exists $tempdsn->{host}) {
                $server = $tempdsn->{host};
            }
            if (exists $tempdsn->{port}) {
                $server .= ":$tempdsn->{port}";
            }
            my @dsn;
            if (defined $server) {
                push @dsn => 'server', $server;
            }

            $evalok = 0;
            eval {
                $testdbh = Redis->new(@dsn);
                $evalok = 1;
            };
        }

        ## Anything else must be something with a standard DBI driver
        else {

          DBICONNTEST: {
                eval {
                    $testdbh = DBI->connect($dsn, $user, $pass, {AutoCommit=>0,RaiseError=>1,PrintError=>0});
                    $evalok = 1;
                };
            }
        }

        ## At this point, we have eval'd a connection

        if (! $evalok) {

            ## For Postgres, we get a little fancy and try to account for instances
            ## where the bucardo user may not exist yet, by reconnecting and
            ## creating said user if needed.
            if ('postgres' eq $dbtype and $user eq 'bucardo' and $@ =~ /FATAL.*"bucardo"/) {
                $user = 'postgres';
                goto DBICONNTEST;
            }

            if ($bcargs->{force}) {
                print "Connection to $fullname failed, but will add anyway.\nError was: $@\n\n";
            }
            else {
                die "Connection to $fullname failed. You may force add it with the --force argument. Error was: $@\n\n";
            }
        }
        else {

            ## If we connected twice to Postgres using a different user,
            ## create the bucardo user now
            if ('postgres' eq $dbtype and $realuser ne $user) {
                if (eval { require Digest::MD5; 1 }) {
                    ## We need a password. If one was not supplied, create one
                    $evalok = 0;
                    eval {
                        my $newpass = $pass;
                        if (! length $pass) {
                            $newpass = generate_password();
                        }
                        my $encpass = Digest::MD5::md5_hex($newpass);
                        $testdbh->do(qq{CREATE USER $realuser SUPERUSER ENCRYPTED PASSWORD '$encpass'});
                        $testdbh->commit();
                        my $extrauser = length $pass ? '' : " with password $newpass";
                        warn "Created superuser '$realuser'$extrauser\n";
                        $evalok = 1;
                    };
                    if (! $evalok) {
                        warn "Unable to create superuser $realuser: $@\n";
                        $bcargs->{force} or exit 1;
                    }
                }
            }

            ## Disconnect
            $testdbh->disconnect() if $module =~ /DBD/;
        }

    } ## end of TESTCONN

    ## If we got a group, process that as well
    if (exists $extra->{dbgroup}) {
        my $gname = $extra->{dbgroup};
        ## We need to store this away as the function below changes the global hash
        my $isnew = exists $DBGROUP->{$gname};

        my ($newgroup, $newrole) = add_db_to_group($item_name, $gname);
        if (! $isnew) {
            $finalmsg .= qq{Created database group "$newgroup"\n};
        }
        $finalmsg .= qq{Added database "$item_name" to group "$newgroup" as $newrole\n};
    }

    ## Adjust the db name so add_all_* can use it
    $bcargs->{db} = $item_name;

    ## Make sure $DB gets repopulated for the add_all_* calls below
    load_bucardo_info(1);

    ## Add in all tables for this database
    $finalmsg .= add_all_tables() if grep /addalltab/i, @nouns;

    ## Add in all sequences for this database
    $finalmsg .= add_all_sequences() if grep /addallseq/i, @nouns;

    if (!$QUIET) {
        print qq{Added database "$item_name"\n};
        $finalmsg and print $finalmsg;
    }

    confirm_commit();

    exit 0;

} ## end of add_database


sub remove_database {

    ## Remove one or more databases. Updates the bucardo.db table
    ## Use the --force argument to clear out related tables and groups
    ## Arguments: one or more
    ## 1+ Name of a database
    ## Returns: undef
    ## Example: bucardo remove db nyc1 nyc2 --force

    ## Grab our generic usage message
    my $usage = usage('remove_database');

    ## Must have at least one name
    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Make sure all named databases exist
    for my $name (@nouns) {
        if (! exists $DB->{$name}) {
            die qq{No such database: $name\n};
        }
    }

    ## Prepare the SQL to delete each database
    $SQL = 'DELETE FROM bucardo.db WHERE name = ?';
    $sth = $dbh->prepare($SQL);

    ## Loop through and attempt to delete each given database
    for my $name (@nouns) {
        ## Wrap in an eval so we can handle known exceptions
        my $evalok = 0;
        eval {
            $sth->execute($name);
            $evalok = 1;
        };
        if (! $evalok) {
            if ($bcargs->{force} and $@ =~ /"goat_db_fk"|"dbmap_db_fk"/) {
                $QUIET or warn qq{Dropping all tables and dbgroups that reference database "$name"\n};
                $dbh->rollback();
                $dbh->do('DELETE FROM bucardo.goat WHERE db = ' . $dbh->quote($name));
                $dbh->do('DELETE FROM bucardo.dbmap WHERE db = ' . $dbh->quote($name));
                ## Try the same query again
                eval {
                    $sth->execute($name);
                };
            }

            ## We've failed: output a reasonable message when possible
            if ($@ =~ /"goat_db_fk"/) {
                die qq{Cannot delete database "$name": must remove all tables that reference it first (try --force)\n};
            }
            if ($@ =~ /"dbmap_db_fk"/) {
                die qq{Cannot delete database "$name": must remove all dbmap references first (try --force)\n};
            }
            $@ and die qq{Could not delete database "$name"\n$@\n};
        }
    }

    for my $name (@nouns) {
        $QUIET or print qq{Removed database "$name"\n};
    }

    confirm_commit();

    exit 0;

} ## end of remove_database


sub update_database {

    ## Update one or more databases.
    ## This may modify the bucardo.db, bucardo.dbgroup, and bucardo.dbmap tables
    ## Arguments: two plus
    ## 1. Name of the database to update. Can be "all" and can have wildcards
    ## 2+ What exactly we are updating.
    ## Returns: undef
    ## Example: bucardo update db nyc1 port=6543 group=nycservers:source,globals

    my @actions = @_;

    ## Grab our generic usage message
    my $usage = usage('update_database');

    if (! @actions) {
        print "$usage\n";
        exit 1;
    }

    my $name = shift @actions;

    ## Recursively call ourselves for wildcards and 'all'
    return if ! check_recurse($DB, $name, @actions);

    ## Make sure this database exists!
    if (! exists $DB->{$name}) {
        die qq{Could not find a database named "$name"\nUse 'list dbs' to see all available.\n};
    }

    ## Everything is a name=value setting after this point
    ## We will ignore and allow noise word "set"
    for my $arg (@actions) {
        next if $arg =~ /set/i;
        next if $arg =~ /\w+=\w+/o;
        print "$usage\n";
        exit 1;
    }

    ## Change the arguments into a hash
    my $args = process_args(join ' ' => @actions);

    ## Track what changes we made
    my %change;

    ## Walk through and handle each argument pair
    for my $setting (sort keys %$args) {

        ## Change the name to a more standard form, to better figure out what they really mean
        ## This also excludes all non-alpha characters
        my $newname = transform_name($setting);

        ## Exclude ones that cannot / should not be changed (e.g. cdate)
        if (exists $column_no_change{$newname}) {
            print "Sorry, the value of $setting cannot be changed\n";
            exit 1;
        }

        ## Standardize the values as well
        my $value = $args->{$setting};
        my $newvalue = transform_value($value);

        ## Handle all the non-standard columns
        if ($newname =~ /^group/) {

            ## Track the changes and publish at the end
            my @groupchanges;

            ## Grab the current hash of groups
            my $oldgroup = $DB->{$name}{group} || '';

            ## Keep track of what groups they end up in, so we can remove as needed
            my %donegroup;

            ## Break apart into individual groups
            for my $fullgroup (split /\s*,\s*/ => $newvalue) {

                my ($group,$role,$extra) = extract_name_and_role($fullgroup);

                ## Note that we've found this group
                $donegroup{$group}++;

                ## Does this group exist?
                if (! exists $DBGROUP->{$group}) {
                    create_dbgroup($group);
                    push @groupchanges => qq{Created database group "$group"};
                }

                ## Are we a part of it already?
                if ($oldgroup and exists $oldgroup->{$group}) {

                    ## Same role?
                    my $oldrole = $oldgroup->{$group}{role};
                    if ($oldrole eq $role) {
                        $QUIET or print qq{No change: database "$name" already belongs to group "$group" as $role\n};
                    }
                    else {
                        change_db_role($role,$group,$name);
                        push @groupchanges => qq{Changed role for database "$name" in group "$group" from $oldrole to $role};
                    }
                }
                else {
                    ## We are not a part of this group yet
                    add_db_to_group($name, "$group:$role");
                    push @groupchanges => qq{Added database "$name" to group "$group" as $role};
                }

                ## Handle any extra modifiers
                if (keys %$extra) {
                    update_dbmap($name, $group, $extra);
                    my $list = join ',' => map { "$_=$extra->{$_}" } sort keys %$extra;
                    push @groupchanges => qq{For database "$name" in group "$group", set $list};
                }

            } ## end each group specified

            ## See if we are removing any groups
            if ($oldgroup) {
                for my $old (sort keys %$oldgroup) {
                    next if exists $donegroup{$old};

                    ## Remove this database from the group, but do not remove the group itself
                    remove_db_from_group($name, $old);
                    push @groupchanges => qq{Removed database "$name" from group "$old"};
                }
            }

            if (@groupchanges) {
                for (@groupchanges) {
                    chomp;
                    $QUIET or print "$_\n";
                }
                confirm_commit();
            }

            ## Go to the next setting
            next;

        } ## end of 'group' adjustments

        ## This must exist in our hash
        if (! exists $DB->{$name}{$newname}) {
            print qq{Cannot change "$newname"\n};
            next;
        }
        my $oldvalue = $DB->{$name}{$newname};

        ## Has this really changed?
        if ($oldvalue eq $newvalue) {
            print "No change needed for $newname\n";
            next;
        }

        ## Add to the queue. Overwrites previous ones
        $change{$newname} = [$oldvalue, $newvalue];

    } ## end each setting

    ## If we have any changes, attempt to make them all at once
    if (%change) {
        my $SQL = 'UPDATE bucardo.db SET ';
        $SQL .= join ',' => map { "$_=?" } sort keys %change;
        $SQL .= ' WHERE name = ?';
        my $sth = $dbh->prepare($SQL);
        eval {
            $sth->execute((map { $change{$_}[1] } sort keys %change), $name);
        };
        if ($@) {
            $dbh->rollback();
            $dbh->disconnect();
            print "Sorry, failed to update the bucardo.db table. Error was:\n$@\n";
            exit 1;
        }

        for my $item (sort keys %change) {
            my ($old,$new) = @{ $change{$item} };
            print "Changed bucardo.db $item from $old to $new\n";
        }

        confirm_commit();
    }

    return;

} ## end of update_database


sub list_databases {

    ## Show information about databases. Queries the bucardo.db table
    ## Arguments: zero or more
    ## 1+ Databases to view. Can be "all" and can have wildcards
    ## Returns: 0 on success, -1 on error
    ## Example: bucardo list db sale%

    ## Grab our generic usage message
    my $usage = usage('list_databases');

    ## Might be no databases yet
    if (! keys %$DB) {
        print "No databases have been added yet\n";
        return -1;
    }

    ## If not doing all, keep track of which to show
    my %matchdb;

    for my $term (@nouns) {

        ## Special case for all: same as no nouns at all, so simply remove them!
        if ($term =~ /\ball\b/i) {
            undef %matchdb;
            undef @nouns;
            last;
        }

        ## Check for wildcards
        if ($term =~ s/[*%]/.*/) {
            for my $name (keys %$DB) {
                $matchdb{$name} = 1 if $name =~ /^$term$/;
            }
            next;
        }

        ## Must be an exact match
        for my $name (keys %$DB) {
            $matchdb{$name} = 1 if $name eq $term;
        }

    } ## end each term

    ## No matches?
    if (@nouns and ! keys %matchdb) {
        print "No matching databases found\n";
        return -1;
    }

    ## We only show the type if they are different from each othera
    my %typecount;

    ## Figure out the length of each item for a pretty display
    my ($maxdb,$maxtype,$maxstat,$maxlim1,$maxlim2,$showlim) = (1,1,1,1,1,0);
    for my $name (sort keys %$DB) {
        next if @nouns and ! exists $matchdb{$name};
        my $info = $DB->{$name};
        $typecount{$info->{dbtype}}++;
        $maxdb   = length $info->{name} if length $info->{name} > $maxdb;
        $maxtype = length $info->{dbtype} if length $info->{dbtype} > $maxtype;
        $maxstat = length $info->{status} if length $info->{status} > $maxstat;
    }

    ## Do we show types?
    my $showtypes = keys %typecount > 1 ? 1 : 0;

    ## Now do the actual printing
    for my $name (sort keys %$DB) {
        next if @nouns and ! exists $matchdb{$name};
        my $info = $DB->{$name};
        my $type = sprintf 'Type: %-*s  ',
            $maxtype, $info->{dbtype};
        printf 'Database: %-*s  %sStatus: %-*s  ',
            $maxdb, $info->{name},
            $showtypes ? $type : '',
            $maxstat, $info->{status};
        my $showhost = length $info->{dbhost} ? " -h $info->{dbhost}" : '';
        my $dbtype = $info->{dbtype};
        if ($dbtype eq 'postgres') {
            print "Conn: psql -p $info->{dbport} -U $info->{dbuser} -d $info->{dbname}$showhost";
            if (! $info->{server_side_prepares}) {
                print ' (SSP is off)';
            }
        }
        if ($dbtype eq 'drizzle') {
            my $showport = (length $info->{dbport} and $info->{dbport} != 3306)
                ? " --port $info->{dbport}" : '';
            printf 'Conn: drizzle -u %s -D %s%s%s',
                $info->{dbuser},
                $info->{dbname},
                $showhost,
                $showport;
        }
        if ($dbtype eq 'flatfile') {
            print "Prefix: $info->{dbname}";
        }
        if ($dbtype eq 'mongo') {
            if (length $info->{dbhost}) {
                print "Host: $info->{dbhost}";
            }
        }
        if ($dbtype eq 'mysql' or $dbtype eq 'mariadb') {
            my $showport = (length $info->{dbport} and $info->{dbport} != 3306)
                ? " --port $info->{dbport}" : '';
            printf 'Conn: mysql -u %s -D %s%s%s',
                $info->{dbuser},
                $info->{dbname},
                $showhost,
                $showport;
        }
        if ($dbtype eq 'oracle') {
            printf 'Conn: sqlplus %s%s',
                $info->{dbuser},
                $showhost ? qq{\@$showhost} : '';
        }
        if ($dbtype eq 'redis') {
            my $server = '';
            if (length $info->{dbhost}) {
                $server .= $info->{dbhost};
            }
            if (length $info->{dbport}) {
                $server .= ":$info->{dbport}";
            }
            if ($server) {
                $server = "server=$server";
                print "Conn: $server";
            }
        }
        if ($dbtype eq 'sqlite') {
            printf 'Conn: sqlite3 %s',
                $info->{dbname};
        }

        print "\n";

        if ($VERBOSE) {

            ## Which database groups is this a member of?
            if (exists $info->{group}) {
                for my $group (sort keys %{ $info->{group} }) {
                    my $i = $info->{group}{$group};
                    my $role = $i->{role};
                    my $gang = $i->{gang};
                    my $pri = $i->{priority};
                    print "  Belongs to database group $group ($role)";
                    $pri and print "  Priority:$pri";
                    print "\n";
                }
            }

            ## Which syncs are using it, and as what role
            if (exists $info->{sync}) {
                for my $syncname (sort keys %{ $info->{sync} }) {
                    print "  Used in sync $syncname in a role of $info->{sync}{$syncname}{role}\n";
                }
            }

            $VERBOSE >= 2 and show_all_columns($info);
        }
    }

    return 0;

} ## end of list_databases


##
## Database-group-related subroutines: add, remove, update, list
##

sub add_dbgroup {

    ## Add one or more database groups. Inserts to the bucardo.dbgroup table
    ## May also insert to the bucardo.dbmap table
    ## Arguments: one plus
    ## 1. The name of the group we are creating
    ## 2+ Databases to add to this group, with optional role information attached
    ## Returns: undef
    ## Example: bucardo add dbgroup nycservers nyc1:source nyc2:source lax1

    ## Grab our generic usage message
    my $usage = usage('add_dbgroup');

    my $name = shift @nouns || '';

    ## Must have a name
    if (!length $name) {
        warn "$usage\n";
        exit 1;
    }

    ## Create the group if it does not exist
    if (! exists $DBGROUP->{$name}) {
        create_dbgroup($name);
        $QUIET or print qq{Created database group "$name"\n};
    }

    ## Add all these databases to the group
    for my $fulldb (@nouns) {

        ## Figure out the optional role
        my ($db,$role) = extract_name_and_role($fulldb);

        ## This database must exist!
        if (! exists $DB->{$db}) {
            print qq{The database "$db" does not exist\n};
            exit 1;
        }

        add_db_to_group($db, "$name:$role");

        $QUIET or print qq{Added database "$db" to group "$name" as $role\n};
    }

    confirm_commit();

    exit 0;

} ## end of add_dbgroup


sub remove_dbgroup {

    ## Remove one or more entries from the bucardo.dbgroup table
    ## Arguments: one or more
    ## 1+ Name of a database group
    ## Returns: undef
    ## Example: bucardo remove dbgroup sales

    ## Grab our generic usage message
    my $usage = usage('remove_dbgroup');

    ## Must have at least one name
    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Make sure all the groups exist
    for my $name (@nouns) {
        if (! exists $DBGROUP->{$name}) {
            die qq{No such database group: $name\n};
        }
    }

    ## Prepare the SQL to delete each group
    $SQL = q{DELETE FROM bucardo.dbgroup WHERE name = ?};
    $sth = $dbh->prepare($SQL);

    for my $name (@nouns) {
        ## Wrap in an eval so we can handle known exceptions
        eval {
            $sth->execute($name);
        };
        if ($@) {
            if ($@ =~ /"sync_dbs_fk"/) {
                if ($bcargs->{force}) {
                    $QUIET or warn qq{Dropping all syncs that reference the dbgroup "$name"\n};
                    $dbh->rollback();
                    $dbh->do('DELETE FROM bucardo.sync WHERE dbs = ' . $dbh->quote($name));
                    eval {
                        $sth->execute($name);
                    };
                    goto NEND if ! $@;
                }
                else {
                    die qq{Cannot remove database group "$name": it is being used by one or more syncs\n};
                }
            }
            die qq{Could not delete database group "$name"\n$@\n};
        }
          NEND:
        $QUIET or print qq{Removed database group "$name"\n};
    }

    confirm_commit();

    exit 0;

} ## end of remove_dbgroup


sub update_dbgroup {

    ## Update one or more database groups
    ## This may modify the bucardo.dbgroup and bucardo.dbmap tables
    ## Arguments: two or more
    ## 1. Group to be updated
    ## 2. Databases to be adjusted, or name change request (name=newname)
    ## Returns: undef
    ## Example: bucardo update dbgroup sales A:slave

    my @actions = @_;

    ## Grab our generic usage message
    my $usage = usage('update_dbgroup');

    if (! @actions) {
        print "$usage\n";
        exit 1;
    }

    my $name = shift @actions;

    ## Recursively call ourselves for wildcards and 'all'
    exit 0 if ! check_recurse($DBGROUP, $name, @actions);

    ## Make sure this database group exists!
    if (! exists $DBGROUP->{$name}) {
        die qq{Could not find a database group named "$name"\nUse 'list dbgroups' to see all available.\n};
    }

    ## From this point on, we have either:
    ## 1. A rename request
    ## 2. A database to add/modify

    ## Track dbs and roles
    my %dblist;

    ## Track if we call confirm_commit or not
    my $changes = 0;

    for my $action (@actions) {
        ## New name for this group?
        if ($action =~ /name=(.+)/) {
            my $newname = $1;
            if ($newname !~ /^$re_dbgroupname$/) {
                die qq{Invalid database group name "$newname"\n};
            }
            next if $name eq $newname; ## Duh
            $SQL = 'UPDATE bucardo.dbgroup SET name=? WHERE name=?';
            $sth = $dbh->prepare($SQL);
            $sth->execute($newname, $name);
            $QUIET or print qq{Changed database group name from "$name" to "$newname"\n};
            $changes++;
            next;
        }

        ## Assume the rest is databases to modify

        ## Default role is always target
        my ($db,$role) = extract_name_and_role($action);
        $dblist{$db} = $role;
    }

    ## Leave now if no databases to handle
    if (! %dblist) {
        $changes and confirm_commit();
        exit 0;
    }

    ## The old list of databases:
    my $oldlist = $DBGROUP->{$name}{db} || {};

    ## Walk through the old and see if any were changed or removed
    for my $db (sort keys %$oldlist) {
        if (! exists $dblist{$db}) {
            remove_db_from_group($db, $name);
            $QUIET or print qq{Removed database "$db" from group "$name"\n};
            $changes++;
            next;
        }
        my $oldrole = $oldlist->{$db}{role};
        my $newrole = $dblist{$db};
        if ($oldrole ne $newrole) {
            change_db_role($newrole, $db, $name);
            $QUIET or print qq{Changed role of database "$db" in group "$name" from $oldrole to $newrole\n};
            $changes++;
        }
    }

    ## Walk through the new and see if any are truly new
    for my $db (sort keys %dblist) {
        next if exists $oldlist->{$db};
        my $role = $dblist{$db};
        add_db_to_group($db, "$name:$role");
        $QUIET or print qq{Added database "$db" to group "$name" as $role\n};
        $changes++;
    }

    confirm_commit() if $changes;

    return;

} ## end of update_dbgroup


sub list_dbgroups {

    ## Show information about all or some subset of the bucardo.dbgroup table
    ## Arguments: zero or more
    ## 1+ Groups to view. Can be "all" and can have wildcards
    ## Returns: 0 on success, -1 on error
    ## Example: bucardo list dbgroups

    ## Grab our generic usage message
    my $usage = usage('list_dbgroups');

    ## Might be no groups yet
    if (! keys %$DBGROUP) {
        print "No database groups have been added yet\n";
        return -1;
    }

    ## If not doing all, keep track of which to show
    my %matchdbg;

    for my $term (@nouns) {

        ## Special case for all: same as no nouns at all, so simply remove them!
        if ($term =~ /\ball\b/i) {
            undef %matchdbg;
            undef @nouns;
            last;
        }

        ## Check for wildcards
        if ($term =~ s/[*%]/.*/) {
            for my $name (keys %$DBGROUP) {
                $matchdbg{$name} = 1 if $name =~ /$term/;
            }
            next;
        }

        ## Must be an exact match
        for my $name (keys %$DBGROUP) {
            $matchdbg{$name} = 1 if $name eq $term;
        }

    } ## end each term

    ## No matches?
    if (@nouns and ! keys %matchdbg) {
        print "No matching database groups found\n";
        return -1;
    }

    ## Figure out the length of each item for a pretty display
    my ($maxlen) = (1);
    for my $name (sort keys %$DBGROUP) {
        next if @nouns and ! exists $matchdbg{$name};
        my $info = $DBGROUP->{$name};
        $maxlen = length $info->{name} if length $info->{name} > $maxlen;
    }

    ## Print it
    for my $name (sort keys %$DBGROUP) {
        next if @nouns and ! exists $matchdbg{$name};
        my $info = $DBGROUP->{$name};
        ## Does it have associated databases?
        my $dbs = '';
        if (exists $DBGROUP->{$name}{db}) {
            $dbs = '  Members:';
            for my $dbname (sort keys %{ $DBGROUP->{$name}{db} }) {
                my $i = $DBGROUP->{$name}{db}{$dbname};
                $dbs .= " $dbname:$i->{role}";
                ## Only show the gang if this group is using multiple gangs
                if ($DBGROUP->{$name}{gangs} >= 2) {
                    $dbs .= ":gang=$i->{gang}";
                }
                ## Only show the priority if <> 0
                if ($i->{priority} != 0) {
                    $dbs .= ":pri=$i->{priority}";
                }
            }
        }
        printf "Database group: %-*s%s\n",
            $maxlen, $name, $dbs;
        $VERBOSE >= 2 and show_all_columns($info);
    }

    return 0;

} ## end of list_dbgroups


##
## Customname-related subroutines: add, exists, remove, list
##

sub add_customname {

    ## Add an item to the customname table
    ## Arguments: none, parses nouns for tablename|goatid, syncname, database name
    ## Returns: never, exits
    ## Examples:
    ## bucardo add customname public.foobar foobarz
    ## bucardo add customname public.foobar foobarz sync=bee
    ## bucardo add customname public.foobar foobarz db=baz
    ## bucardo add customname public.foobar foobarz db=baz sync=bee

    my $item_name = shift @nouns || '';

    my $usage = usage('add_customname');

    ## Must have a second name as well
    my $newname = shift @nouns || '';

    if (!length $item_name or ! length $newname) {
        warn "$usage\n";
        exit 1;
    }

    ## Does this number or name exist?
    if (! exists $GOAT->{$item_name}) {
        print qq{Could not find a matching table for "$item_name"\n};
        exit 1;
    }

    ## Gather the name variations
    my $goat = $GOAT->{$item_name};
    ## If this is a ref due to it being an unqualified name, just use the first one
    $goat = $goat->[0] if ref $goat eq 'ARRAY';
    my ($sname,$tname) = ($goat->{schemaname},$goat->{tablename});

    ## The new name can have a schema. If it does not, use the "old" one
    my $Sname;
    my $Tname = $newname;
    if ($Tname =~ /(.+)\.(.+)/) {
        ($Sname,$Tname) = ($1,$2);
    }
    else {
        $Sname = $sname;
    }

    ## If the new name contains an equal sign, treat as an error
    if ($Tname =~ /=/) {
        warn "$usage\n";
        exit 1;
    }

    ## Names cannot be the same
    if ($sname eq $Sname and $tname eq $Tname) {
        print qq{The new name cannot be the same as the old\n};
        exit 1;
    }

    ## Parse the rest of the arguments
    my (@sync,@db);
    for my $arg (@nouns) {
        ## Name of a sync
        if ($arg =~ /^sync\s*=\s*(.+)/) {
            my $sync = $1;
            if (! exists $SYNC->{$sync}) {
                print qq{No such sync: "$sync"\n};
                exit 1;
            }
            push @sync => $sync;
        }
        elsif ($arg =~ /^(?:db|database)\s*=\s*(.+)/) {
            my $db = $1;
            if (! exists $DB->{$db}) {
                print qq{No such database: "$db"\n};
                exit 1;
            }
            push @db => $db;
        }
        else {
            warn "$usage\n";
            exit 1;
        }
    }

    ## Loop through and start adding rows to customname
    my $goatid = $goat->{id};

    $SQL = "INSERT INTO bucardo.customname(goat,newname,db,sync) VALUES ($goatid,?,?,?)";
    $sth = $dbh->prepare($SQL);

    ## We may have multiple syncs or databases, so loop through
    my $x = 0;
    my @msg;
    {

        ## Setup common message post scripts
        my $message = '';
        defined $db[$x] and $message .= " (for database $db[$x])";
        defined $sync[$x] and $message .= " (for sync $sync[$x])";

        ## Skip if this exact entry already exists
        if (customname_exists($goatid,$newname,$db[$x],$sync[$x])) {
            if (!$QUIET) {
                printf "Already have an entry for %s to %s%s\n",
                    $item_name, $newname, $message;
            }
            next;
        }

        $sth->execute($newname, $db[$x], $sync[$x]);
        push @msg => "Transformed $sname.$tname to $newname$message";

        ## Always go at least one round
        ## We go a second time if there is another sync or db waiting
        $x++;
        redo if defined $db[$x] or defined $sync[$x];
        last;
    }

    if (!$QUIET) {
        for (@msg) {
            chomp; ## Just in case we forgot above
            print "$_\n";
        }
    }

    confirm_commit();

    exit 0;

} ## end of add_customname


sub remove_customname {

    ## Remove one or more entries from the bucardo.customname table
    ## Arguments: one or more
    ## 1+ IDs to be deleted
    ## Returns: undef
    ## Example: bucardo remove customname 7

    ## Grab our generic usage message
    my $usage = usage('remove_customname');

    ## Must have at least one name
    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Make sure each argument is a number
    for my $name (@nouns) {
        if ($name !~ /^\d+$/) {
            warn "$usage\n";
            exit 1;
        }
    }

    ## We want the per-id hash here
    my $cn = $CUSTOMNAME->{id};

    ## Give a warning if a number does not exist
    for my $name (@nouns) {
        if (! exists $cn->{$name}) {
            $QUIET or warn qq{Customname number $name does not exist\n};
        }
    }

    ## Prepare the SQL to delete each customname
    $SQL = 'DELETE FROM bucardo.customname WHERE id = ?';
    $sth = $dbh->prepare($SQL);

    ## Go through and delete any that exist
    for my $number (@nouns) {

        ## We've already handled these in the loop above
        next if ! exists $cn->{$number};

        ## Unlike other items, we do not need an eval,
        ## because it has no cascading dependencies
        $sth->execute($number);

        my $cc = sprintf '%s => %s%s%s',
            $cn->{$number}{tname},
            $cn->{$number}{newname},
            (length $cn->{$number}{sync} ? " Sync: $cn->{$number}{sync}" : ''),
            (length $cn->{$number}{db} ? " Database: $cn->{$number}{db}" : '');

        $QUIET or print qq{Removed customcode $number: $cc\n};

    }

    confirm_commit();

    exit 0;

} ## end of remove_customname


sub customname_exists {

    ## See if an entry already exists in the bucardo.customname table
    ## Arguments: four
    ## 1. Goat id
    ## 2. New name
    ## 3. Database name (can be null)
    ## 4. Sync name (can be null)
    ## Returns: true or false (1 or 0)

    my ($id,$newname,$db,$sync) = @_;

    ## Easy if there are no entries yet!
    return 0 if ! keys %$CUSTOMNAME;

    my $cn = $CUSTOMNAME->{goat};

    ## Quick filtering by the goatid
    return 0 if ! exists $cn->{$id};

    my $matchdb = defined $db ? $db : '';
    my $matchsync = defined $sync ? $sync : '';

    return exists $cn->{$id}{$matchdb}{$matchsync};

} ## end of customname_exists


sub list_customnames {

    ## Show information about all or some subset of the bucardo.customname table
    ## Arguments: zero or more
    ## 1+ Names to view. Can be "all" and can have wildcards
    ## Returns: 0 on success, -1 on error
    ## Example: bucardo list customname

    ## Grab our generic usage message
    my $usage = usage('list_customname');

    ## Might be no entries yet
    if (! keys %$CUSTOMNAME) {
        print "No customnames have been added yet\n";
        return -1;
    }

    my $cn = $CUSTOMNAME->{list};

    ## If not doing all, keep track of which to show
    my $matches = 0;

    for my $term (@nouns) {

        ## Special case for all: same as no nouns at all, so simply remove them!
        if ($term =~ /\ball\b/i) {
            undef @nouns;
            last;
        }

        ## Check for wildcards
        if ($term =~ s/[*%]/.*/) {
            for my $row (@$cn) {
                if ($row->{tname} =~ /$term/) {
                    $matches++;
                    $row->{match} = 1;
                }
            }
            next;
        }

        ## Must be an exact match
        for my $row (@$cn) {
            if ($row->{tname} eq $term) {
                $matches++;
                $row->{match} = 1;
            }
        }

    } ## end each term

    ## No matches?
    if (@nouns and ! $matches) {
        print "No matching customnames found\n";
        return -1;
    }

    ## Figure out the length of each item for a pretty display
    my ($maxid,$maxname,$maxnew,$maxsync,$maxdb) = (1,1,1,1,1);
    for my $row (@$cn) {
        next if @nouns and ! exists $row->{match};
        $maxid   = length $row->{id}      if length $row->{id}      > $maxid;
        $maxname = length $row->{tname}   if length $row->{tname}   > $maxname;
        $maxnew  = length $row->{newname} if length $row->{newname} > $maxnew;
        $maxsync = length $row->{sync}    if length $row->{sync}    > $maxsync;
        $maxdb   = length $row->{db}      if length $row->{db}      > $maxdb;
    }

    ## Now do the actual printing
    ## Sort by tablename, then newname, then sync, then db
    for my $row (sort {
        $a->{tname} cmp $b->{tname}
        or
        $a->{newname} cmp $b->{newname}
        or
        $a->{sync} cmp $b->{sync}
        or
        $a->{db} cmp $b->{db}
        } @$cn) {
        next if @nouns and ! exists $row->{match};
        printf '%-*s Table: %-*s => %-*s',
            1+$maxid, "$row->{id}.",
            $maxname, $row->{tname},
            $maxnew, $row->{newname};
        if ($row->{sync}) {
            printf ' Sync: %-*s',
                $maxsync, $row->{sync};
        }
        if ($row->{db}) {
            printf ' Database: %-*s',
                $maxsync, $row->{db};
        }
        print "\n";

    }

    return 0;

} ## end of list_customnames


##
## Customcols-related subroutines: add, exists, remove, list
##

sub add_customcols {

    ## Add an item to the customcols table
    ## Arguments: none, parses nouns for tablename|goatid, syncname, database name
    ## Returns: never, exits
    ## Examples:
    ## bucardo add customcols public.foobar "select a,b,c"
    ## bucardo add customcols public.foobar "select a,b,c" db=foo
    ## bucardo add customcols public.foobar "select a,b,c" db=foo sync=abc

    my $item_name = shift @nouns || '';

    my $usage = usage('add_customcols');

    ## Must have a clause as well
    my $clause = shift @nouns || '';

    if (!length $item_name or ! length $clause) {
        warn "$usage\n";
        exit 1;
    }

    ## Does this number or name exist?
    if (! exists $GOAT->{$item_name}) {
        print qq{Could not find a matching table for "$item_name"\n};
        exit 1;
    }

    ## Gather the name variations
    my $goat = $GOAT->{$item_name};
    ## If this is a ref due to it being an unqualified name, just use the first one
    $goat = $goat->[0] if ref $goat eq 'ARRAY';
    my ($sname,$tname) = ($goat->{schemaname},$goat->{tablename});

    ## Make sure the clause looks sane
    if ($clause !~ /^\s*SELECT /i) {
        warn "$usage\nThe clause must start with SELECT\n";
        exit 1;
    }

    ## Parse the rest of the arguments
    my (@sync,@db);
    for my $arg (@nouns) {
        ## Name of a sync
        if ($arg =~ /^sync\s*=\s*(.+)/) {
            my $sync = $1;
            if (! exists $SYNC->{$sync}) {
                print qq{No such sync: "$sync"\n};
                exit 1;
            }
            push @sync => $sync;
        }
        elsif ($arg =~ /^(?:db|database)\s*=\s*(.+)/) {
            my $db = $1;
            if (! exists $DB->{$db}) {
                print qq{No such database: "$db"\n};
                exit 1;
            }
            push @db => $db;
        }
        else {
            warn "$usage\n";
            exit 1;
        }
    }

    ## Loop through and start adding rows to customcols
    my $goatid = $goat->{id};

    $SQL = "INSERT INTO bucardo.customcols(goat,clause,db,sync) VALUES ($goatid,?,?,?)";
    $sth = $dbh->prepare($SQL);

    ## We may have multiple syncs or databases, so loop through
    my $x = 0;
    my @msg;
    {
        ## Skip if this exact entry already exists
        next if customcols_exists($goatid,$clause,$db[$x],$sync[$x]);

        $count = $sth->execute($clause, $db[$x], $sync[$x]);
        my $message = qq{New columns for $sname.$tname: "$clause"};
        defined $db[$x] and $message .= " (for database $db[$x])";
        defined $sync[$x] and $message .= " (for sync $sync[$x])";
        push @msg => $message;

        ## Always go at least one round
        ## We go a second time if there is another sync or db waiting
        $x++;
        redo if defined $db[$x] or defined $sync[$x];
        last;
    }

    if (!$QUIET) {
        for (@msg) {
            chomp; ## Just in case we forgot above
            print "$_\n";
        }
    }

    confirm_commit();

    exit 0;

} ## end of add_customcols


sub remove_customcols {

    ## Remove one or more entries from the bucardo.customcols table
    ## Arguments: one or more
    ## 1+ IDs to be deleted
    ## Returns: undef
    ## Example: bucardo remove customcols 7

    ## Grab our generic usage message
    my $usage = usage('remove_customcols');

    ## Must have at least one name
    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Make sure each argument is a number
    for my $name (@nouns) {
        if ($name !~ /^\d+$/) {
            warn "$usage\n";
            exit 1;
        }
    }

    ## We want the per-id hash here
    my $cc = $CUSTOMCOLS->{id};

    ## Give a warning if a number does not exist
    for my $name (@nouns) {
        if (! exists $cc->{$name}) {
            $QUIET or warn qq{Customcols number $name does not exist\n};
        }
    }

    ## Prepare the SQL to delete each customcols
    $SQL = 'DELETE FROM bucardo.customcols WHERE id = ?';
    $sth = $dbh->prepare($SQL);

    ## Go through and delete any that exist
    for my $name (@nouns) {

        ## We've already handled these in the loop above
        next if ! exists $cc->{$name};

        ## Unlike other items, we do not need an eval,
        ## because it has no cascading dependencies
        $sth->execute($name);

        my $cc2 = sprintf '%s => %s%s%s',
            $cc->{$name}{tname},
            $cc->{$name}{clause},
            (length $cc->{$name}{sync} ? " Sync: $cc->{$name}{sync}" : ''),
            (length $cc->{$name}{db} ? " Database: $cc->{$name}{db}" : '');

        $QUIET or print qq{Removed customcols $name: $cc2\n};

    }

    confirm_commit();

    exit 0;

} ## end of remove_customcols


sub customcols_exists {

    ## See if an entry already exists in the bucardo.customcols table
    ## Arguments: four
    ## 1. Goat id
    ## 2. Clause
    ## 3. Database name (can be null)
    ## 4. Sync name (can be null)
    ## Returns: true or false (1 or 0)

    my ($id,$clause,$db,$sync) = @_;

    ## Easy if there are no entries yet!
    return 0 if ! keys %$CUSTOMCOLS;

    my $cc = $CUSTOMCOLS->{goat};

    ## Quick filtering by the goatid
    return 0 if ! exists $cc->{$id};

    ## And by the clause therein
    return 0 if ! exists $cc->{$id}{$clause};

    ## Is there a match for this db and sync combo?
    for my $row (@{ $cc->{$id}{$clause} }) {
        if (defined $db) {
            next if (! length $row->{db} or $row->{db} ne $db);
        }
        else {
            next if length $row->{db};
        }
        if (defined $sync) {
            next if (! length $row->{sync} or $row->{sync} ne $sync);
        }
        else {
            next if length $row->{sync};
        }

        ## Complete match!
        return 1;
    }

    return 0;

} ## end of customcols_exists


sub list_customcols {

    ## Show information about all or some subset of the bucardo.customcols table
    ## Arguments: zero or more
    ## 1+ Names to view. Can be "all" and can have wildcards
    ## Returns: 0 on success, -1 on error
    ## Example: bucardo list customcols

    ## Grab our generic usage message
    my $usage = usage('list_customcols');

    ## Might be no entries yet
    if (! keys %$CUSTOMCOLS) {
        print "No customcols have been added yet\n";
        return -1;
    }

    my $cc = $CUSTOMCOLS->{list};

    ## If not doing all, keep track of which to show
    my $matches = 0;

    for my $term (@nouns) {

        ## Special case for all: same as no nouns at all, so simply remove them!
        if ($term =~ /\ball\b/i) {
            undef @nouns;
            last;
        }

        ## Check for wildcards
        if ($term =~ s/[*%]/.*/) {
            for my $row (@$cc) {
                if ($row->{tname} =~ /$term/) {
                    $matches++;
                    $row->{match} = 1;
                }
            }
            next;
        }

        ## Must be an exact match
        for my $row (@$cc) {
            if ($row->{tname} eq $term) {
                $matches++;
                $row->{match} = 1;
            }
        }

    } ## end each term

    ## No matches?
    if (@nouns and ! $matches) {
        print "No matching customcols found\n";
        return -1;
    }

    ## Figure out the length of each item for a pretty display
    my ($maxid,$maxname,$maxnew,$maxsync,$maxdb) = (1,1,1,1,1);
    for my $row (@$cc) {
        next if @nouns and ! exists $row->{match};
        $maxid   = length $row->{id}     if length $row->{id}      > $maxid;
        $maxname = length $row->{tname}  if length $row->{tname}   > $maxname;
        $maxnew  = length $row->{clause} if length $row->{clause}  > $maxnew;
        $maxsync = length $row->{sync}   if length $row->{sync}    > $maxsync;
        $maxdb   = length $row->{db}     if length $row->{db}      > $maxdb;
    }

    ## Now do the actual printing
    ## Sort by tablename, then newname, then sync, then db
    for my $row (sort {
        $a->{tname} cmp $b->{tname}
        or
        $a->{clause} cmp $b->{clause}
        or
        $a->{sync} cmp $b->{sync}
        or
        $a->{db} cmp $b->{db}
        } @$cc) {
        next if @nouns and ! exists $row->{match};
        printf '%-*s Table: %-*s => %-*s',
            1+$maxid, "$row->{id}.",
            $maxname, $row->{tname},
            $maxnew, $row->{clause};
        if ($row->{sync}) {
            printf ' Sync: %-*s',
                $maxsync, $row->{sync};
        }
        if ($row->{db}) {
            printf ' Database: %-*s',
                $maxsync, $row->{db};
        }
        print "\n";

    }

    return 0;

} ## end of list_customcols


##
## Table-related subroutines: add, remove, update, list
##

sub add_table {

    ## Add one or more tables. Inserts to the bucardo.goat table
    ## May also update the bucardo.herd and bucardo.herdmap tables
    ## Arguments: one or more
    ## 1+ Names of tables to be added
    ## Returns: undef
    ## Example: bucardo add table pgbench_accounts foo% myschema.abc

    ## Grab our generic usage message
    my $usage = usage('add_table');

    ## Must have at least one table
    if (! @nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Inputs and aliases, database column name, flags, default
    my $validcols = q{
        db                       db                   0                null
        ping                     ping                 TF               null
        rebuild_index            rebuild_index        numeric          null
        standard_conflict        standard_conflict    0                null
        analyze_after_copy       analyze_after_copy   TF               null
        herd                     herd                 0                skip
    };

    my ($dbcols,$cols,$phs,$vals,$extra)
        = process_simple_args({cols => $validcols, list => \@nouns, usage => $usage});

    ## Loop through all the args and attempt to add the tables
    ## This returns a hash with the following keys: relations, match, nomatch
    my $goatlist = get_goat_ids(args => \@nouns, dbcols => $dbcols);

    ## The final output. Store it up all at once for a single QUIET check
    my $message = '';

    ## We will be nice and indicate anything that did not match
    if (keys %{ $goatlist->{nomatch} }) {
        $message .= "Did not find matches for the following terms:\n";
        for (sort keys %{ $goatlist->{nomatch} }) {
            $message .= "  $_\n";
        }
    }

    ## Now we need to output which ones were recently added
    if (keys %{ $goatlist->{new} }) {
        $message .= "Added the following tables:\n";
        for (sort keys %{ $goatlist->{new} }) {
            $message .= "  $_\n";
        }
    }

    ## If they requested a herd and it does not exist, create it
    if (exists $extra->{herd}) {
        my $herdname = $extra->{herd};
        if (! exists $HERD->{$herdname}) {
            $SQL = 'INSERT INTO bucardo.herd(name) VALUES(?)';
            $sth = $dbh->prepare($SQL);
            $sth->execute($herdname);
            $message .= qq{Created the herd named "$herdname"\n};
        }
        ## Now load all of these tables into this herd
        $SQL = 'INSERT INTO bucardo.herdmap (herd,priority,goat) VALUES (?,?,'
            . q{ (SELECT id FROM goat WHERE schemaname||'.'||tablename=? AND db=?))};

        $sth = $dbh->prepare($SQL);

        ## Which tables were already in the herd, and which were just added
        my (@oldnames,@newnames);

        for my $name (sort keys %{ $goatlist->{relations} }) {
            ## Is it already part of this herd?
            if (exists $HERD->{$herdname}{goat}{$name}) {
                push @oldnames => $name;
                next;
            }
            my $db = $goatlist->{relations}{$name}{goat}{db};

            my $pri = 0;

            $count = $sth->execute($herdname,$pri,$name, $db);

            push @newnames => $name;
        }

        if (@oldnames) {
            $message .= qq{The following tables were already in the herd "$herdname":\n};
            for (@oldnames) {
                $message .= "  $_\n";
            }
        }

        if (@newnames) {
            $message .= qq{The following tables are now part of the herd "$herdname":\n};
            for (sort numbered_relations @newnames) {
                $message .= "  $_\n";
            }
        }

    } ## end if herd

    if (!$QUIET) {
        print $message;
    }

    confirm_commit();

    exit 0;

} ## end of add_table


sub remove_table {

    ## Usage: remove table tablename [t2 t3 ...]
    ## Arguments: none, parses nouns for tables
    ## Returns: never, exits

    my $usage = usage('remove_table');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Prepare our SQL
    $SQL = q{DELETE FROM bucardo.goat WHERE schemaname||'.'||tablename = ?};
    $sth = $dbh->prepare($SQL);

    for my $name (@nouns) {
        if ($name =~ /^\w[\w\d]*\.\w[\w\d]*$/) {
            if (! exists $GOAT->{$name}) {
                print qq{No such table: $name\n};
            }
            eval {
                $sth->execute($name);
            };
            if ($@) {
                die qq{Could not delete goat "$name"\n$@\n};
            }
        }
        else {
            die qq{Please use the full schema.table name\n};
        }
    }

    print "Removed the following tables:\n";
    for my $name (sort numbered_relations @nouns) {
        print qq{  $name\n};
    }

    confirm_commit();

    exit 0;

} ## end of remove_table

sub remove_sequence {

    die 'Write me';

} ## end of remove_sequence


sub update_table {

    ## Update one or more tables
    ## This may modify the bucardo.goat and bucardo.herdmap tables
    ## Arguments: two or more
    ## 1. Table to be updated
    ## 2+. Items to be adjusted (name=value)
    ## Returns: undef
    ## Example: bucardo update table quad ping=false

    my @actions = @_;

    ## Grab our generic usage message
    my $usage = usage('update_table');

    if (! @actions) {
        print "$usage\n";
        exit 1;
    }

    my $name = shift @actions;

    ## Recursively call ourselves for wildcards and 'all'
    exit 0 if ! check_recurse($GOAT, $name, @actions);

    ## Make sure this table exists!
    if (! exists $GOAT->{$name}) {
        die qq{Could not find a table named "$name"\nUse 'list tables' to see all available.\n};
    }

    ## Store the id so we work with that alone whenever possible
    my $id = $GOAT->{$name}[0]{id};

    ## Everything is a name=value setting after this point
    ## We will ignore and allow noise word "set"
    for my $arg (@actions) {
        next if $arg =~ /set/i;
        next if $arg =~ /\w+=\w+/o;
        print "$usage\n";
        exit 1;
    }

    ## Change the arguments into a hash
    my $args = process_args(join ' ' => @actions);

    ## Track what changes we made
    my %change;

    ## Walk through and handle each argument pair
    for my $setting (sort keys %$args) {

        ## Change the name to a more standard form, to better figure out what they really mean
        ## This also excludes all non-alpha characters
        my $newname = transform_name($setting);

        ## Exclude ones that cannot / should not be changed (e.g. cdate)
        if (exists $column_no_change{$newname}) {
            print "Sorry, the value of $setting cannot be changed\n";
            exit 1;
        }

        ## Standardize the values as well
        my $value = $args->{$setting};
        my $newvalue = transform_value($value);

        ## Handle all the non-standard columns
        if (lc $newname eq 'herd') {

            ## Track the changes and publish at the end
            my @herdchanges;

            ## Grab the current hash of herds
            my $oldherd = $GOAT->{$name}[0]{herd} || '';

            ## Keep track of what groups they end up in, so we can remove as needed
            my %doneherd;

            ## Break apart into individual herds
            for my $herd (split /\s*,\s*/ => $newvalue) {

                ## Note that we've found this herd
                $doneherd{$herd}++;

                ## Does this herd exist?
                if (! exists $HERD->{$herd}) {
                    create_herd($herd);
                    push @herdchanges => qq{Created herd "$herd"};
                }

                ## Are we a part of it already?
                if ($oldherd and exists $oldherd->{$herd}) {
                    $QUIET or print qq{No change: table "$name" already belongs to herd "$herd"\n};
                }
                else {
                    ## We are not a part of this herd yet
                    add_goat_to_herd($herd, $id);
                    push @herdchanges => qq{Added table "$name" to herd "$herd"};
                }

            } ## end each herd specified

            ## See if we are removing any herds
            if ($oldherd) {
                for my $old (sort keys %$oldherd) {
                    next if exists $doneherd{$old};

                    ## We do not want to remove herds here, but maybe in the future
                    ## we can allow a syntax that does
                    next;

                    remove_table_from_herd($name, $old);
                    push @herdchanges => qq{Removed table "$name" from herd "$old"};
                }
            }

            if (@herdchanges) {
                for (@herdchanges) {
                    chomp;
                    $QUIET or print "$_\n";
                }
                confirm_commit();
            }

            ## Go to the next setting
            next;

        } ## end of 'herd' adjustments

        ## This must exist in our hash
        ## We assume it is the first entry for now
        ## Someday be more intelligent about walking and adjusting all matches
        if (! exists $GOAT->{$name}[0]{$newname}) {
            print qq{Cannot change "$newname"\n};
            next;
        }
        my $oldvalue = $GOAT->{$name}[0]{$newname};

        ## May be undef!
        $oldvalue = 'NULL' if ! defined $oldvalue;

        ## Has this really changed?
        if ($oldvalue eq $newvalue) {
            print "No change needed for $newname\n";
            next;
        }

        ## Add to the queue. Overwrites previous ones
        $change{$newname} = [$oldvalue, $newvalue];

    } ## end each setting

    ## If we have any changes, attempt to make them all at once
    if (%change) {
        my $SQL = 'UPDATE bucardo.goat SET ';
        $SQL .= join ',' => map { "$_=?" } sort keys %change;
        $SQL .= ' WHERE id = ?';
        my $sth = $dbh->prepare($SQL);
        eval {
            $sth->execute((map { $change{$_}[1] } sort keys %change), $id);
        };
        if ($@) {
            $dbh->rollback();
            $dbh->disconnect();
            print "Sorry, failed to update the bucardo.goat table. Error was:\n$@\n";
            exit 1;
        }

        for my $item (sort keys %change) {
            my ($old,$new) = @{ $change{$item} };
            print "Changed bucardo.goat $item from $old to $new\n";
        }

        confirm_commit();
    }

    return;

} ## end of update_table


sub list_tables {

    ## Show information about all or some tables in the 'goat' table
    ## Arguments: none (reads nouns for a list of tables)
    ## Returns: 0 on success, -1 on error
    ## Example: bucardo list tables

    ## Grab our generic usage message
    my $usage = usage('list_tables');

    ## Might be no tables yet
    if (! keys %$TABLE) {
        print "No tables have been added yet\n";
        return -1;
    }

    ## If not doing all, keep track of which to show
    my %matchtable;

    for my $term (@nouns) {

        ## Special case for all: same as no nouns at all, so simply remove them!
        if ($term =~ /\ball\b/i) {
            undef %matchtable;
            undef @nouns;
            last;
        }

        ## Check for wildcards
        if ($term =~ s/[*%]/.*/) {
            for my $id (keys %$TABLE) {
                $matchtable{$id} = 1 if $TABLE->{$id}{tablename} =~ /^$term$/;
            }
            next;
        }

        ## Must be an exact match
        for my $id (keys %$TABLE) {
            $matchtable{$id} = 1 if $TABLE->{$id}{tablename} eq $term;
        }

    } ## end each term

    ## No matches?
    if (@nouns and ! keys %matchtable) {
        print "No matching tables found\n";
        return -1;
    }

    ## Figure out the length of each item for a pretty display
    my ($maxid,$maxname,$maxdb,$maxpk) = (1,1,1,1);
    for my $row (values %$TABLE) {
        my $id = $row->{id};
        next if @nouns and ! exists $matchtable{$id};
        $maxid   = length $id if length $id > $maxid;
        my $name = "$row->{schemaname}.$row->{tablename}";
        $maxname = length $name if length $name > $maxname;
        $maxdb = length $row->{db} if length $row->{db} > $maxdb;
        $row->{ppk} = $row->{pkey} ? "$row->{pkey} ($row->{pkeytype})" : 'none';
        $maxpk = length $row->{ppk} if length $row->{ppk} > $maxpk;
    }
    ## Now do the actual printing
    ## Sort by schemaname then tablename
    for my $row (sort numbered_relations values %$TABLE) {
        next if @nouns and ! exists $matchtable{$row->{id}};
        printf '%-*s Table: %-*s  DB: %-*s  PK: %-*s',
            1+$maxid, "$row->{id}.",
            $maxname, "$row->{schemaname}.$row->{tablename}",
            $maxdb, $row->{db},
            $maxpk, $row->{ppk};
        if ($row->{sync}) {
            printf '  Syncs: ';
            print join ',' => sort keys %{ $row->{sync} };
        }
        if (defined $row->{ping}) {
            printf '  ping:%s', $row->{ping} ? 'true' : 'false';
        }
        if ($row->{rebuild_index}) {
            print '  rebuild_index:true';
        }
        print "\n";

        $VERBOSE >= 2 and show_all_columns($row);
    }

    return 0;

} ## end of list_tables


##
## Herd-related subroutines: add, remove, update, list
##

sub add_herd {

    ## Add a herd. Inserts to the bucardo.herd table
    ## May also insert to the bucardo.herdmap and bucardo.goat tables
    ## Arguments: one or more
    ## 1. Name of the herd
    ## 2+ Names of tables or sequences to add. Can have wildcards
    ## Returns: undef
    ## Example: bucardo add herd foobar tab1 tab2

    ## Grab our generic usage message
    my $usage = usage('add_herd');

    my $herdname = shift @nouns || '';

    ## Must have a name
    if (!length $herdname) {
        warn "$usage\n";
        exit 1;
    }

    ## Create the herd if it does not exist
    if (exists $HERD->{$herdname}) {
        print qq{Herd "$herdname" already exists\n};
    }
    else {
        create_herd($herdname);
        $QUIET or print qq{Created herd "$herdname"\n};
    }

    ## Everything else is tables or sequences to add to this herd

    ## How many arguments were we given?
    my $nouncount = @nouns;

    ## No sense going on if no nouns!
    if (! $nouncount) {
        confirm_commit();
        return undef;
    }

    ## Get the list of all requested tables, adding as needed
    my $goatlist = get_goat_ids(args => \@nouns);

    ## The final output. Store it up all at once for a single QUIET check
    my $message = '';

    ## We will be nice and indicate anything that did not match
    if (keys %{ $goatlist->{nomatch} }) {
        $message .= "Did not find matches for the following terms:\n";
        for (sort keys %{ $goatlist->{nomatch} }) {
            $message .= "  $_\n";
        }
    }

    ## Now we need to output which ones were recently added
    if (keys %{ $goatlist->{new} }) {
        $message .= "Added the following tables:\n";
        for (sort keys %{ $goatlist->{new} }) {
            $message .= "  $_\n";
        }
    }


    ## Now load all of these tables into this herd
    $SQL = 'INSERT INTO bucardo.herdmap (herd,priority,goat) VALUES (?,?,'
        . q{ (SELECT id FROM goat WHERE schemaname||'.'||tablename=? AND db=?))};

    $sth = $dbh->prepare($SQL);

    my (@oldnames, @newnames);

    for my $name (sort keys %{ $goatlist->{relations} }) {
        ## Is it already part of this herd?
        if (exists $HERD->{goat}{$name}) {
            push @oldnames => $name;
            next;
        }
        my $db = $goatlist->{relations}{$name}{goat}{db};

        my $pri = 0;

        $count = $sth->execute($herdname,$pri,$name, $db);

        push @newnames => $name;
    }

    if (@oldnames) {
        $message .= qq{The following tables were already in the herd "$herdname":\n};
        for (@oldnames) {
            $message .= "  $_\n";
        }
    }

    if (@newnames) {
        $message .= qq{The following tables are now part of the herd "$herdname":\n};
        for (@newnames) {
            $message .= "  $_\n";
        }
    }

    if (!$QUIET) {
        print $message;
    }

    confirm_commit();

    exit 0;

} ## end of add_herd


sub remove_herd {

    ## Usage: remove herd herdname [herd2 herd3 ...]
    ## Arguments: none, parses nouns
    ## Returns: never, exits

    my $usage = usage('remove_herd');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    my $herd = $global{herd};

    for my $name (@nouns) {
        if (! exists $herd->{$name}) {
            die qq{No such herd: $name\n};
        }
    }

    $SQL = 'DELETE FROM bucardo.herd WHERE name = ?';
    $sth = $dbh->prepare($SQL);
    for my $name (@nouns) {
        eval {
            $sth->execute($name);
        };
        if ($@) {
            if ($@ =~ /"sync_source_herd_fk"/) {
                die qq{Cannot delete herd "$name": must remove all syncs that reference it first\n};
            }
            die qq{Could not delete herd "$name"\n$@\n};
        }
    }

    for my $name (@nouns) {
        print qq{Removed herd "$name"\n};
    }

    $dbh->commit();

    exit 0;

} ## end of remove_herd


sub update_herd {

    ## Update one or more herds
    ## Arguments: none, parses nouns
    ## Returns: never, exits

} ## end of update_herd


sub list_herds {

    ## Show information about all or some subset of the 'herd' table
    ## Arguments: none, parses nouns for herd names
    ## Returns: 0 on success, -1 on error

    my $usage = usage('list_herds');

    ## Any nouns are filters against the whole list
    my $clause = generate_clause({col => 'name', items => \@nouns});
    my $WHERE = $clause ? "WHERE $clause" : '';
    $SQL = "SELECT * FROM bucardo.herd $WHERE ORDER BY name";
    $sth = $dbh->prepare($SQL);
    $count = $sth->execute();
    if ($count < 1) {
        $sth->finish();
        printf "There are no%s entries in the 'herd' table.\n",
            $WHERE ? ' matching' : '';
        return -1;
    }
    $info = $sth->fetchall_arrayref({});

    ## Get sizing information
    my $maxlen = 1;
    for my $row (@$info) {
        $maxlen = length $row->{name} if length $row->{name} > $maxlen;
    }

    for my $row (@$info) {
        my $name = $row->{name};
        my $h = $global{herd}{$name};
        printf 'Herd: %-*s ',
            $maxlen, $name;
        printf ' DB: %s ', $h->{db} if $h->{db};
        ## Got goats?
        if (exists $h->{goat}) {
            print ' Members: ';
            print join ', ' => sort keys %{ $h->{goat} };
        }
        ## Got syncs?
        if (exists $h->{sync}) {
            print "\n  Used in syncs: ";
            print join ', ' => sort keys %{$h->{sync}};
        }
        print "\n";
        $VERBOSE >= 2 and show_all_columns($row);
    }

    return 0;

} ## end of list_herds

##
## Sync-related subroutines: add, remove, update, list
##

sub add_sync {

    ## Add an item to the bucardo.sync table
    ## Arguments: none (uses nouns)
    ## Returns: never, exits

    my $item_name = shift @nouns || '';

    my $usage = usage('add_sync');

    my $batch = 1;

    if (!length $item_name) {
        warn "$usage\n";
        exit 1;
    }

    ## Inputs and aliases, database column name, flags, default
    my $validcols = qq{
        name                     name                 0                $item_name
        herd                     herd                 0                null
        dbs                      dbs                  0                null
        stayalive                stayalive            TF               null
        kidsalive                kidsalive            TF               null
        limitdbs                 limitdbs             numeric          null
        ping                     ping                 TF               null
        do_listen                do_listen            TF               null
        checktime                checktime            interval         null
        status                   status               =active|inactive null
        source_makedelta         source_makedelta     =inherits|on|off null
        target_makedelta         target_makedelta     =inherits|on|off null
        priority                 priority             numeric          null
        analyze_after_copy       analyze_after_copy   TF               null
        overdue                  overdue              interval         null
        expired                  expired              interval         null
        track_rates              track_rates          TF               null
        onetimecopy              onetimecopy          =0|1|2           null
        lifetime                 lifetime             interval         null
        maxkicks                 maxkicks             numeric          null
        rebuild_index|rebuildindex   rebuild_index    numeric          null
        customselect|usecustomselect usecustomselect  TF               null
        tables                   tables               0                skip
    };

    ## Fullcopy syncs get some of their defaults overriden
    ## The controllers and kids never start automatically,
    ## and ping is never on

    my $morph = [
        {
            field => 'synctype',
            value => 'fullcopy',
            new_defaults => 'ping|F stayalive|F kidsalive|F',
        },
    ];

    my ($dbcols,$cols,$phs,$vals)
        = process_simple_args({cols => $validcols, list => \@nouns, usage => $usage, morph => $morph});

    if (! exists $dbcols->{herd}) {
        ## But allow on-the-fly herd creation
        die "Need to specify a herd for this sync\n";
    }
    if (! exists $dbcols->{dbs}) {
        die "Need to specify which databases\n";
    }

    ## Already got a sync by this name?
    if (exists $global{sync}{$item_name}) {
        die qq{A sync with the name "$item_name" already exists\n};
    }

    ## If this is a list of tables, create them as a new herd
    ## Ideally we also re-use the herd if it already exists
    if ($dbcols->{herd} =~ /,/) {
        my @tables = split /\s*,\s*/ => $dbcols->{herd};
        for my $table (sort @tables) {
            if (! exists $global{goat}{$table}) {
                die "No such table or sequence: $table\n";
            }
        }
    }
    else {
        my $name = $dbcols->{herd};
        if (! exists $global{herd}{$name}) {
            my $schema = 'public';

            ## Maybe it's a table?
            if (! exists $global{goat}{$name}) {
                warn "No such herd or table: $name\n";
                warn "To list all herds, use: list herds\n";
                warn "To list all tables, use: list tables\n";
                exit 1;
            }

            ## See if any existing herds match this one
            for my $hname (sort keys %{$global{herd}}) {
                my $number = keys %{$global{herd}{$hname}{hasgoat}{$schema}};
                next unless 1 == $number;
                print qq{Use existing herd "$hname"? Y/N };
                my $ans = $batch ? 'Y' : <STDIN>;
                if ($ans =~ /^y/i) {
                    $dbcols->{herd} = $hname;
                    $vals->{herd} = $hname;
                    last;
                }
            }
            ## If we aren't using an existing herd, create a new one
            $name = $dbcols->{herd};
            if (! exists $global{herd}{$name}) {
                print "Create a new herd with table $name\n";

                ## We will use the name of the sync if free
                ## Otherwise, keep adding numbers to it until we get a free name
                $SQL = 'SELECT 1 FROM bucardo.herd WHERE name = ?';
                $sth = $dbh->prepare($SQL);
                my $gname = $item_name;
                my $x = 1;
                {
                    $count = $sth->execute($gname);
                    $sth->finish();
                    last if $count < 1;
                    $gname  = "${item_name}_$x";
                    $x++;
                    redo;
                }
                $SQL = 'INSERT INTO bucardo.herd (name) VALUES (?)';
                $sth = $dbh->prepare($SQL);
                $count = $sth->execute($gname);
                if (1 != $count) {
                    die qq{Unable to add new herd named "$gname"\n};
                }
                $SQL = 'INSERT INTO bucardo.herdmap(herd,goat) VALUES (?,'
                    . '(SELECT id FROM bucardo.goat WHERE schemaname = ? AND tablename = ?))';
                $sth = $dbh->prepare($SQL);
                $count = $sth->execute($gname, $schema, $name);
                if (1 != $count) {
                    die qq{Unable to add table "$name" to herd "$gname"\n};
                }
                print qq{Created herd "$gname"\n};
                $dbcols->{herd} = $gname;

            }
        } ## end no herd name match

    } ## end single word herd option

    ## Next, the dbs

    ## If this is a single database, turn into a "list" for below
    if (exists $global{db}{$dbcols->{dbs}}) {
        $dbcols->{dbs} .= ',';
    }

    ## Can be a database group, or a list of databases
    ## We want to keep track of how many of each role we have
    my %rolecount;
    if ($dbcols->{dbs} =~ /,/) {
        my @dbs = split /\s*,\s*/ => $dbcols->{dbs};
        my %db;
        my $type = '';
        for my $db (sort @dbs) {
            ## Set the default type of database
            $type = $type eq '' ? 'source' : 'target';
            if ($db =~ s/[=:](.+)//) {
                $type = $1;
            }
            if (! exists $global{db}{$db}) {
                die "No such database: $db\n";
            }
            ## Standardize and check the types
            $type = 'source' if $type =~ /^s/i or $type =~ /^master/i;
            $type = 'target' if $type =~ /^t/i or $type =~ /^rep/i;
            $type = 'fullcopy' if $type =~ /^f/i;
            if ($type !~ /^(source|target|fullcopy)$/) {
                die "Invalid database type: must be source, target, or fullcopy (not $type)\n";
            }
            $db{$db} = $type;
            $rolecount{$type}++;
        }
        ## Do any existing groups match this list exactly?
        my $newlist = join ',' => map { "$_=".$db{$_} } sort keys %db;
        for my $gname (sort keys %{$global{dbgroup}}) {
            my $innerjoin = join ',' => map { "$_=".$global{dbgroup}{$gname}{db}{$_} } sort keys %{$global{dbgroup}{$gname}{db}};
            if ($innerjoin eq $newlist) {
                print qq{That group of databases already exists as dbgroup "$gname"\n};
                print q{Do you wish to use that group instead? Y/N };
                my $ans = $batch ? 'Y' : <STDIN>;
                if ($ans =~ /^y/i) {
                    $dbcols->{dbs} = $gname;
                    $vals->{dbs} = $gname;
                    last;
                }
                ## Don't ask again?
                $newlist = '';
            }
        }

        if ($dbcols->{dbs} =~ /,/) { ## Make sure we still have a list
            ## We will use the name of the sync if free
            ## Otherwise, keep adding numbers to it until we get a free name
            $SQL = 'SELECT 1 FROM bucardo.dbgroup WHERE name = ?';
            $sth = $dbh->prepare($SQL);
            my $gname = $item_name;
            my $x = 1;
            {
                $count = $sth->execute($gname);
                $sth->finish();
                last if $count < 1;
                $gname  = "${item_name}_$x";
                $x++;
                redo;
            }
            $SQL = 'INSERT INTO bucardo.dbgroup (name) VALUES (?)';
            $sth = $dbh->prepare($SQL);
            $count = $sth->execute($gname);
            if (1 != $count) {
                die qq{Unable to add new dbgroup named "$gname"\n};
            }
            $SQL = 'INSERT INTO bucardo.dbmap(dbgroup,db,role) VALUES (?,?,?)';
            $sth = $dbh->prepare($SQL);
            for my $db (sort keys %db) {
                $count = $sth->execute($gname, $db, $db{$db});
                if (1 != $count) {
                    die qq{Unable to add database "$db" to group "$gname"\n};
                }
            }
            print qq{Created database group "$gname"\n};
            $dbcols->{dbs} = $gname;
            $vals->{dbs} = $gname;
        }
    } ## end of multiple databases specified
    else {
        my $dbg = $dbcols->{dbs};
        if (! exists $DBGROUP->{$dbg}) {
            warn "No such database group: $dbg\n";
            warn "To list all database groups, use: list dbgroups\n";
            exit 1;
        }
        for my $db (values %{ $DBGROUP->{$dbg}{db} }) {
            $rolecount{$db->{role}}++;
        }
    }

    ## If this is a pure fullcopy sync, we want to turn stayalive and kidsalive off
    if ($rolecount{'source'} == 1
            and $rolecount{'fullcopy'}
                and ! $rolecount{'target'}) {
        if ($cols !~ /stayalive/) {
            $vals->{stayalive} = 0;
            $phs .= ',?';
            my @cols = split /,/ => $cols;
            $cols = join ',' => sort @cols, 'stayalive';
        }
        if ($cols !~ /kidsalive/) {
            $vals->{kidsalive} = 0;
            $phs .= ',?';
            my @cols = split /,/ => $cols;
            $cols = join ',' => sort @cols, 'kidsalive';
        }
    }

    $dbh->commit();

    ## Attempt to insert this into the database
    $SQL = "INSERT INTO bucardo.sync ($cols) VALUES ($phs)";
    $DEBUG and warn "SQL: $SQL\n";
    $DEBUG and warn Dumper $vals;
    $sth = $dbh->prepare($SQL);
    my @flatvals = map { $vals->{$_} } sort keys %$vals;
    $DEBUG and warn Dumper \@flatvals;
    eval {
        $count = $sth->execute(@flatvals);
    };
    if ($@) {
        die "Failed to add sync: $@\n";
    }

    $dbh->commit();

    if (!$QUIET) {
        print qq{Added sync "$item_name"\n};
    }

    exit 0;

} ## end of add_sync


sub remove_sync {

    ## Usage: remove sync name [name2 name3 ...]
    ## Arguments: none (uses nouns)
    ## Returns: never, exits

    my $usage = usage('remove_sync');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Make sure all named syncs exist
    my $s = $global{sync};
    for my $name (@nouns) {
        if (! exists $s->{$name}) {
            die qq{No such sync: $name\n};
        }
    }

    ## Make sure none of the syncs are currently running
    ## XXX Is there anything we can do to check that the sync is active?

    $SQL = 'DELETE FROM bucardo.sync WHERE name = ?';
    $sth = $dbh->prepare($SQL);

    for my $name (@nouns) {
        eval {
            $sth->execute($name);
        };
        if ($@) {
            if ($@ =~ /"goat_db_fk"/) {
                die qq{Cannot delete sync "$name": must remove all tables that reference it first\n};
            }
            die qq{Could not delete sync "$name"\n$@\n};
        }
    }

    for my $name (@nouns) {
        print qq{Removed sync "$name"\n};
        print "Note: table triggers (if any) are not automatically removed!\n";
    }

    $dbh->commit();

    exit 0;

} ## end of remove_sync

sub update_sync {

    ## Update one or more syncs
    ## Arguments: none (reads nouns for a list of syncs)
    ## Returns: never, exits

    my @actions = @_;

    my $usage = usage('update_sync');

    if (! @actions) {
        print "$usage\n";
        exit 1;
    }

    my $name = shift @actions;

    ## Recursively call ourselves for wildcards and 'all'
    exit 0 if ! check_recurse($SYNC, $name, @actions);

    ## Make sure this sync exists!
    if (! exists $SYNC->{$name}) {
        die qq{Could not find a sync named "$name"\nUse 'list syncs' to see all available.\n};
    }


    my $changes = 0;

    ## Current information about this sync, including column names
    my $syncinfo;

    for my $action (@actions) {
        ## Look for a standard foo=bar or foo:bar format
        if ($action =~ /(.+?)\s*[=:]\s*(.+)/) {
            my ($setting,$value) = (lc $1,$2);

            ## No funny characters please, just boring column names
            $setting =~ /^[a-z_]+$/ or die "Invalid setting: $setting\n";

            ## If we have not already, grab the current information for this sync
            ## We also use this to get the list of valid column names to modify
            if (! defined $syncinfo) {
                $SQL = 'SELECT * FROM sync WHERE name = ?';
                $sth = $dbh->prepare($SQL);
                $count = $sth->execute($name);
                ## Check count
                $syncinfo = $sth->fetchall_arrayref({})->[0];
                for my $col (qw/ cdate /) {
                    delete $syncinfo->{$col};
                }
            }

            ## Is this a valid column?
            if (! exists $syncinfo->{$setting}) {
                die "Invalid setting: $setting\n";
            }

            ## Normalize things like t/f 0/1 others?

            ## Try setting it
            $SQL = "UPDATE sync SET $setting=? WHERE name = ?";
            $sth = $dbh->prepare($SQL);
            $sth->execute($value,$name);
            $changes++;

            next;
        }

        print "Unknown action: $action\n$usage\n";
        exit 1;

    }

    confirm_commit() if $changes;

    return;

} ## end of update_sync


sub list_syncs {

    ## Show information about all or some subset of the 'sync' table
    ## Arguments: none (reads nouns for a list of syncs)
    ## Returns: 0 on success, -1 on error

    my $usage = usage('list_syncs');

    my $syncs = $global{sync};

    ## Do we have at least one name specified (if not, show all)
    my $namefilter = 0;

    for my $term (@nouns) {

        ## Filter out by status: only show active or inactive syncs
        if ($term =~ /^(active|inactive)$/i) {
            my $stat = lc $1;
            for my $name (keys %$syncs) {
                delete $syncs->{$name} if $syncs->{$name}{status} ne $stat;
            }
            next;
        }

        ## Filter out by arbitrary attribute matches
        if ($term =~ /(\w+)\s*=\s*(\w+)/) {
            my ($attrib, $value) = (lc $1,$2);
            for my $name (keys %$syncs) {
                if (! exists $syncs->{$name}{$attrib}) {
                    my $message = "No such sync attribute: $attrib\n";
                    $message .= "Must be one of the following:\n";
                    my $names = join ',' =>
                        sort
                        grep { $_ !~ /\b(?:cdate|name)\b/ }
                        keys %{ $syncs->{$name} };
                    $message .= " $names\n";
                    die $message;
                }
                delete $syncs->{$name} if $syncs->{$name}{$attrib} ne $value;
            }
            next;
        }

        ## Everything else should be considered a sync name
        $namefilter = 1;

        ## Check for wildcards
        if ($term =~ s/[*%]/.*/) {
            for my $name (keys %$syncs) {
                $syncs->{$name}{ok2show} = 1 if $name =~ /$term/;
            }
            next;
        }

        ## Must be an exact match
        for my $name (keys %$syncs) {
            $syncs->{$name}{ok2show} = 1 if $name eq $term;
        }

    }

    ## If we filtered by name, remove all the non-matched ones
    if ($namefilter) {
        for my $name (keys %$syncs) {
            delete $syncs->{$name} if ! exists $syncs->{$name}{ok2show};
        }
    }

    ## Nothing found? We're out of here
    if (! keys %$syncs) {
        print "No syncs found\n";
        return -1;
    }

    ## Determine the size of the output strings for pretty aligning later
    my ($maxname, $maxherd, $maxdbs) = (2,2,2);
    for my $name (keys %$syncs) {
        my $s = $syncs->{$name};
        $maxname = length $name if length $name > $maxname;
        $maxherd = length $s->{herd}{name} if length $s->{herd}{name} > $maxherd;
        $s->{d} = "DB group $s->{dbs}:";
        for (sort keys %{ $s->{dblist} }) {
            $s->{d} .= " $_ ($s->{dblist}{$_}{role})";
        }
        $maxdbs = length $s->{d} if length $s->{d} > $maxdbs;
    }

    ## Now print them out in alphabetic order
    for my $name (sort keys %$syncs) {
        my $s = $syncs->{$name};

        ## Switch to multi-line if database info strings are over this
        my $maxdbline = 50;

        ## Show basic information
        printf "Sync: %-*s  Herd: %-*s %s[%s]\n",
            $maxname, $name,
            $maxherd, $s->{herd}{name},
            $maxdbs > $maxdbline ? '' : "$s->{d}  ",
            ucfirst $s->{status};

        ## Print the second line if needed
        if ($maxdbs > $maxdbline) {
            print "  $s->{d}\n";
        }

        ## Show associated tables if in verbose mode
        if ($VERBOSE >= 1) {
            if (exists $s->{herd}{goat}) {
                my $goathash = $s->{herd}{goat};
                for my $relname (sort {
                                     $goathash->{$b}{priority} <=> $goathash->{$a}{priority}
                                     or $a cmp $b
                                   }
                              keys %{ $goathash }) {
                    printf "  %s %s\n",
                        ucfirst($goathash->{$relname}{reltype}),$relname;
                }
            }
        }

        ## Show all the sync attributes
        $VERBOSE >= 2 and show_all_columns($s);

    } ## end of each sync

    return 0;

} ## end of list_syncs


sub get_goat_ids {

    ## Returns the ids from the goat table for matching relations
    ## Also checks the live database and adds tables to the goat table as needed.
    ## Arguments: key-value pairs:
    ##  - args: arrayref of names to match against. Can have wildcards.
    ##  - dbcols: optional hashref of fields to populate goat table with (e.g. ping=1)
    ## Returns: a hash with:
    ##  - relations: hash of goat objects, key is the fully qualified name
    ##    - original: hash of search term(s) used to find this
    ##    - goat: the goat object
    ##  - nomatch: hash of non-matching terms
    ##  - match: hash of matching terms
    ##  - new: hash of newly added tables

    my %arg = @_;
    my $names = $arg{args} or die;
    my $dbcols = $arg{dbcols} || {};

    ## The final hash we return
    my %relation;

    ## Args that produced a match
    my %match;

    ## Args that produced no matches at all
    my %nomatch;

    ## Keep track of which args we've already done, just in case there are dupes
    my %seenit;

    ## Which tables we added to the goat table
    my %new;

    ## Figure out which database to search in, unless already given
    my $bestdb = exists $dbcols->{db} ? $dbcols->{db} : find_best_db_for_searching();

    ## This check still makes sense: if no databases, there should be nothing in $GOAT!
    if (! defined $bestdb) {
        die "No databases have been added yet, so we cannot add tables!\n";
    }

    my $rdbh = connect_database({name => $bestdb}) or die;

    ## SQL to find a table or a sequence
    ## We do not want pg_table_is_visible(c.oid) here
    my $BASESQL = q{
SELECT nspname||'.'||relname AS name, relkind, c.oid
FROM pg_class c
JOIN pg_namespace n ON (n.oid = c.relnamespace)
WHERE relkind IN ('r')
AND nspname <> 'information_schema'
AND nspname !~ '^pg_'
};

    ## Loop through each argument, and try and find matching goats
  ITEM: for my $item (@$names) {

        ## In case someone entered duplicate arguments
        next if $seenit{$item}++;

        ## Skip if this is not a tablename, but an arguement of the form x=y
        next if index($item, '=') >= 0;

        ## Determine if this item has a dot in it, and/or it is using wildcards
        my $hasadot = index($item,'.') >= 0 ? 1 : 0;
        my $hasstar = (index($item,'*') >= 0 or index($item,'%') >= 0) ? 1 : 0;

        ## Temporary list of matching items
        my @matches;

        ## A list of tables to be bulk added to the goat table
        my %addtable;

        ## We may mutate the arg, so stow away the original
        my $original_item = $item;

        ## On the first pass, we look for matches in the existing $GOAT hash
        ## We may also check the live database afterwards

        ## Wildcards?
        if ($hasstar) {

            ## Change to a regexier form
            $item =~ s/\./\\./g;
            $item =~ s/[*%]/\.\*/g;
            $item = "^$item" if $item !~ /^[\^\.\%]/;
            $item .= '$' if $item !~ /[\$\*]$/;
            ## Pull back all items from the GOAT hash that have a dot in them
            for my $fullname (grep { /\./ } keys %{ $GOAT }) {

                ## We match against the whole thing if we have a dot
                ## in our search term, otherwise we only match the table
                my $searchname = $fullname;
                if (! $hasadot) {
                    (undef,$searchname) = split /\./ => $fullname;
                }

                ## If we got a match, store the item from the GOAT that caused it
                if ($searchname =~ /^$item$/) {
                    push @matches => $fullname;
                }
            }

            ## Setup the SQL to search the live database
            $SQL = $BASESQL . ($hasadot
                ? q{AND nspname||'.'||relname ~ ?}
                : 'AND relname ~ ?');

        } ## end wildcards

        ## A dot with no wildcards: exact match
        ## TODO: Allow foobar. to mean foobar.% ??
        elsif ($hasadot) {

            if (exists $GOAT->{$item}) {
                push @matches => $item;
            }

            ## Setup the SQL to search the live database
            $SQL = $BASESQL . q{AND nspname||'.'||relname = ?};
        }

        ## No wildcards and no dot, so we match all tables regardless of the schema
        else {

            ## Pull back all items from the GOAT hash that have a dot in them
            for my $fullname (grep { /\./ } keys %{ $GOAT }) {
                my ($schema,$table) = split /\./ => $fullname;
                if ($table eq $item) {
                    push @matches => $fullname;
                }
            }

            ## Setup the SQL to search the live database
            $SQL = $BASESQL . 'AND relname = ?';
        }

        ## We do not check the live database if the match was exact
        ## *and* something was found. In all other cases, we go live.
        if ($hasstar or !$hasadot or !@matches) {
            ## Search the live database for matches
            $sth = $rdbh->prepare($SQL);
            ($count = $sth->execute($item)) =~ s/0E0/0/;
            debug(qq{Searched live database "$bestdb" for arg "$item", count was $count});
            for my $row (@{ $sth->fetchall_arrayref({}) }) {

                ## The 'name' is combined "schema.relname"
                my $name = $row->{name};

                ## Don't bother if we have already added this!
                next if exists $GOAT->{$name};

                ## Document the string that led us to this one
                $relation{$name}{original}{$item}++;

                ## Document the fact that we found this on a database
                $new{$name}++;

                ## Mark this item as having produced a match
                $match{$item}++;

                ## Set this table to be added to the goat table below
                $addtable{$name} = {db => $bestdb, reltype => $row->{relkind}, dbcols => $dbcols};

                ## Add this to our matching list
                push @matches => $name;

            }
        }

        ## Add all the tables we just found from searching the live database
        if (keys %addtable) {
            add_items_to_goat_table(\%addtable);
        }

        ## Populate the final hashes based on the match list
        for my $name (@matches) {
            $relation{$name}{original}{$original_item}++;
            $relation{$name}{goat} ||= $GOAT->{$name};
            $match{$item}++;
        }

        ## If this item did not match anything, note that as well
        if (! @matches) {
            $nomatch{$original_item}++;
        }

    } ## end each given needle

    return {
        relations  => \%relation,
        nomatch    => \%nomatch,
        match      => \%match,
        new        => \%new,
    };

} ## end of get_goat_ids


sub add_items_to_goat_table {

    ## Given a list of tables, add them to the goat table as needed
    ## Arguments: one
    ## 1. Hashref where keys are the relnames, and values are additional info:
    ##   - db: the database name (mandatory)
    ##   - reltype: table or sequence (optional, defaults to table)
    ##   - dbcols: optional hashref of goat columns to set
    ## Returns: arrayref with all the new goat.ids

    my $info = shift or die;

    ## Quick check if the entry is already there.
    $SQL = 'SELECT id FROM bucardo.goat WHERE schemaname=? AND tablename=? AND db=?';
    my $isthere = $dbh->prepare($SQL);

    ## SQL to add this new entry in
    my $NEWGOATSQL = 'INSERT INTO bucardo.goat (schemaname,tablename,reltype,db) VALUES (?,?,?,?) RETURNING id';

    my @newid;

    for my $name (sort keys %$info) {
        if ($name !~ /^([\w ]+)\.([\w ]+)$/o) {
            die qq{Invalid name, got "$name", but expected format "schema.relname"};
        }
        my ($schema,$table) = ($1,$2);

        my $db = $info->{$name}{db} or die q{Must provide a database};

        my $reltype = $info->{$name}{reltype} || 't';
        $reltype = $reltype =~ /s/i ? 'sequence' : 'table';

        ## Adjust the SQL as necessary for this goat
        $SQL = $NEWGOATSQL;
        my @args = ($schema, $table, $reltype, $db);
        if (exists $info->{$name}{dbcols}) {
            for my $newcol (sort keys %{ $info->{$name}{dbcols} }) {
                next if $newcol eq 'db';
                $SQL =~ s/\)/,$newcol)/;
                $SQL =~ s/\?,/?,?,/;
                push @args => $info->{$name}{dbcols}{$newcol};
            }
        }
        $sth = $dbh->prepare($SQL);
        ($count = $sth->execute(@args)) =~ s/0E0/0/;

        debug(qq{Added "$schema.$table" to goat table with db "$db", count was $count});

        push @newid => $sth->fetchall_arrayref()->[0][0];
    }

    ## Update the global
    load_bucardo_info('force_reload');

    ## Return a list of goat objects
    my %newlist;
    for my $id (@newid) {
        my $goat = $global{goat}{$id};
        my $name = "$goat->{schemaname}.$goat->{tablename}";
        $newlist{$name} = $goat;
    }

    return \%newlist;


} ## end of add_items_to_goat_table


sub create_dbgroup {

    ## Creates a new entry in the bucardo.dbgroup table
    ## Caller should have alredy checked for existence
    ## Does not commit
    ## Arguments: two
    ## 1. Name of the new group
    ## 2. Boolean: if true, prevents the reload
    ## Returns: undef

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

    $SQL = 'INSERT INTO bucardo.dbgroup(name) VALUES (?)';
    $sth = $dbh->prepare($SQL);
    eval {
        $sth->execute($name);
    };
    if ($@) {
        if ($@ =~ /dbgroup_name_sane/) {
            print "Trying name $name\n";
            print qq{Invalid characters in database group name "$name"\n};
        }
        else {
            print qq{Failed to create database group "$name"\n$@\n};
        }
        exit 1;
    }

    ## Reload our hashes
    $noreload or load_bucardo_info(1);

    return;

} ## end of create_dbgroup


sub get_arg_items {

    ## From an argument list, return all matching items
    ## Arguments: two
    ## 1. Arrayref of source items to match on
    ## 2. Arrayref of arguments
    ## Returns: an arrayref of matches, or an single scalar indicating what arg failed

    my ($haystack, $needles) = @_;

    my %match;

    for my $needle (@$needles) {

        my $hasadot = index($needle,'.') >= 0 ? 1 : 0;
        my $hasstar = (index($needle,'*') >= 0 or index($needle,'%') >= 0) ? 1 : 0;

        ## Wildcards?
        if ($hasstar) {

            ## Change to a regexier form
            $needle =~ s/\*/\.\*/g;

            ## Count matches: if none found, we bail
            my $found = 0;
            for my $fullname (@$haystack) {
                ## If it has a dot, match the whole thing
                if ($hasadot) {
                    if ($fullname =~ /^$needle$/) {
                        $match{$fullname} = $found++;
                    }
                    next;
                }

                ## No dot, so match table part only
                my ($schema,$table) = split /\./ => $fullname;
                if ($table =~ /^$needle$/) {
                    $match{$fullname} = $found++;
                }
            }

            return $needle if ! $found;

            next;

        } ## end wildcards

        ## If it has a dot, it must match exactly
        if ($hasadot) {
            if (grep { $_ eq $needle } @$haystack) {
                $match{$needle} = 1;
                next;
            }
            return $needle;
        }

        ## No dot, so we match all tables regardless of the schema
        my $found = 0;
        for my $fullname (@$haystack) {
            my ($schema,$table) = split /\./ => $fullname;
            if ($table eq $needle) {
                $match{$fullname} = $found++;
            }
        }
        return $needle if ! $found;

    } ## end each given needle


    return \%match; ## May be undefined

} ## end of get_arg_items




sub kick {

    ## Kick one or more syncs
    ## Arguments: none, parses nouns
    ## Returns: never, exits

    my $usage = usage('kick');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    my ($exitstatus, $retries, $do_retry) = (0,0,0);

  RETRY: {
        $dbh->rollback();
        $exitstatus = 0;
      SYNC: for my $sync (@syncs) {
            my $relname = "bucardo_kick_sync_$sync";

            ## If this sync is not active, cowardly refuse to kick it
            if ($SYNC->{$sync}{status} ne 'active') {
                print qq{Cannot kick inactive sync "$sync"\n};
                next SYNC;
            }

            $dbh->do(qq{NOTIFY "bucardo_kick_sync_$sync"});
            my $done = "bucardo_syncdone_$sync";
            my $killed = "bucardo_synckill_$sync";
            if (! defined $adverb) {
                $dbh->commit();
                $QUIET or print qq{Kicked sync $sync\n};
                next;
            }

            $QUIET or print qq{Kick $sync: };
            $dbh->do(qq{LISTEN "$done"});
            $dbh->do(qq{LISTEN "$killed"});
            my $s = $SYNC->{$sync};
            $dbh->do(qq{LISTEN "bucardo_syncdone_$sync"});
            $dbh->commit();

            my $time = time;
            sleep 0.1;

            my $timeout = (defined $adverb and $adverb > 0) ? $adverb : 0;

            my $printstring = $NOTIMER ? '' : '[0 s] ';
            print $printstring unless $QUIET or $NOTIMER;
            my $oldtime = 0;
            local $SIG{ALRM} = sub { die 'Timed out' };
            $do_retry = 0;
            eval {
                if ($timeout) {
                    alarm $timeout;
                }
              WAITIN: {
                    my $lastwait = '';
                    if (time - $time != $oldtime) {
                        $oldtime = time - $time;
                        if (!$QUIET and !$NOTIMER) {
                            print "\b" x length($printstring);
                            $printstring =~ s/\d+/$oldtime/;
                            print $printstring;
                        }
                    }
                    for my $notice (@{ db_get_notices($dbh) }) {
                        my ($name) = @$notice;
                        if ($name eq $done) {
                            $lastwait = 'DONE!';
                        }
                        elsif ($name eq $killed) {
                            $lastwait = 'KILLED!';
                            $exitstatus = 2;
                        }
                        elsif ($name =~ /^bucardo_syncdone_${sync}_(.+)$/) {
                            my $new = sprintf '%s(%ds) ', $1, ceil(time-$time);
                            print $new unless $QUIET;
                            $printstring .= $new;
                        }
                        elsif ($name =~ /^bucardo_synckill_${sync}_(.+)$/) {
                            my $new = sprintf '%s KILLED (%ds) ', $1, ceil(time-$time);
                            print $new unless $QUIET;
                            $printstring .= $new;
                            $exitstatus = 2;
                            $lastwait = ' ';
                        }
                    }
                    $dbh->rollback();
                    if ($lastwait) {
                        print $lastwait unless $QUIET;
                        if ($lastwait ne 'DONE!' and $RETRY and ++$retries <= $RETRY) {
                            print "Retry #$retries\n";
                            $do_retry = 1;
                            die "Forcing eval to exit for retry attempt\n";
                        }
                        last WAITIN;
                    }
                    sleep($WAITSLEEP);
                    redo WAITIN;

                } ## end of WAITIN

                alarm 0 if $timeout;
            };

            alarm 0 if $timeout;
            if ($do_retry) {
                $do_retry = 0;
                redo RETRY;
            }

            if (2 == $exitstatus) {
                my $reason = show_why_sync_killed($sync);
                $reason and print "\n$reason\n";
            }

            if ($@) {
                if ($@ =~ /Timed out/o) {
                    $exitstatus = 1;
                    warn "Timed out!\n";
                }
                else {
                    $exitstatus = 3;
                    warn "Error: $@\n";
                }
                next SYNC;
            }
            next SYNC if $QUIET;

            print "\n";

        } ## end each sync

    } ## end RETRY

    exit $exitstatus;

} ## end of kick


sub show_why_sync_killed {

    ## If a kick results in a "KILLED!" try and show why
    ## Arguments: one
    ## 1. Sync object
    ## Returns: message string

    my $sync = shift;

    $SQL = q{
SELECT * FROM bucardo.q
WHERE sync = ?
AND aborted IS NOT NULL
AND LENGTH(whydie) > 2
AND aborted >= now() - '30 seconds'::interval
ORDER BY aborted DESC LIMIT 1
};
    $sth = $dbh->prepare($SQL);
    $count = $sth->execute($sync);
    if ($count != 1) {
        $sth->finish();
        return '';
    }

    my $result = $sth->fetchall_arrayref({})->[0];
    (my $whydie = $result->{whydie}) =~ s/\\n */\n    /g;
    $whydie =~ s/: ERROR:/:\n    ERROR:/;
    $whydie =~ s/ (at .+ line \d+\.)/\n      $1/g;
    $whydie =~ s/\t/<tab>/g;
    my $message = sprintf "  PID: %s\n  Started: %s\n  Ended: %s\n    %s",
        $result->{pid}, $result->{started}, $result->{aborted}, $whydie;

    return $message;

} ## end of show_why_sync_killed


sub status_all {

    ## Show status of all syncs in the database
    ## Arguments: none
    ## Returns: never, exits

    ## See if the MCP is running and what its PID is
    if (! -e $PIDFILE) {
        #print " (Bucardo does not appear to be currently running)\n";
    }
    else {
        my $fh;
        if (!open $fh, '<', $PIDFILE) {
            print "\nERROR: Could not open $PIDFILE: $!";
        }
        else {
            my $pid = <$fh>;
            chomp $pid;
            close $fh or warn qq{Could not close $PIDFILE: $!\n};
            if ($pid =~ /^\d+$/) {
                print "PID of Bucardo MCP: $pid";
            }
            else {
                print "\nERROR: $PIDFILE contained: $pid";
            }
        }
    }
    print "\n";

    if (! keys %$SYNC) {
        print "No syncs have been created yet.\n";
        exit 0;
    }

    my $orderby = $bcargs->{sort} || '1';
    if ($orderby !~ /^\+?\-?\d$/) {
        die "Invalid sort option, must be +- 1 through 9\n";
    }

    ## Set the status for each sync if possible
    my $max = set_sync_status();

    ## The titles
    my %title = (
        name     => ' Name',
        state    => ' State',
        lastgood => ' Last good',
        timegood => ' Time',
        dit      => ($max->{truncate} ?
                    $max->{conflicts} ? ' Last I/D/T/C' : ' Last I/D/T' :
                    $max->{conflicts} ? ' Last I/D/C' :' Last I/D'),
        lastbad  => ' Last bad',
        timebad  => ' Time',
    );

    ## Set the maximum as needed based on the titles
    for my $name (keys %title) {
        if (! exists $max->{$name}
                or length($title{$name}) > $max->{$name}) {
            $max->{$name} = length $title{$name};
        }
    }

    ## Account for our extra spacing by bumping everything up
    for my $var (values %$max) {
        $var += 2;
    }

    ## Print the column headers
    printf qq{%-*s %-*s %-*s %-*s %-*s %-*s %-*s\n},
        $max->{name},     $title{name},
        $max->{state},    $title{state},
        $max->{lastgood}, $title{lastgood},
        $max->{timegood}, $title{timegood},
        $max->{dit},      $title{dit},
        $max->{lastbad},  $title{lastbad},
        $max->{timebad},  $title{timebad};

    ## Print a fancy dividing line
    printf qq{%s+%s+%s+%s+%s+%s+%s\n},
        '=' x $max->{name},
        '=' x $max->{state},
        '=' x $max->{lastgood},
        '=' x $max->{timegood},
        '=' x $max->{dit},
        '=' x $max->{lastbad},
        '=' x $max->{timebad};

    ## If fancy sorting desired, call the list ourself to sort
    sub sortme {
        my $sortcol = $bcargs->{sort} || 1;

        +1 == $sortcol and return $a cmp $b;
        -1 == $sortcol and return $b cmp $a;

        my ($uno,$dos) = ($SYNC->{$a}, $SYNC->{$b});

        ## State
        +3 == $sortcol and return ($uno->{state} cmp $dos->{state} or $a cmp $b);
        -3 == $sortcol and return ($dos->{state} cmp $uno->{state} or $a cmp $b);

        ## Last good
        +5 == $sortcol and return ($uno->{lastgoodsecs} <=> $dos->{lastgoodsecs} or $a cmp $b);
        -5 == $sortcol and return ($dos->{lastgoodsecs} <=> $uno->{lastgoodsecs} or $a cmp $b);

        ## Good time
        +6 == $sortcol and return ($uno->{lastgoodtime} <=> $dos->{lastgoodtime} or $a cmp $b);
        -6 == $sortcol and return ($dos->{lastgoodtime} <=> $uno->{lastgoodtime} or $a cmp $b);

        if ($sortcol == 7 or $sortcol == -7) {
            my ($total1,$total2) = (0,0);
            while ($uno->{dit} =~ /(\d+)/go) {
                $total1 += $1;
            }
            while ($dos->{dit} =~ /(\d+)/go) {
                $total2 += $1;
            }

            7 == $sortcol and return ($total1 <=> $total2 or $a cmp $b);

            return ($total2 <=> $total1 or $a cmp $b);
        }

        ## Last bad
        +8 == $sortcol and return ($uno->{lastbadsecs} <=> $dos->{lastbadsecs} or $a cmp $b);
        -8 == $sortcol and return ($dos->{lastbadsecs} <=> $uno->{lastbadsecs} or $a cmp $b);

        ## Bad time
        +9 == $sortcol and return ($uno->{lastbadtime} <=> $dos->{lastbadtime} or $a cmp $b);
        -9 == $sortcol and return ($dos->{lastbadtime} <=> $uno->{lastbadtime} or $a cmp $b);


        return $a cmp $b;

    }

    for my $sync (sort sortme keys %$SYNC) {

        my $s = $SYNC->{$sync};

        ## If this has been filtered out, skip it entirely
        next if $s->{filtered};

        ## Populate any missing fields with an empty string
        for my $name (keys %title) {
            if (! exists $s->{$name}) {
                $s->{$name} = '';
            }
        }

        my $X = '|';
        printf qq{%-*s$X%-*s$X%-*s$X%-*s$X%-*s$X%-*s$X%-*s\n},
            $max->{name}," $sync ",
            $max->{state}, " $s->{state} ",
            $max->{lastgood}, " $s->{lastgood} ",
            $max->{timegood}, " $s->{timegood} ",
            $max->{dit}, " $s->{dit} ",
            $max->{lastbad}, " $s->{lastbad} ",
            $max->{timebad}, " $s->{timebad} ";
    }

    exit 0;

} ## end of status_all


sub status_detail {

    ## Show detailed information about one or more syncs
    ## Arguments: none, parses nouns
    ## Returns: never, exits

    ## Walk through and check each given sync
    ## It must either exist, or be a special key word

    my @synclist;
    for my $sync (@nouns) {

        ## Allow a special noise word: 'sync'
        next if $sync eq 'sync';

        ## Add everything if we find the special word 'all'
        if ($sync eq 'all') {
            undef @synclist;
            for my $sync (keys %$SYNC) {
                ## Turn off the filtering that set_sync_status may have added
                $SYNC->{$sync}{filtered} = 0;
                push @synclist => $sync;
            }
            last;
        }

        ## If we don't know about this particular sync, give a warning
        ## We allow another special word: 'all'
        if (!exists $SYNC->{$sync}) {
            ## If a number, skip for ease of "kick name #" toggling
            $sync !~ /^\d+$/ and die "No such sync: $sync\n";
        }
        else {
            push @synclist => $sync;
        }
    }


    ## Verify that all named syncs exist
    my $max = set_sync_status({syncs => \@synclist});

    ## Present each in the order they gave
    my $loops = 0;
    for my $sync (@synclist) {

        my $s = $SYNC->{$sync};

        ## Skip if it has been filtered out
        next if $s->{filtered};

        ## Put a space between multiple entries
        if ($loops++) {
            print "\n";
        }

        print '=' x 70; print "\n";

        my @items;
        my $numtables = keys %{$s->{herd}{goat}};

        my $sourcedb = $s->{herd}{db};

        ## Last good time, and number of rows affected
        if (exists $s->{rowgood}) {
            my $tt = pretty_time($s->{rowgood}{total_time});
            push @items => ['Last good', "$s->{rowgood}{started_time} (time to run: $tt)"];

            ## Space out the numbers
            $s->{dit} =~ s{/}{ / }g;
            ## Pretty comma formatting (based on ENV)
            $s->{dit} =~ s/(\d+)/pretty_number($1)/ge;

            ## Change the title if we have any truncates
            my $extra = $max->{truncates} ? '/truncates' : '';

            ## Change the title if we have any conflicts
            $extra .= $max->{conflicts} ? '/conflicts' : '';

            push @items => ["Rows deleted/inserted$extra", $s->{dit}];
        }

        ## Last bad time, and the exact error
        ## The error should always be last, so we defer adding it to the queue
        my $lasterror = '';
        if (exists $s->{rowbad}) {
            my $tt = pretty_time($s->{rowbad}{total_time});
            push @items => ['Last bad',  "$s->{rowbad}{started_time} (time until fail: $tt)"];

            ## Grab the error message, and strip out trailing whitespace
            ($lasterror = $s->{rowbad}{status}) =~ s/\s+$//;
            ## Add any newlines back in
            $lasterror =~ s/\\n/\n/g;
            ## Remove starting whitespace
            $lasterror =~ s/^\s+//;
        }

        ## Undefined should be written as 'none'
        for (qw/checktime/) {
            $s->{$_} ||= 'none';
        }

        ## Should be 'yes' or 'no'
        for (qw/analyze_after_copy vacuum_after_copy stayalive kidsalive ping do_listen usecustomselect/) {
            $s->{$_} = (defined $s->{$_} and $s->{$_}) ? 'yes' : 'no ';
        }

        ## If currently running, there should be a PID file
        if (exists $s->{PIDFILE}) {
            push @items => ['PID file'         => $s->{PIDFILE}];
            push @items => ['PID file created' => $s->{CREATED}];
        }

        ## Build the display list
        push @items => ['Sync name'            => $sync];
        push @items => ['Current state'        => $s->{state}];
        push @items => ['Source herd/database' => "$s->{herd}{name} / $sourcedb"];
        push @items => ['Tables in sync'       => $numtables];
        push @items => ['Status'               => $s->{status}];
        push @items => ['Check time'           => $s->{checktime}];
        push @items => ['Overdue time'         => $s->{overdue}];
        push @items => ['Expired time'         => $s->{expired}];
        push @items => ['Stayalive/Kidsalive'  => "$s->{stayalive} / $s->{kidsalive}"];
        push @items => ['Rebuild index'        => $s->{rebuild_index} ? 'Yes' : 'No'];
        push @items => ['Ping'                 => $s->{ping}];
        push @items => ['Onetimecopy'          => $s->{onetimecopy} ? 'Yes' : 'No'];

        ## Only show these if enabled
        if ($s->{usecustomselect} eq 'yes') {
            push @items => ['Custom select', $s->{usecustomselect}];
        }
        if ($s->{analyze_after_copy} eq 'yes') {
            push @items => ['Post-copy analyze', 'Yes'];
        }
        if ($s->{vacuum_after_copy} eq 'yes') {
            push @items => ['Post-copy vacuum', 'Yes'];
        }

        ## Final items:
        push @items => ['Last error:' => $lasterror];

        ## Figure out the maximum size of the left-hand items
        my $leftmax = 0;
        for (@items) {
            $leftmax = length $_->[0] if length $_->[0] > $leftmax;
        }

        ## Print it all out
        for (@items) {
            printf "%-*s : %s\n",
                $leftmax, $_->[0], $_->[1];
        }
        print '=' x 70; print "\n";


    }
    exit 0;

} ## end of status_detail




sub append_reason_file {

    ## Add an entry to the 'reason' log file
    ## Arguments: one
    ## 1. Message to store
    ## Returns: undef

    my $event = shift or die;

    my $string = sprintf "%s | %-5s | %s\n", (scalar localtime), $event, $nouns;

    open my $fh, '>', $REASONFILE or die qq{Could not open "$REASONFILE": $!\n};
    print {$fh} $string;
    close $fh or warn qq{Could not close $REASONFILE: $!\n};
    open $fh, '>>', $REASONFILE_LOG or die qq{Could not open "$REASONFILE_LOG": $!\n};
    print {$fh} $string;
    close $fh or warn qq{Could not close $REASONFILE_LOG: $!\n};

    return;

} ## end of append_reason_file




sub set_sync_status {

    ## Set detailed information about syncs from the synrun table
    ## Arguments: zero or one (hashref)
    ## 1. Hashref containing a. syncs=arrarref of syncnames
    ## Returns: hashref indicating maximum lengths of inner information
    ## If a sync is filtered out via the 'syncs' argument, it is set to $s->{filtered} = 1

    my $arg = shift || {};

    ## View the details of the syncs via the syncrun table

    $SQL = qq{
SELECT *,
TO_CHAR(started,'$DATEFORMAT') AS started_time,
CASE WHEN current_date = ended::date
  THEN TO_CHAR(ended,'$SHORTDATEFORMAT')
  ELSE TO_CHAR(ended,'$DATEFORMAT') END AS ended_time,
ROUND(EXTRACT(epoch FROM ended)) AS ended_epoch,
EXTRACT(epoch FROM ended-started) AS total_time,
ROUND(EXTRACT(epoch FROM now()-started)) AS total_time_started,
ROUND(EXTRACT(epoch FROM now()-ended)) AS total_time_ended
FROM syncrun
WHERE sync = ?
AND (   lastgood  IS TRUE
     OR lastbad   IS TRUE
     OR lastempty IS TRUE
     OR ended IS NULL)
};
    $sth = $dbh->prepare($SQL);

    ## Find the maximum lengths of items so we can line things up pretty
    my %max = (
        name      => 1,
        state     => 1,
        dit       => 1,
        lastgood  => 1,
        timegood  => 1,
        lastbad   => 1,
        timebad   => 1,
    );

    for my $sync (keys %$SYNC) {

        my $s = $SYNC->{$sync};

        ## Sometimes we only want some of them
        if ($arg->{syncs}) {
            if (! grep { $_ eq $sync } @{$arg->{syncs}}) { ## no critic (ProhibitBooleanGrep)
                $s->{filtered} = 1;
                next;
            }
        }

        $max{name} = length($sync) if length($sync) > $max{name};

        $count = $sth->execute($sync);
        if ($count < 1) {
            $sth->finish;
            $s->{state} = 'No records found';
            $max{state} = length($s->{state}) if length($s->{state}) > $max{state};
            next;
        }
        for my $row (@{ $sth->fetchall_arrayref({}) }) {
            if ($row->{lastgood}) {
                $s->{rowgood} = $row;
            }
            elsif ($row->{lastempty}) {
                $s->{rowempty} = $row;
            }
            elsif ($row->{lastbad}) {
                $s->{rowbad} = $row;
            }
            else {
                $s->{runningrow} = $row;
            }
        }

        ## What is the state of this sync? First, is it still actively running?
        if (exists $s->{runningrow}) {
            $s->{state} = "$s->{runningrow}{status}";
        }
        else {
            ## What was the most recent run?
            my $highepoch = 0;
            undef $s->{latestrow};
            my $wintype;
            for my $type (qw/ bad good /) {
                my $r = $s->{"row$type"};
                next if ! defined $r;
                my $etime = $r->{ended_epoch};
                if ($etime >= $highepoch) {
                    $s->{latestrow} = $r;
                    $highepoch = $etime;
                    $wintype = $type;
                }
            }
            if (! defined $s->{latestrow}) {
                $s->{state} = 'Unknown';
                $max{state} = length($s->{state}) if length($s->{state}) > $max{state};
                next;
            }
            $s->{state} =
                $wintype eq 'empty' ? 'Empty'
                : $wintype eq 'bad' ? 'Bad'
                : 'Good';
        }

        ## deletes/inserts/truncates/conflicts
        $s->{dit} = '';
        if (exists $s->{rowgood}) {
            $s->{dit} = "$s->{rowgood}{deletes}/$s->{rowgood}{inserts}";
            if ($s->{rowgood}{truncates}) {
                $max{truncates} = 1;
                $s->{dit} .= "/$s->{rowgood}{truncates}";
            }
            if ($s->{rowgood}{conflicts}) {
                $max{conflicts} = 1;
                $s->{dit} .= "/$s->{rowgood}{conflicts}";
            }
        }
        $s->{lastgood} = exists $s->{rowgood} ? $s->{rowgood}{ended_time} : 'none';
        $s->{timegood} = exists $s->{rowgood} ? pretty_time($s->{rowgood}{total_time_ended}) : '';
        $s->{lastbad} = exists $s->{rowbad} ? $s->{rowbad}{ended_time} : 'none';
        $s->{timebad} = exists $s->{rowbad} ? pretty_time($s->{rowbad}{total_time_ended}) : '';

        for my $var (qw/ state dit lastgood timegood lastbad timebad /) {
            $max{$var} = length($s->{$var}) if length($s->{$var}) > $max{$var};
        }
    }

    return \%max;

} ## end of set_sync_status


sub inspect {

    ## Inspect an item in the database
    ## Arguments: none, parses nouns
    ## Returns: never, exits

    my $usage = usage('inspect');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }
    my $thing = shift @nouns;

    inspect_table() if $thing =~ /tab/i  or $thing eq 't';
    inspect_sync()  if $thing =~ /sync/i or $thing eq 's';
    inspect_herd()  if $thing =~ /herd/i or $thing eq 'h';

    warn "$usage\n";

    exit 1;

} ## end of inspect


sub inspect_table {

    ## Inspect an item from the goat table
    ## Arguments: none, parses nouns
    ## Returns: never, exits

    my $usage = usage('inspect_table');

    if (! @nouns) {
        warn "$usage\n";
        exit 1;
    }

    $SQL = q{SELECT * FROM bucardo.goat WHERE tablename=?};
    my $sth_goat = $dbh->prepare($SQL);
    $SQL = q{SELECT * FROM bucardo.goat WHERE schemaname = ? AND tablename=?};
    my $sth_goat_schema = $dbh->prepare($SQL);
    my @tables;
    for my $name (@nouns) {
        my $sthg;
        if ($name =~ /(.+)\.(.+)/) {
            $sthg = $sth_goat_schema;
            $count = $sthg->execute($1,$2);
        }
        else {
            $sthg = $sth_goat;
            $count = $sthg->execute($name);
        }
        if ($count < 1) {
            die "Unknown table: $name\n";
        }

        for my $row (@{$sthg->fetchall_arrayref({})}) {
            push @tables, $row;
        }

    }

    for my $t (@tables) {
        my ($s,$t,$db,$id) = @$t{qw/schemaname tablename db id/};
        print "Inspecting $s.$t on $db\n";
        ## Grab all other tables referenced by this one
        my $tablist = get_reffed_tables($s,$t,$db);

        ## Check that each referenced table is in a herd with this table

        my %seenit;
        for my $tab (@$tablist) {
            my ($type,$tab1,$tab2,$name,$def) = @$tab;
            my $table = $type==1 ? $tab1 : $tab2;
            if ($table !~ /(.+)\.(.+)/) {
                die "Invalid table information\n";
            }
            my $schema = $1;
            $table = $2;
            next if $seenit{"$schema.$table.$type"}++;

            ## Make sure that each herd with this table also has this new table
            my $ggoat = $global{goat};
            my $hherd = $global{herd};
            for my $herd (sort keys %{$ggoat->{$id}{herd}}) {
                $seenit{fktable} = 1;
                next if exists $hherd->{$herd}{hasgoat}{$schema}{$table};
                printf "Table %s.%s is in herd %s, but %s.%s (used as FK%s) is not\n",
                    $s, $t, $herd, $schema, $table,
                        $type == 1 ? '' : ' target';

            }
            if (! exists $seenit{fktable}) {
                printf "Table %s.%s is used as FK%s by %s.%s\n",
                    $s,$t,$type==1 ? '' : ' target', $schema, $table;
                delete $seenit{fktable};
            }
        }
    }

    exit 0;

} ## end of inspect_table


sub inspect_herd {

    ## Inspect an item from the herd table
    ## Arguments: none, parses nouns
    ## Returns: never, exits

    my $usage = usage('inspect_herd');

    if (! @nouns) {
        warn "$usage\n";
        exit 1;
    }

    die "Not implemented yet\n";

} ## end of inspect_herd


sub inspect_sync {

    ## Inspect an item from the sync table
    ## Arguments: none, parses nouns
    ## Returns: never, exits

    my $usage = usage('inspect_sync');

    if (! @nouns) {
        warn "$usage\n";
        exit 1;
    }

    die "Not implemented yet\n";

} ## end of inspect_sync


sub get_reffed_tables {

    ## Find all tables that are references by the given one
    ## Arguments: three
    ## 1. Schema name
    ## 2. Table name
    ## 3. Database name
    ## Returns: arrayref of tables from the database

    my ($s,$t,$db) = @_;

    my $rdbh = connect_database({name => $db});

    ## So we get the schemas
    $rdbh->do('SET search_path = pg_catalog');

    $SQL= q{
SELECT CASE WHEN conrelid=x.toid THEN 1 ELSE 2 END,
 confrelid::regclass,
 conrelid::regclass,
 conname,
 pg_get_constraintdef(oid, true)
FROM pg_constraint,
(SELECT c.oid AS toid FROM pg_class c JOIN pg_namespace n
   ON (n.oid=c.relnamespace) WHERE nspname=? AND relname=?
) x
WHERE contype = 'f' AND
(confrelid = x.toid OR conrelid = x.toid)
};

    $sth = $rdbh->prepare($SQL);
    $count = $sth->execute($s,$t);
    return $sth->fetchall_arrayref();

} ## end of get_reffed_tables




sub show_all_columns {

    ## Give a detailed listing of a particular row in the bucardo database
    ## Arguments: one
    ## 1. Hashref of information to display
    ## Returns: formatted, sorted, and indented list as a single string

    my $row = shift or die;

    my $maxkey = 1;
    for my $key (keys %$row) {
        next if ref $row->{$key};
        $maxkey = length $key if length $key > $maxkey;
    }
    for my $key (sort {
        ($a eq 'src_code' and $b ne 'src_code' ? 1 : 0)
        or
        ($a ne 'src_code' and $b eq 'src_code' ? -1 : 0)
        or
        $a cmp $b } keys %$row
     ) {
        next if ref $row->{$key};
        printf "    %-*s = %s\n", $maxkey, $key,
            defined $row->{$key} ? $row->{$key} : 'NULL';
    }

    return;

} ## end of show_all_columns


sub process_args {

    ## Break apart a string of args, return a clean hashref
    ## Arguments: one
    ## 1. List of arguments
    ## Returns: hashref

    my $string = shift or return {};
    $string .= ' ';

    my %arg;

    while ($string =~ m/(\w+)\s*=\s*"(.+?)" /g) {
        $arg{lc $1} = $2;
    }
    $string =~ s/\w+\s*=\s*".+?" / /g;

    while ($string =~ m/(\w+)\s*=\s*'(.+?)' /g) {
        $arg{lc $1} = $2;
    }
    $string =~ s/\w+\s*=\s*'.+?' / /g;

    while ($string =~ m/(\w+)\s*=\s*(\S+)/g) {
        $arg{lc $1} = $2;
    }
    $string =~ s/\w+\s*=\s*\S+/ /g;

    if ($string =~ /\S/) {
        $string =~ s/^\s+//;
        $arg{extraargs} = [split /\s+/ => $string];
    }

    ## Clean up and standardize the names
    if (exists $arg{type}) {
        $arg{type} = standardize_rdbms_name($arg{type});
    }

    return \%arg;

} ## end of process_args





sub list_customcodes {

    ## Show information about all or some subset of the 'customcode' table
    ## Arguments: none, parses nouns for customcodes
    ## Returns: 0 on success, -1 on error

    my $usage = usage('list_customcodes');

    ## Any nouns are filters against the whole list
    my $clause = generate_clause({col => 'name', items => \@nouns});
    my $WHERE = $clause ? "WHERE $clause" : '';
    $SQL = "SELECT * FROM bucardo.customcode $WHERE ORDER BY name";
    $sth = $dbh->prepare($SQL);
    $count = $sth->execute();
    if ($count < 1) {
        $sth->finish();
        printf "There are no%s entries in the 'customcode' table.\n",
            $WHERE ? ' matching' : '';
        return -1;
    }

    $info = $sth->fetchall_arrayref({});

    my ($maxname,$maxwhen) = (1,1);
    for my $row (@$info) {
        $maxname = length $row->{name} if length $row->{name} > $maxname;
        $maxwhen = length $row->{whenrun} if length $row->{whenrun} > $maxwhen;
    }

    for my $row (@$info) {
        my $name = $row->{name};
        printf "Code: %-*s  When run: %-*s  Get dbh: %s  Get rows: %s Trigrules: %s\n",
            $maxname, $name,
            $maxwhen, $row->{whenrun},
            $row->{getdbh}, $row->{getrows}, $row->{trigrules};
        if (defined $row->{about} and $VERBOSE) {
            (my $about = $row->{about}) =~ s/(.)^/$1    /gsm;
            print "  About: $about\n";
        }
        $VERBOSE >= 2 and show_all_columns($row);
    }

    return 0;

} ## end of list_customcodes

sub update_customcode {

    die 'Write me';

} ## end of update_customcode




sub list_sequences {

    ## Show information about all or some sequences in the 'goat' table
    ## Arguments: none (reads nouns for a list of sequences)
    ## Returns: 0 on success, -1 on error

    my $usage = usage('list_sequences');

    my $clause = generate_clause({col => 'tablename', items => \@nouns});
    my $WHERE = $clause ? "AND $clause" : '';
    $SQL = "SELECT * FROM bucardo.goat WHERE reltype = 'sequence' $WHERE ORDER BY schemaname, tablename";
    $sth = $dbh->prepare($SQL);
    $count = $sth->execute();
    if ($count < 1) {
        $sth->finish();
        printf "There are no%s entries in the 'goat' table.\n",
            $WHERE ? ' matching' : '';
        return -1;
    }

    $info = $sth->fetchall_arrayref({});
    my $maxq = 1;
    for my $row (@$info) {
        my $len = length "$row->{schemaname}.$row->{tablename}";
        $maxq = $len if $len > $maxq;
    }

    for my $row (@$info) {
        printf "Sequence: %-*s  DB: %s\n",
            $maxq, "$row->{schemaname}.$row->{tablename}",
              $row->{db};
        $VERBOSE >= 2 and show_all_columns($row);
    }


    return 0;

} ## end of list_sequences


sub pretty_time {

    ## Change seconds to a prettier display with hours, minutes, etc.
    ## Arguments: one
    ## 1. Number of seconds
    ## Returns: formatted string

    my $secs = shift;

    ## Round up to the nearest second if decimal places are given
    $secs = ceil($secs);

    ## If we cannot figure out the seconds, give up and return a question mark
    return '?' if ! defined $secs or $secs !~ /^\-?\d+$/o or $secs < 0;

    ## Initialize days, hours, minutes, and seconds
    my ($D,$H,$M,$S) = (0,0,0,0);

    ## Some people do not want days shown, so leave it as an option
    if ($bcargs->{showdays}) {
        if ($secs > 60*60*24) {
            $D = int $secs/(60*60*24);
            $secs -= $D*60*60*24;
        }
    }

    ## Show hours if there is > 1 hour
    if ($secs > 60*60) {
        $H = int $secs/(60*60);
        $secs -= $H*60*60;
    }

    ## Show minutes if there is > 1 minute
    if ($secs > 60) {
        $M = int $secs/60;
        $secs -= $M*60;
    }
    $secs = int $secs;
    my $answer = sprintf "%s%s%s${secs}s",$D ? "${D}d " : '',$H ? "${H}h " : '',$M ? "${M}m " : '';

    ## Detailed listings get compressed
    if ((defined $COMPRESS and $COMPRESS) or (!defined $COMPRESS and !@nouns)) {
        $answer =~ s/ //g;
    }

    return $answer;

} ## end of pretty_time


sub pretty_number {

    ## Format a raw number in a more readable style
    ## Arguments: one
    ## 1. Number
    ## Returns: formatted number

    my $number = shift;

    return $number if $number !~ /^\d+$/ or $number < 1000;

    ## If this is our first time here, find the correct separator
    if (! defined $bcargs->{tsep}) {
        my $lconv = localeconv();
        $bcargs->{tsep} = $lconv->{thousands_sep} || ',';
    }

    ## No formatting at all
    return $number if '' eq $bcargs->{tsep} or ! $bcargs->{tsep};

    (my $reverse = reverse $number) =~ s/(...)(?=\d)/$1$bcargs->{tsep}/g;
    $number = reverse $reverse;
    return $number;

} ## end of pretty_number



sub vate_sync {

    ## Activate or deactivate a sync
    ## Arguments: none (reads verbs and nouns)
    ## Returns: never, exits

    my $name = lc $verb;
    my $ucname = ucfirst $name;
    @nouns or die qq{${name}_sync requires at least one sync name\n};

    my $wait = (defined $adverb and $adverb eq '0') ? 1 : 0;
    for my $sync (@syncs) {
        (my $vname = $ucname) =~ s/e$/ing/;
        $QUIET or print qq{$vname sync $sync};
        my $done = "bucardo_${name}d_sync_$sync";
        $dbh->do(qq{NOTIFY "bucardo_${name}_sync_$sync"});
        if ($wait) {
            print '...';
            $dbh->do(qq{LISTEN "$done"});
        }
        $dbh->commit();
        if (!$wait) {
            print "\n";
            next;
        }
        sleep 0.1;
        wait_for_notice($dbh, $done);
        print "OK\n";
    } ## end each sync

    exit 0;

} ## end of vate_sync




















sub add_customcode {

    ## Add an entry to the bucardo.customcode table
    ## Arguments: none (uses nouns)
    ## Returns: never, exits

    my $item_name = shift @nouns || '';

    my $usage = usage('add_customcode');

    if (!length $item_name) {
        warn "$usage\n";
        exit 1;
    }

    ## Inputs and aliases, database column name, flags, default
    my $whenrunlist = 'before_txn before_check_rows before_trigger_drop after_trigger_drop'
        . ' after_table_sync exception conflict before_trigger_enable after_trigger_enable'
        . ' after_txn before_sync after_sync';
    my $whenrun = join '|' => split /\s+/ => $whenrunlist;
    my $validcols = qq{
        name                     name                 0                $item_name
        about                    about                0                null
        whenrun|when_run         whenrun              =$whenrun        null
        getdbh                   getdbh               TF               null
        getrows                  getrows              TF               null
        trigrules                trigrules            TF               null
        sync                     sync                 0                skip
        goat                     goat                 0                skip
        active                   active               TF               skip
        priority                 priority             number           skip
        src_code                 src_code             0                skip
    };

    my ($dbcols,$cols,$phs,$vals)
        = process_simple_args({cols => $validcols, list => \@nouns, usage => $usage});

    my $newname = $dbcols->{name};

    ## Does this already exist?
    if (exists $CUSTOMCODE->{$newname}) {
        warn qq{Cannot create: customcode "$newname" already exists\n};
        exit 2;
    }

    ## We must have a "whenrun"
    if (! $dbcols->{whenrun}) {
        warn "$usage\n";
        exit 1;
    }

    ## We must have a src_code as a file
    if (! exists $dbcols->{src_code} or ! $dbcols->{src_code}) {
        warn "$usage\n";
        exit 1;
    }
    my $tfile = $dbcols->{src_code};
    if (! -e $tfile) {
        warn qq{Could not find a file named "$tfile"\n};
        exit 2;
    }
    open my $fh, '<', $tfile or die qq{Could not open "$tfile": $!\n};
    my $src = '';
    { local $/; $src = <$fh>; } ## no critic (RequireInitializationForLocalVars)
    close $fh or warn qq{Could not close "$tfile": $!\n};

    ## Attempt to insert this into the database
    $SQL = "INSERT INTO bucardo.customcode ($cols,src_code) VALUES ($phs,?)";
    $DEBUG and warn "SQL: $SQL\n";
    $DEBUG and warn Dumper $vals;
    $sth = $dbh->prepare($SQL);
    eval {
        $count = $sth->execute(@$vals, $src);
    };
    if ($@) {
        die "Failed to add customcode: $@\n";
    }

    my $finalmsg = '';

    ## See if any updates to customcode_map need to be made

    ## Only one of sync or goat can be specified
    if ($dbcols->{sync} and $dbcols->{goat}) {
        die qq{Sorry, you must specify a sync OR a goat, not both\n};
    }

    ## Makes no sense to specify priority or active if no goat or sync
    if (($dbcols->{priority} or $dbcols->{active}) and !$dbcols->{sync} and ! $dbcols->{goat}) {
        die qq{You must specify a sync or a goat when using priority or active\n};
    }

    ## Is this a valid sync?
    if ($dbcols->{sync} and ! exists $global{sync}{$dbcols->{sync}}) {
        die qq{Unknown sync: $dbcols->{sync}\n};
    }

    ## Is this a valid gaot?
    if ($dbcols->{goat} and ! exists $global{goat}{$dbcols->{goat}}) {
        die qq{Unknown goat: $dbcols->{goat}\n};
    }

    ## Add to the customcode_map table
    if ($dbcols->{sync} or $dbcols->{goat}) {
        $SQL = 'INSERT INTO customcode_map(code,';
        my @vals;
        for my $col (qw/sync goat priority active/) {
            if ($dbcols->{$col}) {
                $SQL .= "$col,";
                push @vals => $dbcols->{$col};
            }
        }
        my $phs2 = '?,' x @vals;
        $SQL .= ") VALUES ((SELECT currval('customcode_id_seq')),$phs2)";
        $SQL =~ s/,\)/)/g;
        $sth = $dbh->prepare($SQL);
        eval {
            $count = $sth->execute(@vals);
        };
        if ($@) {
            die "Failed to add customcode_map: $@\n";
        }
    }

    confirm_commit();

    if (!$QUIET) {
        print qq{Added customcode "$newname"\n};
        $finalmsg and print $finalmsg;
    }

    exit 0;

} ## end of add_customcode
















sub _list_databases {

    ## Quick list of available databases
    ## Arguments: none
    ## Returns: list of databases as a single string

    return if ! keys %{ $DB };
    warn "The following databases are available:\n";
    for (sort keys %{ $DB }) {
        next if $DB->{$_}{dbtype} ne 'postgres';
        print "$_\n";
    }
    return;

} ## end of _list_databases


sub add_all_tables {

    ## Add all tables, returns output from add_all_goats
    ## Arguments: none
    ## Returns: output of inner sub

    return add_all_goats('table');

} ## end of add_all_tables


sub add_all_sequences {

    ## Add all tables, returns output from add_all_goats
    ## Arguments: none
    ## Returns: output of inner sub

    return add_all_goats('sequence');

} ## end of add_all_sequences


sub add_all_goats {

    ## Add all tables, or sequences
    ## Arguments: one
    ## 1. The type, table or sequence
    ## Returns: Srting indicating what was done

    my $type = shift;

    ## Usage: add all table(s) | add all sequence(s)
    ## Options:
    ## --db: use this database (internal name from the db table)
    ## --schema or -n: limit to one or more comma-separated schemas
    ## --exclude-schema or -N: exclude these schemas
    ## --table or -t: limit to the given tables
    ## --exclude-table or -t: exclude these tables
    ## --herd: name of the herd to add new tables to
    ## pkonly: exclude tables with no pkey
    ## Returns: text string of results, with a newline

    ## Transform command-line args to traditional format
    ## e.g. db=foo becomes the equivalent of --db=foo
    ## foo becomes foo=1
    for my $noun (@nouns) {
        if ($noun =~ /(\w+)=(\w+)/) {
            $bcargs->{$1} = $2;
        }
        else {
            $bcargs->{$noun} = 1;
        }
    }

    ## If no databases, cowardly refuse to continue
    if (! keys %$DB) {
        die "Sorry, cannot add any ${type}s until at least one database has been added\n";
    }

    ## If there is more than one database, it must be selected via db=
    my $db;
    my $showdbs = 0;
    if (exists $bcargs->{db}) {
        if (! exists $DB->{$bcargs->{db}}) {
            warn qq{Sorry, could not find a database named "$bcargs->{db}"\n};
            $showdbs = 1;
        }
        else {
            $db = $DB->{$bcargs->{db}};
        }
    }
    elsif (keys %$DB == 1) {
        ($db) = values %$DB;
    }
    else {
        ## Grab the most likely candidate
        my $bestdb = find_best_db_for_searching();
        if (! $bestdb) {
            warn "Please specify which database to use with the db=<name> option\n";
            $showdbs = 1;
        }
        else {
            $db = $DB->{$bestdb};
        }
    }

    if ($showdbs) {
        _list_databases();
        exit 1;
    }

    ## Connect to the remote database
    my $dbh2 = connect_database({name => $db->{name}});

    ## Query to pull all tables we may possibly need
    my $kind = $type eq 'table' ? 'r' : 'S';
    $SQL = q{SELECT nspname, relname FROM pg_catalog.pg_class c }
        . q{JOIN pg_catalog.pg_namespace n ON (n.oid = c.relnamespace) }
        . qq{WHERE relkind = '$kind' };

    ## We always exclude information_schema, system, and bucardo schemas
    $SQL .= q{AND n.nspname <> 'information_schema' AND nspname !~ '^pg' AND nspname !~ '^bucardo'};

    my @clause;

    ## If they gave a schema option, restrict the query by namespace
    push @clause => generate_clause({col => 'nspname', items => [$bcargs->{schema}]});

    ## If they have asked to exclude schemas, append that to the namespace clause
    push @clause => generate_clause({col => 'nspname', items => [$bcargs->{'exclude-schema'}], not => 1});

    ## If they gave a table option, restrict the query by relname
    push @clause => generate_clause({col => 'relname', items => [$bcargs->{table}]});

    ## If they have asked to exclude tables, append that to the relname clause
    push @clause => generate_clause({col => 'relname', items => [$bcargs->{'exclude-table'}], not => 1});

    for my $c (@clause) {
        next if ! $c;
        $SQL .= "\nAND ($c)";
    }

    ## Fetch all the items, warn if no matches are found
    $VERBOSE >= 2 and warn "Query: $SQL\n";
    $sth = $dbh2->prepare($SQL);
    $count = $sth->execute();
    if ($count < 1) {
        warn "Sorry, no ${type}s were found\n";
    }

    ## Grab all current tables or sequences from the goat table.
    $SQL = qq{SELECT schemaname, tablename FROM bucardo.goat WHERE reltype= '$type' AND db = '$db->{name}'};
    my %hastable;
    for my $row (@{$dbh->selectall_arrayref($SQL)}) {
        $hastable{$row->[0]}{$row->[1]}++;
    }

    ## Do we have a herd request? Process it if so
    my $herd = '';
    my $addtoherd;
    if (exists $bcargs->{herd}) {
        $herd = $bcargs->{herd};
        $SQL = 'SELECT 1 FROM bucardo.herd WHERE name = ?';
        my $herdcheck = $dbh->prepare($SQL);
        $count = $herdcheck->execute($herd);
        $herdcheck->finish();
        if ($count < 1) {
            print "Creating herd: $herd\n";
            $SQL = 'INSERT INTO bucardo.herd(name) VALUES (?)';
            $herdcheck = $dbh->prepare($SQL);
            $herdcheck->execute($herd);
        }
        else {
            $VERBOSE >= 1 and warn "Herd already exists: $herd\n";
        }
        $SQL = 'INSERT INTO bucardo.herdmap(herd,goat) VALUES (?,?)';
        $addtoherd = $dbh->prepare($SQL);
    }

    ## Get ready to add tables or sequences to the goat table
    $SQL = q{INSERT INTO bucardo.goat (db,schemaname,tablename,reltype) VALUES (?,?,?,?)};
    my $addtable = $dbh->prepare($SQL);

    ## Walk through all returned tables from the remote database
    my %count = (seenit => 0, added => 0);
    my (%old, %new, %fail, $id);
    for my $row (@{$sth->fetchall_arrayref()}) {
        my ($S,$T) = @$row;
        my $tinfo;
        ## Do we already have this one?
        if (exists $hastable{$S}{$T}) {
            $VERBOSE >= 2 and warn "Skipping $type already in goat: $S.$T\n";
            $count{seenit}++;
            $old{$S}{$T} = 1;
            if ($herd) {
                ## In case this is not already in the herd, grab the id and jump down
                $SQL = 'SELECT * FROM goat WHERE db=? AND schemaname=? AND tablename=? AND reltype=?';
                $sth = $dbh->prepare($SQL);
                $count = $sth->execute($db->{name},$S,$T,$type);
                if ($count < 1) {
                    die qq{Could not find $type $S.$T in database "$db->{name}"!\n};
                }
                $tinfo = $sth->fetchall_arrayref({})->[0];
                $id = $tinfo->{id};
                goto HERD;
            }
            next;
        }
        $VERBOSE >= 2 and warn "Attempting to add $S.$T to the goat table\n";
        eval {
            $count = $addtable->execute($db->{name},$S,$T,$type);
        };
        if ($@) {
            warn "$@\n";
            if ($@ =~ /prepared statement.+already exists/) {
                warn "This message usually indicates you are using pgbouncer\n";
                warn "You can probably fix this problem by running:\n";
                warn "$progname update db $db->{name} server_side_prepares=false\n";
                warn "Then retry your command again\n\n";
            }
            exit 1;
        }
        if ($count != 1) {
            $addtable->finish();
            warn "Failed to add $type $S.$T to the goat table!\n";
            $fail{$S}{$T} = 1;
            next;
        }
        $SQL = q{SELECT currval('bucardo.goat_id_seq')};
        $id = $dbh->selectall_arrayref($SQL)->[0][0];
        $VERBOSE >= 2 and warn "ID of new table $S.$T is $id\n";

        $count{added}++;
        $new{$S}{$T} = 1;
      HERD:
        if ($herd) {
            if (! defined $tinfo) {
                $SQL = 'SELECT * FROM goat WHERE id = ?';
                $sth = $dbh->prepare($SQL);
                $sth->execute($id);
                $tinfo = $sth->fetchall_arrayref({})->[0];
            }
            if ($bcargs->{pkonly} and 'table' eq $type and ! length $tinfo->{pkey}) {
                warn "Not adding table $S.$T to herd: no pkey\n";
            }
            else {
                $SQL = 'SELECT 1 FROM herdmap WHERE herd=? AND goat = ?';
                $sth = $dbh->prepare($SQL);
                $count = $sth->execute($herd, $id);
                if ($count < 1) {
                    $addtoherd->execute($herd, $id);
                    print "Added $type $S.$T to herd $herd\n";
                }
            }
        }

    }

    ## Disconnect from the remote database
    $dbh2->disconnect();

    if ($VERBOSE >= 1) {
        if (%new) {
            print "New ${type}s:\n";
            for my $s (sort keys %new) {
                for my $t (sort keys %{$new{$s}}) {
                    print "  $s.$t\n";
                }
            }
        }
        if (%fail) {
            print "Failed to add ${type}s:\n";
            for my $s (sort keys %fail) {
                for my $t (sort keys %{$fail{$s}}) {
                    print "  $s.$t\n";
                }
            }
        }
    }

    my $message = "New ${type}s added: $count{added}\n";
    if ($count{seenit}) {
        $message .= "Already added: $count{seenit}\n";
    }

    return $message;

} ## end of add_all_goats




sub remove_customcode {

    ## Usage: remove customcode name [name2 name3 ...]
    ## Arguments: none (uses nouns)
    ## Returns: never, exits

    my $usage = usage('remove_customcode');

    if (!@nouns) {
        warn "$usage\n";
        exit 1;
    }

    ## Make sure all named codes exist
    my $code = $global{cc};
    for my $name (@nouns) {
        if (! exists $code->{$name}) {
            die qq{No such code: $name\n};
        }
    }

    $SQL = 'DELETE FROM bucardo.customcode WHERE name = ?';
    $sth = $dbh->prepare($SQL);

    for my $name (@nouns) {
        eval {
            $sth->execute($name);
        };
        if ($@) {
            die qq{Could not delete customcode "$name"\n$@\n};
        }
    }

    for my $name (@nouns) {
        print qq{Removed customcode "$name"\n};
    }

    $dbh->commit();

    exit 0;


} ## end of remove_customcode











sub clog {

    ## Output a message to stderr
    ## Arguments: one
    ## 1. Message
    ## Returns: undef

    my $message = shift;
    chomp $message;

    warn "$message\n";

    return;

} ## end of clog


sub schema_exists {

    ## Determine if a named schema exists
    ## Arguments: one
    ## 1. Schema name
    ## Returns: 0 or 1

    my $schema = shift;

    my $SQL = 'SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = ?';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($schema);
    $sth->finish();

    return $count < 1 ? 0 : 1;

} ## end of schema_exists


sub relation_exists {

    ## Determine if a named relation exists
    ## Arguments: two
    ## 1. Schema name
    ## 2. Relation name
    ## Returns: OID of the relation, or 0 if it does not exist

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

    my $SQL = 'SELECT c.oid FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n '.
        'WHERE n.oid=c.relnamespace AND n.nspname = ? AND c.relname = ?';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($schema,$name);
    if ($count == 1) {
        return $sth->fetchall_arrayref()->[0][0];
    }
    $sth->finish();

    return 0;

} ## end of relation_exists


sub domain_exists {

    ## Determine if a named domain exists
    ## Arguments: two
    ## 1. Schema name
    ## 2. Domain name
    ## Returns: 0 or 1

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

    my $SQL =
          q{SELECT 1 FROM pg_catalog.pg_type t }
        . q{JOIN pg_namespace n ON (n.oid = t.typnamespace) }
        . q{WHERE t.typtype = 'd' AND n.nspname = ? AND t.typname = ?};
    my $sth = $dbh->prepare_cached($SQL);
    $count = $sth->execute($schema,$name);
    $sth->finish();

    return $count < 1 ? 0 : 1;

} ## end of domain_exists


sub config_exists {

    ## Checks if a configuration settings exists
    ## Arguments: one
    ## 1. Name of the setting
    ## Returns: 0 or 1

    my $name = shift;

    my $SQL = 'SELECT 1 FROM bucardo.bucardo_config WHERE name = ?';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($name);
    $sth->finish();

    return $count < 1 ? 0 : 1;

} ## end of config_exists


sub constraint_exists {

    ## Determine if a named constraint exists
    ## Arguments: three
    ## 1. Schema name
    ## 2. Table name
    ## 3. Constraint name
    ## Returns: 0 or 1

    my ($schema,$table,$constraint) = @_;

    my $SQL = 'SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, pg_catalog.pg_constraint o '.
        'WHERE n.oid=c.relnamespace AND c.oid=o.conrelid AND n.nspname = ? AND c.relname = ? AND o.conname = ?';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($schema,$table,$constraint);
    $sth->finish();

    return $count < 1 ? 0 : 1;

} ## end of constraint_exists


sub column_exists {

    ## Determine if a named column exists
    ## Arguments: three
    ## 1. Schema name
    ## 2. Table name
    ## 3. Column name
    ## Returns: 0 or 1

    my ($schema,$table,$column) = @_;

    my $SQL = 'SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, '.
        'pg_catalog.pg_attribute a WHERE n.oid=c.relnamespace AND n.nspname = ? AND c.relname = ? '.
        'AND a.attname = ? AND a.attrelid = c.oid';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($schema,$table,$column);
    $sth->finish();

    return $count < 1 ? 0 : 1;

} ## end of column_exists


sub trigger_exists {

    ## Determine if a named trigger exists
    ## Arguments: one
    ## 1. Trigger name
    ## Returns: 0 or 1

    my $name = shift;
    my $SQL = 'SELECT 1 FROM pg_catalog.pg_trigger WHERE tgname = ?';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($name);
    $sth->finish();
    return $count < 1 ? 0 : 1;

} ## end of trigger_exists


sub function_exists {

    ## Determine if a named function exists
    ## Arguments: three
    ## 1. Schema name
    ## 2. Function name
    ## 3. Function arguments (as one CSV string)
    ## Returns: MD5 of the function source if found, otherwise an empty string

    my ($schema,$name,$args) = @_;

    $name = lc $name;
    $SQL = 'SELECT md5(prosrc) FROM pg_proc p, pg_language l '.
        'WHERE p.prolang = l.oid AND proname = ? AND oidvectortypes(proargtypes) = ?';
    $sth = $dbh->prepare($SQL);
    $count = $sth->execute($name,$args);
    if ($count < 1) {
        $sth->finish();
        return '';
    }

    return $sth->fetchall_arrayref()->[0][0];

} ## end of function_exists


sub column_default {

    ## Return the default value for a column in a table
    ## Arguments: three
    ## 1. Schema name
    ## 2. Table name
    ## 3. Column name
    ## Returns: default value if available, otherwise an empty string

    my ($schema,$table,$column) = @_;
    my $SQL = 'SELECT pg_get_expr(adbin,adrelid) FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, '.
        'pg_catalog.pg_attribute a, pg_attrdef d '.
        'WHERE n.oid=c.relnamespace AND n.nspname = ? AND c.relname = ? '.
        'AND a.attname = ? AND a.attrelid = c.oid AND d.adnum = a.attnum AND d.adrelid = a.attrelid';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($schema,$table,$column);
    if ($count < 1) {
        $sth->finish();
        return '';
    }
    return $sth->fetchall_arrayref()->[0][0];

} ## end of column_default


sub column_value {

    ## Return the value of a table's column
    ## Arguments: four
    ## 1. Schema name
    ## 2. Table name
    ## 3. Column name
    ## 4. Where clause
    ## Returns: value if available, otherwise an empty string

    my ($schema,$table,$column,$where) = @_;

    my $SQL = "SELECT $column FROM $schema.$table WHERE $where";
    return $dbh->selectall_arrayref($SQL)->[0][0];

} ## end of column_value


sub column_type {

    ## Return the data type of a table column
    ## Arguments: three
    ## 1. Schema name
    ## 2. Table name
    ## 3. Column name
    ## Returns: data type if available, otherwise an empty string

    my ($schema,$table,$column) = @_;
    my $SQL = 'SELECT  pg_catalog.format_type(a.atttypid, a.atttypmod) '.
        'FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, '.
        'pg_catalog.pg_attribute a '.
        'WHERE n.oid=c.relnamespace AND n.nspname = ? AND c.relname = ? '.
        'AND a.attname = ? AND a.attrelid = c.oid';
    my $sth = $dbh->prepare_cached($SQL);
    my $count = $sth->execute($schema,$table,$column);
    if ($count eq '0E0') {
        $sth->finish();
        return '';
    }
    return $sth->fetchall_arrayref()->[0][0];

} ## end of column_type


sub constraint_definition {

    ## Return the definition for a constraint
    ## Arguments: one
    ## 1. Constraint name
    ## Returns: definition if found, otherwise an empty string

    my $name = shift;

    my $SQL = qq{SELECT pg_get_constraintdef(oid) FROM pg_constraint WHERE conname = '$name'};
    my $def = $dbh->selectall_arrayref($SQL)->[0][0];

    ## Nothing found? Just return an empty string
    return '' if ! defined $def;

    ## Do some cleanups to standardize across versions and match bucardo.schema cleanly
    $def =~ s/\((\(.+\))\)/$1/;
    $def =~ s/\"?(\w+)\"? = ANY \(ARRAY\[(.+)\]\)/$1 IN ($2)/;
    $def =~ s/<> ALL \(ARRAY\[(.+)\]\)/NOT IN ($1)/;
    $def =~ s/::text//g;
    $def =~ s/(\w+) ~ '/$1 ~ E'/g;
    $def =~ s/CHECK \(\((\w+)\) <>/CHECK ($1 <>/;

    return $def;

} ## end of constraint_definition


sub table_comment {

    ## Return the comment of a table
    ## Arguments: two
    ## 1. Schema name
    ## 2. Table name
    ## Returns: comment if available, otherwise an empty string

    my ($schema,$relation) = @_;

    my $SQL = q{SELECT description FROM pg_description WHERE objoid = (}
        . q{ SELECT c.oid FROM pg_class c JOIN pg_namespace n ON (n.oid = c.relnamespace)}
        . q{ WHERE n.nspname = ? AND c.relname = ?)};

    my $sth = $dbh->prepare($SQL);
    $count = $sth->execute($schema,$relation);
    if ($count < 1) {
        $sth->finish();
        return '';
    }
    return $sth->fetchall_arrayref()->[0][0];

} ## end of table_comment


sub domain_comment {

    ## Return the comment of a domain
    ## Arguments: two
    ## 1. Schema name
    ## 2. Domain name
    ## Returns: comment if available, otherwise an empty string

    my ($schema,$relation) = @_;

    my $SQL = q{SELECT description FROM pg_description WHERE objoid = (}
        . q{ SELECT t.oid FROM pg_type t JOIN pg_namespace n ON (n.oid = t.typnamespace)}
        . q{ WHERE t.typtype = 'd' AND n.nspname = ? AND t.typname = ?)};

    my $sth = $dbh->prepare($SQL);
    $count = $sth->execute($schema,$relation);
    if ($count < 1) {
        $sth->finish();
        return '';
    }
    return $sth->fetchall_arrayref()->[0][0];

} ## end of domain_comment


sub find_bucardo_schema {

    ## Locate the best bucardo.schema file and return a file handle and name for it
    ## Arguments: none
    ## Returns: file handle and location of the file

    my $fh;

    ## Start by checking the current directory
    my $schema_file = 'bucardo.schema';
    return ($fh, $schema_file) if open $fh, '<', $schema_file;

    ## Try /usr/local/share/bucardo
    $schema_file = '/usr/local/share/bucardo/bucardo.schema';
    return ($fh, $schema_file) if open $fh, '<', $schema_file;

    ## Try /usr/share/bucardo
    $schema_file = '/usr/share/bucardo/bucardo.schema';
    return ($fh, $schema_file) if open $fh, '<', $schema_file;

    die "Could not find the bucardo.schema file!\n";

} ## end of find_bucardo_schema


sub table_definition {

    ## Pull the complete table definition from the bucardo.schema file
    ## Returns an arrayref of sequences, and the textual table def
    ## Arguments: one
    ## 1. Name of the table
    ## Returns: arrayref of sequences used, table definition

    my $name = shift;

    my $def = '';

    my ($fh, $schema_file) = find_bucardo_schema();
    my @seq;
    while (<$fh>) {
        if (!$def) {
            if (/^CREATE TABLE $name/) {
                $def .= $_;
            }
        }
        else {
            $def .= $_;
            last if /^\);/;
        }
    }
    close $fh or die qq{Could not close "$schema_file": $!\n};
    while ($def =~ /nextval\('(.+?)'/g) {
        push @seq => $1;
    }

    if (! length($def)) {
        die "Could not find the table definition for $name\n";
    }

    return \@seq, $def;

} ## end of table_definition


sub generate_clause {

    ## Generate a snippet of SQL for a WHERE clause
    ## Arguments: one
    ## 1. Hashref of information
    ## Returns: new clause

    my $arg = shift or die;
    return '' if ! $arg->{items} or ! defined $arg->{items}[0];

    my $col = $arg->{col} or die;
    my $items = $arg->{items};
    my ($NOT,$NOTR) = ('','');
    if (exists $arg->{not}) {
        $NOT = 'NOT ';
        $NOTR = '!';
    }
    my $andor = exists $arg->{andor} ? uc($arg->{andor}) : $NOT ? 'AND' : 'OR';

    my (@oneitem,@itemlist);
    for my $name (@{$items}) {
        $name =~ s/^\s*(.+?)\s*$/$1/;
        ## Break into schema and relation
        my $schema = '';
        if ($col eq 'tablename' and $name =~ s/(.+\w)\.(\w.+)/$2/) {
            $schema = $1;
        }

        my $one = 1;
        ## Contains:
        if ($name =~ s/^\*(.+)\*$/$1/) {
            push @oneitem => "$col ${NOTR}~ " . qquote($1);
        }
        ## Starts with:
        elsif ($name =~ s/^\*(.+)/$1/) {
            push @oneitem => "$col ${NOTR}~ " . qquote("$1\$");
        }
        ## Ends with:
        elsif ($name =~ s/(.+)\*$/$1/) {
            push @oneitem => "$col ${NOTR}~ " . qquote("^$1");
        }
        else {
            push @itemlist => qquote($name);
            $one = 0;
        }
        if ($schema) {
            my $col2 = 'schemaname';
            my $old = $one ? pop @oneitem : pop @itemlist;
            if ($schema =~ s/^\*(.+)\*$/$1/) {
                push @oneitem => "($old AND $col2 ${NOTR}~ " . qquote($1) . ')';
            }
            elsif ($schema =~ s/^\*(.+)/$1/) {
                push @oneitem => "($old AND $col2 ${NOTR}~ " . qquote("$1\$") . ')';
            }
            elsif ($schema =~ s/(.+)\*$/$1/) {
                push @oneitem => "($old AND $col2 ${NOTR}~ " . qquote("^$1") . ')';
            }
            else {
                push @oneitem => "($col = $old AND $col2 = " . qquote($schema) . ')';
            }
        }
    }
    if (@itemlist) {
        my $list = sprintf '%s %s%s (%s)', $col, $NOT, 'IN', (join ',' => @itemlist);
        push @oneitem => $list;
    }
    my $SQL = join " $andor " => @oneitem;

    return $SQL;

} ## end of generate_clause


sub qquote {

    ## Quick SQL quoting
    ## Arguments: one
    ## 1. String to be quoted
    ## Returns: modified string

    my $thing = shift;

    $thing =~ s/'/''/g;

    return qq{'$thing'};

} ## end of qquote


sub upgrade {

    ## Make upgrades to an existing Bucardo schema to match the current version
    ## Arguments: none
    ## Returns: never, exits

    ## Ensure the bucardo.schema file is available and the correct version
    my ($fh, $schema_file) = find_bucardo_schema();

    my $schema_version = 0;
    while (<$fh>) {
        if (/\-\- Version (\d+\.\d+\.\d+)/) {
            $schema_version = $1;
            last;
        }
    }
    if (! $schema_version) {
        die qq{Could not find version number in the file "$schema_file"!\n};
    }
    if ($schema_version ne $VERSION) {
        die qq{Cannot continue: bucardo is version $VERSION, but $schema_file is version $schema_version\n};
    }

    $dbh->do(q{SET escape_string_warning = 'OFF'});
    $dbh->do(q{SET standard_conforming_strings = 'ON'});

    my $changes = 0;

    ## Quick sanity to make sure we don't try to cross the 4/5 boundary
    if (!relation_exists('bucardo', 'syncrun')) {
      print "Sorry, but Bucardo version 4 cannot be upgraded to version 5\n";
      print "You will have to recreate your information (dbs, syncs, etc.)\n";
      exit 1;
    }

    ## Make sure the upgrade_log table is in place

    if (!relation_exists('bucardo', 'upgrade_log')) {
        my ($seqlist, $tabledef) = table_definition('bucardo.upgrade_log');
        upgrade_and_log($tabledef,'CREATE TABLE bucardo.upgrade_log');
        $dbh->commit();
    }

    my @old_sequences = (
        'dbgroup_id_seq',
    );

    my @old_configs = (
        'pidfile',
        'upsert_attempts',
    );

    my @old_constraints = (
        ['bucardo', 'goat', 'goat_pkeytype_check'],
        ['bucardo', 'sync', 'sync_replica_allornone'],
        ['bucardo', 'sync', 'sync_disable_triggers_method'],
        ['bucardo', 'sync', 'sync_disable_rules_method'],
    );

    my @old_columns = (
        ['bucardo', 'sync', 'disable_rules'],
        ['bucardo', 'sync', 'disable_triggers'],
    );

    my @old_functions = (
        ['create_child_q', 'text'],
    );

    my @old_indexes = (
        ['bucardo', 'sync', 'sync_source_targetdb'],
        ['bucardo', 'sync', 'sync_source_targetgroup'],
    );

    my @old_views = (
        'goats_in_herd',
    );

    my @new_columns = (
    );

    my @altered_columns = (
        ['bucardo', 'goat', 'rebuild_index', 'BOOL2SMALLINT1'],
        ['bucardo', 'goat', 'schemaname',    'NO DEFAULT'],
        ['bucardo', 'sync', 'rebuild_index', 'BOOL2SMALLINT1'],
    );

    my @row_values = (

        ['bucardo_config','about',q{name = 'log_showtime'}, 1,
         'Show timestamp in the log output?  0=off  1=seconds since epoch  2=scalar gmtime  3=scalar localtime'],
    );

    my @drop_all_rules = (
    );

    ## Drop all existing rules from a table:
    for my $row (@drop_all_rules) {
        my ($schema,$table) = @$row;
        my $oid = relation_exists($schema,$table);
        if (!$oid) {
            warn "Could not find table $schema.$table to check!\n";
            next;
        }
        $SQL = 'SELECT rulename FROM pg_catalog.pg_rewrite WHERE ev_class = ? ORDER BY rulename';
        $sth = $dbh->prepare($SQL);
        $count = $sth->execute($oid);
        if ($count < 1) {
            $sth->finish();
            next;
        }
        for my $rule (map { $_->[0] } @{$sth->fetchall_arrayref()}) {
            upgrade_and_log(qq{DROP RULE "$rule" ON $schema.$table});
            clog "Dropped rule $rule on table $schema.$table";
            $changes++;
        }
    }

    ## Drop any old views
    for my $name (@old_views) {
        next if !relation_exists('bucardo', $name);
        upgrade_and_log("DROP VIEW $name");
        clog "Dropped view $name";
        $changes++;
    }

    ## Drop any old sequences
    for my $sequence (@old_sequences) {
        next if !relation_exists('bucardo', $sequence);
        upgrade_and_log("DROP SEQUENCE bucardo.$sequence");
        clog "Dropped sequence: $sequence";
        $changes++;
    }

    ## Drop any old constraints
    for my $con (@old_constraints) {
        my ($schema, $table, $constraint) = @$con;
        next if !constraint_exists($schema, $table, $constraint);
        upgrade_and_log(qq{ALTER TABLE $schema.$table DROP CONSTRAINT "$constraint"});
        clog "Dropped constraint $constraint ON $schema.$table";
        $changes++;
    }

    ## Parse the bucardo.schema file and verify the following types of objects exist:
    ## Functions, triggers, constraints, sequences, indexes, comments, and domains
    my (@flist, @tlist, @ilist, @clist, @clist2, @slist, @tablelist, @comlist, @domlist, @collist);
    my ($fname,$args,$fbody) = ('','','');
    my ($tname,$tbody) = ('','');
    my ($tablename,$tablebody) = ('','');
    my ($altername,$alterbody,$alterstat) = ('','','');
    seek $fh, 0, 0;
    while (<$fh>) {
        if ($fbody) {
            if (/^(\$bc\$;)/) {
                $fbody .= $1;
                push @flist, [$fname, $args, $fbody];
                $fbody = $fname = $args = '';
            }
            else {
                $fbody .= $_;
            }
            next;
        }
        if ($tbody) {
            $tbody .= $_;
            if (/;/) {
                push @tlist, [$tname, $tbody];
                $tbody = $tname = '';
            }
            next;
        }
        if ($tablebody) {
            $tablebody .= $_;
            if (/^\s*CONSTRAINT\s+(\w+)\s+(.+?)\s*$/) {
                my ($cname,$def) = ($1,$2);
                $def =~ s/,$//;
                $def =~ s/\bbucardo\.//;
                push @clist2, [$tablename, $cname, $def];
            }
            elsif (/^\s+([a-z_]+)\s+([A-Z]+)\s*(NOT)? NULL/) {
                my ($colname,$coltype,$isnull) = ($1, $2, $3 ? 1 : 0);
                push @collist, ['bucardo', $tablename, $colname, $_];
            }
            elsif (/;/) {
                push @tablelist, [$tablename, $tablebody];
                $tablebody = $tablename = '';
            }
            else {
                die qq{Could not parse table definition: invalid column at line $. ($_)\n};
            }
            next;
        }
        if ($altername) {
            $alterbody =~ s/\s+$//;
            $alterbody ? s/^\s+/ / : s/^\s+//;
            s/\s+$/ /;
            $alterbody .= $_;
            $alterstat .= $_;
            if ($alterbody =~ s/;\s*$//) {
                push @clist, [$altername->[0], $altername->[1], $alterbody, $alterstat];
                $alterbody = $altername = $alterstat = '';
            }
            next;
        }
        if (/^CREATE (?:OR REPLACE )?FUNCTION\s+bucardo\.(.+?\))/) {
            $fname = $1;
            $fbody .= $_;
            $fname =~ s/\((.*)\)// or die "No args found for function: $_\n";
            $args = $1;
            $args =~ s/,(\S)/, $1/g;
            next;
        }
        if (/^CREATE TRIGGER (\S+)/) {
            $tname = $1;
            $tbody .= $_;
            next;
        }
        if (/^CREATE TABLE bucardo\.(\w+)/) {
            $tablename = $1;
            $tablebody .= $_;
            next;
        }
        if (/^CREATE (UNIQUE )?INDEX (\S+)/) {
            push @ilist, [$1, $2, $_];
            next;
        }
        if (/^ALTER TABLE bucardo\.(\S+)\s+ADD CONSTRAINT\s*(\S+)\s*(\S*.*)/) {
            $altername = [$1,$2];
            $alterbody = $3 || '';
            $alterstat = $_;
            next;
        }
        if (/^CREATE SEQUENCE bucardo\.(\w+)/) {
            push @slist, [$1, $_];
            next;
        }
        if (/^COMMENT ON (\w+) (\w+)\.(\w+) IS \$\$(.+)\$\$/) {
            push @comlist, [lc $1, $2, $3, $4, $_];
            next;
        }
        if (/^CREATE DOMAIN bucardo\.(\w+) (.+)/) {
            push @domlist, [$1, $2];
            next;
        }
    }

    ## Add any new domains, verify existing ones
    for my $row (@domlist) {
        my ($name,$def) = @$row;
        next if domain_exists('bucardo', $name);
        upgrade_and_log("CREATE DOMAIN bucardo.$name $def");
        clog("Created domain: $name");
        $changes++;
    }

    ## Check for any added sequences
    for my $row (@slist) {
        my ($sname,$body) = @$row;
        next if relation_exists('bucardo', $sname);
        upgrade_and_log($body);
        clog "Created sequence $sname";
        $changes++;
    }

    ## Check for any added tables
    for my $row (@tablelist) {
        my ($name,$body) = @$row;
        next if relation_exists('bucardo', $name);
        upgrade_and_log($body);
        clog "Created table $name";
        $changes++;
    }

    ## Add new columns as needed from the schema
    for my $row (@collist) {
        my ($schema,$table,$column,$def) = @$row;
        next if column_exists($schema, $table, $column);
        $def =~ s/\-\-.+$//;
        $def =~ s/,\s*$//;
        $def =~ s/\s+/ /g;
        upgrade_and_log("ALTER TABLE $schema.$table ADD COLUMN $def");
        clog "Created column: $schema.$table.$column";
        $changes++;
    }

    ## Add specific new columns as needed
    for my $row (@new_columns) {
        my ($schema,$table,$column,$def) = @$row;
        next if column_exists($schema, $table, $column);
        $def =~ s/\s+/ /g;
        upgrade_and_log("ALTER TABLE $schema.$table ADD COLUMN $def");
        clog "Created column: $schema.$table.$column";
        $changes++;
    }

    ## Change any altered columns
    for my $row (@altered_columns) {
        my ($schema,$table,$column,$change) = @$row;
        next if ! column_exists($schema, $table, $column);
        if ($change eq 'NO DEFAULT') {
            my $def = column_default($schema, $table, $column);
            next if !$def;
            upgrade_and_log("ALTER TABLE $schema.$table ALTER COLUMN $column DROP DEFAULT");
            clog "Removed DEFAULT ($def) from $schema.$table.$column";
            $changes++;
        }
        elsif ($change =~ /^RENAME\s+(\w+)/) {
            my $newname = $1;
            upgrade_and_log("ALTER TABLE $schema.$table RENAME COLUMN $column TO $newname");
            clog("Renamed $schema.$table.$column to $newname");
            $changes++;
        }
        elsif ($change =~ /^DEFAULT\s+(.+)/) {
            my $newname = $1;
            my $oldname = column_default($schema, $table, $column);
            next if $newname eq $oldname;
            upgrade_and_log("ALTER TABLE $schema.$table ALTER COLUMN $column SET DEFAULT $newname");
            clog("Changed DEFAULT on $schema.$table.$column to $newname");
            $changes++;
        }
        elsif ($change =~ /BOOL2SMALLINT(\d)/) {
            my $defval = $1;
            my $oldtype = column_type($schema, $table, $column);
            next if $oldtype eq 'smallint';
            upgrade_and_log("ALTER TABLE $schema.$table ALTER COLUMN $column DROP DEFAULT");
            upgrade_and_log("ALTER TABLE $schema.$table ALTER COLUMN $column TYPE smallint "
                            . "USING CASE WHEN $column IS NULL OR $column IS FALSE THEN 0 ELSE $defval END");
            upgrade_and_log("ALTER TABLE $schema.$table ALTER COLUMN $column SET DEFAULT 0");
            clog("Changed type of $schema.$table.$column to smallint");
            $changes++;
        }
        else {
            die qq{Do not know how to handle altered column spec of "$change"};
        }
    }

    ## Special cases
    if (! column_exists('bucardo', 'goat', 'source_makedelta')) {
        #upgrade_and_log(q{ALTER TABLE bucardo.goat ADD COLUMN source_makedelta ONOFF NOT NULL DEFAULT 'inherits'});
        #upgrade_and_log(q{ALTER TABLE bucardo.goat ADD COLUMN target_makedelta ONOFF NOT NULL DEFAULT 'inherits'});
        if (column_exists('bucardo', 'goat', 'makedelta')) {
        #    upgrade_and_log(q{UPDATE goat SET source_makedelta='on', target_makedelta='on' WHERE makedelta IS TRUE});
        #    upgrade_and_log(q{UPDATE goat SET source_makedelta='off', target_makedelta='off' WHERE makedelta IS FALSE});
        #    upgrade_and_log('ALTER TABLE bucardo.goat DROP COLUMN makedelta');
        }
        #clog('Added columns goat.source_makedelta and goat.target_makedelta');
        #$changes++;
    }
    if (! column_exists('bucardo', 'sync', 'source_makedelta')) {
        #upgrade_and_log(q{ALTER TABLE bucardo.sync ADD COLUMN source_makedelta ONOFF NOT NULL DEFAULT 'inherits'});
        #upgrade_and_log(q{ALTER TABLE bucardo.sync ADD COLUMN target_makedelta ONOFF NOT NULL DEFAULT 'inherits'});
        if (column_exists('bucardo', 'sync', 'makedelta')) {
         #   upgrade_and_log(q{UPDATE sync SET source_makedelta='on', target_makedelta='on' WHERE makedelta IS TRUE});
         #   upgrade_and_log(q{UPDATE sync SET source_makedelta='off', target_makedelta='off' WHERE makedelta IS FALSE});
         #   upgrade_and_log('ALTER TABLE bucardo.sync DROP COLUMN makedelta');
        }
        #clog('Added columns sync.source_makedelta and sync.target_makedelta');
        #$changes++;
    }

    ## Drop any old columns
    for my $row (@old_columns) {
        my ($schema,$table,$column) = @$row;
        next if !column_exists($schema, $table, $column);
        upgrade_and_log("ALTER TABLE $schema.$table DROP COLUMN $column");
        clog "Dropped column: $schema.$table.$column";
        $changes++;
    }

    ## Drop any old indexes
    for my $row (@old_indexes) {
        my ($schema,$table,$name) = @$row;
        next if !relation_exists($schema, $name);
        upgrade_and_log("DROP INDEX $name");
        clog "Dropped index $name";
        $changes++;
    }

    ## Drop any old functions
    for my $row (@old_functions) {
        my ($name, $largs) = @$row;
        next if ! function_exists('bucardo', $name, $largs);
        clog "Dropped function $name($largs)";
        upgrade_and_log(qq{DROP FUNCTION bucardo."$name"($largs)});
        $changes++;
    }

    ## Drop any old config items
    for my $name (@old_configs) {
        next if ! config_exists($name);
        clog "Removed old bucardo_config name: $name";
        upgrade_and_log(qq{DELETE FROM bucardo.bucardo_config WHERE name = '$name'});
        $changes++;
    }

    ## Check for any new config items
    $SQL = 'SELECT setting FROM bucardo.bucardo_config WHERE lower(name) = ?';
    my $cfgsth = $dbh->prepare($SQL);
    $SQL = 'INSERT INTO bucardo.bucardo_config(name,setting,about) VALUES (?,?,?)';
    my $newcfg = $dbh->prepare($SQL);
    my %config;
    my $inside = 0;
    seek $fh, 0, 0;
    while (<$fh>) {
        if (!$inside) {
            if (/^WITH DELIMITER/) {
                $inside = 1;
            }
            next;
        }
        if (/^\\/) {
            $inside = 0;
            next;
        }
        ## Scoop
        my ($name,$setting,$about) = split /\|/ => $_;
        $config{$name} = [$setting,$about];
        $count = $cfgsth->execute($name);
        $cfgsth->finish();
        if ($count eq '0E0') {
            clog "Added new bucardo_config name: $name";
            $changes++;
            $newcfg->execute($name,$setting,$about);
        }
    }
    close $fh or die qq{Could not close file "$file": $!\n};

    ## Apply any specific row changes
    for my $row (@row_values) {
        my ($table,$column,$where,$force,$value) = @$row;
        my $val = column_value('bucardo',$table,$column,$where);
        if (!defined $val) {
            die "Failed to find $table.$column where $where!\n";
        }
        next if $val eq $value;
        $SQL = sprintf "UPDATE bucardo.$table SET $column=%s WHERE $where",
            $dbh->quote($value);
        upgrade_and_log($SQL);
        clog "New value set for bucardo.$table.$column WHERE $where";
        $changes++;
    }


    $SQL = 'SELECT pg_catalog.md5(?)';
    my $md5sth = $dbh->prepare($SQL);
    for my $row (@flist) {
        my ($name,$arg,$body) = @$row;
        next if $name =~ /plperlu_test/;
        my $oldbody = function_exists('bucardo',$name,$arg);
        if (!$oldbody) {
            upgrade_and_log($body,"CREATE FUNCTION $name($arg)");
            clog "Added function $name($arg)";
            $changes++;
            next;
        }
        my $realbody = $body;
        $realbody =~ s/.*?\$bc\$(.+)\$bc\$;/$1/sm;
        $md5sth->execute($realbody);
        my $newbody = $md5sth->fetchall_arrayref()->[0][0];
        next if $oldbody eq $newbody;
        $body =~ s/^CREATE FUNCTION/CREATE OR REPLACE FUNCTION/;
        (my $short = $body) =~ s/^(.+?)\n.*/$1/s;
        $dbh->do('SAVEPOINT bucardo_upgrade');
        eval { upgrade_and_log($body,$short); };
        if ($@) {
            $dbh->do('ROLLBACK TO bucardo_upgrade');
            (my $dropbody = $short) =~ s/CREATE OR REPLACE/DROP/;
            $dropbody .= ' CASCADE';
            upgrade_and_log($dropbody);
            upgrade_and_log($body,$short);
        }
        else {
            $dbh->do('RELEASE bucardo_upgrade');
        }
        clog "Updated function: $name($arg)";
        $changes++;
    }

    ## Check for any added triggers
    for my $row (@tlist) {
        my ($name,$body) = @$row;
        next if trigger_exists($name);
        upgrade_and_log($body);
        clog "Created trigger $name";
        $changes++;
    }

    ## Check for any added indexes
    for my $row (@ilist) {
        my ($uniq,$name,$body) = @$row;
        next if relation_exists('bucardo',$name);
        upgrade_and_log($body);
        clog "Created index $name";
        $changes++;
    }

    ## Check for any added constraints
    for my $row (@clist) {
        my ($tcname,$cname,$cdef,$body) = @$row;
        if (! constraint_exists('bucardo', $tcname, $cname)) {
            upgrade_and_log($body);
            clog "Created constraint $cname on $tcname";
            $changes++;
            next;
        }

        ## Clean up the constraint to make it match what comes back from the database:
        $cdef =~ s/;$//;
        $cdef =~ s/','/', '/g;
        $cdef =~ s{\\\\}{\\}g;
        if ($cdef =~ s/([^)]) (OR|AND) (\w)/$1) $2 ($3/g) {
            $cdef =~ s/CHECK (.+)/CHECK ($1)/;
        }
        my $condef = constraint_definition($cname);
        if ($condef ne $cdef) {
            upgrade_and_log("ALTER TABLE $tcname DROP CONSTRAINT $cname");
            upgrade_and_log("ALTER TABLE $tcname ADD CONSTRAINT $cname $cdef");
            clog "Altered constraint $cname on $tcname";
            $changes++;
        }
    }

    ## Check that any bare constraints (e.g. foreign keys) are unchanged
    for my $row (@clist2) {
        my ($tcname,$cname,$cdef) = @$row;
        my $condef = constraint_definition($cname);
        next if ! $condef or $condef eq $cdef;
        if ($condef and $condef ne $cdef) {
            upgrade_and_log("ALTER TABLE $tcname DROP CONSTRAINT $cname");
        }
        upgrade_and_log("ALTER TABLE $tcname ADD CONSTRAINT $cname $cdef");
        my $action = $condef ? 'Altered' : 'Added';
        clog "$action constraint $cname on $tcname";
        $changes++;
    }

    ## Check that object comments exist and match
    for my $row (@comlist) {
        my ($type,$schema,$relation,$comment,$full) = @$row;
        my $current_comment =
            $type eq 'table' ? table_comment($schema,$relation)
            : $type eq 'domain' ? domain_comment($schema,$relation)
            : 'Unkonwn type';
        if ($current_comment ne $comment) {
            upgrade_and_log($full);
            clog (length $current_comment
                ? "Changed comment on $type $schema.$relation"
                : "Added comment for $type $schema.$relation");
            $changes++;
        }
    }

    ## The freezer.q_staging table is no longer needed, but we must empty it before dropping
    if (relation_exists('freezer','q_staging')) {
        upgrade_and_log('INSERT INTO freezer.master_q SELECT * FROM freezer.q_staging');
        upgrade_and_log('DROP TABLE freezer.q_staging');
        clog 'Dropped deprecated table freezer.q_staging';
        $changes++;
    }

    ## Make sure bucardo_config has the new schema version
    $count = $cfgsth->execute('bucardo_current_version');
    if ($count eq '0E0') {
        $cfgsth->finish();
        warn "Weird: could not find bucardo_current_version in the bucardo_config table!\n";
    }
    else {
        my $curval = $cfgsth->fetchall_arrayref()->[0][0];
        if ($curval ne $schema_version) {
            $SQL = 'UPDATE bucardo.bucardo_config SET setting = ? WHERE name = ?';
            my $updatecfg = $dbh->prepare($SQL);
            $updatecfg->execute($schema_version, 'bucardo_current_version');
            clog "Set bucardo_config.bucardo_current_version to $schema_version";
            $changes++;
        }
    }

    ## Run the magic updater
    $SQL = 'SELECT bucardo.magic_update()';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $message = $sth->fetchall_arrayref()->[0][0];
    if (length $message) {
        clog $message;
        $changes++;
    }

    if ($changes) {
        printf "Okay to commit $changes %s? ", $changes==1 ? 'change' : 'changes';
        exit if <STDIN> !~ /Y/i;
        $dbh->commit();
        print "Changes have been commited\n";
    }
    else {
        print "No schema changes were needed\n";
        exit 1;
    }

    print "Don't forget to run '$progname validate all' as well: see the UPGRADE file for details\n";

    exit 0;

} ## end of upgrade


sub upgrade_and_log {

    ## Put an entry in the bucardo.upgrade_log table
    ## Arguments: two
    ## 1. Type of action
    ## 2. Optional message
    ## Returns: undef

    my $action = shift;
    my $short = shift || $action;

    eval {
        $dbh->do($action);
    };
    if ($@) {
        my $line = (caller)[2];
        die "From line $line, action $action\n$@\n";
    }

    $SQL = 'INSERT INTO bucardo.upgrade_log(action,version,summary) VALUES (?,?,?)';
    eval {
        $sth = $dbh->prepare($SQL);
        $sth->execute($action,$VERSION,$short);
    };
    if ($@) {
        my $line = (caller)[2];
        die "From line $line, insert to upgrade_log failed\n$@\n";
    }

    return;

} ## end of upgrade_and_log


sub usage {

    ## Grab the help string for a specific item
    ## Arguments: one
    ## 1. The thing we want help on
    ## Returns: help string or an empty string for no matches

    my $name = shift or die;

    ## no critic (RequireInterpolationOfMetachars)
    if ('add' eq $name) {
        return qq{Usage: add <type> <name> [options]
Adds an item to the internal Bucardo database.
The type is one of: db, dbgroup, herd, sync, table, sequence,
  customname, customcols, or customcode
For more information, run: $progname help add <type>};
    }
    if ('add_customcode' eq $name) {
        return q{Usage: add customcode <name> <whenrun=value> <src_code=filename> [optional information]};
    }
    if ('add_customname' eq $name) {
        return q{Usage: add customname tablename newtablename [sync=x db=x]};
    }
    if ('add_customcols' eq $name) {
        return q{Usage: add customcols tablename select_clause [sync=x db=x]};
    }
    if ('add_database' eq $name) {
        return q{Usage: add db <name> <dbname=value> [optional information]
Adds a database

Required information
dbname=name         (database name to connect to)

Optional information:
 dbtype=type of database (defaults to postgres. Others: drizzle,mongo,mariadb,mysql,oracle,redis,sqlite)
 dbuser=username    (defaults to 'bucardo')
 dbpass=name        (avoid if possible; use .pgpass or equivalent instead)
 dbport=#           (defaults to type default, e.g. 5432 for postgres)
 dbhost=hostname    (defaults to none - which is Unix socket for some types)
 dbconn=string      (extra connection parameters, e.g. "sslmode=require")
 status=status      (active or inactive, defaults to 'active')
 dbgroup=name       (database group; created if it does not exist)
 addalltables       (automatically adds all tables in the database)
 addallsequences    (automatically adds all sequences in the database)
 ssp=#              (if server-side prepares are used, defaults to 1 (on) - postgres only);
 dbservice=name     (the service name to use for Postgres databases)
 makedelta=onoff    (whether this database needs makedelta magic)
};
    }
    if ('add_dbgroup' eq $name) {
        return q{Usage: add dbgroup <name> [database1 database2 ...]
Adds a database group, and optionally which databases are members of it.
You can specify the role a database has with a colon and role name.
If none is given, the default is a role of 'target'
Example: add dbgroup foobar d1:source d2:source d3 d4
This can also be used to assign databases to an existing group.};
    }
    if ('add_herd' eq $name) {
        return q{Usage: add herd <name> [table1 table2 ...]
Adds a herd, and optionally which tables are members of it.
This can also be used to assign tables to an existing herd.};
    }
    if ('add_sync' eq $name) {
        return q{Usage: add sync <name> herd=x dbs=dbgroup [options]
Adds a sync. Required args:
  herd=name          (the herd containing the tables and sequences to be replicated)
  dbs=name           (the database group, or a comma-separated list of databases)

Optional information:
  status=x           (One of 'active' or 'inactive', defaults to 'active')
  rebuild_index      (whether to rebuild indexes after every sync, defaults to 0 (off))
  lifetime=number    (how long a kid can live before being reaped, defaults to null (no limit))
  maxkicks=number    (how many kicks a kid can process before being reaped, defaults to 0 (no limit))
  onetimecopy=number (controls whether we switch to a fullcopy mode for one run,
                      0=off, 1=always full copy, 2=only copy tables which are empty on the target)
  tables=list        (list of tables to add to this new sync,
                      will create a herd with the same name as the sync)};
    }
    if ('add_sequence' eq $name) {
        return q{Usage: add sequence <name> [name2 name3 ...]
Adds one or more sequences to the internal Bucardo database.};
    }
    if ('add_table' eq $name) {
        return q{Usage: add table <name> [name2 name3 ...]
Adds one or more tables to the internal Bucardo database.
Use "table" to find all matching tables, or be specific with "schema.table"
Can use wildcards, e.g. "foobar%"
Use "add table all" to add in all tables};
    }
    if ('config' eq $name) {
        return q{Usage: config show [all | name]
Usage: config set foo=bar [foo2=bar2]
Displays or sets items from the bucardo_config table.
Using 'show' will display all items or a subset based on the given name.
Using 'set' will change one or more items.};
    }
    if ('inspect' eq $name) {
        return q{Usage: inspect <type> <name>
Inspects an object to see if it has dependency issues.
The only supported type is currently 'table'.};
    }
    if ('inspect_herd' eq $name) {
        return q{Usage: inspect herd <name>
Inspects a herd to see if it has dependency issues.};
    }
    if ('inspect_sync' eq $name) {
        return q{Usage: inspect sync <name>
Inspects a sync to see if it has dependency issues.};
    }
    if ('inspect_table' eq $name) {
        return q{Usage: inspect table <name>
Inspects a table to see if it has dependency issues.};
    }
    if ('kick' eq $name) {
        return q{Usage: kick <name> [# of seconds] [name2 name3...] [--notimer]
Kicks one or more named syncs.
If a number is given, that is how long to wait for the sync to signal that it is finished.
If the number 0 is given, we wait as long as it takes.
By default a graphical timer is given: this can be turned off with the --notimer option.};
    }
    if ('list' eq $name) {
        return qq{Usage: list <type> [options]
Shows information about items in the internal Bucardo database.
The type is one of: db, dbgroup, table, sequence, herd, sync,
  customcode, customcols, customname, or all to see everything at once.
For more information, run: $progname help list <type>};
    }
    if ('list_customcode' eq $name) {
        return q{Usage: list customcode name [options]
Lists information about each customcode.
If ''verbose' is added, the 'About' for each code is shown.
If '--verbose --verbose' is added, the internal database columns for the 'customcode' table are shown.};
    }
    if ('list_databases' eq $name) {
        return q{Usage: list db [name] [--verbose] [--verbose]
Lists information about each database Bucardo knows about.

Without a name, all databases are listed.
The name can have wildcards with '%'
If '--verbose' is added, information is shown about which groups and syncs are involved.
If a second '--verbose' is added, the internal database columns for the 'db' table are shown.};
    }
    if ('list_dbgroups' eq $name) {
        return q{Usage: list dbgroup [name] [--verbose --verbose]
Lists database groups, and which databases (if any) are members.

Without a name, all database groups are listed.
The name can have wildcards with a '%'
If '--verbose --verbose' is added, the internal database columns for the 'dbgroup' table are shown.};
    }
    if ('list_herds' eq $name) {
        return q{Usage: list herd [name] [--verbose --verbose]
Lists herds, and which tables (if any) are members.

Without a name, all herds are listed.
The name can have wildcards with a '%'
If '--verbose --verbose' is added, the internal database columns for the 'herd' table are shown.};
    }
    if ('list_syncs' eq $name) {
        return q{Usage: list sync [name] [--verbose --verbose]
Lists syncs

Without a name, all syncs are listed.
The name can have wildcards with a '%'
If '--verbose --verbose' is added, the internal database columns for the 'sync' table are shown.};
    }
    if ('list_sequences' eq $name) {
        return q{Usage: list sequence [name] [--verbose --verbose]
Lists sequences

Without a name, all sequences are listed.
The name can have wildcards with a '%'
If '--verbose --verbose' is added, the internal database columns for the 'goat' table are shown.};
    }
    if ('list_tables' eq $name) {
        return q{Usage: list table [name] [--verbose --verbose]
Lists tables

Without a name, all tables are listed.
The name can have wildcards with a '%'
If '--verbose --verbose' is added, the internal database columns for the 'goat' table are shown.};
    }
    if ('message' eq $name) {
        return q{Usage: message "text"
Asks a running Bucardo daemon to insert the given text into the Bucardo log files.};
    }
    if ('ping' eq $name) {
        return q{Usage: ping [# of seconds]
Sends a ping notice to the main Bucardo process (MCP) to see if it will respond.
By default, it will wait 15 seconds for a response.
If a number is given, it will wait that number of seconds instead.
Returns a Nagios like message starting with "OK" or "CRITICAL" for success or failure.\n};
    }
    if ('reload' eq $name) {
        return q{Usage: reload sync [sync2 sync3 ...]
Reloads one or more syncs.
This sends a message to Bucardo to stop the named syncs, reload their
information from the database, and start them up again.
Useful if you have changed a setting for an active syncs and need
to put the change in place right away.};
    }
    if ('reload_config' eq $name) {
        return q{Usage: reload_config
Instructs the Bucardo daemon to reload its configuration file and restart.};
    }
    if ('remove' eq $name or 'delete' eq $name) {
        return qq{Usage: remove <type> <name> [options]
Removes an item from the internal Bucardo database.
The type is one of: db, dbgroup, table, sequence, herd, sync,
  customcode, customcols, or customname
For more information, run: $progname help remove <type>};
    }
    if ('remove_customcode' eq $name) {
        return qq{Usage: $verb customcode <name> [name2 name3 ...]
Removes one or more customcodes.};
    }
    if ('remove_customname' eq $name) {
        return qq{Usage: $verb customname number [number2 number3 ...]
Removes one or more customnames. Determine the numbers with list customnames};
    }
    if ('remove_customcols' eq $name) {
        return qq{Usage: $verb customcols number [number2 number3 ...]
Removes one or more customcols. Determine the numbers with list customcols};
    }
    if ('remove_database' eq $name) {
        return qq{Usage: $verb db <name> [name2 name3 ...]
Removes one or more databases.
Use the --force option to remove all related tables};
    }
    if ('remove_dbgroup' eq $name) {
        return qq{Usage: $verb dbgroup <name> [name2 name3 ...]
Removes one or more database groups.};
    }
    if ('remove_herd' eq $name) {
        return qq{Usage: $verb herd <name> [name2 name3 ...]
Removes one or more herds.};
    }
    if ('remove_sync' eq $name) {
        return qq{Usage: $verb sync <name> [name2 name3 ...]
Removes one or more syncs.};
    }
    if ('remove_sequence' eq $name) {
        return qq{Usage: $verb sequence <name> [name2 name3 ...]
Removes one or more sequences.};
    }
    if ('remove_table' eq $name) {
        return qq{Usage: $verb table <name> [name2 name3 ...]
Removes one or more tables.};
    }
    if ('restart' eq $name) {
        return q{Usage: restart ["reason"]
Restarts Bucardo by stopping, then starting it up again.
An optional reason can be given.};
    }
    if ('start' eq $name) {
        return q{Usage: start ["reason"]
Starts Bucardo.
An optional reason can be given.};
    }
    if ('status' eq $name) {
        return q{Usage: status [sync1 sync2]
Displays information about the status of syncs.
With no arguments, shows a summary of all syncs.
If one or more named syncs are given, detailed information
about each sync is given.}
    }
    if ('stop' eq $name) {
        return q{Usage: stop ["reason"]
Stops Bucardo and all of its child processes.
An optional reason can be given.
Active children will not stop until they have finished their current task.};
    }
    if ('update' eq $name) {
        return q{Usage: update <type> <name> col1=val [col2=val2 col3=val3 ...]
Updates an item from the internal Bucardo database.
The type is one of: db, dbgroup, table, sequence, herd, sync, or customname};
    }
    if ('update_database' eq $name) {
        return q{Usage: update database <name> col1=val [col2=val2 col3=val3 ...]
Updates a database
    };
    }
    if ('update_table' eq $name) {
        return q{Usage: update table <name> col1=val [col2=val2 col3=val3 ...] herd=herd1,herd2
Updates a table
    };
    }
    if ('upgrade' eq $name) {
        return q{Usage: upgrade
Upgrades an existing Bucardo database to the current version.};
    }
    if ('validate' eq $name) {
        return q{Usage: validate [all] [sync1 sync2]
Validates one or more named syncs.
Use 'all' to validate all known syncs at once};
    }
    if ('purge' eq $name) {
        return q{Usage: purge [all] [database] [database.table]
Purges one or more tables in one or more databases
Use 'all' to validate all known tables};
    }

    ## use critic

    return '';

} ## end of usage


sub connect_database {

    ## Connect to a datbase and return a dbh
    ## Arguments: one
    ## 1. Hashref of connection arguments (optional)
    ## Returns: database handle

    my $dbh2;

    my $opt = shift || {};

    ## If given just a name, transform to a hash
    if (! ref $opt) {
        $opt = { name => $opt };
    }

    if (exists $opt->{name}) {
        $SQL = qq{SELECT bucardo.db_getconn('$opt->{name}')};
        my $conn = $dbh->selectall_arrayref($SQL)->[0][0];
        my ($type,$dsn,$user,$pass) = split /\n/ => $conn;

        if ($type ne 'postgres') {
            return "Cannot return a handle for database type $type";
        }

        eval {
            $dbh2 = DBI->connect_cached($dsn, $user, $pass, {AutoCommit=>0,RaiseError=>1,PrintError=>0});
        };
        if ($@) {
            ## The bucardo user may not exist yet.
            if ($user eq 'bucardo' and $@ =~ /FATAL/ and $@ =~ /bucardo/) {
                $user = 'postgres';
                $dbh2 = DBI->connect_cached($dsn, $user, $pass, {AutoCommit=>0,RaiseError=>1,PrintError=>0});
                $dbh2->do('CREATE USER bucardo SUPERUSER');
                $dbh2->commit();
                $user = 'bucardo';
                $dbh2 = DBI->connect_cached($dsn, $user, $pass, {AutoCommit=>0,RaiseError=>1,PrintError=>0});
            }
        }
    }

    return $dbh2;

} ## end of connect_database


sub config {

    ## View or change a value inside the bucardo_config table
    ## Arguments: none, reads nouns
    ## Returns: never, exits

    my $setusage = "Usage: $progname set setting=value [setting=value ...]\n";

    ## Allow for old syntax
    if ($verb eq 'config') {
        @nouns or die "Usage: $progname config [set|show]\n";
        $verb = shift @nouns;
    }

    if (!@nouns) {
        $verb eq 'set' and die $setusage;
        die "Usage: $progname show <all|setting1> [settting2 ...]\n";
    }

    $SQL = 'SELECT * FROM bucardo.bucardo_config';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $config = $sth->fetchall_hashref('name');
    if ($verb eq 'show') {
        my $all = $nouns[0] =~ /\ball\b/i ? 1 : 0;
        my $maxsize = 3;
        for my $s (keys %$config) {
            next if ! $all and ! grep { $s =~ /$_/i } @nouns;
            $maxsize = length $s if length $s > $maxsize;
        }
        for my $s (sort keys %$config) {
            next if ! $all and ! grep { $s =~ /$_/i } @nouns;
            printf "%-*s = %s\n", $maxsize, $s, $config->{$s}{setting};
        }
        exit 1;
    }

    $SQL = 'UPDATE bucardo.bucardo_config SET setting = ? WHERE name = ?';
    $sth = $dbh->prepare($SQL);

    for my $noun (@nouns) {
        $noun =~ /(\w+)=(.+)/ or die $setusage;
        my ($setting,$val) = (lc $1,$2);

        if (! exists $config->{$setting}) {
            die qq{Unknown setting "$setting"\n};
        }

        ## Sanity checks
        if ($setting eq 'log_level') {
            if ($val !~ /^(?:terse|normal|verbose|debug)$/oi) {
                die "Invalid log_level, must be terse, normal, verbose, or debug\n";
            }
        }
        if ($setting eq 'default_standard_conflict') {
            if ($val !~ /^(?:source|target|skip|random|latest|none)$/oi) {
                ## FIXME
                #die "Invalid default_standard_conflict, must be none, source, target, skip, random, or latest\n";
            }
            if ($val =~ /none/i) {
                $val = '';
            }
        }

        $sth->execute($val,$setting);
        print qq{Set "$setting" to "$val"\n};

    }

    $dbh->commit();

    exit 0;

} ## end of config


sub message {

    ## Add a message to the Bucardo logs, via the bucardo_log_message table
    ## Note: If no MCP processes are listening, the message will hang out until an MCP processes it
    ## Arguments: none (reads in nouns)
    ## Returns: never, exits

    if (! length($nouns)) {
        die qq{Usage: bucardo message "Some message to send to the logs"\n};
    }

    $SQL = 'INSERT INTO bucardo.bucardo_log_message(msg) VALUES (?)';
    $sth = $dbh->prepare($SQL);
    $sth->execute($nouns);
    $dbh->commit();
    $VERBOSE and print "Message added\n";

    exit 0;

} ## end of message


sub db_get_notices {

    ## Gather up and return a list of asynchronous notices received since the last check
    ## Arguments: one
    ## 1. Database handle
    ## Returns: arrayref of notices, each an arrayref of name and pid
    ## If using 9.0 or greater, the payload becomes the name

    my ($ldbh) = @_;

    my ($n, @notices);

    while ($n = $ldbh->func('pg_notifies')) {
        my ($name, $pid, $payload) = @$n;
        if ($ldbh->{pg_server_version} >= 9999990000) {
            next if $name ne 'bucardo';
            $name = $payload; ## presto!
        }
        push @notices => [$name, $pid];
    }

    return \@notices;

} ## end of db_get_notices


sub install {

    ## Install Bucardo into a database
    ## Arguments: none
    ## Returns: never, exits

    if (! $bcargs->{batch}) {
        print "This will install the bucardo database into an existing Postgres cluster.\n";
        print "Postgres must have been compiled with Perl support,\n";
        print "and you must connect as a superuser\n\n";
    }

    ## Setup our default arguments for the installation choices
    my $host = $bcargs->{dbhost} || $ENV{PGHOST} || '<none>';
    my $port = $bcargs->{dbport} || $ENV{PGPORT} || 5432;
    my $user = $ENV{DBUSER} || 'postgres';
    my $dbname = $ENV{DBNAME} || 'postgres';

    ## Make sure the bucardo.schema file is available, and extract some config items
    my ($fh, $schema_file) = find_bucardo_schema();
    my %confvar = (piddir => '');
    while (<$fh>) {
        for my $string (keys %confvar) {
            if (/^$string\|(.+?)\|/) {
                $confvar{$string} = $1;
            }
        }
    }
    close $fh or warn qq{Could not close "$schema_file": $!\n};

    ## Make sure each item has a default value
    for my $key (keys %confvar) {
        if (!$confvar{$key}) {
            warn "Could not find default configuration for $key!\n";
        }
    }

    ## If the PID directory was not provided on the command line,
    ## use the value from the bucardo.schema file
    my $piddir = $bcargs->{piddir} || $confvar{piddir};

    ## Keep looping until they are happy with the settings
  GOOEY:
    {

        print "Current connection settings:\n";

        print "1. Host:           $host\n";
        print "2. Port:           $port\n";
        print "3. User:           $user\n";
        print "4. Database:       $dbname\n";
        print "5. PID directory:  $piddir\n";

        ## If in batch mode, we accept everything right away and move on
        last GOOEY if $bcargs->{batch};

        print 'Enter a number to change it, P to proceed, or Q to quit: ';

        my $ans = <>;
        print "\n";

        ## If the answer starts with a number, try and apply it
        ## Can also provide the value right away
        if ($ans =~ /^\s*(\d+)(.*)/) {
            my ($num,$text) = (int $1,$2);
            $text =~ s/^\s*(\S+)\s*$/$1/;
            my $new = length $text ? $text : '';

            ## Host: allow anything
            ## Change empty string to '<none>';
            if (1 == $num) {
                if (!length $new) {
                    print 'Change the host to: ';
                    $new = <>;
                    print "\n";
                    chomp $new;
                }
                $host = length $new ? $new : '<none>';
                print "Changed host to: $host\n";
            }

            ## Port: only allow numbers
            elsif (2 == $num) {
                if (!length $new) {
                    print 'Change the port to: ';
                    $new = <>;
                    print "\n";
                    chomp $new;
                }
                if ($new !~ /^\d+$/) {
                    print "-->Sorry, but the port must be a number\n\n";
                    redo GOOEY;
                }
                $port = $new;
                print "Changed port to: $port\n";
            }

            ## User: allow anything except an empty string
            elsif (3 == $num) {
                if (!length $new) {
                    print 'Change the user to: ';
                    $new = <>;
                    print "\n";
                    chomp $new;
                }
                if (! length $new) {
                    print "-->Sorry, you must specify a user\n\n";
                    redo GOOEY;
                }
                $user = $new;
                print "Changed user to: $user\n";
            }

            ## Database: allow anything except an empty string
            elsif (4 == $num) {
                if (!length $new) {
                    print 'Change the database name to: ';
                    $new = <>;
                    print "\n";
                    chomp $new;
                }
                if (! length $new) {
                    print "-->Sorry, you must specify a database name\n\n";
                    redo GOOEY;
                }
                $dbname = $new;
                print "Changed database name to: $dbname\n";
            }

            ## PID directory: allow anything, as long as it starts with a slash
            elsif (5 == $num) {
                if (!length $new) {
                    print 'Change the PID directory to: ';
                    $new = <>;
                    print "\n";
                    chomp $new;
                }
                if (! length $new) {
                    print "-->Sorry, you must specify a directory\n\n";
                    redo GOOEY;
                }
                if ($new !~ m{^/}) {
                    print "-->Sorry, the PID directory must be absolute (start with a slash)\n";
                    redo GOOEY;
                }
                if (! -d $new) {
                    print "-->Sorry, that is not a valid directory\n";
                    redo GOOEY;
                }
                $piddir = $new;
                print "Changed PID dir to: $piddir\n";
            }
        }
        elsif ($ans =~ /^\s*Q/i) {
            die "Goodbye!\n";
        }
        elsif ($ans =~ /^\s*P/i) {
            ## Check on the PID directory before going any further
            ## This is the only item that can be easily checked here
            if (! -d $piddir) {
                print "-->Sorry, that is not a valid PID directory\n";
                redo GOOEY;
            }
            last GOOEY;
        }
        else {
            print "-->Please enter Q to quit, P to proceed, or enter a number to change a setting\n";
        }

        redo GOOEY;

    }

    ## Try to connect
    my $PSQL = "psql -p $port -U $user -d $dbname";
    $host !~ /</ and $PSQL .= " --host=$host";

    ## We also want the version, so we grab that as the initial connection test
    my $COM = "$PSQL -c 'SELECT version()'";

    my $res = qx{$COM 2>&1};

    ## Dump any problems verbatim to stderr
    if ($res =~ /FATAL|ERROR/ or $res =~ /psql:/) {
        warn $res;
    }

    ## Check for some common errors
    if ($res =~ /role ".+" does not exist/) {
        print "-->Sorry, please try using a different user\n\n";
        goto GOOEY;
    }

    if ($res !~ /(\d+)\.(\d+)(\S+)/) {
        print "-->Sorry, unable to connect to the database\n\n";
        goto GOOEY;
    }

    ## At this point, we assume a good connection
    ## Assign the version variables
    my ($maj,$min,$rev) = ($1,$2,$3);
    ## We need to be able to handle things such as 9.2devel
    $rev =~ s/^\.//;
    $rev =~ s/(\d+)\.\d+/$1/;

    print "Postgres version is: $maj.$min\n";

    ## Bare minimum for the install is 8.1
    if ($maj < 8 or (8 == $maj and $min < 1)) {
        die "Sorry, Bucardo requires Postgres version 8.1 or higher. This is only $maj.$min\n";
    }

    ## Determine if we need to create the bucardo user
    $COM = qq{$PSQL -c "SELECT 1 FROM pg_user WHERE usename = 'bucardo'"};
    $res = qx{$COM 2>&1};

    ## If no number 1 seen, no bucardo user, so create it
    if ($res !~ /1/) {
      print "Creating superuser 'bucardo'\n";

      ## Generate a new random password
      my $pass = generate_password();
      $SQL = qq{CREATE USER bucardo SUPERUSER PASSWORD '$pass'};
      $COM = qq{$PSQL -c "$SQL"};
      $res = qx{$COM 2>&1};

      ## Put the new password into the .pgpass file
      my $passfile = "$ENV{HOME}/.pgpass";
      my $pfh;
      if (open my $pfh, '>>', $passfile) {
        printf {$pfh} "%s:%s:%s:%s:%s\n",
          $host =~ /^\w/ ? $host : '*',
          $port =~ /^\d/ ? $port : '*',
          '*',
          'bucardo',
          $pass;
        close $pfh or warn qq{Could not close file "$passfile": $!\n};
        chmod 0600, $passfile;
      }
      else {
        print qq{Could not append password information to file "$passfile"\n};
        print qq{Password for user bucardo is: $pass\n};
        print qq{You probably want to change it or put into a .pgpass file\n};
      }
    }

    ## Now we apply the bucardo.schema to the new database
    $COM = "$PSQL -AX -qt -f $schema_file 2>&1";

    print "Attempting to create and populate the bucardo database and schema\n"
        if ! $bcargs->{batch};

    $res= qx{$COM};
    chomp $res;

    ## Detect case where bucardo is already there
    ## This probably needs to be i18n safe
    if ($res =~ /relation .* already exists/) {
        warn "\nINSTALLATION FAILED! Looks like you already have Bucardo installed there.\n";
        warn "Try running 'bucardo upgrade' instead\n\n";
        exit 1;
    }

    ## This can actually happen for many reasons: lack of this message
    ## simply means something went wrong somewhere
    ## TODO: Parse other errors more gracefully
    if ($res !~ m{Pl/PerlU was successfully installed}) {
        warn "\nINSTALLATION FAILED! ($res)\n\n";
        warn "This is often caused by the Pl/PerlU language not being available\n";
        warn "This is usually available as a separate package\n";
        warn "For example, you might try: yum install postgresql-plperl\n";
        warn "If compiling from source, add the --with-perl option to your ./configure command\n\n";
        exit 1;
    }

    ## We made it! All downhill from here
    print "Database creation is complete\n\n" if ! $bcargs->{batch};

    ## Whether or not we really need to, change some bucardo_config items:
    my $BDSN  = 'dbi:Pg:dbname=bucardo';
    $host and $host ne '<none>' and $BDSN .= ";host=$host";
    $port and $BDSN .= ";port=$port";
    $dbh = DBI->connect($BDSN, 'bucardo', '', {AutoCommit=>0,RaiseError=>1,PrintError=>0});
    $dbh->do('SET search_path = bucardo');

    $SQL = 'UPDATE bucardo.bucardo_config SET setting = ? WHERE name = ?';
    $sth = $dbh->prepare($SQL);
    $confvar{piddir} = $piddir;
    for my $key (sort keys %confvar) {
        $count = $sth->execute($confvar{$key}, $key);
        if ($count != 1) {
            warn "!! Failed to set $key to $confvar{$key}\n";
        }
        else {
            print qq{Updated configuration setting "$key"\n} if ! $bcargs->{batch};
        }
    }
    $dbh->commit();

    print "Installation is now complete.\n";
    ## A little less verbose if in batch mode
    if (! $bcargs->{batch}) {
        print "If you see errors or need help, please email bucardo-general\@bucardo.org\n\n";

        print "You may want to check over the configuration variables next, by running:\n";
        print "$progname show all\n";
        print "Change any setting by using: $progname set foo=bar\n\n";
    }

    exit 0;

} ## end of install


##
## Internal helper subs
##

sub debug {

    ## Print a debug line if needed
    ## Arguments: one
    ## 1. String to be printed
    ## Returns: undef

    return if ! $DEBUG;

    my $string = shift;

    chomp $string;

    print " |DEBUG| $string\n";

    return;

} ## end of debug


sub standardize_name {

    ## Return canonical version of certain names
    ## Normalizes abbreviations, misspelling, plurals, case, etc.
    ## Arguments: one
    ## 1. Name
    ## Returns: canonical name

    my $name = shift;

    return 'customcode' if $name =~ /^c?code/i or $name =~ /^custom_?code/i;

    return 'customname' if $name =~ /^cname/i or $name =~ /^custom_?name/i;

    return 'customcols' if $name =~ /^ccol/i or $name =~ /^custom_?col/i;

    return 'dbgroup'    if $name =~ /^dbg/i or $name =~ /^d.+group/i;

    return 'database'   if $name =~ /^db/i or $name =~ /^database/i;

    return 'herd'       if $name =~ /^herd/i;

    return 'sync'       if $name =~ /^s[yi]n[ck]/i;

    return 'table'      if $name =~ /^tab/i or $name =~ /^tbale/i;

    return 'sequence'   if $name =~ /^seq/i;

    return 'all'        if $name =~ /^all$/i;

    return '';

} ## end of standardize_name


sub generate_password {

    ## Generate a random 42 character password
    ## Arguments: none
    ## Returns: new password

    my @chars = split // => q!ABCDEFGHJKMNPQRSTWXYZabcdefghjkmnpqrstwxyz23456789@#%^&(){}[];./!;
    my $pass = join '' => @chars[map{ rand @chars }(1..42)];

    return $pass;

} ## end of generate_password


sub process_simple_args {

    ## Process args to an inner function in the style of a=b
    ## Arguments: one
    ## 1. Custom hashref
    ## Returns: db column hashref, columns string, placeholders string,
    ##    values string, and 'extra' hashref

    my $arg = shift;
    my $validcols = $arg->{cols} or die 'Need a list of valid cols!';
    my $list = $arg->{list}      or die 'Need a list of arguments!';
    my $usage = $arg->{usage}    or die 'Need a usage!';

    my %item;
    my %dbcol;
    my %extra;
    my %othername;

    ## Transform array of x=y into a hashref
    my $xyargs = process_args(join ' ' => @$list);

    ## Parse the validcols string, and setup any non-null defaults
    for my $row (split /\n/ => $validcols) {
        next if $row !~ /\w/ or $row =~ /^#/;
        $row =~ /^\s*(\S+)\s+(\S+)\s+(\S+)\s+(.+)/ or die "Invalid valid cols ($row)";
        my ($args,$dbcol,$flag,$default) = ([split /\|/ => $1],$2,$3,$4);
        my $alias = @{$args}[-1];
        for my $name (@$args) {
            $item{$name} = [$dbcol,$flag,$default];
            $othername{$name} = $alias;
        }
        ## Process environment variable default
        if ($default =~ s/^ENV://) {
            for my $env (split /\|/ => $default) {
                if ($ENV{$env}) {

                    ## Skip if it starts with PG and this is not postgres
                    next if $env =~ /^PG/ and exists $xyargs->{type} and $xyargs->{type} ne 'postgres';

                    $dbcol{$dbcol} = $ENV{$env};
                    last;
                }
            }
        }
        elsif ($default ne 'null' and $default ne 'skip') {
            $dbcol{$dbcol} = $default;
        }
    }

    for my $arg (sort keys %$xyargs) {

        next if $arg eq 'extraargs';

        if (! exists $item{$arg}) {
            die "Unknown option '$arg'\n$usage\n";
        }

        (my $val = $xyargs->{$arg}) =~ s/^\s*(\S+)\s*$/$1/;

        if ($item{$arg}[2] eq 'skip') {
            $extra{$othername{$arg}} = $val;
            next;
        }

        my ($dbcol,$flag,$default) = @{$item{$arg}};
        if ($flag eq '0') {
            ## noop
        }
        elsif ($flag eq 'TF') {
            $val =~ s/^\s*t(?:rue)*\s*$/1/i;
            $val =~ s/^\s*f(?:alse)*\s*$/0/i;
            if ($val !~ /^[01]$/) {
                die "Invalid value for '$arg': must be true of false\n";
            }
        }
        elsif ($flag eq 'numeric') {
            if ($val !~ /^\d+$/) {
                die "Invalid value for '$arg': must be numeric\n";
            }
        }
        elsif ($flag =~ /^=(.+)/) {
            my $ok = 0;
            for my $okval (split /\|/ => $1) {
                if ($okval =~ /~/) { ## aliases - force to the first one
                    my @alias = split /~/ => $okval;
                    for my $al (@alias) {
                        if ($val eq $al) {
                            $ok = 1;
                            last;
                        }
                    }
                    if ($ok) {
                        $val = $alias[0];
                        last;
                    }
                }
                elsif (lc $val eq lc $okval) {
                    $ok = 1;
                    last;
                }
            }
            if (!$ok) {
                (my $arglist = $flag) =~ s/\|/ or /g;
                $arglist =~ s/^=//;
                $arglist =~ s/~\w+//g;
                die "Invalid value for '$arg': must be one of $arglist\n";
            }
        }
        elsif ($flag eq 'interval') {
            ## Nothing for now
        }
        else {
            die "Unknown flag '$flag' for $arg";
        }

        ## Value has survived our minimal checking. Store it and clobber any default
        $dbcol{$dbcol} = $val;

    }

    ## Apply any magic
    if (exists $arg->{morph}) {
        for my $mline (@{$arg->{morph}}) {
            if (exists $mline->{field}) {
                next unless exists $dbcol{$mline->{field}};
                next if $dbcol{$mline->{field}} ne $mline->{value};
                for my $change (split /\s+/ => $mline->{new_defaults}) {
                    my ($f,$v) = split /\|/ => $change;
                    next if exists $dbcol{$f};
                    $dbcol{$f} = $v;
                }
            }
            else {
                die "Do not know how to handle that morph!\n";
            }
        }
    }


    ## Build the lists of columns and placeholders for the SQL statement
    my ($cols,$phs,$vals) = ('','',{});
    for my $col (sort keys %dbcol) {
        $cols .= "$col,";
        $phs .= '?,';
        $vals->{$col} = $dbcol{$col};
    }
    $cols =~ s/,$//;
    $phs =~ s/,$//;

    return \%dbcol, $cols, $phs, $vals, \%extra;

} ## end of process_simple_args


sub check_recurse {

    ## Call a sub recursively depending on first argument
    ## Arguments: three or more
    ## 1. Type of thing (e.g. database)
    ## 2. Name of the thing
    ## 3. Any additional actions
    ## Returns: 0 or 1

    my ($thing, $name, @actions) = @_;

    my $caller = (caller(1))[3];

    ## If the name is 'all', recursively call on all objects of this type
    if ($name =~ /all/i) {
        for my $item (sort keys %$thing) {
            &$caller($item, @actions);
        }
        return 0;
    }

    ## If we have a wildcard, recursively call all matching databases
    if ($name =~ s/[*%]/\.*/g) {
        my @list = grep { $_ =~ /^$name$/ } keys %$thing;
        if (! @list) {
            die qq{No matching items found\n};
        }
        for my $item (sort @list) {
            &$caller($item, @actions);
        }
        return 0;
    }

    return 1;

} ## end of check_recurse


sub extract_name_and_role {

    ## Given a group or db name with optional role information, return both
    ## Also returns optional list of other items, e.g. ABC:slave:pri=2:gangs=2
    ## Arguments: one
    ## 1. Group or database name: 'foo' or 'foo:master'
    ## Returns: name, role name, and hashref of 'extra' info

    my $name = shift or die;

    ## Role always defaults to 'target'
    my $role = 'target';

    ## Check for a role attached to the group name
    if ($name =~ s/:([^:]+)//) {
        $role = lc $1;
    }

    ## Look for any additional items
    my %extra;
    while ($name =~ s/:([^:]+)//) {
        my $extra = $1;
        if ($extra !~ /(\w+)=([\w\d]+)/) {
            die qq{Invalid value "$extra"\n};
        }
        my ($lname,$val) = ($1,$2);
        if ($lname =~ /make?delta/i) {
            $extra{'makedelta'} = make_boolean($val);
        }
        elsif ($lname =~ /gang/i) {
            $extra{'gang'} = $val;
        }
        elsif ($lname =~ /pri/i) {
            $extra{'priority'} = $val;
        }
        else {
            die qq{Unknown value "$lname": must be priority, gang, or makedelta\n};
        }
    }

    ## Valid group name?
    if ($name !~ /^[\w\d]+$/) {
        die "Invalid name: $name\n";
    }

    ## Valid role name?
    if ($role !~ /^(?:master|target|slave|rep|replica|source|fullcopy)$/) {
        die "Invalid database role: $role\n";
    }

    ## Standardize the names
    $role = 'source' if $role =~ /master/;
    $role = 'target' if $role =~ /slave|rep|tar/;

    return $name, $role, \%extra;

} ## end of extract_name_and_role


sub load_bucardo_info {

    ## Load of all information from the database into global hashes
    ## Arguments: one
    ## 1. Boolean: if true, force run even if we've run once already
    ## Returns: undef

    my $force = shift || 0;

    return if exists $global{db} and ! $force;

    ## Grab all database information
    $SQL = 'SELECT *, EXTRACT(epoch FROM cdate) AS epoch FROM bucardo.db';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $db = $sth->fetchall_hashref('name');

    ## Grab all database information
    $SQL = 'SELECT * FROM bucardo.dbgroup';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $dbgroup = $sth->fetchall_hashref('name');

    ## Figure out if each dbgroup is using multiple gangs
    my %gang;

    ## Map databases to their groups
    $SQL = 'SELECT * FROM bucardo.dbmap';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    for my $row (@{$sth->fetchall_arrayref({})}) {
        $db->{$row->{db}}{group}{$row->{dbgroup}} = $row;

        ## Tally up the roles each database fills
        $db->{$row->{db}}{roles}{$row->{role}}++;

        ## Mark if this db is ever used as a source, for help in adding table
        $db->{$row->{db}}{issource}++ if $row->{role} eq 'source';

        $dbgroup->{$row->{dbgroup}}{db}{$row->{db}} = $row;
        ## Figure out how many gangs each group has
        $gang{$row->{dbgroup}}{$row->{gang}}++;
    }

    for my $group (keys %$dbgroup) {
        $dbgroup->{$group}{gangs} = keys %{ $gang{$group} };
    }

    ## Grab all goat information
    $SQL = 'SELECT * FROM bucardo.goat';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $goat = $sth->fetchall_hashref('id');
    ## Since relations cannot start with a number, we can also safely add the name to the hash
    for my $key (%$goat) {
        next if $key !~ /^\d/;
        my $tname = $goat->{$key}{tablename};
        my $name = "$goat->{$key}{schemaname}.$tname";
        $goat->{$name} = $goat->{$key};
        ## Also want a table-only version:
        push @{$goat->{$tname}} => $goat->{$key};
    }

    ## Grab all herd information
    $SQL = 'SELECT * FROM bucardo.herd';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $herd = $sth->fetchall_hashref('name');

    ## Grab all herdmap information, stick into previous hashes
    $SQL = 'SELECT * FROM bucardo.herdmap ORDER BY priority DESC, goat ASC';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    for my $row (@{$sth->fetchall_arrayref({})}) {
        my ($g,$h,$p) = @$row{qw/goat herd priority/};
        $goat->{$g}{herd}{$h} = $p;
        $herd->{$h}{goat}{"$goat->{$g}{schemaname}.$goat->{$g}{tablename}"} = {
            id       => $g,
            priority => $p,
            reltype  => $goat->{$g}{reltype},
            schema   => $goat->{$g}{schemaname},
            table    => $goat->{$g}{tablename},
        };
        my ($s,$t) = @{$goat->{$g}}{qw/schemaname tablename/};
        $herd->{$h}{hasgoat}{$s}{$t} = $p;
        ## Assign each herd to a datbase via its included goats
        $herd->{$h}{db} = $goat->{$g}{db};
    }

    ## Grab all sync information
    $SQL = 'SELECT * FROM bucardo.sync';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $sync;
    for my $row (@{$sth->fetchall_arrayref({})}) {
        my ($name,$p,$sherd,$dbs) = @$row{qw/name priority herd dbs/};
        $sync->{$name} = $row;
        ## Add in herd information
        $sync->{$name}{herd} = $herd->{$sherd};
        ## Add this sync back to the herd
        $herd->{$sherd}{sync}{$name}++;
        ## Grab the databases used by this sync
        $sync->{$name}{dblist} = $dbgroup->{$dbs}{db};
        ## Map each database back to this sync, along with its type (source/target)
        for my $dbname (keys %{ $sync->{$name}{dblist} }) {
            $db->{$dbname}{sync}{$name} = $sync->{$name}{dblist}{$dbname};
        }
        ## Note which syncs are used by each goat
        for my $row2 (sort keys %{$row->{herd}{goat}}) {
            $goat->{$row2}{sync}{$name} = 1;
        }
    }

    ## Grab all customcode information
    $SQL = 'SELECT * FROM bucardo.customcode';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my $cc = $sth->fetchall_hashref('name');
    $SQL = 'SELECT * FROM bucardo.customcode_map';
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    my %codename;
    for my $row (values %$cc) {
        $codename{$row->{id}} = $row->{name};
    }
    for my $row (@{$sth->fetchall_arrayref({})}) {
        my $codename = $codename{$row->{code}};
        push @{$cc->{$codename}{map}} => $row;
    }

    ## Grab all customname information
    $SQL = q{SELECT c.id, c.goat, c.newname,
COALESCE(c.sync, '') AS sync,
COALESCE(c.db, '') AS db,
g.schemaname || '.' || g.tablename AS tname
FROM bucardo.customname c
JOIN goat g ON (g.id = c.goat)
};
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    $CUSTOMNAME = {};
    for my $row (@{ $sth->fetchall_arrayref({}) }) {
        ## We store three versions

        ## Look things up by the internal customname id: used for 'delete customname'
        ## Only one entry per id
        $CUSTOMNAME->{id}{$row->{id}} = $row;

        ## Look things up by the goat id: used to check for existing entries
        ## Can have multiple entries per goat
        $CUSTOMNAME->{goat}{$row->{goat}}{$row->{db}}{$row->{sync}} = $row;

        ## A simple list of all rows: used for 'list customnames'
        push @{ $CUSTOMNAME->{list} } => $row;
    }

    ## Grab all customcols information
    $SQL = q{SELECT c.id, c.goat, c.clause,
COALESCE(c.sync, '') AS sync,
COALESCE(c.db, '') AS db,
g.schemaname || '.' || g.tablename AS tname
FROM bucardo.customcols c
JOIN goat g ON (g.id = c.goat)
};
    $sth = $dbh->prepare($SQL);
    $sth->execute();
    $CUSTOMCOLS = {};
    for my $row (@{ $sth->fetchall_arrayref({}) }) {
        ## We store three versions: one for quick per-goat lookup,
        ## one by the assigned id, and one just a big list
        push @{ $CUSTOMCOLS->{goat}{$row->{goat}}{$row->{clause}} } => $row;
        $CUSTOMCOLS->{id}{$row->{id}} = $row;
        push @{ $CUSTOMCOLS->{list} } => $row;
    }

    $global{cc}      = $CUSTOMCODE = $cc;
    $global{dbgroup} = $DBGROUP = $dbgroup;
    $global{db}      = $DB   = $db;
    $global{goat}    = $GOAT = $goat;
    $global{herd}    = $HERD = $herd;
    $global{sync}    = $SYNC = $sync;

    ## Separate goat into tables and sequences
    for my $id (keys %$GOAT) {
        ## Ids only please
        next if $id !~ /^\d+$/;
        my $type = $GOAT->{$id}{reltype};
        if ($type eq 'table') {
            $TABLE->{$id} = $GOAT->{$id};
        }
        elsif ($type eq 'sequence') {
            $SEQUENCE->{$id} = $GOAT->{$id};
        }
        else {
            die "Unknown goat type $type!";
        }
    }

    return;

} ## end of load_bucardo_info


sub transform_name {

    ## Change a given word to a more standard form
    ## Generally used for database column names, which follow some simple rules
    ## Arguments: one
    ## 1. Name to transform
    ## Returns: transformed name

    my $name = shift;

    ## Complain right away if these are not standard characters
    if ($name !~ /^[\w ]+$/) {
        die "Invalid name: $name\n";
    }

    ## Change to lowercase
    $name = lc $name;

    ## Change dashes and spaces to underscores
    $name =~ s{[- ]}{_}go;

    ## Compress all underscores
    $name =~ s{__+}{_}go;

    ## Fix common spelling errors
    $name =~ s{perpare}{prepare}go;

    ## Look up standard abbreviations
    if (exists $alias{$name}) {
        $name = $alias{$name};
    }

    return $name;

} ## end of transform_name


sub transform_value {

    ## Change a value to a more standard form
    ## Used for database column SET actions
    ## Arguments: one
    ## 1. Value
    ## Returns: transformed value

    my $value = shift;

    ## Remove all whitespace on borders
    $value =~ s/^\s*(\S+)\s*$/$1/;

    ## Change booleans to 0/1
    $value =~ s/^(?:t|true)$/1/io;
    $value =~ s/^(?:f|false)$/0/io;

    return $value;

} ## end of transform_value


sub make_boolean {

    ## Transform some string into a strict boolean value
    ## Arguments: one
    ## 1. String to be analyzed
    ## Returns: the string literals 'true' or 'false' (unquoted)

    my $value = shift;

    $value = lc $value;

    return 'true' if $value =~ /^(?:t|true|1|yes)$/o;

    return 'false' if $value =~ /^f|false|0|no$/o;

    die "Invalid value: must be 'true' of 'false'\n";

} ## end of make_boolean


sub standardize_rdbms_name {

    ## Make the database types standard: account for misspellings, case, etc.
    ## Arguments: one
    ## 1. Name of a database type
    ## Returns: modified name

    my $name = shift;

    $name =~ s/postgres.*/postgres/io;
    $name =~ s/pg.*/postgres/io;
    $name =~ s/driz?zle.*/drizzle/io;
    $name =~ s/mongo.*/mongo/io;
    $name =~ s/mysql.*/mysql/io;
    $name =~ s/maria.*/mariadb/io;
    $name =~ s/oracle.*/oracle/io;
    $name =~ s/redis.*/redis/io;
    $name =~ s/sqll?ite.*/sqlite/io;

    return $name;

} ## end of standardize_rdbms_name


sub find_best_db_for_searching {

    ## Returns the db from $DB most likely to contain tables to add
    ## Basically, we use source ones first, then the date added
    ## Arguments: none
    ## Returns: database name or undef if no databases defined yet

    for my $db (
        map { $_->[0] }
        sort {
            ## Source databases are always first
            $a->[1] <=> $b->[1]
            ## First created are first
            or $a->[2] <=> $b->[2]
            ## All else fails, sort by name (should not happen)
            or $b->[0] cmp $a->[0] }
        map { [
               $_,
               exists $DB->{$_}{issource} ? 0 : 1,
               $DB->{$_}{epoch},
               lc $_,
              ]
            }
        keys %{ $DB } ) {
        return $db;
    }

    ## Probably an error, but let the caller handle it:

    return undef;

} ## end of find_best_db_for_searching


##
## Subs to perform common SQL actions
##

sub confirm_commit {

    ## Perform a database commit unless the user does not want it
    ## Arguments: none
    ## Returns: true for commit, false for rollback

    ## The dryrun option overrides everything else: we never commit
    if ($bcargs->{dryrun}) {
        $VERBOSE and print "In dryrun mode, so no going to commit database changes\n";
        return 0;
    }

    if ($bcargs->{confirm}) {
        print 'Commit the changes? Y/N ';
        if (<STDIN> !~ /Y/i) {
            $dbh->rollback();
            print "Changes have been rolled back\n";
            return 0;
        }
        else {
            $dbh->commit();
            print "Changes have been committed\n";
        }
    }
    else {
        $dbh->commit();
    }

    return 1;

} ## end of confirm_commit


sub add_db_to_group {

    ## Add a database to a group
    ## Will create the group as needed
    ## Does not commit
    ## Arguments: two
    ## 1. Database name
    ## 2. Group name (may have :role specifier)
    ## Returns: group name and role name

    my ($db,$fullgroup) = @_;

    ## Figure out the role. Defaults to target
    my ($group,$role) = extract_name_and_role($fullgroup);

    if (! exists $DBGROUP->{$group}) {
        ## Extra argument prevents load_bucardo_info from being called by the sub
        create_dbgroup($group, 1);
    }

    $SQL = 'INSERT INTO bucardo.dbmap(db,dbgroup,role) VALUES (?,?,?)';
    $sth = $dbh->prepare($SQL);
    eval {
        $sth->execute($db,$group,$role);
    };
    if ($@) {
        my $message = qq{Cannot add database "$db" to group "$group"};
        if ($@ =~ /"dbmap_unique"/) {
            die qq{$message: already part of the group\n};
        }
        die qq{$message: $@\n};
    }

    ## Reload our hashes
    load_bucardo_info(1);

    return $group, $role;

} ## end of add_db_to_group


sub remove_db_from_group {

    ## Removes a database from a group: deletes from bucardo.dbmap
    ## Does not commit
    ## Arguments: two
    ## 1. Database name
    ## 2. Group name
    ## 3. Boolean: if true, prevents the reload
    ## Returns: undef

    my ($db,$group,$noreload) = @_;

    $SQL = 'DELETE FROM bucardo.dbmap WHERE db=? AND dbgroup=?';
    $sth = $dbh->prepare_cached($SQL);
    $sth->execute($db, $group);

    ## Reload our hashes
    $noreload or load_bucardo_info(1);

    return;

} ## end of remove_db_from_group


sub change_db_role {

    ## Changes the role of a database: updates bucardo.dbmap
    ## Does not commit
    ## Arguments: four
    ## 1. New role
    ## 2. Name of the dbgroup
    ## 3. Name of the database
    ## 4. Boolean: if true, prevents the reload
    ## Returns: undef

    my ($role,$group,$db,$noreload) = @_;

    $SQL = 'UPDATE bucardo.dbmap SET role=? WHERE dbgroup=? AND db=?';
    $sth = $dbh->prepare_cached($SQL);
    $sth->execute($role,$group,$db);

    ## Reload our hashes
    $noreload or load_bucardo_info(1);

    return;

} ## end of change_db_role


sub update_dbmap {

    ## Update the values in the bucardo.dbmap table
    ## Arguments: three
    ## 1. Name of the database
    ## 2. Name of the database group
    ## 3. Hashref of things to change
    ## Returns: undef

    my ($db,$group,$changes) = @_;

    ## This should not need quoting as they are all [\w\d]
    my $list = join ',' => map { "$_=$changes->{$_}" } sort keys %$changes;

    $SQL = "UPDATE bucardo.dbmap SET $list WHERE db=? AND dbgroup=?";
    $sth = $dbh->prepare($SQL);
    $sth->execute($db, $group);

    return;

} ## end of update_dbmap


sub create_herd {

    ## Creates a new entry in the bucardo.herd table
    ## Caller should have alredy checked for existence
    ## Does not commit
    ## Arguments: two
    ## 1. Name of the new herd
    ## 2. Boolean: if true, prevents the reload
    ## Returns: undef

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

    $SQL = 'INSERT INTO bucardo.herd(name) VALUES (?)';
    $sth = $dbh->prepare($SQL);
    eval {
        $sth->execute($name);
    };
    if ($@) {
        print qq{Failed to create database group "$name"\n$@\n};
        exit 1;
    }

    ## Reload our hashes
    $noreload or load_bucardo_info(1);

    return;

} ## end of create_herd


__END__

=head1 NAME

bucardo - utility script for controlling the Bucardo program

=head1 VERSION

This document describes version 4.99.5 of bucardo

=head1 SYNOPSIS

  ./bucardo install

  ./bucardo list dbs

  ./bucardo add sync testsync source=herd1 type=pushdelta targetdb=B

  ./bucardo add sync testsync source=herd1 type=pushdelta targetdb=B tables=tab1,tab2,tab3

  ./bucardo add database newdb dbname=internal_name port=5432 host=myserver

  ./bucardo add all tables db=foo [herd=x] [pkonly]

  ./bucardo add all sequences db=foo [herd=x]

  ./bucardo add herd newherd table1 table2 table3 ...

  ./bucardo add dbgroup name db1 db2 db3 ...

  ./bucardo start "Starting up - Greg"

  ./bucardo stop "Bringing down for debugging - Raul E."

  ./bucardo ping

  ./bucardo status

  ./bucardo status sync1 sync2

  ./bucardo kick sync1 sync2

  ./bucardo kick sync1 0

  ./bucardo reload_config

  ./bucardo upgrade

  ./bucardo reload sync

  ./bucardo validate sync

  ./bucardo message "Your message here"

  ./bucardo config show

  ./bucardo config set foo=bar baz=123


=head1 DESCRIPTION

The bucardo script is the main interaction to a running Bucardo instance. It can 
be used to start and stop Bucardo, add new items, kick syncs, and even install and 
upgrade Bucardo itself. For more complete documentation, please view the wiki at:

http://bucardo.org/

=head1 COMMANDS

=over 4

=item B<install>

Usage: ./bucardo install

Attempts to install the Bucardo schema from the file 'bucardo.schema' into an existing 
Postgres cluster. The user 'bucardo' and database 'bucardo' will be created first as needed.
This is an interactive installer, but you can supply the following values from the command 
line:

=over 2

=item --dbuser (defaults to postgres)

=item --dbname (defaults to postgres)

=item --dbport (defaults to 5432)

=item --piddir (defaults to /var/run/bucardo/)

=back

=item B<upgrade>

Usage: ./bucardo upgrade

Upgrades an existing Bucardo installation to the current version of the bucardo script.
Requires that the bucardo script and the bucardo.schema file be the same version. All
changes should be backwards compatible, but you may need to re-validate existing scripts
to make sure changes get propagated to all databases.

=item B<start>

Usage: ./bucardo start "Reason --name"

Restarts Bucardo cleanly by first issuing the equivalent of a stop to ask any existing Bucardo
processes to exit, and then starting a new Bucardo MCP process. A short reason and name should
be provided - these are logged in the reason_file file and sent in the email sent when Bucardo
has been started up.

Before attempting to kill any old processes, a ping command with a timeout of 5 seconds is issued.
If this returns successfully (indicating an active MCP process already running), the script will
exit with a return value of 2.

=item B<stop>

Usage: ./bucardo stop "Reason --name"

Forces Bucardo to quit by creating a stop file which all MCP, CTL, and KID processes should
detect and cause them to exit. Note that active syncs will not exit right away, as they
will not look for the stop file until they have finished their current run. Typically,
you should scan the list of processes after running this program to make sure that all Bucardo
processes have stopped. One should also provide a reason for issuing the stop - usually
this is a short explanation and your name. This is logged in the reason_file file and
is also used by Bucardo when it exits and sends out mail about its death.

=item B<list>

Usage: ./bucardo list <type> <regex>

Lists summary information about dbs, dbgroups, tables, sequences,
syncs, herds, customnames, customcols or customcode. Adding anything
after the type will look up all matching entries.

=item B<add>

Usage:  add <item_type> <item_name>

Usage:  add db <dbname> dbname=internal_name port=xxx host=xxx user=xxx pass=xxx service=xxx conn=xxx ssp=1/0

Usage:  add dbgroup name db1 db2 db3 ...

Usage:  add table [schema].table db=internal_db_name ping=bool standard_conflict=xxx herd=xxx

Usage:  add all tables [herd=xxx] [pkonly]

Usage:  add sequence [schema].table herd=xxx

Usage:  add all sequences herd=xxx

Usage:  add sync syncname options

Usage:  add herd name

Usage:  add customname tablename newtablename [sync=x db=x]

Usage:  add customcols tablename select_clause [sync=x db=x]

Usage:  add customcode <name> <whenrun=value> <src_code=filename> [optional information]

Tells Bucardo about new objects it should know about. These commands can
replace direct manipulation of the tables in the bucardo schema for the
supported object types (you'll still need to add things like the mappings between objects on your own).

=item B<remove>

Usage:  remove <item_type> <item_name>

Removes one or more items from the Bucardo database. Valid item types are database, 
dbgroup, herd, sync, table, and sequence.

=item B<kick>

Usage: ./bucardo kick <syncname(s)> [timeout]

Tells one or more named syncs to fire as soon as possible. Note that this simply sends a request that 
the sync fire: it may not start right away if the same sync is already running, or if the source or 
target database has exceeded the number of allowed Bucardo connections. If the final argument is a 
number, it is treated as a timeout. If this number is zero, the bucardo command will not return 
until the sync has finished. For any other number, the sync will wait at most that number of seconds. 
If any sync has not finished before the timeout, a false value is returned. In all other cases, a 
true value is returned.

If a timeout is given, the total completion time in seconds is also displayed. If the sync is going to 
multiple targets, the time that each target takes from the start of the kick is also shown as each 
target finishes.

=item B<reload_config>

Forces Bucardo to reload the bucardo_config file, and then restart all processes to ensure that the new 
information is loaded.

=item B<show>

Usage: ./bucardo show <all|setting1> [setting2..]

Shows the current settings in the bucardo_config table. Use the keyword 'all' to see all the settings, or 
specify one or more search terms.

=item B<set>

Usage: ./bucardo set setting1=value [setting2=value]

Sets one or more items inside the bucardo_config table. Setting names are case-insensitive.

=item B<ping>

Sends a ping notice to the MCP process to see if it will respond. By default, it will wait 15 seconds. A 
numeric argument will change this timeout. Using a 0 as the timeout indicates waiting forever. If a response 
was returned, the program will exit with a value of 0. If it times out, the value will be 1.

=item B<status>

Usage: ./bucardo status [syncname(s)] [--sort=#] [--showdays] [--compress]

Shows the brief status of all known syncs in a tabular format. If given one or more syncnames, 
shows detailed information for each one. To see detailed information for all syncs, simply 
use 'status all'.

When showing brief information, the columns are:

=over 8

=item 1. B<Name>

The name of the sync

=item 2. B<State>

The state of the sync. Can be 'Good', 'Bad', 'Empty', or 'No records found'

=item 3. B<Last good>

When the sync last successfully ran.

=item 4. B<Time>

How long it has been since the last sync success

=item 5. B<Last I/U>

The number of insert and deletes performed by the last successful sync. May also show 
the number of rows truncated (T) or conflicted (C), if applicable.

=item 6. B<Last bad>

When the sync last failed.

=item 7. B<Time>

How long it has been since the last sync failure

=back


=item B<activate> syncname [syncname2 syncname3 ...] [timeout]

Activates one or more named syncs. If given a timeout argument, it will wait until it has received 
confirmation from Bucardo that each sync has been successfully activated.

=item B<deactivate> syncname [syncname2 syncname3 ...] [timeout]

Deactivates one or more named syncs. If given a timeout argument, it will wait until it has received 
confirmation from Bucardo that the sync has been successfully deactivated.

=item B<message>

Adds a message to the running Bucardo logs. This message will appear prefixed with "MESSAGE: ". If 
Bucardo is not running, the message will go to the logs the next time Bucardo is running and someone 
adds another message.

=back

=head1 OPTIONS

It is usually easier to set most of these options at the top of the script, or make an alias for them, 
as they will not change very often if at all.

=over 4

=item B<--dbport=number>

=item B<--dbhost=string>

=item B<--dbname=string>

=item B<--dbuser=string>

=item B<--dbpass=string>

The port, host, and name of the Bucardo database, the user to connect as, and the password to use.

=item B<--verbose>

Makes bucardo run verbosely. Default is off.

=item B<--quiet>

Tells bucardo to be as quiet as possible. Default is off.

=item B<--help>

Shows a brief summary of usage for bucardo.

=back

=head2 Kick arguments

The following arguments are only used with the 'kick' command:

=over 4

=item B<--retry=#>

The number of times to retry a sync if it fails. Defaults to 0.

=item B<--retrysleep>

How long to sleep, in seconds, between each retry attempt.

=item B<--notimer>

By default, kicks with a timeout argument give a running real-time summary of time elapsed by 
using the backspace character. This may not be wanted if running a kick, for example, 
via a cronjob, so turning --notimer on will simply print the entire message without backspaces.

=back

=head2 Status arguments

The following arguments are only used with the 'status' command:

=over 4

=item B<--showdays>

Specifies whether or not do list the time interval with days, or simply show the hours. For example, 
"3d 12h 6m 3s" vs. "48h 6m 3s"

=item B<--compress>

Specifies whether or not to compress the time interval by removing spaces. Mostly used to limit 
the width of the 'status' display.

=item B<--sort=#>

Requests sorting of the 'status' output by one of the nine columns. Use a negative number to reverse 
the sort order.

=back

=head2 Startup arguments

The following arguments are only applicable when using the "start" command:

=over 4

=item B<--sendmail>

Tells Bucardo whether or not to send mail on interesting events: startup, shutdown, and errors. Default is on.
Only applicable when using ./bucardo start.

=item B<--extraname=string>

A short string that will be appended to the version string as output by the Bucardo process names. Mostly 
useful for debugging.

=item B<--debugfilesep>

Forces creation of separate log files for each Bucardo process of the form "log.bucardo.X.Y", 
where X is the type of process (MCP, CTL, or KID), and Y is the process ID.

=item B<--debugsyslog>

Sends all log messages to the syslog daemon. On by default. The facility used is controlled by 
the row "syslog_facility" in the bucardo_config table, and defaults to "LOG_LOCAL1".

=item B<--debugfile>

If set, writes detailed debugging information to one or more files.

=item B<--debugdir=directory name>

Directory where the debug files should go.

=item B<--debugname=string>

Appends the given string to the end of the default debug file name, "log.bucardo". A dot is added 
before the name as well, so a debugname of "rootdb" would produce a log file named "log.bucardo.rootdb".

=item B<--cleandebugs>

Forces removal of all old debug files before running.

=back

=head1 FILES

In addition to command-line configurations, you can put any options inside of a file. The file F<.bucardorc> in 
the current directory will be used if found. If not found, then the file F<~/.bucardorc> will be used. Finally, 
the file /etc/bucardorc will be used if available. The format of the file is option = value, one per line. Any 
line starting with a '#' will be skipped. Any values loaded from a bucardorc file will be overwritten by 
command-line options. All bucardorc files can be ignored by supplying a C<--no-bucardorc> argument. A specific 
file can be forced with the C<--bucardorc=file> option; if this option is set, bucardo will refuse to run 
unless that file can be read.

=head1 ENVIRONMENT VARIABLES

The bucardo script uses I<$ENV{HOME}> to look for a F<.bucardorc> file.

=head1 BUGS

Bug reports and feature requests are always welcome, please visit http://bucardo.org or email bucardo-general@bucardo.org.

=head1 SEE ALSO

Bucardo

=head1 COPYRIGHT

Copyright 2006-2012 Greg Sabino Mullane <greg@endpoint.com>

This program is free to use, subject to the limitations in the LICENSE file.

=cut