# Copyright (c) 2004  Joshua Hoblitt
#
# $Id: SQL.pm,v 1.63 2008-08-21 02:26:58 jhoblitt 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,
            assigned_vol_id,
            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,
            assigned_vol_id
        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 = ?
    },
    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(mountedvol.vol_id) as available_instances,
            count(mountedvol.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 mountedvol
            USING(vol_id)
        WHERE mountedvol.available = 1
--        WHERE storage_object_xattr.name = 'user.copies'
        GROUP BY so_id
        HAVING available_instances < instances OR instances < copies
    },
    get_mounted_volumes => qq{
        SELECT * FROM mountedvol ORDER BY vol_id
    },
);

{
    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)
) 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,
    sha1sum CHAR(40) ASCII,
    assigned_vol_id INT,
    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 cur1 CURSOR FOR SELECT vol_id, name, host, path, allocate, available, xattr FROM volume;
    DECLARE CONTINUE HANDLER FOR SQLSTATE '02000' SET done = TRUE;

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

    COMMIT;
END 
