/*********************************************************************************
 *
 *  The storage manager interface
 *  Copyright (c) 2002, 2003 by Leonidas Fegaras, the University of Texas at
 *     Arlington. All rights reserved.
 *  Programmer: Leonidas Fegaras
 *  Date: 1/10/02
 *
 ********************************************************************************/


#ifndef __ldm_sm_h__
#define __ldm_sm_h__

#include "rpc.h"
#include "gc_cpp.h"
#include "basic.h"


#ifdef LDB_SERVER
  #include "sm_vas.h"
  #define SM_DO(rc) { if (rc != RCOK) sm_error(__FILE__,__LINE__,rc); }
  extern lvid_t current_volume;
  extern serial_t root_catalog;
  extern ss_m* ssm;
  #define RPC_CLASS(name,arg) class name { public: arg;
#else  // LDB_CLIENT
  // clients should never use SHORE code
  #include <stream.h>
  #include <string.h>
  #define RPC_CLASS(name,arg) class name: public gc { public: int ref;
#endif


void copy_n_bytes ( char* dest, const char* src, int n );

void ldb_abort ( const char* message );

void sm_error ( const char* file, int line, const char* msg );

#define SM_ERROR(msg) sm_error(__FILE__,__LINE__,msg)


#ifdef LDB_SERVER
class oid_t: public serial_t {
public:
  oid_t ( serial_t s ) : serial_t(s) {};
  oid_t () : serial_t() {};
  short compare ( oid_t x ) const;
#else // LDB_CLIENT
class oid_t {
public:
  int oid;   // for 64 bit OIDs, use long
public:
  oid_t () : oid(0) {};
  oid_t ( int x ) : oid(x) {};
  short compare ( oid_t x ) const { return (oid==x.oid) ? 0 : (oid>x.oid) ? 1 : -1; };
  inline bool operator== ( oid_t x ) const { return compare(x.oid)==0; };
  inline bool operator<= ( oid_t x ) const { return compare(x.oid)<=0; };
#endif
  friend rpc_stream& operator>> ( rpc_stream& o, oid_t &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const oid_t &s );
  friend ostream& operator<< ( ostream& o, const oid_t &x );
};


class sm_string;
class sm_sequence;
class sm_ref;


class raw_string {
public:
  size_t length;
  int    offset;
public:
  raw_string () : length(0), offset(0) {};
  raw_string ( size_t length, int offset ) : length(length), offset(offset) {};
  raw_string ( const sm_string &s );
  raw_string& operator= ( const sm_string &s );
  friend ostream& operator<< ( ostream& o, const raw_string &x );
  friend rpc_stream& operator>> ( rpc_stream& o, raw_string &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const raw_string &s );
};


class raw_sequence {
public:
  size_t element_size;
  size_t cardinality;
  int    offset;
public:
  raw_sequence () : element_size(0) {};
  raw_sequence& operator= ( const sm_sequence &s );
  size_t size () const { return element_size; };
  friend ostream& operator<< ( ostream& o, const raw_sequence &x );
  friend rpc_stream& operator>> ( rpc_stream& o, raw_sequence &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const raw_sequence &s );
};


class raw_ref {
friend class sm_ref;
private:
  oid_t ref;
public:
  raw_ref () {};
  raw_ref ( oid_t o ) : ref(o) {};
  raw_ref ( sm_ref &x );
  raw_ref& operator= ( const sm_ref &s );
  oid_t oid () const { return ref; };
  friend ostream& operator<< ( ostream& o, const raw_ref &x );
};


/********************************************************************************/


class object_cache: public gc {
private:
  static const int memo_size = 956;
  typedef struct element {
     int                 key;
     sm_string*          info;
     struct element*     next;
  } element;
  element*      memo[memo_size];
public:
  object_cache () { clear(); };
  void clear ();
  bool find ( int key, sm_string &info );
  void remove ( int key );
  void store ( int key, const sm_string info );
};


class pinned_object: public gc {
  friend class sm_ref;
private:
  enum { TRANSIENT, PINNED } tag;
  union {
     struct {
        size_t length;
        char*  content;
     }                    value;  // transient object
#ifdef LDB_SERVER
     pin_i*    po;                // pinned persistent object
#endif
  } data;
#ifndef LDB_SERVER
  int  ref;
#endif
public:
  pinned_object ();
  pinned_object ( oid_t oid );
  pinned_object ( size_t size );
  pinned_object ( size_t size, const char* content );
#ifdef LDB_SERVER
  pinned_object ( pin_i* p );
#endif
  bool pinnedp () { return tag == PINNED; };
  char** content () { return &data.value.content; };
  int body_size () const;
  oid_t oid ();
  void repin ();
  const char* get ( int offset, size_t size );
  void set ( int offset, size_t size, const char* data );
  int replace ( int offset, size_t size, const char* data, size_t data_size );
  int append ( const char* data, size_t data_size );
  int append ( int offset, size_t size, const char* data, size_t data_size );
  void remove ( int offset, size_t data_size, int remaining_size );
  void print ();
  sm_string pinned_get ( int offset, size_t size );
  void pinned_set ( int offset, size_t size, sm_string data );
  int pinned_replace ( int offset, size_t size, sm_string data, size_t data_size );
  int pinned_append ( sm_string data, size_t data_size );
  int pinned_append ( int offset, size_t size, sm_string data, size_t data_size );
  friend sm_string get_pinned_data ( int offset, size_t size );
  friend sm_string object_deref ( oid_t ref, size_t size );
  friend rpc_stream& operator>> ( rpc_stream& o, pinned_object &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const pinned_object &s );
};


void set_current_object ( const size_t offset, const size_t size, const char* data );

extern pinned_object* current_object;


typedef enum { TRANSIENT = 0, PINNED = -1, RAW = -2 } sm_storage_type;


class sm_string: public gc {
private:
  size_t            length;   // total number of bytes in string
                              // or the offset of the string header in pinned object
  sm_storage_type   tag;
  union {
     char*          content;  // TRANSIENT: in-memory string
     pinned_object* pinned;   // PINNED: string header in a pinned persistent object
  } data;
  size_t body_offset () const;
public:
  sm_string ();
  sm_string ( const char* c );
  sm_string ( const char* c, int n );
  sm_string ( pinned_object* po, size_t header_offset );
  sm_string ( const raw_string &x );    // the only way to access a raw_string is to coerce it
  sm_string& operator= ( const sm_string &source );
  sm_string& operator= ( const raw_string &source );
  const sm_string operator+ ( const sm_string source ) const;
  const sm_string operator+ ( const char* source ) const;
  const char* content () const;
  const char* scontent () const;
  size_t size () const;
  const sm_string substring ( int start, size_t size ) const;
  const char operator[] ( int n ) const;
  short strcmp ( const sm_string source ) const;
  bool valid () const { return tag <= 0 && tag >= -2; };
  bool operator == ( const sm_string s ) const { return strcmp(s)==0; };
  bool operator != ( const sm_string s ) const { return strcmp(s)!=0; };
  bool operator <= ( const sm_string s ) const { return strcmp(s)<=0; };
  bool operator >= ( const sm_string s ) const { return strcmp(s)>=0; };
  bool operator < ( const sm_string s ) const { return strcmp(s)<0; };
  bool operator > ( const sm_string s ) const { return strcmp(s)>0; };
  friend ostream& operator<< ( ostream& o, const sm_string &s );
  friend rpc_stream& operator>> ( rpc_stream& o, sm_string &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const sm_string &s );
};


class sm_sequence: public gc {
public:
  size_t    element_size;     // size of each sequence element
                              // or the offset of the string header in pinned object
  size_t    seq_cardinality;  // number of elements in the sequence
  sm_storage_type   tag;
  union {
     char*          sequence; // TRANSIENT: in-memory sequence
     pinned_object* pinned;   // PINNED: string header in a pinned persistent object
  } data;
  size_t body_offset () const;
public:
  sm_sequence ( size_t element_size );
  sm_sequence ();
  sm_sequence ( pinned_object* po, size_t header_offset );
  sm_sequence ( const raw_sequence &x );    // the only way to access a raw_sequence is to coerce it
  int cardinality () const { return seq_cardinality; };
  size_t esize () const;
  const char* access ( int n ) const;
  const char* sequence () const;
  sm_sequence subsequence ( int n, size_t size ) const;
  sm_sequence& operator= ( const sm_sequence &source );
  bool valid () const { return tag <= 0 && tag >= -2; };
  bool member ( const char* data ) const;
  bool member ( const char* data, bool* (*eq)(const char*,const char*) ) const;
  int find ( const char* data ) const;
  int find ( const char* data, bool* (*eq)(const char*,const char*) ) const;
  void set ( int n, const char* data );
  void append ( const char* data );
  void append ( const char* data, int n );
  void insert ( const char* data );
  void insert ( const char* data, bool* (*eq)(const char*,const char*) );
  void remove ( int n );
  void remove ( const char* data ) { remove(find(data)); };
  void remove ( const char* data, bool* (*eq)(const char*,const char*) ) { remove(find(data,eq)); };
  friend ostream& operator<< ( ostream& o, const sm_sequence &x );
};


template<class T>
class Sequence: public sm_sequence {
public:
  Sequence () : sm_sequence(sizeof(T)) {};
  Sequence ( const raw_sequence &r ) : sm_sequence(r) {};
  template<class S>
     Sequence ( const Sequence<S> &source );
  T operator[] ( int n ) { return *(T*) access(n); };
  Sequence subsequence ( int n, size_t size ) const { return (Sequence) sm_sequence::subsequence(n,size); };
  Sequence& operator= ( const sm_sequence &source ) { sm_sequence::operator=(source); return *this; };
  template<class S>
     Sequence& operator= ( const Sequence<S> &source );
  bool member ( const T data ) const { return sm_sequence::member((char*) &data); };
  bool member ( const T data, bool(eq)(const T,const T) ) const;
  int find ( const T data ) const { return sm_sequence::find((char*) &data); };
  int find ( const T data, bool(eq)(const T,const T) ) const;
  void set ( int n, T data ) { sm_sequence::set(n,(char*) &data); };
  void append ( T data ) { sm_sequence::append((char*) &data); };
  void append ( T data[], int n ) { sm_sequence::append((char*) &data,n); };
  void insert ( T data ) { sm_sequence::insert((char*) &data); };
  void insert ( T data, bool(eq)(const T,const T) ) { if (!member(data,eq)) append(data); };
  void remove ( int n ) { sm_sequence::remove(n); };
  void remove ( T data ) { remove(find(data)); };
  void remove ( T data, bool(eq)(const T,const T) ) { remove(find(data,eq)); };
};


template<class T>
bool Sequence<T>::member ( const T c, bool(eq)(const T,const T) ) const {
  for(int i=0; i<cardinality(); i++)
     if (eq(c,access(i)))
        return true;
  return false;
};


template<class T>
int Sequence<T>::find ( const T c, bool(eq)(const T,const T) ) const {
  for(int i=0; i<cardinality(); i++)
     if (eq(c,access(i)))
        return i;
  return -1;
};


template<class T>
template<class S>
Sequence<T>::Sequence ( const Sequence<S> &source ) : sm_sequence(sizeof(T)) {
  size_t size = sizeof(T)*source.seq_cardinality;
  seq_cardinality = source.seq_cardinality;
  data.sequence = new (GC) char[size];
  for(int i=0; i<seq_cardinality; i++)
     (*(T*) access(i)) = (*(S*) source.access(i));
};


template<class T>
template<class S>
Sequence<T>& Sequence<T>::operator= ( const Sequence<S> &source ) {
  size_t size = sizeof(T)*source.seq_cardinality;
  seq_cardinality = source.seq_cardinality;
  if (tag == TRANSIENT) {
     element_size = sizeof(T);
     data.sequence = new (GC) char[size];
  } else {
     raw_sequence nh;
     nh.element_size = sizeof(T);
     nh.cardinality = source.seq_cardinality;
     nh.offset = data.pinned->append(new (GC) char[size],size);
     data.pinned->set(element_size,sizeof(raw_sequence),(char*) &nh);
     current_object = data.pinned;
  };
  for(int i=0; i<seq_cardinality; i++)
     (*(T*) access(i)) = (*(S*) source.access(i));
  return *this;
};


template<class T>
class Bag: public Sequence<T> {
public:
  Bag () : Sequence<T>() {};
  Bag subsequence ( int n, size_t size ) { return (Bag) Sequence<T>::subsequence(n,size); };
};


template<class T>
class Set: public Sequence<T> {
public:
  Set () : Sequence<T>() {};
  Set subsequence ( int n, size_t size ) { return (Set) Sequence<T>::subsequence(n,size); };
};


template<class T,int N>
class Vector: public Sequence<T> {
public:
  Vector () : Sequence<T>() { append(new T[N],N); };
};


template<class T,int N,int M>
class Array: public Vector<T,N*M> {
public:
  Array () : Vector<T,N*M>() {};
  T access ( int i, int j ) { return *(T*) sm_sequence::access(i*N+j); };
  void set ( int i, int j, const T data ) { sm_sequence::set(i*N+j,(char*) &data); };
};


class sm_ref: public gc {
  friend class raw_ref;
  union {
     oid_t*         oid;      // RAW: the oid of a newly created object
     char*          pointer;  // TRANSIENT: in memory ref
     pinned_object* pinned;   // PINNED: string header in a pinned persistent object
  } data;
  sm_storage_type   tag;
  size_t            header;   // the offset of the ref header in pinned object
public:
  sm_ref ();                                        // new empty transient
  sm_ref ( const char* content );                   // new transient
  sm_ref ( size_t size );                           // new uninitialized transient
  sm_ref ( pinned_object* po, int header_offset );  // new pinned object
  sm_ref ( const oid_t x );                         // new raw oid
  sm_ref ( const raw_ref &s );                      // the only way to access a raw_ref is to coerce it
  const oid_t oid () const;
  const char* deref ( size_t size ) const;
  bool valid () const { return !((tag == TRANSIENT) && (data.pointer == NULL)); };
  size_t size () const;
  friend rpc_stream& operator>> ( rpc_stream& o, sm_ref &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const sm_ref &s );
  friend ostream& operator<< ( ostream& o, const sm_ref &x );
};


template<class T>
class Ref: public sm_ref {
public:
  Ref () : sm_ref(new (GC) char[sizeof(T)]) {};
  Ref ( const T data ) : sm_ref((char*) &data) {};
  Ref ( const sm_ref &s ) : sm_ref(s) {};
  Ref ( const raw_ref &s ) : sm_ref(s) {};
  T* operator-> () { return (T*) deref(sizeof(T)); };
  T& operator* () { return *(T*) deref(sizeof(T)); };
};


RPC_CLASS(sm_file,oid_t file_oid)
public:
  sm_file () {};
  sm_file ( oid_t fid );
  sm_file ( bool temp );
  oid_t oid () const;
  sm_ref append ( size_t size );
  sm_ref append ( sm_string data );
  sm_ref append ( const char* data, size_t size ) { return append(sm_string(data,size)); };
  void remove ( sm_ref object );
  void delete_file ();
  friend rpc_stream& operator>> ( rpc_stream& o, sm_file &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const sm_file &s );
  friend ostream& operator<< ( ostream& o, const sm_file &x );
};


template<class T>
class Pool: public sm_file {
public:
  Pool () : sm_file() {};
  Pool ( const sm_file &f ) : sm_file(f.oid()) {};
  Pool ( bool temp ) : sm_file(temp) {};
  Ref<T> new_record ( const T data ) { return (Ref<T>) sm_file::append((const char*) &data,sizeof(T)); };
  Ref<T> new_record () { return sm_file::append((const char*) new (GC) T(),sizeof(T)); };
  void remove ( Ref<T> object ) { sm_file::remove((Ref<T>) object); };
};


RPC_CLASS(sm_file_scan,scan_file_i* oid)
public:
  sm_file_scan () {};
  sm_file_scan ( sm_file file );
  void close ();
  bool next ( sm_ref &object );
  bool eof ();
};


template<class T>
class PoolScan: public sm_file_scan {
public:
  PoolScan ( Pool<T> file ) : sm_file_scan(file) {};
  bool next ( Ref<T> &object ) { return sm_file_scan::next(object); };
};


RPC_CLASS(sm_file_append,append_file_i* oid)
public:
  sm_file_append () {};
  sm_file_append ( sm_file file );
  void close ();
  void append ( sm_string data );
  void append ( const char* data, size_t size ) { append(sm_string(data,size)); };
  bool eof ();
};


RPC_CLASS(sm_index,oid_t index_oid)
public:
  bool unique_index;
  sm_index () {};
  sm_index ( oid_t iid, bool uniquep );
  sm_index ( bool temp, bool uniquep, sm_string kdesc );
  oid_t oid () const;
  void add ( sm_string key, raw_ref data );
  void remove ( sm_string key, raw_ref data );
  bool member ( sm_string key );
  bool find ( sm_string key, raw_ref &data );
  void delete_index ();
  friend rpc_stream& operator>> ( rpc_stream& o, sm_index &s );
  friend rpc_stream& operator<< ( rpc_stream& o, const sm_index &s );
  friend ostream& operator<< ( ostream& o, const sm_index &x );
};


RPC_CLASS(sm_index_scan,scan_index_i* oid)
public:
  sm_index_scan () {};
  sm_index_scan ( sm_index index, sm_string low, sm_string high );
  void close ();
  bool next ( sm_string &key, raw_ref &data );
  bool eof ();
};


RPC_CLASS(sm_sort_scan,sort_stream_i* oid)
public:
  sm_sort_scan () {};
  sm_sort_scan ( bool uniquep );
  void close ();
  void put ( sm_string key, sm_string data );
  bool next ( sm_string &key, sm_string &data );
  bool eof ();
};


RPC_CLASS(storage_manager,oid_t catalog)
  bool in_transaction;
public:
  storage_manager () { in_transaction = false; };
  storage_manager ( sm_string device_pathname, bool format_device, long size = 0 );
  void run ( void (*program) () );    // run a program
  void run ();                        // run client transactions
  void begin_transaction ();
  void commit_transaction ();
  void abort_transaction ();
  void add_declaration ( sm_string name, sm_string value );
  void remove_declaration ( sm_string name );
  bool find_declaration ( sm_string name, sm_string &value );
#ifndef LDB_SERVER
  friend storage_manager* current_storage_manager ();
#endif
};


class LDBobject {
public:
  int ref;
  LDBobject ( int id ) { ref=id; };
  LDBobject () { ref=-1; };
  int getID () { return ref; };
  void flush ();
  int getInt ();
  sm_string getString ();
  float getFloat ();
  int size ();
  sm_string toString ();
  sm_string getType ();
  LDBobject nth ( int index );
  void* getvalue ();
  Expr gettype ();
  friend rpc_stream& operator>> ( rpc_stream& o, LDBobject &s );
};


void open ( sm_string username, sm_string password, sm_string schema );

void close ();

LDBobject query ( sm_string OQLquery );

void statement ( sm_string stmt );

void update ( sm_string dest, sm_string OQLquery );

void insert ( sm_string dest, sm_string OQLquery );

void remove ( sm_string dest, sm_string OQLquery );

void create_index ( sm_string index_name, sm_string class_name, sm_string attribute );

void remove_index ( sm_string index_name );

void change_password ( sm_string oldp, sm_string newp );

sm_string tcl_schema ();

void begin_transaction ();

void commit_transaction ();

void abort_transaction ();


extern storage_manager* ldb_sm;

extern int ldb_sm_location;


void push_current_object ();
void pop_current_object ();
void set_current_object ();
void set_current_object ( oid_t ref );

void process_definition ( const char* definition );

void initialize_LDB ();


#define MAIN		 							\
* _xxxx;                                                                        \
int main_argc;									\
char** main_argv;								\
rc_t init_config_options ( const char* opt_file, option_group_t& options );     \
void sm_error ( const char* file, int line, rc_t rc );                          \
void initialize_database () {							\
  ldb_sm->begin_transaction();							\
  create_user(sm_string("admin"),sm_string("pass"));				\
  ldb_sm->commit_transaction();							\
};										\
void process_main ( int argc, char* argv[] );					\
void process_main () {								\
  process_main(main_argc,main_argv);						\
};										\
int main ( int argc, char* argv[] ) {						\
  main_argc = argc;								\
  main_argv = argv;								\
  option_t* volume_opt;								\
  option_t* volumesize_opt;							\
  const int option_level_cnt = 3;						\
  option_group_t options(option_level_cnt);					\
  SM_DO(options.add_option("sm_volume", "unix pathname",			\
			   "./database", "the database volume",			\
			   false, option_t::set_value_charstr,			\
			   volume_opt));					\
  SM_DO(options.add_option("sm_volumesize", "size in KBytes",			\
			   "1000", "the database volume size",			\
			   false, option_t::set_value_long,			\
			   volumesize_opt));					\
  SM_DO(options.add_class_level("ssm"));					\
  SM_DO(options.add_class_level("server"));					\
  SM_DO(options.add_class_level("ldb"));					\
  SM_DO(init_config_options(LDBconfig,options));                                \
  if (argc==2 && strcmp(argv[1],"-build")==0)					\
  {  cout << "Do you really want to format the entire database? ";	        \
     char answer;								\
     cin >> answer;								\
     if (answer != 'y' && answer != 'Y')					\
     {  cerr << "Please try again without the -build option\n";			\
        exit(1);                                                                \
     };                                                                         \
     ldb_sm = new storage_manager(sm_string(volume_opt->value()),true,		\
				  strtol(volumesize_opt->value(),0,0));		\
     ldb_sm->run(initialize_database);						\
  } else {									\
     ldb_sm = new storage_manager(sm_string(volume_opt->value()),false);	\
     ldb_sm_location = store_object(ldb_sm);                                    \
     try {                                                                      \
        ldb_sm->run(process_main);						\
     } catch (odmg_error_class(m)) {                                            \
        cerr << m.message;                                                      \
     };                                                                         \
  };										\
};										\
void process_main


#endif