(use ugarit-backend)
(use sql-de-lite)
(use matchable)
(use miscmacros)
(use srfi-9)

;; Clustered storage backend, as per https://www.kitten-technologies.co.uk/project/ugarit/tktview?name=f1f2ce8cdc

(define cache-sql-schema
  (list
   "CREATE TABLE shards (shard_id INTEGER PRIMARY KEY NOT NULL, storage TEXT NOT NULL, trust INTEGER NOT NULL, write_weight INTEGER NOT NULL, read_priority INTEGER NOT NULL, read_weight INTEGER NOT NULL);"
   "CREATE TABLE blocks (key_id INTEGER NOT NULL PRIMARY KEY, \"key\" TEXT NOT NULL, UNIQUE INDEX (\"key\");"
   "CREATE TABLE replicas (key_id INTEGER NOT NULL, shard_id INTEGER NOT NULL, PRIMARY KEY (key_id, shard_id)) WITHOUT ROWID;"))

(define-record-type shard
  (make-shard id storage-desc storage trust write-weight read-priority read-weight)
  shard?
  (id shard-id)
  (storage-desc shard-storage-desc (setter shard-storage-desc))
  (storage shard-storage (setter shard-storage))
  (trust shard-trust (setter shard-trust))
  (write-weight shard-write-weight (setter shard-write-weight))
  (read-priority shard-read-priority (setter shard-read-priority))
  (read-weight shard-read-weight (setter shard-read-weight)))

(define (backend-cache cachepath be)
   (define *db* (open-database cachepath))
   (change-file-mode cachepath (bitwise-ior perm/irusr perm/iwusr))
   (set-busy-handler! *db* (busy-timeout 100000))
   (when (null? (schema *db*))
         (for-each (lambda (statement)
                     (exec (sql *db* statement)))
                   cache-sql-schema))
   (exec (sql *db* "BEGIN;"))

   (define add-block-query (sql *db* "INSERT OR REPLACE INTO blocks (\"key\") VALUES (?);"))
   (define add-replica-query (sql *db* "INSERT OR REPLACE INTO replicas (key_id,shard_id) VALUES (?,?);"))
   (define replicas-get-query (sql *db* "SELECT shard_id FROM (SELECT * FROM blocks where \"key\" = ?)
                                           INNER JOIN replicas ON blocks.key_id = replicas.key_id;"))
   (define delete-block-query (sql *db* "DELETE FROM blocks WHERE \"key\" = ?;"))
   (define delete-block-replicas-query (sql *db* "DELETE FROM replicas WHERE key_id IN (SELECT key_id FROM blocks WHERE \"key\" = ?);"))

   ;; Load list of shards
   (define *shards*
     (alist->hash-table
      (map (lambda (r)
             (cons (first r)
                   (make-shard (first r)
                               (second r)
                               #f
                               (third r)
                               (fourth r)
                               (fifth r)
                               (sixth r))))
           (query fetch-all (sql *db* "SELECT shard_id, storage, trust, write_weight, read_priority, read_weight FROM shards;")))))

   (define (connect! shard)
     ;; FIXME: Maybe retry once or twice?
     (set! (shard-storage shard)
           (import-storage (shard-storage-desc shard)))

     (when ((storage-writable? (shard-storage shard)))
           ((storage-set-tag! (shard-storage shard))
            "#ugarit-non-vault-storage"
            "(cluster-shard \"This is a cluster shard, and will only contain a fraction of the contents of the actual vault.\")")))

   (define (disconnect! shard)
     (when (shard-storage shard)
           ((storage-close! (shard-storage shard)))
           (set! (shard-storage shard) #f)))
   
   ;; Apply thunk to a storage pointing to that shard.
   ;; Reconnects a few times if anything fails, before giving up.
   (define (call-with-shard shard thunk)
     (let loop ((attempt 1))
       (unless (shard-storage shard)
               (connect! shard))
       (handle-exceptions
        exn
        (begin
          (disconnect! shard)
          (if (< attempt *attempts*)
              (loop (+ attempt 1))
              (raise exn)))
        (thunk shard))))

   (define commit-interval 1000)
   (define *updates-since-last-commit* 0)
   (define (flush!)
     (when (> *updates-since-last-commit* 0)
      (inc! *flushes*)
      (exec (sql *db* "COMMIT;"))
      (exec (sql *db* "BEGIN;"))
      (set! *updates-since-last-commit* 0)))
   (define (maybe-flush!)
     (inc! *updates-since-last-commit*)
     (when (> *updates-since-last-commit* commit-interval)
           ((storage-flush! be))
           (flush!)))

   (define (cache-set! key type)
      (when type
            (begin
              (exec cache-set-query key (symbol->string type))
              (maybe-flush!)))
      type)

   (define (cache-get key)
      (let ((result
             (query fetch cache-get-query key)))
        (if (pair? result)
            (string->symbol (car result))
            #f)))

   (define (cache-delete! key)
     (exec cache-delete-query key)
     (maybe-flush!))

   (define (cache-clear!)
     (exec (sql *db* "DELETE FROM cache")))

   (make-storage
      (storage-max-block-size be)
      (storage-writable? be)
      (storage-unlinkable? be)
      (lambda (key data type) ; put!
         (begin
            ((storage-put! be) key data type)
            (cache-set! key type)
	    (void)))
      (lambda ()                        ; flush!
        (begin
          ((storage-flush! be))
          (flush!)
          (void)))
      (lambda (key) ; exists?
        (let ((cached-result (cache-get key)))
          (if cached-result
              (begin
                (inc! *hits*)
                cached-result)
              (begin
                (inc! *misses*)
                (cache-set! key ((storage-exists? be) key))))))
      (lambda (key) ; get
         ((storage-get be) key))
      (lambda (key) ; link!
         ((storage-link! be) key))
      (lambda (key) ; unlink!
         (let ((result ((storage-unlink! be) key)))
            (if result
               (begin
                  (cache-delete! key)
                  result)
               result)))
      (lambda (tag key) ; set-tag!
        ((storage-set-tag! be) tag key)
        ((storage-flush! be))
        (flush!))
      (lambda (tag) ; tag
         ((storage-tag be) tag))
      (lambda () ; all-tags
         ((storage-all-tags be)))
      (lambda (tag) ; remove-tag!
         ((storage-remove-tag! be) tag)
         ((storage-flush! be))
         (flush!))
      (lambda (tag) ; lock-tag!
         (let ((result ((storage-lock-tag! be) tag)))
           ((storage-flush! be))
           (flush!)
           result))
      (lambda (tag) ; tag-locked?
         ((storage-tag-locked? be) tag))
      (lambda (tag) ; unlock-tag!
         ((storage-unlock-tag! be) tag)
          ((storage-flush! be))
          (flush!))
      (lambda (command) ; admin!
        (match command
               (('info)
                (list (cons 'backend "cache")
                      (cons 'block-size (storage-max-block-size be))
                      (cons 'writable? (storage-writable? be))
                      (cons 'unlinkable? (storage-unlinkable? be))
                      (cons 'cache-file cachepath)
                      (cons 'commit-interval commit-interval)))
               (('help)
                (list (cons 'info "Return information about the archive")
                      (cons 'help "List available admin commands")
                      (cons 'stats "Examine the cache and report back statistics")
                      (cons 'clear! "Clear the cache")))
               (('stats)
                (list (cons 'entries (car (query fetch (sql *db* "SELECT COUNT(*) FROM cache"))))))
               (('clear!)
                (cache-clear!)
                (flush!)
                (list (cons 'result "Done")))
               (else (error "Unknown admin command"))))
      (lambda () ; close!
        (begin
          ((backend-log!) 'info (sprintf "Cache hits: ~A misses: ~A flushes: ~A" *hits* *misses* *flushes*))
          ((storage-close! be))
          (exec (sql *db* "COMMIT;"))
          (close-database *db*)))))


(define backend
  (match (command-line-arguments)
	 ((cachepath backend)
	  (lambda () (backend-cache cachepath (import-storage backend))))

	 (else
          (export-storage-error! "Invalid arguments to backend-cache")
	  (printf "USAGE:\nbackend-cache <path-to-cache-file> \"<backend command line>\"\n")
	  #f)))

(if backend
    (export-storage! backend))

                      *global-rules*
                      #f ; Unset version
                      #f ; Unset configuration alist
                      #f)) ; Not changed yet
              (conf-tag
               (vault-tag vault "#ugarit-vault-configuration")))

         (let ((nvs-tag (vault-tag vault "#ugarit-non-vault-storage")))
          (when nvs-tag
                (signal (make-property-condition
                         'exn
                         'location 'open-vault
                         'message (sprintf 
                                   "The storage is not a vault: ~S" (tag-key nvs-tag))
                         'arguments (list (tag-type nvs-tag) (tag-key nvs-tag))))))
         
         (unless conf-tag
                 ;; Create default v1 tag
                 (receive (conf-key conf-reused?)
                          (store-sexpr! vault '(1) 'ugarit-vault-configuration '())

                          (vault-flush! vault)
                          (set! conf-tag (make-tag "#ugarit-vault-configuration" 'ugarit-vault-configuration conf-key))