# Copyright (c) 2004  Joshua Hoblitt
#
# $Id: SQL.pm,v 1.78 2008-12-14 22:54:25 eugene Exp $

package Nebulous::Server::SQL;

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

our $VERSION = '0.02';

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

my %sql = (
    set_transaction_model   => qq{
        SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED
    },
    update_instance_uri => qq{
        UPDATE instance
        SET vol_id = ?, uri = ?
        WHERE ins_id = ?
    },
    last_insert_id      => qq{
        SELECT LAST_INSERT_ID()
    },
    new_object          => qq{
        INSERT INTO storage_object
        (so_id, ext_id, type)
        VALUES (?, ?, 'REG_FILE')
    },
    new_object_attr  => qq{
        INSERT INTO storage_object_attr
        (so_id, read_lock, write_lock)
        VALUES (?, 0, NULL)
    },
    delete_object       => qq{
        DELETE FROM storage_object
        WHERE storage_object.so_id = ?
    },
    new_object_instance => qq{
        INSERT INTO instance 
        (so_id, vol_id, uri)
        VALUES (LAST_INSERT_ID(), ?, 'error')
    },
    new_instance        => qq{
        INSERT INTO instance
        (so_id, vol_id, uri)
        VALUES (?, ?, 'error')
    },
    get_object          => qq{
        SELECT
            so_id,
            ext_id,
            read_lock,
            write_lock,
            epoch,
            mtime
        FROM storage_object
        JOIN storage_object_attr
        USING (so_id)
        WHERE ext_id = ?
    },
    stat_object          => qq{
        SELECT
            storage_object.so_id,
            storage_object.ext_id,
            read_lock,
            write_lock,
            storage_object_attr.epoch,
            storage_object_attr.mtime,
            COUNT(instance.so_id) as instances
        FROM storage_object
        JOIN storage_object_attr
            USING (so_id)
        JOIN instance
            USING (so_id)
        WHERE ext_id = ?
        GROUP BY storage_object.so_id
    },
    # Note: this sets an update lock
    get_object_locks    => qq{
        SELECT
            storage_object.so_id,
            read_lock,
            write_lock 
        FROM storage_object
        JOIN storage_object_attr
        USING (so_id)
        WHERE ext_id = ?
        FOR UPDATE
    },
    new_object_xattr  => qq{
        INSERT INTO storage_object_xattr
            SELECT
                so_id,
                ?,
                ?        
            FROM storage_object
            WHERE ext_id = ?
    },
    replace_object_xattr  => qq{
        REPLACE INTO storage_object_xattr
            SELECT
                so_id,
                ?,
                ?        
            FROM storage_object
            WHERE ext_id = ?
    },
    list_object_xattr    => qq{
        SELECT storage_object_xattr.name
        FROM storage_object
        JOIN storage_object_xattr
        USING (so_id)
        WHERE ext_id = ?
    },
    get_object_xattr    => qq{
        SELECT storage_object_xattr.value
        FROM storage_object
        JOIN storage_object_xattr
        USING (so_id)
        WHERE ext_id = ?
            AND name = ?
    },
    remove_object_xattr    => qq{
        DELETE FROM storage_object_xattr
        WHERE so_id = (SELECT so_id from storage_object where ext_id = ?)
            AND name = ?
    },
    set_write_lock      => qq{
        UPDATE storage_object_attr
        JOIN storage_object
        USING(so_id)
        SET write_lock = 'write'
        WHERE storage_object.ext_id = ?
    },
    delete_write_lock   => qq{
        UPDATE storage_object_attr
        JOIN storage_object
        USING(so_id)
        SET write_lock = NULL
        WHERE ext_id = ?
    },
    increment_read_lock => qq{
        UPDATE storage_object_attr
        JOIN storage_object
        USING(so_id)
        SET read_lock = read_lock + 1
        WHERE ext_id = ?
    },
    decrement_read_lock => qq{
        UPDATE storage_object_attr
        JOIN storage_object
        USING(so_id)
        SET read_lock = read_lock - 1
        WHERE ext_id = ?
    },
    delete_instance     => qq{
        DELETE FROM instance
        WHERE uri = ?
    },
    get_object_from_uri   => qq{
        SELECT so_id
        FROM instance
        WHERE uri = ?
    },
    get_instance_count   => qq{
        SELECT count(ins_id)
        FROM instance
        WHERE so_id = ?
    },
    get_object_instances    => qq{
        SELECT
            storage_object.so_id,
            uri,
            available
        FROM storage_object
        JOIN instance
            USING (so_id)
        JOIN mountedvol
            USING(vol_id)
        WHERE ext_id = ?
            AND available = ?
    },
    get_object_instances_by_vol_name => qq{
        SELECT
            storage_object.so_id,
            uri
        FROM storage_object
        JOIN instance
            USING (so_id)
        JOIN mountedvol
            USING(vol_id)
        WHERE ext_id = ?
            AND name = ?
            AND available = ?
    },
    get_storage_volume_by_name   => qq{
        SELECT
            vol_id,
            host,
            path,
            xattr,
            total - used as free
        FROM mountedvol
        WHERE
            used / total < ?
            AND name = ?
            AND available = ?
            AND allocate = ?
        ORDER BY free DESC
        LIMIT 1
    },
    get_replication_volume_for_ext_id   => qq{
        SELECT
            vol_id,
            host,
            path,
            xattr,
            total - used as free
        FROM (
-- This query works but is slow...
--             SELECT
--                  mountedvol.*
--             FROM mountedvol
--             WHERE
--                 vol_id NOT IN(
--                     SELECT vol_id
--                     FROM storage_object
--                     JOIN instance
--                         USING(so_id)
--                     WHERE ext_id = --
--                  )
                SELECT
                    m.*
                FROM mountedvol AS m
                LEFT JOIN instance AS i
                    ON m.vol_id = i.vol_id
                    AND i.so_id = (
                        SELECT so_id
                        FROM storage_object
                        WHERE ext_id = ?
                    )
                WHERE
                    i.vol_id IS NULL
        ) as Foo 
        WHERE
            used / total < ? 
            AND available = ? 
            AND allocate = ? 
        ORDER BY RAND()
        LIMIT 1
    },
    get_storage_volume          => qq{
        SELECT
            vol_id,
            host,
            path,
            xattr,
            total - used as free
        FROM mountedvol
        WHERE
            used / total < ?
            AND available = ?
            AND allocate = ?
        ORDER BY free DESC
        LIMIT 1
    },
    new_volume          => qq{
        INSERT INTO volume (name, host, path, allocate, available, xattr)
        VALUES (?, ?, ?, TRUE, TRUE, FALSE)
    },
    get_volume_by_name => qq{
        SELECT vol_id, name, host, path
        FROM volume
        WHERE name = ?
    },
    get_volumes => qq{
        SELECT
            v.vol_id,
            v.name,
            v.host,
            v.path,
            v.allocate,
            v.available,
            v.xattr,
            mountedvol.vol_id IS NOT NULL as mounted
        FROM volume AS v
        LEFT JOIN mountedvol
            USING(vol_id)
    },
    find_objects => qq{
        SELECT *
        FROM storage_object
        WHERE ext_id REGEXP ?
    },
    rename_object => qq{
        UPDATE storage_object
        SET ext_id = ?
        WHERE ext_id = ?
    },
    find_objects_with_unavailable_instances => qq{
        SELECT
            storage_object.so_id,
            ext_id,
            count(ins_id) as instances,
            volume.name as volume_name,
            volume.host as volume_host,
            count(mymountedvol.vol_id) as available_instances,
            count(mymountedvol.vol_id) > 0 as recoverable,
            storage_object_xattr.value as copies
        FROM storage_object
        JOIN instance
            USING(so_id)
        JOIN volume
            USING(vol_id)
        LEFT JOIN storage_object_xattr
            ON storage_object.so_id = storage_object_xattr.so_id
        JOIN mymountedvol
            USING(vol_id)
        WHERE mymountedvol.available = 1
--        WHERE storage_object_xattr.name = 'user.copies'
        GROUP BY so_id
        HAVING available_instances < instances OR instances < copies
    },
    find_objects_with_extra_instances => qq{
        SELECT
            storage_object.so_id,
            ext_id,
            count(ins_id) as instances,
            volume.name as volume_name,
            volume.host as volume_host,
            count(mymountedvol.vol_id) as available_instances,
            count(mymountedvol.vol_id) > 0 as recoverable,
            storage_object_xattr.value as copies
        FROM storage_object
        JOIN instance
            USING(so_id)
        JOIN volume
            USING(vol_id)
        LEFT JOIN storage_object_xattr
            ON storage_object.so_id = storage_object_xattr.so_id
        JOIN mymountedvol
            USING(vol_id)
        WHERE
            mymountedvol.available = 1
            AND storage_object_xattr.name = 'user.copies'
        GROUP BY so_id
        HAVING available_instances > copies
    },
    get_mounted_volumes => qq{
        SELECT * FROM mountedvol ORDER BY host, name
    },
);

{
    my @schema;

    local $/ = '###';

    foreach my $statement (<DATA>) {
        last unless ( $statement =~ /\S+/ );
        $statement =~ s/###//g;
        push @schema, $statement;
    }

    $sql{get_db_schema} = \@schema;
}

{
    my @clear = split /;/, <<END;
SET FOREIGN_KEY_CHECKS=0;
DROP TABLE IF EXISTS storage_object;
DROP TABLE IF EXISTS storage_object_attr;
DROP TABLE IF EXISTS storage_object_xattr;
DROP TABLE IF EXISTS instance;
DROP TABLE IF EXISTS lock_record;
DROP TABLE IF EXISTS volume;
DROP TABLE IF EXISTS mount;
DROP TABLE IF EXISTS log;
DROP TABLE IF EXISTS mountedvol;
DROP PROCEDURE IF EXISTS getmountedvol;
SET FOREIGN_KEY_CHECKS=1
END
    $sql{get_db_clear} = \@clear;
}

__PACKAGE__->mk_ro_accessors( keys %sql );

sub new {
    my $class = shift;

    my $self = \%sql;

    bless $self, $class || ref $class;

    return $self;
}

1;

__DATA__
CREATE TABLE storage_object (
    so_id BIGINT NOT NULL AUTO_INCREMENT,
    ext_id VARCHAR(255) NOT NULL UNIQUE,
    type enum('REG_FILE'),
    PRIMARY KEY(so_id),
    KEY(ext_id(64)),
    KEY(type)
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE storage_object_attr (
    so_id BIGINT NOT NULL AUTO_INCREMENT,
    FOREIGN KEY(so_id) REFERENCES storage_object(so_id) ON DELETE CASCADE,
    read_lock TINYINT DEFAULT 0 NOT NULL,
    write_lock ENUM( 'write' ),
    epoch TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    mtime TIMESTAMP,
    PRIMARY KEY(so_id)
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE storage_object_xattr (
    so_id BIGINT NOT NULL AUTO_INCREMENT,
    FOREIGN KEY(so_id) REFERENCES storage_object(so_id) ON DELETE CASCADE,
    name VARCHAR(255),
    value BLOB,
    PRIMARY KEY(so_id, name),
    KEY(so_id),
    KEY(name(64))
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE lock_record (
    so_id BIGINT NOT NULL,
    FOREIGN KEY(so_id) REFERENCES storage_object(so_id),
    type ENUM( 'read', 'write' ) NOT NULL,
    epoch TIMESTAMP,
    KEY(so_ID)
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE volume (
    vol_id INT NOT NULL AUTO_INCREMENT,
    name VARCHAR(255) UNIQUE NOT NULL,
    host VARCHAR(255) NOT NULL,
    path VARCHAR(255) NOT NULL,
    allocate BOOLEAN DEFAULT FALSE,
    available BOOLEAN DEFAULT FALSE,
    xattr BOOLEAN DEFAULT FALSE,
    PRIMARY KEY(vol_id),
    KEY(name(16)),
    KEY(host(16)),
    KEY(path(255)),
    KEY(allocate),
    KEY(available)
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE instance (
    ins_id BIGINT NOT NULL AUTO_INCREMENT,
    so_id BIGINT NOT NULL,
    FOREIGN KEY(so_id) REFERENCES storage_object(so_id),
    vol_id INT NOT NULL,
    FOREIGN KEY(vol_id) REFERENCES volume(vol_id),
    uri VARCHAR(255) NOT NULL UNIQUE,
    epoch TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    mtime TIMESTAMP,
    PRIMARY KEY(ins_id),
    KEY(so_id),
    KEY(vol_id),
    KEY(uri(64))
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE mount (
    mountpoint VARCHAR(255) NOT NULL,
    total BIGINT NOT NULL,
    used BIGINT NOT NULL,
    PRIMARY KEY(mountpoint)
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE log (
    timestamp TIMESTAMP,
    hostname VARCHAR(255),
    level VARCHAR(255),
    sub VARCHAR(255),
    message VARCHAR(2048) NOT NULL,
    PRIMARY KEY(timestamp)
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE TABLE mountedvol(
    mountpoint VARCHAR(255) NOT NULL,
    FOREIGN KEY(mountpoint) REFERENCES mount(mountpoint) ON DELETE CASCADE,
    total BIGINT NOT NULL,
    used BIGINT NOT NULL,
    vol_id INT NOT NULL,
    FOREIGN KEY(vol_id) REFERENCES volume(vol_id) ON DELETE CASCADE,
    name VARCHAR(255) NOT NULL,
    host VARCHAR(255) NOT NULL,
    path VARCHAR(255) NOT NULL,
    FOREIGN KEY(path) REFERENCES volume(path) ON DELETE CASCADE,
    allocate BOOLEAN DEFAULT FALSE,
    available BOOLEAN DEFAULT FALSE,
    xattr BOOLEAN DEFAULT FALSE,
    KEY(vol_id),
    KEY(allocate),
    KEY(available)
) ENGINE=innodb DEFAULT CHARSET=latin1;

###

CREATE PROCEDURE getmountedvol() DETERMINISTIC
BEGIN
    DECLARE done BOOLEAN DEFAULT FALSE;
    DECLARE vol_idvar INT;
    DECLARE namevar VARCHAR(255);
    DECLARE hostvar VARCHAR(255);
    DECLARE pathvar VARCHAR(255);
    DECLARE allocatevar BOOLEAN;
    DECLARE availablevar BOOLEAN;
    DECLARE xattrvar BOOLEAN;
    DECLARE trans_level VARCHAR(255);
    DECLARE key_checks BOOLEAN;
    DECLARE cur1 CURSOR FOR SELECT vol_id, name, host, path, allocate, available, xattr FROM myvolume;
    DECLARE CONTINUE HANDLER FOR SQLSTATE '02000' SET done = TRUE;

    -- store the okey checking state
    SELECT @@FOREIGN_KEY_CHECKS INTO key_checks; 
    -- disable foregin check checks to prevent deadlocks on the mountedvol table
    SET FOREIGN_KEY_CHECKS=0;

    -- make sure the temp table does not already exist... this can happy if the
    -- stored proc fails for some reason
    DROP TABLE IF EXISTS myvolume;
    CREATE TEMPORARY TABLE myvolume LIKE volume;
    INSERT INTO myvolume SELECT * FROM volume;

    -- store the current transaction level
--    SELECT @@session.tx_isolation INTO trans_level; 
    -- set trans level to repeatable-read so the volume table does not change
    -- out from under our cursor
--    SET @@session.tx_isolation = 'REPEATABLE-READ';

    -- iterate over the volume table finding the coresponding entry in the
    -- mount table and inserting union of the volume & mount row into the
    -- mountedvol table
    OPEN cur1;

    DELETE FROM mountedvol;

    myloop: LOOP
        FETCH cur1 INTO vol_idvar, namevar, hostvar, pathvar, allocatevar, availablevar, xattrvar;
        IF `done` THEN LEAVE myloop; END IF;
        INSERT INTO mountedvol
            SELECT mountpoint, total, used, vol_idvar, namevar, hostvar, pathvar, allocatevar, availablevar, xattrvar
            FROM
                (SELECT *, INSTR(pathvar, mountpoint) = 1 as substring
                FROM mount
                HAVING substring = 1
                ORDER BY substring DESC, LENGTH(mountpoint) DESC
                LIMIT 1) as bar;
    END LOOP myloop;
    
    CLOSE cur1;

    -- restore the original transaction level
--    SET @@session.tx_isolation = trans_level;

    -- restore the original key checking state
    SET @@FOREIGN_KEY_CHECKS = key_checks; 

    DROP TABLE IF EXISTS myvolume;

    SET FOREIGN_KEY_CHECKS=1;

    COMMIT;
END 
