(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 length type) ; put!
        (begin
          ((storage-put! be) key data length 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 buffer) ; get
         ((storage-get be) key buffer))
      (lambda (key) ; link!
         ((storage-link! be) key))
      (lambda (key buffer) ; unlink!
         (let ((result ((storage-unlink! be) key buffer)))
            (if result
               (begin
                  (cache-delete! key)
                  result)
               result)))
      (lambda (tag key) ; set-tag!
        ((storage-set-tag! be) tag key)

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

   (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 length type) ; put!
         (begin
            ((storage-put! be) key data length 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 buffer) ; get
         ((storage-get be) key buffer))
      (lambda (key) ; link!
         ((storage-link! be) key))
      (lambda (key buffer) ; unlink!
         (let ((result ((storage-unlink! be) key buffer)))
            (if result
               (begin
                  (cache-delete! key)
                  result)
               result)))
      (lambda (tag key) ; set-tag!
        ((storage-set-tag! be) tag key)

 (define (backend-nullwrap be)
   (make-storage
      (storage-max-block-size be)
      (storage-writable? be)
      (storage-unlinkable? be)
      (lambda (key data length type) ; put!
         ((storage-put! be) key data length type))
      (lambda () ; flush!
        ((storage-flush! be)))
      (lambda (key) ; exists?
         ((storage-exists? be) key))
      (lambda (key buffer) ; get
         ((storage-get be) key buffer))
      (lambda (key) ; link!
         ((storage-link! be) key))
      (lambda (key buffer) ; unlink!
         ((storage-unlink! be) key buffer))
      (lambda (tag key) ; set-tag!
         ((storage-set-tag! be) tag key))
      (lambda (tag) ; tag
         ((storage-tag be) tag))
      (lambda () ; all-tags
         ((storage-all-tags be)))
      (lambda (tag) ; remove-tag!

         ((storage-close! be)))))

(define (backend-limit-block-size be max-block-size)
   (make-storage
      (min max-block-size (storage-max-block-size be))
      (storage-writable? be)
      (storage-unlinkable? be)
      (lambda (key data length type) ; put!
         ((storage-put! be) key data length type))
      (lambda () ; flush!
        ((storage-flush! be)))
      (lambda (key) ; exists?
         ((storage-exists? be) key))
      (lambda (key buffer) ; get
         ((storage-get be) key buffer))
      (lambda (key) ; link!
         ((storage-link! be) key))
      (lambda (key buffer) ; unlink!
         ((storage-unlink! be) key buffer))
      (lambda (tag key) ; set-tag!
         ((storage-set-tag! be) tag key))
      (lambda (tag) ; tag
         ((storage-tag be) tag))
      (lambda () ; all-tags
         ((storage-all-tags be)))
      (lambda (tag) ; remove-tag!

   (define block-size (* 1024 1024))

   (make-storage
    block-size
      #t ; We are writable
      #t ; We support unlink!
      (lambda (key data length type) ; put!
         (if (file-read-access? (make-name key ".type"))
            (signal (make-property-condition 'exn 'message "Duplicate block: put! should not be called on an existing hash" 'arguments (list key type)))
            (begin
               (ensure-directory! key)
               ; Note: We save to ...~ files then mv them into place, so as to avoid ending up with a partial block
               ; in the archive if it dies in mid-write. We move the .type file in last, since the existance of that is what
               ; makes the block "official".
               ; The only thing we need worry about is a race between two snapshots writing the same block at once...
               ; However, since we can't easily provide atomicity on link!, we just say "don't do that" for now.
               (with-output-to-file (make-name key ".data~")
                  (lambda () (write-u8vector data (current-output-port) 0 length)))
               (with-output-to-file (make-name key ".type~")
                  (lambda () (write type)))
               (with-output-to-file (make-name key ".refcount~")
                  (lambda () (write 1)))
               (rename-file (make-name key ".data~") (make-name key ".data"))
               (rename-file (make-name key ".refcount~") (make-name key ".refcount"))
               (rename-file (make-name key ".type~") (make-name key ".type"))
               (void))))
      (lambda () (void)) ; flush! - a no-op for us
      (lambda (key) ; exists?
         (if (file-read-access? (make-name key ".data"))
            (with-input-from-file (make-name key ".type")
               (lambda () (read)))
            #f))
      (lambda (key buffer) ; get
         (if (file-read-access? (make-name key ".data"))
            (with-input-from-file (make-name key ".data")
               (lambda () (read-u8vector! #f buffer)))
            #f))
      (lambda (key) ; link!
         (if
            (file-read-access? (make-name key ".data"))
            (let
               ((current-refcount
                  (with-input-from-file (make-name key ".refcount")
                     (lambda () (read)))))
               (begin
                  (with-output-to-file (make-name key ".refcount~")
                     (lambda () (write (+ current-refcount 1))))
                     (rename-file (make-name key ".refcount~") (make-name key ".refcount"))))))
      (lambda (key buffer) ; unlink!
         (and-let*
            (((file-read-access? (make-name key ".data")))
            (current-refcount
               (with-input-from-file (make-name key ".refcount")
                  (lambda () (read))))
            (new-refcount (- current-refcount 1)))
            (if (zero? new-refcount)
               (let
                  ((data (with-input-from-file (make-name key ".data")
                     (lambda () (read-u8vector! #f buffer)))))
                  (begin
                     (delete-file (make-name key ".data"))
                     (delete-file (make-name key ".type"))
                     (delete-file (make-name key ".refcount"))
                     (delete-dir-if-empty! key)
                     data)) ; returned in case of deletion
               (begin

                     (void))))

      (make-storage
         block-size
         writable?
         #f ; We DO NOT support unlink!

         (lambda (key data length type) ; put!
           (check-writable)
           (when (pair? (get-block-data key))
                 (error "Duplicate block" key type))

           (set-file-position! *log* 0 seek/end)

           (let ((header (sprintf "(block ~S ~S ~S)" key type length))
                 (posn (file-position *log*)))
             (if (and (not (zero? posn)) (> (+ length (string-length header) posn) max-logpart-size))
                 (begin
                   (file-close *log*)
                   (set! posn 0)
                   (set-logcount! (+ *logcount* 1))
                   (set! *log* (file-open (string-append logdir "/log" (number->string *logcount*))
                                          (bitwise-ior open/creat open/rdwr open/append) (bitwise-ior perm/irusr perm/iwusr)))))
             (file-write *log* header)
             (file-write *log* (u8vector->blob/shared data) length)
             (set-block-data! key type *logcount* (+ (string-length header) posn) length)
             (void)))

         (lambda ()                     ; flush!
           (flush!)
           (void))

         (lambda (key) ; exists?
           (let ((bd (get-block-data key)))
             (if (pair? bd)
                 (car bd)
                 #f)))

         (lambda (key buffer) ; get
            (let* ((entry (get-block-data key)))
              (if (pair? entry)
               (let* ((type (first entry))
                      (index (second entry))
                      (position (third entry))
                      (length (fourth entry))

                      (logpart (get-log index)))
                 (set-file-position! logpart position seek/set)
                 (file-read logpart length (u8vector->blob/shared buffer))

                 length)
               #f)))

         (lambda (key) ; link!
           (check-writable)
           (void))

         (lambda (key buffer) ; unlink!
           (check-writable)
           (error "splitlog archives do not support unlinkined"))

         (lambda (tag key) ; set-tag!
           (check-writable)
           (file-write *log* (sprintf "(tag ~S ~S)" tag key))
           (set-tag! tag key)

	 (else
          (export-storage-error! "Invalid arguments to backend-fs")
	  (printf "USAGE:\nbackend-fs fs <basedir-path>\nbackend-fs splitlog <logdir-path> <metadata-file-path>\n")
	  #f)))

(if backend
    (export-storage! backend #f))

(use ugarit-backend)
(use matchable)

(define (backend-log name logfile include-bulk-data? be)
  (let ((logport (open-output-file logfile)))
    (set-buffering-mode! logport #:none)
    (fprintf logport "(use ugarit-backend)\n")
    (fprintf logport "(use test)\n")
    (fprintf logport "(let ((storage (import-storage ~S)))\n" name)
    (make-storage
     (storage-max-block-size be)
     (storage-writable? be)
     (storage-unlinkable? be)
     (lambda (key data length type)               ; put!
       (begin
         (fprintf logport "  (test (void) ((storage-put! storage) ~S '~S '~S))\n"
                  key (if include-bulk-data? data '...) type)
         ((storage-put! be) key data length type)))

     (lambda ()                            ; flush!
       (begin
         (fprintf logport "  (test (void) ((storage-flush! storage)))\n")
         ((storage-flush! be))))

     (lambda (key)                         ; exists?
       (let ((result ((storage-exists? be) key)))
         (begin
           (fprintf logport "  (test '~S ((storage-exists? storage) ~S))\n"
                    result key)
           result)))
     (lambda (key buffer)                         ; get
       (let ((result ((storage-get be) key buffer)))
         (begin
           (if include-bulk-data?
               (fprintf logport "  (test '~S ((storage-get storage) ~S))\n"
                        result key)
               (fprintf logport "  ((storage-get storage) ~S)\n"
                        key))
           result)))
     (lambda (key)                         ; link!
       (begin
         (fprintf logport "  (test (void) ((storage-link! storage) ~S))\n" key)
         ((storage-link! be) key)))
     (lambda (key buffer)                         ; unlink!
       (let ((result ((storage-unlink! be) key buffer)))
         (begin
           (if include-bulk-data?
               (fprintf logport "  (test '~S ((storage-unlink! storage) ~S))\n"
                        result key)
               (fprintf logport "  ((storage-unlink! storage) ~S)\n"
                        key))
           result)))

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

(if backend
    (export-storage! backend #f))

                           (exec set-block-data-query key (symbol->string type) (u8vector->blob/shared content))
                           (maybe-flush!)))

         (link-block! (lambda (key)
                           (exec link-block-query key)
                           (maybe-flush!)))

         (unlink-block! (lambda (key buffer)
                           (exec unlink-block-query key)
                           (maybe-flush!)
                           (let ((rc (query fetch get-block-refcount-query key)))
                             (if (pair? rc)
                                 (if (< (car rc) 1)
                                     (let ((contents (second (get-block-data key))))
                                       (exec delete-block-query key)
                                       (move-memory! contents buffer 0 (u8vector-length contents))
                                       (u8vector-length contents))
                                     #f)
                                 #f))))

         (set-tag! (lambda (tag key)
                    (exec set-tag-query tag key)
                    (flush!)))

                     (map car (query fetch-all get-tags-query)))))

      (make-storage
         block-size
         writable?
         #t ; We DO support unlink!

         (lambda (key data length type) ; put!
           (check-writable)
           (when (get-block-metadata key)
                 (error "Duplicate block" key type))

           (set-block-data! key type (subu8vector data 0 length)) ; FIXME: COPY
           (void))

         (lambda ()                     ; flush!
           (flush!)
           (void))

         (lambda (key) ; exists?
           (let ((bmd (get-block-metadata key)))
             bmd))

         (lambda (key buffer) ; get
            (let* ((entry (get-block-data key)))
              (if (pair? entry)
               (let* ((type (first entry))
                      (content (second entry)))
                 (move-memory! content buffer (u8vector-length content))
                 (u8vector-length content))
               #f)))

         (lambda (key) ; link!
           (check-writable)
           (link-block! key)
           (void))

         (lambda (key buffer) ; unlink!
           (check-writable)
           (unlink-block! key buffer))

         (lambda (tag key) ; set-tag!
           (check-writable)
           (set-tag! tag key)
           (void))

         (lambda (tag) ; tag

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

(if backend
    (export-storage! backend #f))

(storage "backend-sqlite /tmp/ugarit-test-4.vault")
;(hash tiger "foods")
;(file-cache "/tmp/ugarit-test-cache-4")

       (define var (void)) ...
       (test-no-errors name (set!-values (var ...) expr))))))

;; Test utilities

; Basic smoke test of a backend
(define (test-backend w)
  (let ((buffer (make-u8vector 5)))
      (parameterize
       ((backend-log! (lambda (type message) (void))))
       (test-assert "Storage block size is valid" (exact? (storage-max-block-size w))) 
       (test-assert "Storage writable" (storage-writable? w))
       (test-assert "Storage is empty" (not ((storage-exists? w) "TEST")))

       (test "Load a block" (void) ((storage-put! w) "TEST" (list->u8vector (list 1 2 3 4 5)) 5 'test))
       (test-assert "Block successfully loaded" ((storage-exists? w) "TEST"))
       (test "Block reads back" 5 ((storage-get w) "TEST" buffer))
       (test "Block contents reads back" (list 1 2 3 4 5) (u8vector->list buffer))

    (let ((long-name "caeef4a6ffe0cc5e25f9966c922366ec36b2bcf0dcd40754991ffe107b49fb33"))

      (test "Nonexistant block with a long name reacts correctly" #f ((storage-get w) long-name buffer))
      (test "Load a block with a long name" (void) ((storage-put! w) long-name (list->u8vector (list 6 7 8 9 10)) 5 'test))
      (test-assert "Block with a long name successfully loaded" ((storage-exists? w) long-name))
      (test "Block with a long name reads back" 5 ((storage-get w) long-name buffer))
      (test "Block contents with a long name reads back" (list 6 7 8 9 10) (u8vector->list buffer)))

    (test "Nonexistant block reacts correctly" #f ((storage-get w) "NONEXISTANT" buffer))
    (test-error "Cannot update existing blocks" ((storage-put! w) "TEST" (list->u8vector (list 1 2 3 4 5 6)) 6 'test))
    (if (storage-unlinkable? w)
        (begin
          (test "Unlink returns data" 5 ((storage-unlink! w) "TEST" buffer))
          (test "Unlink returns correct data" (list 1 2 3 4 5) (u8vector->list buffer))
          (test-assert "Unlinked block is gone" (not ((storage-exists? w) "TEST")))))

    (test "Set a tag" (void) ((storage-set-tag! w) "TEST" "TEST123"))
    (test "Tag is not locked" #f ((storage-tag-locked? w) "TEST"))
    (test "Lock a tag" #t ((storage-lock-tag! w) "TEST"))
    (test "Tag is now locked" #t ((storage-tag-locked? w) "TEST"))
    (test "Lock a tag again" #f ((storage-lock-tag! w) "TEST"))
    (test "Tag is still locked" #t ((storage-tag-locked? w) "TEST"))
    (test "Unlock a tag" (void) ((storage-unlock-tag! w) "TEST"))
    (test "Tag is no longer locked" #f ((storage-tag-locked? w) "TEST"))
    (test "Tag reads back" "TEST123" ((storage-tag w) "TEST"))
    (test "Tag list works" (list "TEST") ((storage-all-tags w)))
    (test "Remove tag" (void) ((storage-remove-tag! w) "TEST"))

    (test "Nonexistant tag is not locked" #f ((storage-tag-locked? w) "NONEXISTANT"))
    (test "Lock a nonexistant tag" #t ((storage-lock-tag! w) "NONEXISTANT"))
    (test "Nonexistant tag is now locked" #t ((storage-tag-locked? w) "NONEXISTANT"))
    (test "Lock a nonexistant tag again" #f ((storage-lock-tag! w) "NONEXISTANT"))
    (test "Nonexistant tag is still locked" #t ((storage-tag-locked? w) "NONEXISTANT"))
    (test "Unlock a locked nonexistant tag" (void) ((storage-unlock-tag! w) "NONEXISTANT"))
    (test "Nonexistant tag is no longer locked" #f ((storage-tag-locked? w) "NONEXISTANT"))

    (test "Close storage" (void) ((storage-close! w))))))

(define (key-stream-cat a ks-hash ks-type level)
   (define type (vault-exists? a ks-hash))
   (if (eq? type ks-type)
      (begin
         (printf "ks(~A): ~A (~A)\n" level ks-hash type)
            (for-each (lambda (subkey)

                (test "Extract a directory" (void)
                      (extract-directory! a dir4-key extract4-dir))
                (check-extract-results extract4-dir "Hello world again!" "Hello world 2")

                ;; Tidy up
                (when (vault-unlinkable? a)
                      (begin
                        (printf "Get dir: ~s\n" (vault-get a dir1-key))
                        (printf "Unlink dir: ~s\n" (vault-unlink! a dir1-key))
                        (exit 0)

                        (test "Delete the first directory" (void) (unlink-directory! a dir1-key))
                        (check-extract-results extract2-dir "Hello world" "Hello world 2")
                        (test "Delete the second directory" (void) (unlink-directory! a dir2-key))
                        (check-extract-results extract3-dir "Hello world again!" "Hello world 2")
                        (test "Delete the third directory" (void) (unlink-directory! a dir3-key))
                        (check-extract-results extract4-dir "Hello world again!" "Hello world 2")
                        (test "Delete the fourth directory" (void) (unlink-directory! a dir4-key))

           ;; This bit will throw an error if the vault's encryption or
           ;; hashing is set up incorrectly.
           

         (let ((configuration
                (handle-exceptions
                 exn
                 (signal exn)
                 #;(signal (make-property-condition
                          'exn
                          'location 'open-vault
                          'message "Reading the vault header failed. Most likely, your hash or encryption settings are incorrect, or there is a problem with the vault."))

                 (read-sexpr vault (tag-key conf-tag) 'ugarit-vault-configuration))))
           (match configuration
                  (((? integer? ver) . alist)

(use ports)
(use matchable)
(use posix)
(use srfi-4)
(use data-structures)
(use miscmacros)














; Backends can call the procedure found in this paramter to log
; things. type should be 'warning, 'error or 'info. message should
; be any string.
(define backend-log! (make-parameter
                      (lambda (type message)
                        (error "No backend log handler has been defined"))))

(define-record-type storage
  (make-storage*
   name
   max-block-size        ; Integer: largest size of block we can store
   writable? ; Boolean: Can we call put!, link!, unlink!, set-tag!, lock-tag!, unlock-tag!?
   unlinkable?                          ; Boolean: Can we call unlink?
   put! ; Procedure: (put! key data length type) - stores the data (u8vector) under the key (string) with the given type tag (symbol) and a refcount of 1. Does nothing of the key is already in use.
   flush! ; Procedure: (flush!) - all previous changes must be flushed to disk by the time the continuation is applied.
   exists? ; Procedure: (exists? key) - returns the type of the block with the given key if it exists, or #f otherwise
   get ; Procedure: (get key buffer) - returns the contents of the block with the given key (string) into the u8vector buffer and returns the length used if it exists, or #f otherwise
   link! ; Procedure: (link key) - increments the refcount of the block
   unlink! ; Procedure: (unlink key buffer) - decrements the refcount of the block. If it's now zero, deletes it but loads it into the u8vector buffer and returns the length. If not, returns #f.
   set-tag! ; Procedure: (set-tag! name key) - assigns the given key (string) to the given tag (named with a string). Creates a new tag if the name has not previously been used, otherwise updates an existing tag
   tag ; Procedure: (tag name) - returns the key assigned to the given tag, or #f if it does not exist.
   all-tags ; Procedure: (all-tags) - returns a list of all existing tag names
   remove-tag! ; Procedure: (remove-tag! name) - removes the named tag
   lock-tag! ; Procedure: (lock-tag! name) - locks the named tag, returning #t if all went well, or #f if it can't be locked.
   tag-locked? ; Procedure: (tag-locked? name) - returns #t if the tag is locked, #f otherwise
   unlock-tag! ; Procedure: (unlock-tag! name) - unlocks the named tag

                     (lambda (type message)
                       (queue-add! log (cons type message))
                       (void))))
                   (let ((result (begin body ...)))
                     (write (list (queue->list log) result)))))))

; Return the result of the body as a data block, and any logs
(define-syntax-rule (with-error-reporting-and-block buffer body ...)
  (handle-exceptions
   exn (write (list "error" (describe-exception exn)))
   (let ((log (make-queue)))
     (parameterize ((backend-log!
                     (lambda (type message)
                       (queue-add! log (cons type message))
                       (void))))
                   (let ((length (begin body ...)))
                     (if length
                         (begin
                           (write (list (queue->list log) length))
                           (write-u8vector buffer (current-output-port) 0 length))
                         (write (list (queue->list log) #f))))))))

; Return any logs
(define-syntax-rule (with-error-reporting body ...)
  (handle-exceptions
   exn (write (list "error" (describe-exception exn)))
   (let ((log (make-queue)))

  ; Write the error header
  (write *magic-v2*) (newline)
  (write (list "error" message)))

;; Given a storage object, provide the storage remote access protocol
;; via current-input-port / current-output-port until the storage is closed
;; via the protocol.
(define (export-storage! storage-thunk debug)
  (set-buffering-mode! (current-output-port) #:none)

  ; Write the header
  (write *magic-v2*) (newline)
  (let ((storage #f))

    (with-error-reporting-and-result ; Initialise and send the header
     (let ((storage* (storage-thunk)))
       (set! storage storage*)        ; This feels hacky
       (list (storage-max-block-size storage)
             (storage-writable? storage)
             (storage-unlinkable? storage))))

                                        ; Engage command loop
    (if storage
        (let ((*buffer* (make-u8vector (storage-max-block-size storage))))
         (let loop ()
           (newline)
           (let ((command (read)))
             (if (eof-object? command)
                 (begin
                   (if debug (printf "~a: EOF~%" debug))
                   (with-error-reporting
                    ((storage-close! storage)))
                   (void))
                 (match
                  command

                  (('put! key type length)
                   (if debug (printf "~a: put! ~a ~a ~a~%" debug key type length))
                   (let ((bytes-read (read-u8vector! length *buffer* (current-input-port))))
                     (with-error-reporting
                      ;; FIXME: assert (= bytes-read length) in case of short read due to EOF
                      ((storage-put! storage) key *buffer* length type)))
                   (loop))

                  (('flush!)
                   (if debug (printf "~a: flush!~%" debug))
                   (with-error-reporting
                    ((storage-flush! storage)))
                   (loop))

                  (('exists? key)
                   (if debug (printf "~a: exists? ~a~%" debug key))
                   (with-error-reporting-and-result
                    ((storage-exists? storage) key))
                   (loop))

                  (('get key)
                   (if debug (printf "~a: get ~a~%" debug key))
                   (with-error-reporting-and-block *buffer*
                                                   ((storage-get storage) key *buffer*))
                   (loop))

                  (('link! key)
                   (if debug (printf "~a: link! ~a~%" debug key))
                   (with-error-reporting
                    ((storage-link! storage) key))
                   (loop))

                  (('unlink! key)
                   (if debug (printf "~a: unlink! ~a~%" debug key))
                   (with-error-reporting-and-block *buffer*
                                                   ((storage-unlink! storage) key *buffer*))
                   (loop))

                  (('set-tag! name key)
                   (if debug (printf "~a: set-tag! ~a ~a~%" debug name key))
                   (with-error-reporting
                    ((storage-set-tag! storage) name key))
                   (loop))

                  (('tag name)
                   (if debug (printf "~a: tag ~a~%" debug name))
                   (with-error-reporting-and-result
                    ((storage-tag storage) name))
                   (loop))

                  (('all-tags)
                   (if debug (printf "~a: all-tags~%" debug))
                   (with-error-reporting-and-result
                    ((storage-all-tags storage)))
                   (loop))

                  (('remove-tag! name)
                   (if debug (printf "~a: remove-tag! ~a~%" debug name))
                   (with-error-reporting
                    ((storage-remove-tag! storage) name))
                   (loop))

                  (('lock-tag! name)
                   (if debug (printf "~a: lock-tag! ~a~%" debug name))
                   (with-error-reporting-and-result
                    ((storage-lock-tag! storage) name))
                   (loop))

                  (('tag-locked? name)
                   (if debug (printf "~a: tag-locked? ~a~%" debug name))
                   (with-error-reporting-and-result
                    ((storage-tag-locked? storage) name))
                   (loop))

                  (('unlock-tag! name)
                   (if debug (printf "~a: unlock-tag! ~a~%" debug name))
                   (with-error-reporting
                    ((storage-unlock-tag! storage) name))
                   (loop))

                  (('admin! command)
                   (if debug (printf "~a: admin! ~a~%" debug command))
                   (with-error-reporting-and-result
                    ((storage-admin! storage) command))
                   (loop))

                  (('close!)
                   (if debug (printf "~a: close!~%" debug))
                   (with-error-reporting
                    ((storage-close! storage)))
                   (void))

                  (else
                   (write (list "error" (sprintf "Bad command ~s" command)))
                   (loop))))))))))

;; Importing a storage - taking a command line to a backend protocol
;; server and turning it into a storage record

(define (protocol-error message backend operation . irritants)
  (abort
   (make-composite-condition

(define (read-response-v1 port backend operation)
  (let ((response (read port)))
   (match response
          (("error" err) (protocol-error "Backend protocol error" backend operation err))
          (else response))))

(define (read-response-v1-body port backend operation buffer)
  (let ((response (read-response-v1 port backend operation)))
    (if response
        (read-u8vector! (car response) buffer port)
        #f)))

(define (import-storage-v1 command-line debug responses commands pid)
  (let ((header
         (rewrite-i/o-errors command-line 'read-header
          (lambda () (read responses)))))
    (if debug (printf "~a: read header ~a~%" command-line header))
    (if (not (list? header))
        (protocol-error "Invalid backend protocol header" command-line 'read-header header))
    (if (not (= (length header) 3))
        (protocol-error "Invalid backend protocol header" command-line 'read-header header))
    (let* ((max-block-size (car header))
           (writable? (cadr header))
           (unlinkable? (caddr header))
           (buffer (make-u8vector max-block-size)))
      (make-storage*
       command-line
       max-block-size
       writable?
       unlinkable?

       (lambda (key data length type)  ; put!
         (rewrite-i/o-errors command-line 'put!
          (lambda ()
            (if debug (printf "~a: put!" command-line))
            (write `(put! ,key ,type ,(u8vector-length data)) commands)
            (write-u8vector data commands 0 length)
            (read-response-v1 responses command-line 'put!)
            (void))))

       (lambda ()                  ; flush!
         (rewrite-i/o-errors command-line 'flush!
          (lambda ()
            (if debug (printf "~a: flush!" command-line))
            (write `(flush!) commands)
            (read-response-v1 responses command-line 'flush!)
            (void))))

       (lambda (key)            ; exists?
         (rewrite-i/o-errors command-line 'exists?
          (lambda ()
            (if debug (printf "~a: exists?" command-line))
            (write `(exists? ,key) commands)
            (read-response-v1 responses command-line 'exists?))))

       (lambda (key buffer)            ; get
         (rewrite-i/o-errors command-line 'get
          (lambda ()
            (if debug (printf "~a: get" command-line))
            (write `(get ,key) commands)
            (read-response-v1-body responses command-line 'get buffer))))

       (lambda (key)            ; link!
         (rewrite-i/o-errors command-line 'link!
          (lambda ()
           (if debug (printf "~a: link!" command-line))
           (write `(link! ,key) commands)
           (read-response-v1 responses command-line 'link!)
           (void))))

       (lambda (key buffer)            ; unlink!
         (rewrite-i/o-errors command-line 'unlink!
          (lambda ()
            (if debug (printf "~a: unlink! ~s" command-line key))
            (write `(unlink! ,key) commands)
            (read-response-v1-body responses command-line 'unlink! buffer))))

       (lambda (name key)               ; set-tag!
         (rewrite-i/o-errors command-line 'set-tag!
          (lambda ()
            (if debug (printf "~a: set-tag!" command-line))
            (write `(set-tag! ,name ,key) commands)
            (read-response-v1 responses command-line 'set-tag!)

          ((log value)
           (for-each (lambda (logentry)
                       ((backend-log!) (car logentry) (cdr logentry)))
                     log)
           value)
          (else (error "Malformed response from backend" response)))))

(define (read-response-v2-body port backend operation buffer)
  (let ((response-length (read-response-v2-result port backend operation)))
    (if response-length
        (read-u8vector! response-length buffer port)
        #f)))

(define (import-storage-v2 command-line debug responses commands pid)
  (let ((header
         (rewrite-i/o-errors command-line 'read-header
          (lambda ()
            (read-response-v2-result responses command-line 'read-header)))))
    (if debug (printf "~a: read header ~a~%" command-line header))
    (if (not (list? header))
        (protocol-error "Invalid backend protocol header" command-line 'read-header header))
    (if (not (= (length header) 3))
        (protocol-error "Invalid backend protocol header" command-line 'read-header header))
    (let* ((max-block-size (car header))
           (writable? (cadr header))
           (unlinkable? (caddr header))
           (buffer (make-u8vector max-block-size)))
      (make-storage*
       command-line
       max-block-size
       writable?
       unlinkable?

       (lambda (key data length type)  ; put!
         (rewrite-i/o-errors command-line 'put!
          (lambda ()
            (if debug (printf "~a: put!" command-line))
            (write `(put! ,key ,type ,length) commands)
            (write-u8vector data commands 0 length)
            (read-response-v2 responses command-line 'put!)
            (void))))

       (lambda ()                  ; flush!
         (rewrite-i/o-errors command-line 'flush!
          (lambda ()
            (if debug (printf "~a: flush!" command-line))
            (write `(flush!) commands)
            (read-response-v2 responses command-line 'flush!)
            (void))))

       (lambda (key)            ; exists?
         (rewrite-i/o-errors command-line 'exists?
          (lambda ()
            (if debug (printf "~a: exists?" command-line))
            (write `(exists? ,key) commands)
            (read-response-v2-result responses command-line 'exists?))))

       (lambda (key buffer)            ; get
         (rewrite-i/o-errors command-line 'get
          (lambda ()
            (if debug (printf "~a: get ~s" command-line key))
            (write `(get ,key) commands)
            (read-response-v2-body responses command-line 'get buffer))))

       (lambda (key)            ; link!
         (rewrite-i/o-errors command-line 'link!
          (lambda ()
            (if debug (printf "~a: link!" command-line))
            (write `(link! ,key) commands)
            (read-response-v2 responses command-line 'link!)
            (void))))

       (lambda (key buffer)            ; unlink!
         (rewrite-i/o-errors command-line 'unlink!
          (lambda ()
            (if debug (printf "~a: unlink! ~s" command-line key))
            (write `(unlink! ,key) commands)
            (read-response-v2-body responses command-line 'unlink! buffer))))

       (lambda (name key)               ; set-tag!
         (rewrite-i/o-errors command-line 'set-tag!
          (lambda ()
            (if debug (printf "~a: set-tag!" command-line))
            (write `(set-tag! ,name ,key) commands)
            (read-response-v2 responses command-line 'set-tag!)

            (if debug (printf "~a: close!!" command-line))
            (write '(close!) commands)
            (read-response-v2 responses command-line 'close!)
            (close-input-port responses)
            (close-output-port commands)
            (void))))))))
























;; Given the command line to a storage remote access protocol server,
;; activate it and return a storage object providing access to the
;; server.
(define (import-storage command-line . args)
  (let-optionals args ((debug #f))
   (let-values (((responses commands pid)
                 (process command-line)))




     (if debug (printf "~a: process opened~%" command-line))
     (let ((magic
            (rewrite-i/o-errors command-line 'read-magic
             (lambda () (read responses)))))
       (if debug (printf "~a: read magic ~a~%" command-line magic))
       (cond
        ((equal? magic *magic-v1*)
         (import-storage-v1 command-line debug responses commands pid))
        ((equal? magic *magic-v2*)
         (import-storage-v2 command-line debug responses commands pid))
        (else (protocol-error (sprintf "Unrecognised backend protocol header magic: ~A" magic) command-line 'read-magic magic))))))))

 (define cache-commit-interval 1000)

 (define (vault-cache-updated! vault)
   (inc! (vault-cache-updates-uncommitted vault))
   (when (> cache-commit-interval (vault-cache-updates-uncommitted vault))
         (vault-cache-flush! vault)))


;; Take a block, and return a compressed and encrypted block
 (define (wrap-block vault block)
   ((vault-encrypt vault)
    ((vault-compress vault) block)))

 ;; Take a compressed and encrypted block, and recover the original data
 (define (unwrap-block vault block)

      " "
      (string-pad (number->string (vector-ref localtime 2)) 2 #\0)
      ":"
      (string-pad (number->string (vector-ref localtime 1)) 2 #\0)
      ":"
      (string-pad (number->string (vector-ref localtime 0)) 2 #\0))))

 (define (vault-put! vault key data type) ; FIXME: Update to reuse buffer
   (unless (vault-writable? vault)
           (error 'vault-put! "This isn't a writable vault"))
   (let ((wrapped-block (wrap-block vault data)))
     (with-backend-logging
      ((storage-put! (vault-storage vault))
       key
       wrapped-block
       (u8vector-length wrapped-block)
       type)))
   (inc! (job-blocks-stored (current-job)))
   (inc! (job-bytes-stored (current-job)) (u8vector-length data))
   (void))

(define (vault-conf-get vault key default)
  (let ((result (assq key (vault-conf-alist vault))))
    (if result

   ;; vault-cache-flush also flushes the backend
   (vault-cache-flush! vault))

 (define (vault-exists? vault key)
   (with-backend-logging ((storage-exists? (vault-storage vault)) key)))

 (define (vault-get vault key type) ; FIXME: Update to reuse buffer
   (let* ((block-size (vault-max-block-size vault))
          (buffer (make-u8vector block-size))
          (length (with-backend-logging ((storage-get (vault-storage vault)) key buffer)))
          (data (if length
                    (unwrap-block vault (subu8vector buffer 0 length))
                    (error 'vault-get (sprintf "Nonexistant block ~A ~A" key type)))))
     (unless (string=? key ((vault-hash vault) data type))
             (error 'vault-get (sprintf "Consistency check failure: asked for ~A, got ~A" key ((vault-hash vault) data type))))
     data))

 (define (vault-link! vault key)
   (unless (vault-writable? vault)
           (error 'vault-link! "This isn't a writable vault"))
   (with-backend-logging ((storage-link! (vault-storage vault)) key)))

 (define (vault-unlink! vault key) ;; FIXME: Update to reuse buffer
   (unless (vault-writable? vault)
           (error 'vault-unlink! "This isn't a writable vault"))
   (let* ((buffer (make-u8vector (vault-max-block-size vault)))
          (length (with-backend-logging ((storage-unlink! (vault-storage vault)) key buffer))))
     (if length
         (unwrap-block vault (subu8vector buffer 0 length)) ;; FIXME: COPIES
         #f)))

 (define (vault-admin! vault command)
   (with-backend-logging ((storage-admin! (vault-storage vault)) command)))

 (define-record-type tag
   (make-tag name type key)

  (let ((source-file (string-append name ".scm"))
        (so-file (string-append name ".so"))
        (import-file (string-append name ".import.scm"))
        (import-so-file (string-append name ".import.so"))
        (o-file (string-append name ".o")))

    (when (newer source-file so-file)
          (compile -s -optimize-level 3 -debug-level 2 ,(string->symbol source-file) -j ,(string->symbol name))
          (compile -s -optimize-level 3 -debug-level 2 ,(string->symbol import-file))
          (compile -c -optimize-level 3 -debug-level 2 ,(string->symbol source-file) -unit ,(string->symbol name)))

    (install-extension (string->symbol name) `(,so-file ,o-file ,import-so-file)
                       `((version ,*version*)
                         (static o-file)))))

(define (build-program name)
  (let ((source-file (string-append name ".scm"))
        (exec-file name))
   (when (newer source-file exec-file)
         (compile -optimize-level 3 -debug-level 2 ,(string->symbol source-file)))
   (install-program (string->symbol name) exec-file
                    `((version ,*version*)))))

(build-module "directory-rules")
(build-module "ugarit-mime")
(build-module "ugarit-backend")
(build-module "ugarit-core")