(module
 ugarit-snapshot
 (init-snapshot-subsystem

  tag-snapshot!

  update-snapshot-cache-for-tag!
  get-snapshot-parents
  get-snapshot-children
  get-snapshot-props)

 (import scheme)
 (import chicken)
 (use srfi-1)
 (use sql-de-lite)
 (use ugarit-core)
 (use ugarit-streams)

(define-record-type vault-subsys
  (make-vault-subsys)
  vault-subsys?)

(define (init-snapshot-subsystem cache)
  (ensure-table
   cache "snapshots"
   "CREATE TABLE snapshots (tag TEXT, snapshot_id INTEGER PRIMARY KEY, key TEXT);")
  (ensure-table
   cache "snapshot_props"
   "CREATE TABLE snapshot_props (snapshot_id INTEGER, prop TEXT, value TEXT);"))

;; Accept an even number of arguments. Each pair of arguments
;; is a boolean and a value. The result is a list of all values
;; whose booleans were non-#f
(define (conditional-list . args)
  (let loop ((args-left args)
             (result '()))
    (cond
     ((null? args-left)
      (reverse result))
     ((null? (cdr args-left))
      (error 'conditional-list "An even number of arguments is required" args))
     ((car args-left)
      (loop (cddr args-left)
            (cons (cadr args-left) result)))
     (else
      (loop (cddr args-left)
            result)))))

 ;; SNAPSHOT STORAGE
 ;; A snapshot is a single block containing an alist
 ;; Keys are 'ctime (in seconds since the epoch),
 ;; 'contents (hash of root directory),
 ;; 'hostname (name of host snapshotted)
 ;; 'prefix (prefix of filesystem on host)
 ;; 'notes (user-supplied notes)
 ;; 'previous (hash of previous snapshot)
 ;; 'stats (alist of stats:
 ;;         'blocks-stored
 ;;         'bytes-stored
 ;;         'blocks-skipped
 ;;         'bytes-skipped
 ;;         'file-cache-hits
 ;;         'file-cache-bytes
 ;; 'log (list of log events, each being a (type timestamp path message) list
 ;; Returns the snapshot's key.

 (define (tag-snapshot! vault tag contents-key contents-reused? snapshot-properties job)
   (check-vault-writable vault)
   (vault-lock-tag! vault tag) ;; Lock BEFORE reading previous state of the tag, to avoid races.
   (receive (job-log-key job-log-key-reused?)
            (if job
                (store-job-log! vault job)
                (values #f #f))
            (let* ((previous (vault-tag vault tag))
                   (stats (if job
                              (job-stats-alist job)
                              #f))
                   (snapshot
                    (append
                     (conditional-list
                      #t (cons 'mtime (current-seconds))
                      #t (cons 'contents contents-key)
                      (list? stats) (cons 'stats stats)
                      job-log-key (cons 'log job-log-key))
                     snapshot-properties))
                   (keys
                    (conditional-list
                     ; We do not list the previous snapshot - since we are about to overwrite the tag that points to it, which would be a decrement.
                     #t (cons contents-key contents-reused?)
                     job-log-key (cons job-log-key job-log-key-reused?))))
              (when (tag? previous)
                    (when (not (eq? (tag-type previous) 'snapshot))
                          (error (sprintf "It is illegal to add a snapshot to the tag ~A, which is of type ~A" tag (tag-type previous))))
                    (set! snapshot (cons
                                    (cons 'previous (tag-key previous))
                                    snapshot)))
              (let-values (((snapshot-key snapshot-reused?)
                            (store-sexpr! vault snapshot 'snapshot keys)))
                (vault-flush! vault) ; After this point we can be sure that the snapshot and all blocks it refers to are stably stored
                (vault-set-tag! vault (make-tag tag 'snapshot snapshot-key)) ; Therefore, we can be confident in saving it in a tag.
                (vault-unlock-tag! vault tag)
                (when snapshot-reused? ; Rare, but possible; fork a tag then snapshot the same FS state to both at the same second.
                      (vault-link! vault snapshot-key))
                snapshot-key))))

(define (cache-snapshot*! vault tag-name key)
  (let ((db (vault-cache vault)))
    (when (zero? (car (query fetch (sql db "SELECT COUNT(*) FROM snapshots WHERE tag = ? AND key = ?;") tag-name key)))
          (let ((snapshot-block (read-sexpr vault key 'snapshot)))
            (exec (sql db "INSERT INTO snapshots (tag,key) VALUES (?,?);")
                  tag-name key)

            (let ((snapshot-id (last-insert-rowid db)))
             (for-each (lambda (se)
                         ;; For each entry in the property alist...
                         (exec (sql db "INSERT INTO snapshot_props (snapshot_id, prop, value) VALUES (?,?,?);") snapshot-id (symbol->string (car se)) (serialise (cdr se))))
                       snapshot-block))

            ; Cache all parent references (recurses here)
            (for-each
             (lambda (ie)
               (when (eq? (car ie) 'previous)
                     (cache-snapshot*! vault tag-name (cdr ie))))
             snapshot-block)))))

(define (cache-snapshot! vault tag-name key)
  (let ((db (vault-cache vault)))
    (when (zero? (car (query fetch (sql db "SELECT COUNT(*) FROM snapshots WHERE tag = ? AND key = ?;") tag-name key)))
          (vault-cache-flush! vault) ; Start a new txn
          (cache-snapshot*! vault tag-name key)
          (vault-cache-flush! vault)))) ; Flush that txn

(define (update-snapshot-cache-for-tag! vault tag-name)
  (let ((tag (vault-tag vault tag-name)))
    (when tag
          (cache-snapshot! vault tag-name (tag-key tag)))))

(define (get-snapshot-parents vault tag-name key)
  (let* ((db (vault-cache vault))
         (r (query fetch-all
                   (sql db "SELECT value FROM snapshot_props WHERE snapshot_id IN (SELECT snapshot_id FROM snapshots WHERE tag = ? AND key = ?) AND prop = 'previous';")
                   tag-name key)))
    (map (lambda (x)
           (deserialise (car x)))
         r)))

(define (get-snapshot-children vault tag-name key)
  (let* ((db (vault-cache vault))
         (r (query fetch-all
                   (sql db "SELECT key FROM snapshots WHERE snapshot_id IN (SELECT snapshot_id FROM snapshot_props WHERE prop = 'previous' AND value = ?) AND tag = ?")
                   (serialise key) tag-name)))
    (map car r)))

(define (get-snapshot-props vault tag-name key)
  (let* ((db (vault-cache vault))
         (r (query fetch-all
                   (sql db "SELECT prop, value FROM snapshot_props WHERE snapshot_id IN (SELECT snapshot_id FROM snapshots WHERE tag = ? AND key = ?);")
                   tag-name key)))
    (map (lambda (x)
           (cons
            (string->symbol (first x))
            (deserialise (second x))))
         r)))

)