(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)))

 (defun opt-no-setter-p (k)

   (let ((k (typecase k

              (string (string-downcase k))

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

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

     (member t

             (mapcar (lambda (x) (equal k x)) (list "parallelism" "enable-statistics")))))

 (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

     ;; initialize slots - remember, initargs doesn't refer to slot

     ;; names, they're opt names.

     (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)

           ;; note how we don't handle any special cases here - we can

           ;; always set an opt but sometimes we can't get it.

           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)

           unless (opt-no-setter-p k)

           do (pull-sap self k))

     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*

             unless (opt-no-setter-p k)

             do (pull-sap self k))

       (pull-sap* self))

   (rdb-opts-table self))

 (defun default-rdb-opts () 

   (make-rdb-opts :create-if-missing t :create-missing-column-families t

                  :parallelism (num-cpus)))

 (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

 (defclass rdb-iter (sequence)

   ((sap :initform nil :initarg :sap :type (or null alien) :accessor rdb-iter-sap)))

 (defmethod iter-valid-p ((self rdb-iter))

   (rocksdb-iter-valid (rdb-iter-sap self)))

 (defmethod iter-seek-to-first ((self rdb-iter))

   (rocksdb-iter-seek-to-first (rdb-iter-sap self))) 

 (defmethod iter-seek-to-last ((self rdb-iter))

   (rocksdb-iter-seek-to-last (rdb-iter-sap self)))

 (defmethod iter-seek-for-prev ((self rdb-iter) (key vector) &key)

   (rocksdb-iter-seek-for-prev (rdb-iter-sap self) key (length key)))

 (defmethod iter-seek ((self rdb-iter) (key simple-vector) &key)

   (rocksdb-iter-seek (rdb-iter-sap self) key (length key)))

 (defmethod iter-next ((self rdb-iter))

   (rocksdb-iter-next (rdb-iter-sap self)))

 (defmethod iter-prev ((self rdb-iter))

   (rocksdb-iter-prev (rdb-iter-sap self)))

 (defmethod iter-key ((self rdb-iter))

   (with-alien ((klen size-t))

     (let ((key (rocksdb-iter-key (rdb-iter-sap self) (addr klen))))

       (let ((k (make-array klen :element-type 'octet)))

         (clone-octets-from-alien key k klen)

         (values

          k

          klen)))))

 (defmethod iter-val ((self rdb-iter))

   (with-alien ((vlen size-t))     

     (let ((val (rocksdb-iter-value (rdb-iter-sap self) (addr vlen))))

       (let ((v (make-array vlen :element-type 'octet)))

         (clone-octets-from-alien val v vlen)

         (values

          v

          vlen)))))

 (defmethod iter-kv ((self rdb-iter))

   (make-kv (iter-key self) (iter-val self)))

 (defmethod iter-timestamp ((self rdb-iter))

   (with-alien ((tslen size-t))

     (values

      (rocksdb-iter-timestamp (rdb-iter-sap self) (addr tslen))

      tslen)))

 ;;; column family

 (defstruct (rdb-cf (:constructor make-rdb-cf (name &key key-type val-type sap)))

   "RDB Column Family structure. Contains a name, key-type, val-type,

 and a system-area-pointer to the underlying rocksdb_cf_t handle.

 A NIL key-type or val-type indicates an unitialized value which defaults to

 'octet-vector. This is needed to distinguish the value 'octet-vector being

 supplied by the user from the default value."

   (name "" :type string)

   (key-type nil :type (or list symbol))

   (val-type nil :type (or list symbol))

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

 (defmethod close-cf ((self rdb-cf) &optional error)

   (if-let ((sap (rdb-cf-sap self)))

     (setf (rdb-cf-sap self) (rocksdb:rocksdb-column-family-handle-destroy sap))

     (when error (rdb-error "column family is already closed."))))

 ;;; 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 (vector 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 :open ~A" (length (rdb-cfs self)) (db-open-p self))))

 (defmethod db ((self rdb))

   (rdb-db self))

 (defmethod db-open-p ((self rdb))

   (when (db self) t))

 (defmethod db-closed-p ((self rdb))

   (unless (db self) t))

 (defun translate-cf-to-field (cf)

   (let ((vt (or (rdb-cf-val-type cf) 'octet-vector))

         (kt (unless (rdb-cf-val-type cf) (or (rdb-cf-key-type cf) 'octet-vector))))

     (make-field :name (rdb-cf-name cf)

                 :type (if kt

                           (cons kt vt)

                           vt))))

 (defmethod load-field ((self rdb-cf) (field field))

   (let ((type (field-type field)))

   (typecase type

     ;; note that this means you can't use LOAD-SCHEMA to reset an

     ;; rdb schema as you may expect.

     (null nil)

     (atom (setf (rdb-cf-val-type self) type))

     (list (setf (rdb-cf-key-type self) (car type)

                 (rdb-cf-val-type self)

                 (if (and (listp (cdr type))

                          (= 1 (length (cdr type))))

                     (cadr type)

                     (cdr type)))))

     self))

 (defmethod load-schema ((self rdb) (schema schema))

   "Load SCHEMA into rdb database object SELF. This will add any missing rdb-cfs

 and update existing key/value types for cfs with the same name. Existing cfs

 only get their their type slots updated on non-nil values."

   (loop for field across (fields schema)

         do (if-let ((cf (find-cf (field-name field) self)))

              (load-field cf field)

              (push-cf

               (load-field (make-rdb-cf (field-name field)) field)

               self)))

   self)

 (defmethod derive-schema ((self rdb))

   (apply 'make-schema

          (loop for cf across (rdb-cfs self)

                collect (translate-cf-to-field cf))))

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

   "Construct a new RDB instance from NAME.

 OPTS = rdb-opts