#!/usr/bin/perl -w
use strict;

=head1 NAME

    update_postgresql_tables_1015_to_1016 -- Bacula System upgrade tool

=head1 SYNOPIS

    update_postgresql_tables_1015_to_1016 
                --bin=/opt/bacula/bin 
                --conf=/opt/bacula/etc/bacula-dir.conf

                -U|--user|-u      user
                -D|--database|-d  database
                -h|--host         hostname
                -T|--type         database type (Pg|mysql)
                -P|--port         database port

                --work-mem=20GB   work mem setting for Pg (30% of the ram)
                --merge-join      use merge join and hash join

=cut

eval "use DBI;";

die "\nERROR: Please install Perl(DBI) and Perl(DBD::Pg) packages\n\tperl-DBI and perl-DBD-Pg or libdbi-perl and libdbd-pg-perl to run this script\n" if $@;
use Getopt::Long qw/:config no_ignore_case/;
use Data::Dumper;
use Pod::Usage;
use POSIX qw(strftime);

my $bin_path = "/opt/bacula/bin";
my $dir_conf = "/opt/bacula/etc/bacula-dir.conf";
my $org_version  = 1015;
my $dest_version = 1016;
my $file_temp = "File${dest_version}";

my $user='';
my $pass='';
my $host='';
my $dbname='';
my $dbport='';
my $dbtype='postgresql';

# Can be set to 10 or 20G if possible (using --work-mem= option)
my $work_mem='2000MB';

# Cannot be > to 2GB
my $maintenance_work_mem='2000MB';

# Force merge join
my $use_mergejoin = 0;

GetOptions('bin=s'    => \$bin_path,
           'conf=s'   => \$dir_conf,
           'U=s'      => \$user,
           'user|u=s' => \$user,
           'host|h=s' => \$host,
           'd|database|D=s' => \$dbname,
           'type|T=s' => \$dbtype,
           'port|P=i' => \$dbport,
           'work-mem=s' => \$work_mem,
           'merge-join' => \$use_mergejoin,
    ) || Pod::Usage::pod2usage(-exitval => 2, -verbose => 2) ;

my $df = "df -Ph";

if ($^O ne 'linux') {
    $df = "df -k";
}

my %dbi;
my $SQL;

my %PG_SQL = (type => 'Pg',
other => [
"ALTER TABLE Job ADD COLUMN FileTable text default 'File'",
"ALTER TABLE JobHisto ADD COLUMN FileTable text default 'File'",
"ALTER TABLE Media ALTER VolWrites TYPE BIGINT",
"CREATE TABLE Snapshot (
  SnapshotId      serial,
  Name            text not null,
  CreateTDate     bigint default 0,
  CreateDate      timestamp without time zone not null,
  ClientId        int default 0,
  Volume          text not null,
  Device          text not null,
  Type            text not null,
  Retention       integer default 0,
  Comment         text,
  primary key (SnapshotId)
)",
"CREATE UNIQUE INDEX snapshot_idx ON Snapshot (Device text_pattern_ops, 
                                               Volume text_pattern_ops, 
                                               Name text_pattern_ops)"
              ],
create_file => "
CREATE TABLE $file_temp
(
    FileId            bigint      not null,
    FileIndex         integer     not null  default 0,
    JobId             integer     not null,
    PathId            integer     not null,
    Filename          text        not null  default '',
    DeltaSeq          smallint    not null  default 0,
    MarkId            integer     not null  default 0,
    LStat             text        not null,
    Md5               text        not null
)",
create_index => ["CREATE INDEX ${file_temp}_jpfid_idx on ${file_temp} (JobId, PathId, Filename text_pattern_ops)",
                 "ALTER TABLE ONLY ${file_temp} ADD CONSTRAINT ${file_temp}_pkey PRIMARY KEY (FileId)"],

# ALTER SEQUENCE is useful to avoid dropping the sequence when dropping the table

rename_object => [
"ALTER TABLE ${file_temp} RENAME TO file",
"ALTER INDEX ${file_temp}_pkey RENAME TO file_pkey",
"ALTER INDEX ${file_temp}_jpfid_idx RENAME TO file_jpfid_idx",
"CREATE SEQUENCE file_fileid_seq OWNED BY file.fileid",
"ALTER TABLE File ALTER COLUMN FileId SET DEFAULT nextval('file_fileid_seq'::regclass)",
"SELECT setval('file_fileid_seq', (SELECT MAX(FileId) FROM File), true)"],

check_file_temp => "SELECT relname FROM pg_class WHERE relkind = 'r' AND relname ilike '$file_temp'",
check_nb_file_temp => "SELECT relname FROM pg_class WHERE relkind = 'r' AND relname ilike 'nb_file_temp'",
check_file_temp_idx => "SELECT relname FROM pg_class WHERE relkind = 'i' AND relname ilike '${file_temp}_pkey'",
);

my %MY_SQL = (
type => 'mysql',
other => [
"ALTER TABLE Media CHANGE VolWrites VolWrites BIGINT UNSIGNED",
"ALTER TABLE Job ADD COLUMN FileTable CHAR(20) default 'File'",
"ALTER TABLE JobHisto ADD COLUMN FileTable CHAR(20) default 'File'",
"CREATE TABLE Snapshot (
  SnapshotId      INTEGER UNSIGNED AUTO_INCREMENT,
  Name            TINYBLOB NOT NULL,
  CreateTDate     BIGINT   NOT NULL,
  CreateDate      DATETIME NOT NULL,
  ClientId        INTEGER DEFAULT 0,
  Volume          TINYBLOB NOT NULL,
  Device          TINYBLOB NOT NULL,
  Type            TINYBLOB NOT NULL,
  Retention       INTEGER DEFAULT 0,
  Comment         BLOB,
  primary key (SnapshotId)
)",
"CREATE UNIQUE INDEX snapshot_idx ON Snapshot (Device(255), 
                                               Volume(255), 
                                               Name(255))"
],
create_file => "
CREATE TABLE IF NOT EXISTS $file_temp (
   FileId BIGINT UNSIGNED NOT NULL,
   FileIndex INTEGER UNSIGNED DEFAULT 0,
   JobId INTEGER UNSIGNED NOT NULL REFERENCES Job,
   PathId INTEGER UNSIGNED NOT NULL REFERENCES Path,
   Filename BLOB NOT NULL,
   DeltaSeq SMALLINT UNSIGNED DEFAULT 0,
   MarkId INTEGER UNSIGNED DEFAULT 0,
   LStat TINYBLOB NOT NULL,
   MD5 TINYBLOB
   )
",
create_index => [
"ALTER TABLE $file_temp ADD PRIMARY KEY (FileId)",
"ALTER TABLE $file_temp MODIFY COLUMN FileId BIGINT UNSIGNED NOT NULL AUTO_INCREMENT",
"CREATE INDEX JobId on ${file_temp} (JobId)",
"CREATE INDEX JobId_2 on ${file_temp} (JobId, PathId, Filename(255))"
],

rename_object => ["ALTER TABLE ${file_temp} RENAME TO File"],
check_file_temp => "SHOW TABLES LIKE '$file_temp'",
check_nb_file_temp => "SHOW TABLES LIKE 'nb_file_temp'",
check_file_temp_idx => "SHOW INDEX FROM $file_temp",
);

sub p
{
    print strftime("%H:%M:%S ", localtime);
    print @_;
}

sub connect_db
{
    my ($dbi, $user, $password) = @_;

    if (!$dbi) {
        ($dbi, $user, $password) = get_bacula_db();
    }

    my $dbh;

    if ($dbi !~ /Pg|mysql|SQLite/) {
        die("ERROR: It looks you have a case problem with your DBI string.\n" . 
            "$dbi doesn't contain Pg, mysql or SQLite");
    }
    
    $dbh = DBI->connect($dbi, $user, $password);

    {
       no warning;
       die("ERROR: Can't connect to your database:\n$DBI::errstr\n\n" .
           "You can try to run this script as 'postgres'\n" .
           "For example: $0 -d bacula\n"
           ) unless ($dbh);
    }

    $dbh->{FetchHashKeyName} = 'NAME_lc';
    $dbh->{PrintError} = 1;

    if ($dbi =~ /Pg/) {
        $SQL = \%PG_SQL;
        $dbh->do("set work_mem='$work_mem';");
        $dbh->do("set maintenance_work_mem='$maintenance_work_mem';");

    } else {
        $SQL = \%MY_SQL;
    }

    return $dbh;
}

sub get_bacula_db
{
    if ($user || $host || $dbname || $dbport) {
        $dbi{db_user}    = $user;
        $dbi{db_address} = $host;
        $dbi{db_name}    = $dbname;
        $dbi{db_port}    = $dbport;
        $dbi{db_type}    = $dbtype;

        die "Unable to find database name in argument"
            unless $dbi{db_name};

    } else {

        die "ERROR: Unable to find or access $bin_path/dbcheck, you can try to use command line arguments"
            unless (-x "$bin_path/dbcheck");
        
        die "ERROR: Unable to find or access $dir_conf, you can try to run this script as 'bacula'"
            unless (-r $dir_conf);
        
        my @conf = `$bin_path/dbcheck -B -c $dir_conf`;
        if ($? != 0) {
            die "ERROR: dbcheck returned an error\n@conf";
        }
        
        my $found=0;
        
        foreach my $l (@conf) {
            chomp($l);
            if ($l !~ /=/) {
                print "Strange line on dbcheck output $l\n";
                next;
            }
            if ($l =~ /catalog=/) {
                if ($found++ > 0) {
                    print "Will choose the first catalog, discarding $l\n";
                    print "To migrate $l, use command line parameters\n";
                    last;
            }
            }
            my ($k, $v) = split(/=/, $l);
            $dbi{$k} = $v;
        }

        die "ERROR: Unable to get information from dbcheck"
            unless ($dbi{db_type} and $dbi{db_name});
    }

    my $dbi_string;
    if (lc($dbi{db_type}) eq 'postgresql') {
        $dbi_string = "DBI:Pg";

    } elsif (lc($dbi{db_type}) eq 'mysql') {
        $dbi_string = "DBI:mysql";

    } else {
        die "Database type $dbi{db_type} is not supported by this script";
    }

    # Is present!
    if ($dbi{db_name}) {
        $dbi_string .= ":database=$dbi{db_name}";
    }

    if ($dbi{db_address}) {
        $dbi_string .= ";host=$dbi{db_address}";
    }

    if ($dbi{db_port}) {
        $dbi_string .= ";port=$dbi{db_port}";
    }

    if ($dbi{db_socket}) {
        $dbi_string .= ";host=$dbi{db_socket}";
    }

    return ($dbi_string, $dbi{db_user}, $dbi{db_password});
}

my $ret;
my $nb_file=0;
my $nb_file_temp=0;
my $max_fileid=0;
my $max_fileid_temp=0;
my $dbh = connect_db();

p "INFO: Connexion to the catalog OK\n";

################################################################

$ret = $dbh->selectrow_hashref("SELECT VersionId as ver FROM Version");
if (!$ret || !$ret->{ver} || $ret->{ver} != $org_version) {
    die "ERROR: The current version of the catalog is $ret->{ver}, " . 
        "was expecting $org_version";
}

p "INFO: Found catalog version $org_version\n";

################################################################

$ret = $dbh->selectrow_hashref("SELECT sum(JobFiles) as nb FROM Job WHERE Type='B'");
$nb_file = $ret->{nb};

if (1 || $ret->{nb} > 700000000) {
    p "INFO: Your catalog is already pretty big.\n",
      "      Make sure to stay in touch with Bacula Systems Support team\n",
      "      during the catalog migration\n\n",
      "      The best way to proceed with your Catalog is to create a second\n",
      "      File table while your Bacula is running. This method requires to\n",
      "      store at least 2 times the size of your database during the\n",
      "      migration.\n\n",

      "      PLEASE MAKE SURE YOUR CATALOG FILESYSTEM IS BIG ENOUGH!!\n\n";
    system("$df | head -1");
    system("$df | grep '^/dev/'");
    print "\n",
          " -- Press enter key to continue or press Ctrl-C to cancel ---\n";
    $ret = <>;

}

################################################################

################################################################

$ret = $dbh->do($SQL->{check_file_temp});

my $do_drop=0;
my $do_create=1;

if ($ret > 0) {

    p "INFO: We found a previous table, do you want to update it\n",
      "      with new records? (Y/n): ";

    $ret = <>;

    if ($ret =~ m/y/i) {
        $do_drop = 0;
        $do_create = 0;

    } else {
        print "\n";
        p "INFO: Are you sure you want to drop the previous table?\n";
        print "\n",
              " -- Press enter key to continue or press Ctrl-C to cancel ---\n";

        $ret = <>;

        $do_drop = 1;
        $do_create = 1;
    }
}

################################################################

$dbh->begin_work();

################################################################

if ($do_drop) {
    p "INFO: Found a previous try, removing it\n";
    $dbh->do("DROP TABLE $file_temp");
    $dbh->do("DROP TABLE nb_file_temp");
}


################################################################
# count(1) or Max(FileId) may take hours.... So keep the record

# May fail
$ret = $dbh->do($SQL->{check_nb_file_temp});

if ($ret eq '0E0') {
    $ret = 0;
}

if ($ret != 0 && $do_drop) {
    $dbh->do("DROP TABLE nb_file_temp");
} 

if ($ret == 0) {
    p "INFO: Creating the temporary table 'nb_file_temp'\n";
    $dbh->do("CREATE TABLE nb_file_temp (maxid bigint)");
}

$ret = $dbh->selectrow_hashref("SELECT maxid FROM nb_file_temp");
if ($ret) {
    $max_fileid_temp = $ret->{maxid} || 0;
}

################################################################

if ($do_create) {
    p "INFO: Creating the temporary $file_temp table\n";
    $ret = $dbh->do($SQL->{create_file});

    if (!$ret) {
        print "ERROR: Unable to create the temporary table, check permission\n",
              "       on the database for the current user\n",
              "       ALTER DATABASE $dbi{db_name} OWNER TO $dbi{db_user};\n",
              "or\n",
              "       GRANT ALL TO $dbi{db_user} on $dbi{db_name};\n";
        exit;
    }
}

################################################################
p "INFO: Will update the temp File table with records\n";

# Need $max_fileid and $max_fileid_temp
sub update_file_temp 
{
    # We choose nested loop plan to avoid extra sort operation
    if (!$use_mergejoin) {
        $dbh->do("set enable_mergejoin to off");
        $dbh->do("set enable_hashjoin to off");
    }
    $ret = $dbh->do("
   INSERT INTO $file_temp (FileId, FileIndex, JobId, PathId, 
                           Filename, DeltaSeq, MarkId, LStat, Md5)
   SELECT FileId, FileIndex, JobId, PathId, 
          Filename.Name, DeltaSeq, MarkId, LStat, Md5
    FROM File JOIN Filename USING (FilenameId)
   WHERE FileId > $max_fileid_temp");

    if ($ret eq '0E0') {
        $ret = 0;
    }

    if (!$use_mergejoin) {
        $dbh->do("set enable_mergejoin to default");
        $dbh->do("set enable_hashjoin to default");
    }

    p "INFO: $ret records have been inserted in the new table\n";

    $ret = $dbh->selectrow_hashref("SELECT MAX(FileId) as max FROM File");
    $max_fileid = $ret->{max} || 0;

    # Keep track for the next run
    $dbh->do("DELETE FROM nb_file_temp");
    $ret = $dbh->do("INSERT INTO nb_file_temp (maxid) VALUES ($max_fileid)");

    if (!$ret) {
        print "ERROR: Unable to update temporary file table...\n";
        exit 1;
    }

    $max_fileid_temp = $max_fileid;
}

update_file_temp();

################################################################

$ret = $dbh->commit();
if (!$ret) {
    p "ERROR: Transaction aborted, please retry this script\n";
    exit 1;
}

################################################################

$dbh->begin_work();

$ret = $dbh->do($SQL->{check_file_temp_idx});

if ($ret eq '0E0') {
    $ret = 0;
}

if (!$ret) {
    p "INFO: Creating indexes if not already created\n";
    map { $dbh->do($_) } @{$SQL->{create_index}};
}

$ret = $dbh->commit();                 # this operation may fail

if (!$ret) {
    print "ERROR: unable to create indexes properly\n";
    exit 1;
}

################################################################

p "INFO: Now it's time to STOP bacula and start the final part of the\n", 
  "      upgrade\n\n",
  "      Everything was commited to the database, so you can stop here\n",
  "      and restart the process later.\n"
;
print "\n",
    " -- Press enter key to continue or press Ctrl-C to cancel ---\n";

$ret = <>;

################################################################

$dbh->begin_work();

p "INFO: Do a last update from file to temporary file table\n";

# should be pretty fast now
update_file_temp();

$ret = $dbh->commit();

if (!$ret) {
    print "ERROR: unable to update file properly\n";
    exit 1;
}

my %users;
if ($SQL->{type} eq 'Pg') {

    p "INFO: Copying GRANT information\n";
    $dbh->begin_work();

    $ret = $dbh->selectall_arrayref("
SELECT grantee, privilege_type
  FROM information_schema.role_table_grants 
 WHERE table_name='file' and grantee <> 'postgres' 
 GROUP BY grantee, privilege_type");

    if ($ret) {
        foreach my $elt (@$ret) {
            $dbh->do("GRANT $elt->[1] ON TABLE ${file_temp} TO $elt->[0]");
            $users{$elt->[0]} = 1;
        }
    }
    
    $dbh->commit();
}

p "INFO: Dropping original File table\n";
$dbh->begin_work();
$dbh->do("DROP TABLE File");
$dbh->do("DROP TABLE Filename");
map { $dbh->do($_) } @{ $SQL->{rename_object} };
$dbh->do("DROP TABLE nb_file_temp");
map { $dbh->do($_) } @{ $SQL->{other} };
$dbh->do("UPDATE Version SET VersionId=$dest_version");
$ret = $dbh->commit();

# Time to restore GRANT information for the new File sequence
if ($SQL->{type} eq 'Pg') {
    if (scalar(keys %users) > 0) {
        p "INFO: Restoring GRANT information for File sequence\n";
        $dbh->begin_work();
        foreach my $u (keys %users) {
            $dbh->do("GRANT ALL ON file_fileid_seq TO $u");
        }
        $dbh->commit();
    }

    my $bacula="bacula";

    # If the current user is postgres, we can change the owner of the table to bacula
    my $row = $dbh->selectrow_arrayref("SELECT current_user");
    if ($row && $row->[0] eq 'postgres') {
        p "INFO: current user is $row->[0]\n";

        # Should give the bacula user name
        $row = $dbh->selectrow_arrayref("
SELECT pg_user.usename 
  FROM pg_class 
    JOIN pg_user ON (pg_user.usesysid = pg_class.relowner) 
 WHERE relname = 'restoreobject'");

        if ($row && $row->[0]) {
            $bacula = $row->[0];
            p "INFO: owner of bacula tables is $bacula\n";
        }
        if ($bacula ne "postgres") {
            p "INFO: Trying to change the OWNER of the file table to $bacula\n";
            $dbh->do("ALTER TABLE File OWNER TO $bacula");
            $dbh->do("ALTER SEQUENCE file_fileid_seq OWNER TO $bacula");
        }
    }
}

if (!$ret) {
    print "ERROR: unable to drop old tables and/or activate new ones\n";
    exit 1;
}

$dbh->begin_work();

p "INFO: It is possible to upgrade bacula at this point...\n";
p "INFO: Deleting orphan records\n";
my $total=0;

$ret = $dbh->do("DELETE FROM File WHERE NOT EXISTS " .
                " (SELECT 1 FROM Job WHERE Job.JobId = File.JobId)");

if ($ret ne '0E0') {
    $total = $ret;
}

$ret = $dbh->do("DELETE FROM File WHERE JobId IN " .
                 " (SELECT JobId FROM Job WHERE Job.PurgedFiles = 1)");

if ($ret ne '0E0') {
    $ret += $total;
}

p "INFO: $total records deleted\n";

$ret = $dbh->commit();

if (!$ret) {
    print "ERROR: unable to update file properly\n";
    exit 1;
}

p "INFO: You can now upgrade bacula itself with your package manager\n";


################################################################


exit 0;

