(in-package :rdb)

 ;;; rdb-opts

 (defvar *rdb-opts-lookup-table*

   (let ((table (make-hash-table :test #'equal)))

     (mapc (lambda (x) (setf (gethash (car x) table) (cdr x)))

           (loop for y across *rocksdb-options*

                 collect (cons y (format nil "~:@(rocksdb-options-set-~x~)" y))))

     table))

 (defun %set-rocksdb-option (opt key val)

   (funcall (rdb-opt-setter key) opt val))

 (defun %get-rocksdb-option (opt key)

   (if-let ((g (rdb-opt-getter key)))

     (funcall g opt)

     (warn 'opt-handler-missing :message key)))

 (defclass rdb-opts ()

   ((table :initarg :table :type hash-table :accessor rdb-opts-table)

    (sap :initarg :sap :type (or null alien) :accessor rdb-opts-sap)))

 (defmethod initialize-instance ((self rdb-opts) &rest initargs &key &allow-other-keys)

   (with-slots (sap table) self

     (unless (getf initargs :table) (setf table (make-hash-table :test #'equal)))

     (unless (getf initargs :sap) (setf sap (rocksdb-options-create)))

     (loop for (k v) on initargs by #'cddr while v

           do (let ((k (typecase k

                         (string (string-downcase k))

                         (symbol (string-downcase (symbol-name k)))

                         (t (string-downcase (format nil "~s" k))))))

                (set-opt self k v)))

     self))

 (defun make-rdb-opts (&rest values)

   (let ((opts (apply #'make-instance 'rdb-opts values)))

     (push-sap* opts)

     opts))

 (defmethod get-opt ((self rdb-opts) key)

   "Return the current value of KEY in SELF if found, else return nil."

   (gethash key (rdb-opts-table self)))

 (defmethod set-opt ((self rdb-opts) key val &key push)

   "Set the VAL of KEY in SELF with '(setf (gethash SELF KEY) VAL)'."

   (prog1

       (setf (gethash key (rdb-opts-table self)) val)

     (when push (push-sap self key))))

 (defmethod push-sap ((self rdb-opts) key)

   "Push KEY from slot :TABLE to the instance :SAP."

   (%set-rocksdb-option (rdb-opts-sap self) key (get-opt self key)))

 (defmethod push-sap* ((self rdb-opts))

   "Initialized the SAP slot with values from TABLE."

   (with-slots (table) self

     (loop for k in (hash-table-keys table)

           do (push-sap self k))))

 (defmethod pull-sap ((self rdb-opts) key)

   (setf (gethash key (rdb-opts-table self)) (%get-rocksdb-option (rdb-opts-sap self) key)))

 (defmethod pull-sap* ((self rdb-opts))

   (with-slots (table) self

     (loop for k in (hash-table-keys table)

     table))

 (defmethod backfill-opts ((self rdb-opts) &key full)

   "Backfill the TABLE slot with values from SAP.

 When FULL is non-nil, retrieve the full set of options available, not

 just the keys currently present in TABLE."

   (if full

       (loop for k across *rocksdb-options*

             do (pull-sap self k))

       (pull-sap* self))

   (rdb-opts-table self))

 (defun default-rdb-opts () 

 (defclass rdb-kv ()

   ((key :initarg :key :type octet-vector :accessor rdb-key)

    (val :initarg :val :type octet-vector :accessor rdb-val)))

 (defmethod make-kv (key val)

   (make-instance 'rdb-kv 

     :key (make-key key) 

     :val (make-val val)))

 (defvar *default-rdb-kv* (make-kv #() #()))

 ;;; iterator

 ;;; column family

 (defstruct (rdb-cf (:constructor make-rdb-cf (name &key #+nil kv sap)))

   "RDB Column Family structure. Contains a name, a cons of (rdb-key-type

 . rdb-val-type), and a system-area-pointer to the underlying

 rocksdb_cf_t handle."

   (name "" :type string)

   ;; (kv *default-rdb-kv* :type rdb-kv)

   (sap nil :type (or null alien)))

 ;;; rdb-stats

 (defstruct (rdb-stats (:constructor make-rdb-stats (&optional sap)))

   (sap nil :type (or null alien)))

 ;;; metadata

 (defstruct rdb-cf-metadata

   (name "default" :type string)

   (size 0 :type fixnum)

   (level-count 7 :type fixnum)

   (file-count 0 :type fixnum)

   (sap nil :type (or null alien)))

 (defmethod get-metadata ((self rdb-cf-metadata) &optional (level 0))

   (with-slots (sap) self

     (if (null sap)

         (warn 'metadata-missing :message "ignoring attempt to pull fields from null sap.")

         (make-rdb-level-metadata :sap (rocksdb-column-family-metadata-get-level-metadata sap level)))))

 (defmethod print-object ((self rdb-cf-metadata) stream)

   (print-unreadable-object (self stream :type t)

     (with-slots (name size level-count file-count) self

       (format stream "~A :size ~A :levels ~A :files ~A" name size level-count file-count))))

 (defmethod pull-sap* ((self rdb-cf-metadata))

   (with-slots (name size level-count file-count sap) self

     (if (null sap)

         (warn 'metadata-missing :message "ignoring attempt to pull fields from null sap.")

         (setf name (rocksdb-column-family-metadata-get-name sap)

               size (rocksdb-column-family-metadata-get-size sap)

               level-count (rocksdb-column-family-metadata-get-level-count sap)

               file-count (rocksdb-column-family-metadata-get-file-count sap)))

     self))

 (defstruct rdb-level-metadata

   (level 0 :type fixnum)

   (size 0 :type fixnum)

   (file-count 0 :type fixnum)

   (sap nil :type (or null alien)))

 (defmethod get-metadata ((self rdb-level-metadata) &optional (file 0))

   (with-slots (sap) self

     (if (null sap)

         (warn 'metadata-missing :message "ignoring attempt to pull fields from null sap.")

         (make-rdb-sst-file-metadata :sap (rocksdb-level-metadata-get-sst-file-metadata sap file)))))

 (defmethod print-object ((self rdb-level-metadata) stream)

   (print-unreadable-object (self stream :type t)

     (with-slots (level size file-count) self

       (format stream "~A :size ~A :files ~A" level size file-count))))

 (defmethod pull-sap* ((self rdb-level-metadata))

   (with-slots (level size file-count sap) self

     (if (null sap)

         (warn 'metadata-missing :message "ignoring attempt to pull fields from null sap.")

         (setf level (rocksdb-level-metadata-get-level sap)

               size (rocksdb-level-metadata-get-size sap)

               file-count (rocksdb-level-metadata-get-file-count sap)))

     self))

 ;; NOTE: we only store the sizes of largest and smallest key, not the

 ;; keys themselves. This may change in the future.

 (defstruct rdb-sst-file-metadata

   (relative-filename "" :type string)

   (directory "" :type string)

   (size 0 :type fixnum)

   (smallestkey 0 :type fixnum)

   (largestkey 0 :type fixnum)

   (sap nil :type (or null alien)))

 (defmethod print-object ((self rdb-sst-file-metadata) stream)

   (print-unreadable-object (self stream :type t)

     (with-slots (relative-filename directory size smallestkey largestkey) self

       (format stream "~A :dir ~A :size ~A :smallest ~A :largest ~A"

               relative-filename directory size smallestkey largestkey))))

 (defmethod pull-sap* ((self rdb-sst-file-metadata))

   (with-slots (relative-filename directory size smallestkey largestkey sap) self

     (if (null sap)

         (warn 'metadata-missing :message "ignoring attempt to pull fields from null sap.")

         (with-alien ((ssize size-t 0)

                      (lsize size-t 0))

           (rocksdb-sst-file-metadata-get-largestkey sap (addr lsize))

           (rocksdb-sst-file-metadata-get-smallestkey sap (addr ssize))

           (setf relative-filename (rocksdb-sst-file-metadata-get-relative-filename sap)

                 directory (rocksdb-sst-file-metadata-get-directory sap)

                 size (rocksdb-sst-file-metadata-get-size sap)

                 largestkey lsize

                 smallestkey ssize)))

     self))

 ;;; rdb

 (defstruct (rdb (:constructor make-rdb (name opts &optional cfs db)))

   (name "" :type string)

   (opts (default-rdb-opts) :type rdb-opts)

   (cfs (make-array 0 :element-type 'rdb-cf :adjustable t :fill-pointer 0) :type (array rdb-cf))

   (db nil :type (or null alien))

   (backup nil :type (or null alien))

   (snapshots #() :type (array alien)))

 (defvar *default-rdb-opts* (default-rdb-opts))

 (defmethod print-object ((self rdb) stream)

   (print-unreadable-object (self stream :type t :identity t)

     (format stream ":cfs ~A" (length (rdb-cfs self)))))

 (defun create-db (name &key opts cfs open)

   "Construct a new RDB instance from NAME.

 OPTS = rdb-opts

 CFS = (sequence rdb-cf)

 OPEN = boolean

 When OPEN is non-nil, the database and all column families are opened

 and internal sap slots are initialized."

   (when (probe-file name) (log:warn! "directory already exists: " name))

   (let* ((opts (or opts (default-rdb-opts)))

          (obj

            (make-rdb (string-right-trim '(#\/)

                                         (typecase name

                                           (pathname (namestring name))

                                           (string name)

                                           (t (error "invalid NAME: ~S" name))))

                      opts

                      (or (when cfs

                            (typecase cfs

                              (list (coerce cfs 'vector))

                              ((array rdb-cf) cfs)

                              (rdb-cf (vector cfs))

                              (t (log:warn! "invalid CF passed to create-db"))))

                          (make-array 0 :element-type 'rdb-cf :fill-pointer 0)))))

     (when open

       (open-db obj))

     obj))

 (defmethod push-cf ((cf rdb-cf) (db rdb))

   (vector-push cf (rdb-cfs db)))

 ;; TODO: fix

 (defmethod create-cf ((db rdb) (cf rdb-cf))

   (setf (rdb-cf-sap cf)

         (create-cf-raw (rdb-db db) (rdb-cf-name cf) (rdb-opts-sap (rdb-opts db)))))

 (defmacro unless-null-db (slots self &body body)

   `(with-slots (db ,@slots) ,self

      (unless (null db)

        ,@body)))

 (defmethod destroy-cf ((cf rdb-cf))

   (with-slots (sap) cf

     (unless (null sap)

       (setf sap (destroy-cf-raw sap)))))

 (defmethod set-opt ((self rdb) key val &key push)

   (with-slots (opts) self

     (set-opt opts key val :push push)))

 (defmethod get-opt ((self rdb) key)

   (with-slots (opts) self

     (get-opt opts key)))