path: root/guix-build-coordinator/datastore/sqlite.scm

(define-module (guix-build-coordinator datastore sqlite)
  #:use-module (oop goops)
  #:use-module (srfi srfi-1)
  #:use-module (srfi srfi-19)
  #:use-module (srfi srfi-43)
  #:use-module (srfi srfi-71)
  #:use-module (ice-9 match)
  #:use-module (ice-9 format)
  #:use-module (ice-9 threads)
  #:use-module (ice-9 exceptions)
  #:use-module (sqlite3)
  #:use-module (fibers)
  #:use-module (prometheus)
  #:use-module (guix base16)
  #:use-module (guix derivations)
  #:use-module (guix-build-coordinator utils)
  #:use-module (guix-build-coordinator utils fibers)
  #:use-module (guix-build-coordinator config)
  #:use-module (guix-build-coordinator datastore abstract)
  #:export (sqlite-datastore
            datastore-optimize
            datastore-spawn-fibers
            datastore-initialise-metrics!
            datastore-update-metrics!
            datastore-update
            make-transaction-rollback-exception
            datastore-call-with-transaction
            datastore-store-derivation
            datastore-build-exists-for-derivation-outputs?
            datastore-build-required-by-another?
            datastore-list-related-derivations-with-no-build-for-outputs
            datastore-list-failed-builds-with-blocking-count
            datastore-list-builds-for-derivation-recursive-inputs
            datastore-insert-build
            datastore-cancel-build
            datastore-update-unprocessed-builds-with-lower-derived-priorities
            datastore-update-build-priority
            datastore-count-builds
            datastore-for-each-build
            datastore-find-build
            datastore-list-builds
            datastore-insert-build-tags
            datastore-fetch-build-tags
            datastore-find-build-result
            datastore-find-build-derivation-system
            datastore-count-builds-for-derivation
            datastore-count-build-results
            datastore-insert-build-result
            datastore-update-unprocessed-builds-for-build-success
            datastore-remove-build-allocation
            datastore-mark-build-as-processed
            datastore-delete-relevant-outputs-from-unbuilt-outputs
            datastore-store-output-metadata
            datastore-list-unbuilt-derivation-outputs
            datastore-list-build-outputs
            datastore-new-agent
            datastore-list-agents
            datastore-set-agent-active
            datastore-find-agent-status
            datastore-update-agent-status
            datastore-find-agent
            datastore-find-agent-by-name
            datastore-insert-dynamic-auth-token
            datastore-dynamic-auth-token-exists?
            datastore-fetch-agent-tags
            datastore-store-build-start
            datastore-find-build-starts
            datastore-count-setup-failures
            datastore-list-setup-failures-for-build
            datastore-fetch-setup-failures
            datastore-store-setup-failure
            datastore-store-setup-failure/missing-inputs
            datastore-list-setup-failure-missing-inputs
            datastore-find-derivation
            datastore-find-derivation-system
            datastore-find-derivation-inputs
            datastore-find-derivation-for-output
            datastore-find-derivation-outputs
            datastore-find-derivation-output-details
            datastore-list-builds-for-output
            datastore-list-builds-for-output-and-system
            datastore-new-agent-password
            datastore-agent-password-exists?
            datastore-agent-list-passwords
            datastore-replace-agent-tags
            datastore-list-processed-builds
            datastore-list-unprocessed-builds
            datastore-find-first-unallocated-deferred-build
            datastore-fetch-prioritised-unprocessed-builds
            datastore-insert-unprocessed-hook-event
            datastore-count-unprocessed-hook-events
            datastore-list-unprocessed-hook-events
            datastore-find-unprocessed-hook-event
            datastore-delete-unprocessed-hook-event
            datastore-list-agent-builds
            datastore-agent-for-build
            datastore-count-build-allocation-plan-entries
            datastore-replace-build-allocation-plan
            datastore-remove-build-from-allocation-plan
            datastore-count-allocated-builds
            datastore-agent-requested-systems
            datastore-update-agent-requested-systems
            datastore-fetch-build-to-allocate
            datastore-insert-to-allocated-builds
            datastore-remove-builds-from-plan
            datastore-list-allocation-plan-builds))

(define-class <sqlite-datastore> (<abstract-datastore>)
  database-file
  worker-reader-thread-channel
  worker-writer-thread-channel
  metrics-registry)

(define* (sqlite-datastore database-uri
                           #:key
                           update-database?
                           metrics-registry)
  (define database-file
    (string-drop database-uri
                 (string-length "sqlite://")))

  (when update-database?
    (run-sqitch database-file))

  (let ((db (db-open database-file)))
    (sqlite-exec db "PRAGMA journal_mode=WAL;")
    (sqlite-exec db "PRAGMA optimize;")
    (sqlite-exec db "PRAGMA wal_checkpoint(TRUNCATE);")
    (sqlite-close db))

  (let ((datastore (make <sqlite-datastore>)))

    (slot-set! datastore 'database-file database-file)
    (slot-set! datastore 'metrics-registry metrics-registry)

    (slot-set!
     datastore
     'worker-writer-thread-channel
     (make-worker-thread-channel
      (lambda ()
        (let ((db
               (db-open database-file)))
          (sqlite-exec db "PRAGMA busy_timeout = 5000;")
          (sqlite-exec db "PRAGMA synchronous = NORMAL;")
          (sqlite-exec db "PRAGMA temp_store = MEMORY;")
          (sqlite-exec db "PRAGMA foreign_keys = ON;")
          (list db)))
      #:destructor
      (let ((writer-thread-destructor-counter
             (make-gauge-metric metrics-registry
                                "datastore_writer_thread_close_total")))
        (lambda (db)
          (db-optimize db
                       database-file
                       metrics-registry
                       #:maybe-truncate-wal? #f)

          (metric-increment writer-thread-destructor-counter)
          (sqlite-close db)))
      #:lifetime 500

      ;; SQLite doesn't support parallel writes
      #:parallelism 1
      #:delay-logger (let ((delay-metric
                            (make-histogram-metric
                             metrics-registry
                             "datastore_write_delay_seconds")))
                       (lambda (seconds-delayed)
                         (metric-observe delay-metric
                                         ;; TODO exact->inexact to work around
                                         ;; a bug in guile-prometheus where
                                         ;; the metric sum will output in the
                                         ;; exact form including the /q
                                         (exact->inexact seconds-delayed))
                         (log-delay "datastore write" seconds-delayed)
                         (when (> seconds-delayed 1)
                           (format
                            (current-error-port)
                            "warning: database write delayed by ~1,2f seconds~%"
                            seconds-delayed))))))

    (slot-set!
     datastore
     'worker-reader-thread-channel
     (make-worker-thread-channel
      (lambda ()
        (let ((db
               (db-open database-file #:write? #f)))
          (sqlite-exec db "PRAGMA temp_store = MEMORY;")
          (sqlite-exec db "PRAGMA busy_timeout = 4000;")
          (sqlite-exec db "PRAGMA cache_size = -16000;")
          (list db)))
      #:destructor
      (let ((reader-thread-destructor-counter
             (make-gauge-metric metrics-registry
                                "datastore_reader_thread_close_total")))
        (lambda (db)
          (metric-increment reader-thread-destructor-counter)
          (sqlite-close db)))
      #:lifetime 50000

      ;; Use a minimum of 2 and a maximum of 8 threads
      #:parallelism
      (min (max (current-processor-count)
                2)
           8)
      #:delay-logger (let ((delay-metric
                            (make-histogram-metric
                             metrics-registry
                             "datastore_read_delay_seconds")))
                       (lambda (seconds-delayed)
                         (metric-observe delay-metric
                                         ;; TODO exact->inexact to work around
                                         ;; a bug in guile-prometheus where
                                         ;; the metric sum will output in the
                                         ;; exact form including the /q
                                         (exact->inexact seconds-delayed))
                         (log-delay "datastore read" seconds-delayed)
                         (when (> seconds-delayed 1)
                           (format
                            (current-error-port)
                            "warning: database read delayed by ~1,2f seconds~%"
                            seconds-delayed))))))

    datastore))

(define (sqlite-step-and-reset statement)
  (let ((val (sqlite-step statement)))
    (sqlite-reset statement)
    val))

(define (sqlite-step-and-finalize statement)
  (let ((val (sqlite-step statement)))
    (sqlite-finalize statement)
    val))

(define* (db-optimize db db-filename metrics-registry
                      #:key (maybe-truncate-wal? #t))
  (define (wal-size)
    (let ((db-wal-filename
           (string-append db-filename "-wal")))

      (stat:size (stat db-wal-filename))))

  (define MiB (* (expt 2 20) 1.))
  (define wal-size-threshold
    (* 5 MiB))

  (define extreme-wal-size-threshold
    (* 30 MiB))

  (and
   (let ((checkpoint-duration-metric-name
          "datastore_wal_checkpoint_duration_seconds"))

     (if (and maybe-truncate-wal?
              (> (wal-size) wal-size-threshold))
         (call-with-duration-metric
          metrics-registry
          checkpoint-duration-metric-name
          (lambda ()
            (if (> (wal-size) extreme-wal-size-threshold)
                ;; Since the WAL is really getting too big, wait for much longer
                (sqlite-exec db "PRAGMA busy_timeout = 300000;")
                (sqlite-exec db "PRAGMA busy_timeout = 100;"))

            (let* ((statement
                    (sqlite-prepare
                     db
                     "PRAGMA wal_checkpoint(TRUNCATE);"))
                   (result
                    (match (sqlite-step-and-finalize statement)
                      (#(blocked? modified-page-count pages-moved-to-db)
                       (if (= blocked? 1)
                           (begin
                             (simple-format
                              (current-error-port)
                              "warning: wal checkpoint blocked\n")
                             #f)
                           (begin
                             (simple-format
                              (current-error-port)
                              "wal checkpoint completed (~A, ~A)\n"
                              modified-page-count
                              pages-moved-to-db)
                             #t))))))
              (sqlite-exec db "PRAGMA busy_timeout = 5000;")

              result)))
         #t))

   (call-with-duration-metric
    metrics-registry
    "datastore_optimize_duration_seconds"
    (lambda ()
      (sqlite-exec db
                   "
PRAGMA analysis_limit=1000;
PRAGMA optimize;")
      #t))))

(define-method (datastore-optimize
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (db-optimize
      db
      (slot-ref datastore 'database-file)
      (slot-ref datastore 'metrics-registry)))))

(define-method (datastore-spawn-fibers
                (datastore <sqlite-datastore>))
  (spawn-fiber
   (lambda ()
     (while #t
       (sleep (* 60 3))                 ; 3 minutes
       (with-exception-handler
           (lambda (exn)
             (simple-format (current-error-port)
                            "exception when performing WAL checkpoint: ~A\n"
                            exn))
         (lambda ()
           (with-time-logging
               "performing regular database maintenance"
             (let loop ((result (datastore-optimize datastore))
                        (retry-count 0))
               (if result
                   #t
                   (if (< retry-count 300)
                       (begin
                         (sleep 0.1)
                         (loop (datastore-optimize datastore)
                               (+ 1 retry-count)))
                       (error "unable to perform WAL checkpoint"))))))
         #:unwind? #t)))
   #:parallel? #t))

(define-method (datastore-initialise-metrics!
                (datastore <sqlite-datastore>))
  (define registry
    (slot-ref datastore 'metrics-registry))

  (let ((setup-failures-total
         (make-gauge-metric registry
                            "setup_failures_total"
                            #:labels '(agent_id reason))))

    (letpar& ((setup-failure-counts
               (with-time-logging "counting setup failures"
                 (datastore-count-setup-failures datastore))))

      (for-each (match-lambda
                  (((agent-id reason) . count)
                   (metric-set setup-failures-total
                               count
                               #:label-values
                               `((agent_id . ,agent-id)
                                 (reason   . ,reason)))))
                setup-failure-counts)))
  #t)

(define-method (datastore-update-metrics!
                (datastore <sqlite-datastore>))
  (let* ((db-filename
          (slot-ref datastore 'database-file))
         (db-wal-filename
          (string-append db-filename "-wal"))

         (registry
          (slot-ref datastore 'metrics-registry))
         (builds-total
          (or (metrics-registry-fetch-metric registry "builds_total")
              (make-gauge-metric registry
                                 "builds_total"
                                 #:labels '(system))))
         (build-results-total
          (or (metrics-registry-fetch-metric registry "build_results_total")
              (make-gauge-metric registry
                                 "build_results_total"
                                 #:labels '(agent_id result))))
         (db-bytes
          (or (metrics-registry-fetch-metric registry
                                             "datastore_bytes")
              (make-gauge-metric
               registry "datastore_bytes"
               #:docstring "Size of the SQLite database file")))
         (db-wal-bytes
          (or (metrics-registry-fetch-metric registry
                                             "datastore_wal_bytes")
              (make-gauge-metric
               registry "datastore_wal_bytes"
               #:docstring "Size of the SQLite Write Ahead Log file"))))

    (letpar& ((build-counts
               (datastore-count-builds datastore))
              (build-result-counts
               (datastore-count-build-results datastore)))
      (for-each (match-lambda
                  ((system . count)
                   (metric-set builds-total
                               count
                               #:label-values
                               `((system . ,system)))))
                build-counts)
      (for-each (match-lambda
                  (((agent-id result) . count)
                   (metric-set build-results-total
                               count
                               #:label-values
                               `((agent_id . ,agent-id)
                                 (result   . ,result)))))
                build-result-counts))

    (metric-set db-bytes (stat:size (stat db-filename)))
    (metric-set db-wal-bytes (stat:size (stat db-wal-filename))))
  #t)

(define (call-with-time-tracking datastore thing thunk)
  (define registry (slot-ref datastore 'metrics-registry))
  (define metric-name
    (string-append "datastore_" thing "_duration_seconds"))

  (if registry
      (let* ((metric
              (or (metrics-registry-fetch-metric registry metric-name)
                  (make-histogram-metric registry
                                         metric-name)))
             (start-time (get-internal-real-time)))
        (call-with-values
            thunk
          (lambda vals
            (metric-observe metric
                            (/ (- (get-internal-real-time) start-time)
                               internal-time-units-per-second))
            (apply values vals))))
      (thunk)))

(define (call-with-worker-thread/delay-logging channel proc)
  (call-with-worker-thread channel
                           proc
                           #:duration-logger
                           (lambda (duration)
                             (log-delay proc duration))))

(define-exception-type &transaction-rollback-exception &exception
  make-transaction-rollback-exception
  transaction-rollback-exception?
  (return-value transaction-rollback-exception-return-value))

(define %current-transaction-proc
  (make-parameter #f))

(define* (datastore-call-with-transaction datastore proc
                                          #:key
                                          readonly?
                                          (immediate? (not readonly?))
                                          duration-metric-name)
  (define (run-proc-within-transaction db)
    (define (attempt-begin)
      (with-exception-handler
          (lambda (exn)
            (match (exception-args exn)
              (('sqlite-exec 5 msg)
               (simple-format
                (current-error-port)
                "warning: issue starting transaction (code: 5, proc: ~A): ~A\n"
                proc msg)
               #f)
              (_
               (simple-format (current-error-port)
                              "exception starting transaction\n")
               (raise-exception exn))))
        (lambda ()
          (sqlite-exec db (if immediate?
                              "BEGIN IMMEDIATE TRANSACTION;"
                              "BEGIN TRANSACTION;"))
          #t)
        #:unwind? #t))

    (define (attempt-commit)
      (with-exception-handler
          (lambda (exn)
            (match (exception-args exn)
              (('sqlite-exec 5 msg)
               (simple-format
                (current-error-port)
                "warning: attempt commit (code: 5, proc: ~A): ~A\n"
                proc msg)
               #f)
              (_
               (simple-format (current-error-port)
                              "exception committing transaction\n")
               (raise-exception exn))))
        (lambda ()
          (sqlite-exec db "COMMIT TRANSACTION;")
          #t)
        #:unwind? #t))

    (if (attempt-begin)
        (call-with-values
            (lambda ()
              (with-exception-handler
                  (lambda (exn)
                    (if (transaction-rollback-exception? exn)
                        (begin
                          (sqlite-exec db "ROLLBACK TRANSACTION;")
                          (transaction-rollback-exception-return-value exn))
                        (begin
                          (simple-format (current-error-port)
                                         "error: sqlite rolling back transaction (~A)\n"
                                         exn)
                          (sqlite-exec db "ROLLBACK TRANSACTION;")
                          (raise-exception exn))))
                (lambda ()
                  (parameterize ((%current-transaction-proc proc))
                    (proc-with-duration-timing db)))
                #:unwind? #t))
          (lambda vals
            (let loop ((success? (attempt-commit)))
              (if success?
                  (apply values vals)
                  (loop (attempt-commit))))))

        ;; Database is busy, so retry
        (run-proc-within-transaction db)))

  (define (proc-with-duration-timing db)
    (let ((start-time (get-internal-real-time)))
      (call-with-values
          (lambda ()
            (if duration-metric-name
                (call-with-time-tracking
                 datastore
                 duration-metric-name
                 (lambda ()
                   (proc db)))
                (proc db)))
        (lambda vals
          (let ((duration-seconds
                 (/ (- (get-internal-real-time) start-time)
                    internal-time-units-per-second)))
            (when (and (not readonly?)
                       (> duration-seconds 2))
              (display
               (format
                #f
                "warning: ~a:\n  took ~4f seconds in transaction\n"
                proc
                duration-seconds)
               (current-error-port)))

            (cons duration-seconds vals))))))

  (match (call-with-worker-thread
          (slot-ref datastore (if readonly?
                                  'worker-reader-thread-channel
                                  'worker-writer-thread-channel))
          (lambda (db)
            (if (%current-transaction-proc)
                (proc-with-duration-timing db) ; already in transaction
                (run-proc-within-transaction db))))
    ((duration vals ...)
     (log-delay proc duration)

     (apply values vals))))

(define-method (datastore-find-agent
                (datastore <sqlite-datastore>)
                uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT description FROM agents WHERE id = :id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:id uuid)

       (let ((result
              (match (sqlite-map
                      (match-lambda
                        (#(description)
                         `((description . ,description))))
                      statement)
                (() #f)
                ((agent) agent))))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-find-agent-by-name
                (datastore <sqlite-datastore>)
                name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT id FROM agents WHERE name = :name"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:name name)

       (let ((result
              (match (sqlite-map
                      (match-lambda
                        (#(id) id))
                      statement)
                (() #f)
                ((agent) agent))))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-insert-dynamic-auth-token
                (datastore <sqlite-datastore>)
                token)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
INSERT INTO dynamic_auth_tokens (token) VALUES (:token)"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:token token)

       (sqlite-step-and-reset statement)))))

(define-method (datastore-dynamic-auth-token-exists?
                (datastore <sqlite-datastore>)
                token)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT 1 FROM dynamic_auth_tokens WHERE token = :token"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:token token)

       (let ((result
              (match (sqlite-map
                      (match-lambda
                        (#(1) #t))
                      statement)
                ((#t) #t)
                (() #f))))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-fetch-agent-tags
                (datastore <sqlite-datastore>)
                agent-id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT key, value
FROM tags
INNER JOIN agent_tags
  ON tags.id = agent_tags.tag_id
WHERE agent_tags.agent_id = :agent_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:agent_id agent-id)

       (let ((result
              (list->vector
               (sqlite-fold
                (lambda (row result)
                  (match row
                    (#(key value)
                     `((,key . ,value)
                       ,@result))))
                '()
                statement))))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-new-agent
                (datastore <sqlite-datastore>)
                uuid
                name
                description)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (insert-agent db uuid name description)))
  #t)

(define-method (datastore-list-agents
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT id, name, description, active FROM agents ORDER BY id"
             #:cache? #t)))

       (let ((agents (sqlite-map
                       (match-lambda
                         (#(id name description active)
                          `((uuid        . ,id)
                            (name        . ,name)
                            (description . ,description)
                            (active      . ,(= active 1)))))
                       statement)))
         (sqlite-reset statement)

         agents)))))

(define-method (datastore-set-agent-active
                (datastore <sqlite-datastore>)
                agent-uuid
                active?)
  (unless (boolean? active?)
    (error "datastore-set-agent-active called with non-boolean"))

  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
UPDATE agents SET active = :active WHERE id = :uuid"
             #:cache? #f)))

       (sqlite-bind-arguments statement
                              #:uuid agent-uuid
                              #:active (if active? 1 0))

       (sqlite-step-and-finalize statement))))
  active?)

(define-method (datastore-find-agent-status
                (datastore <sqlite-datastore>)
                agent-id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT status, load_average_1min, timestamp, processor_count
FROM agent_status
WHERE agent_id = :agent_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:agent_id agent-id)

       (match (sqlite-step-and-reset statement)
         (#(status 1min_load_average timestamp
                   processor_count)
          `((status            . ,status)
            (1min_load_average . ,1min_load_average)
            (timestamp         . ,timestamp)
            (processor_count   . ,processor_count)))
         (_ #f))))))

(define-method (datastore-update-agent-status
                (datastore <sqlite-datastore>)
                agent-uuid
                status
                1min-load-average
                processor-count)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
DELETE FROM agent_status WHERE agent_id = :uuid"
             #:cache? #t)))

       (sqlite-bind-arguments statement
                              #:uuid agent-uuid)
       (sqlite-step-and-reset statement))

     (let ((statement
            (sqlite-prepare
             db
             "
INSERT INTO agent_status (agent_id, status, load_average_1min, processor_count)
  VALUES (:uuid, :status, :load, :processor_count)"
             #:cache? #t)))

       (sqlite-bind-arguments statement
                              #:uuid agent-uuid
                              #:status status
                              #:load 1min-load-average
                              #:processor_count processor-count)

       (sqlite-step-and-reset statement))))

  #t)

(define-method (datastore-new-agent-password
                (datastore <sqlite-datastore>)
                agent-uuid
                password)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (insert-agent-password db agent-uuid password)))
  #t)

(define-method (datastore-agent-password-exists?
                (datastore <sqlite-datastore>)
                uuid
                password)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT 1 FROM agent_passwords \
WHERE agent_id = :agent_id AND password = :password"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:agent_id uuid
        #:password password)

       (match (sqlite-step-and-reset statement)
         (#f #f)
         (#(1) #t))))))

(define-method (datastore-agent-list-passwords
                (datastore <sqlite-datastore>)
                uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT password FROM agent_passwords WHERE agent_id = :agent_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:agent_id uuid)

       (let ((result (sqlite-map
                      (match-lambda
                        (#(password) password))
                      statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-replace-agent-tags
                (datastore <sqlite-datastore>)
                agent-id
                tags)
  (datastore-call-with-transaction
   datastore
   (lambda (db)
     (let ((delete-agent-tags-statement
            (sqlite-prepare
             db
             "
DELETE FROM agent_tags WHERE agent_id = :agent_id"
             #:cache? #t))
           (insert-tag-statement
            (sqlite-prepare
             db
             "
INSERT INTO tags (\"key\", \"value\") VALUES (:tagkey, :tagvalue)"
             #:cache? #t))
           (find-tag-statement
            (sqlite-prepare
             db
             "
SELECT id FROM tags WHERE key = :tag_key AND value = :tag_value"
             #:cache? #t))
           (agent-tags-statement
            (sqlite-prepare
             db
             "
INSERT INTO agent_tags (agent_id, tag_id) VALUES (:agent_id, :tag_id)"
             #:cache? #t)))

       (define (tag->id key value)
         (sqlite-bind-arguments find-tag-statement
                                #:tag_key key
                                #:tag_value value)
         (match (sqlite-step-and-reset find-tag-statement)
           (#(id) id)
           (#f
            (sqlite-bind-arguments insert-tag-statement
                                   #:tagkey key
                                   #:tagvalue value)
            (sqlite-step-and-reset insert-tag-statement)
            (last-insert-rowid db))))

       (define (insert-tag key value)
         (sqlite-bind-arguments agent-tags-statement
                                #:agent_id agent-id
                                #:tag_id (tag->id key value))
         (sqlite-step-and-reset agent-tags-statement))

       (sqlite-bind-arguments delete-agent-tags-statement
                              #:agent_id agent-id)
       (sqlite-step-and-reset delete-agent-tags-statement)
       (for-each
        (match-lambda
          ((('key   . key)
            ('value . value))
           (insert-tag key value))
          ((key . value) (insert-tag key value)))
        (if (vector? tags)
            (vector->list tags)
            tags)))))
  #t)

(define %derivation-outputs-cache
  (make-hash-table))

(define-method (datastore-store-derivation
                (datastore <sqlite-datastore>)
                derivation)
  (datastore-call-with-transaction
   datastore
   (lambda (db)
     (insert-derivation-and-return-outputs db derivation)
     (hash-clear! %derivation-outputs-cache))
   #:duration-metric-name "store_derivation")

  #t)

(define-method (datastore-build-exists-for-derivation-outputs?
                (datastore <sqlite-datastore>)
                derivation)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT 1
FROM derivation_outputs
INNER JOIN derivation_outputs AS all_derivation_outputs
  ON all_derivation_outputs.output_id = derivation_outputs.output_id
INNER JOIN derivations
  ON derivations.id = all_derivation_outputs.derivation_id
INNER JOIN builds
  ON builds.derivation_id = derivations.id
WHERE derivation_outputs.derivation_id = :derivation_id
  AND (SELECT system_id FROM derivations WHERE id = :derivation_id) =
      derivations.system_id
  AND builds.canceled = 0
"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:derivation_id (db-find-derivation-id db derivation))

       (if (sqlite-step-and-reset statement)
           #t
           #f)))))

(define-method (datastore-build-required-by-another?
                (datastore <sqlite-datastore>)
                uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT 1
FROM builds
INNER JOIN derivation_outputs
  ON builds.derivation_id = derivation_outputs.derivation_id
INNER JOIN outputs ON derivation_outputs.output_id = outputs.id
INNER JOIN derivation_outputs AS all_derivation_outputs
  ON outputs.id = all_derivation_outputs.output_id
INNER JOIN derivation_inputs
  ON derivation_inputs.derivation_output_id = all_derivation_outputs.id
INNER JOIN derivations AS dependent_derivations
  ON dependent_derivations.id = derivation_inputs.derivation_id
INNER JOIN builds AS dependent_builds
  ON dependent_builds.derivation_id = dependent_derivations.id
 AND dependent_builds.processed = 0
 AND dependent_builds.canceled = 0
WHERE builds.uuid = :uuid
"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:uuid uuid)

       (if (sqlite-step-and-reset statement)
           #t
           #f)))))

(define-method (datastore-list-related-derivations-with-no-build-for-outputs
                (datastore <sqlite-datastore>)
                derivation)
  (define (get-input-derivations-with-no-builds db derivation-id)
    (let ((statement
           (sqlite-prepare
            db
            "
SELECT derivations.id
FROM derivation_inputs
INNER JOIN derivation_outputs
  ON derivation_inputs.derivation_output_id = derivation_outputs.id
INNER JOIN derivations
  ON derivations.id = derivation_outputs.derivation_id
WHERE derivation_inputs.derivation_id = :derivation_id
  AND NOT EXISTS (
    SELECT 1
    FROM builds
    INNER JOIN derivation_outputs AS other_derivation_derivation_outputs
      ON other_derivation_derivation_outputs.derivation_id =
         builds.derivation_id
    INNER JOIN derivations AS other_derivations
      ON other_derivation_derivation_outputs.derivation_id =
         other_derivations.id
    INNER JOIN derivation_outputs AS all_other_derivation_derivation_outputs
      ON all_other_derivation_derivation_outputs.output_id =
         other_derivation_derivation_outputs.output_id
    WHERE all_other_derivation_derivation_outputs.derivation_id = derivations.id
      AND other_derivations.system_id = derivations.system_id
      AND builds.canceled = 0
  )
"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:derivation_id derivation-id)

      (let ((result (sqlite-map
                     (match-lambda
                       (#(derivation-id) derivation-id))
                     statement)))
        (sqlite-reset statement)

        result)))

  (define (get-derivation-name db derivation-id)
    (let ((statement
           (sqlite-prepare
            db
            "
SELECT name FROM derivations WHERE id = :id"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:id derivation-id)

      (match (sqlite-step-and-reset statement)
        (#(name) name))))

  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let loop ((derivation-ids (list (db-find-derivation-id db derivation)))
                (result '()))
       (let ((new-ids
              (delete-duplicates!
               (append-map! (lambda (derivation-id)
                              (get-input-derivations-with-no-builds
                               db
                               derivation-id))
                            derivation-ids))))

         (if (null? new-ids)
             (map (lambda (derivation-id)
                    (get-derivation-name db derivation-id))
                  (delete-duplicates! result))
             (loop new-ids
                   (append! result new-ids))))))))

(define-method (datastore-list-failed-builds-with-blocking-count
                (datastore <sqlite-datastore>)
                .
                args)
  (apply
   (lambda* (system #:key include-cancelled?)
     (call-with-worker-thread
      (slot-ref datastore 'worker-reader-thread-channel)
      (lambda (db)
        (let ((statement
               (sqlite-prepare
                db
                (string-append
                 "
SELECT *
FROM (
  SELECT
    builds.uuid,
    derivations.name,
    (
      WITH RECURSIVE related_derivations(id) AS (
        VALUES(builds.derivation_id)
      UNION
        SELECT derivation_inputs.derivation_id
        FROM derivation_outputs
        INNER JOIN related_derivations
          ON derivation_outputs.derivation_id = related_derivations.id
        INNER JOIN derivation_inputs
          ON derivation_outputs.id = derivation_inputs.derivation_output_id
      )
      SELECT COUNT(DISTINCT blocked_builds.uuid)
      FROM related_derivations
      INNER JOIN builds AS blocked_builds
        ON related_derivations.id = blocked_builds.derivation_id
      AND blocked_builds.processed = 0
      AND blocked_builds.canceled = 0
    ) AS blocking_count
  FROM builds
  INNER JOIN derivations
    ON derivations.id = builds.derivation_id
  LEFT JOIN build_results
    ON builds.id = build_results.build_id
  WHERE
    (
      (
        builds.processed = 1
        AND build_results.result = 'failure'"
                 (if include-cancelled?
                     "
      ) OR (
        builds.canceled = 1"
                     "")
                 "
      )
    )"
                 (if system
                     "
    AND derivations.system_id = :system_id"
                     "")
                 "
    AND NOT EXISTS (
      SELECT 1
      FROM derivation_outputs
      INNER JOIN derivation_outputs AS other_build_derivation_outputs
        ON derivation_outputs.output_id = other_build_derivation_outputs.output_id
      INNER JOIN builds AS other_builds
        ON other_build_derivation_outputs.derivation_id = other_builds.derivation_id
      INNER JOIN build_results AS other_build_results
        ON other_builds.id = other_build_results.build_id
      WHERE derivation_outputs.derivation_id = builds.derivation_id
      AND other_build_results.result = 'success'
    )
  ) AS data
  WHERE blocking_count > 0
  ORDER BY 3 DESC, 2, 1")
                #:cache? #t)))
          (when system
            (sqlite-bind-arguments statement
                                   #:system (db-system->system-id db system)))

          (let ((result (sqlite-map
                         (match-lambda
                           (#(uuid derivation-name blocked-count)
                            `((uuid            . ,uuid)
                              (derivation_name . ,derivation-name)
                              (blocked_count   . ,blocked-count))))
                         statement)))
            (sqlite-reset statement)

            result)))))
   args))

(define-method (datastore-list-builds-for-derivation-recursive-inputs
                (datastore <sqlite-datastore>)
                derivation)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
WITH RECURSIVE related_derivations(id) AS (
  VALUES(:derivation_id)
UNION
  SELECT derivation_outputs.derivation_id
  FROM derivation_outputs
  INNER JOIN derivation_inputs
    ON derivation_outputs.id = derivation_inputs.derivation_output_id
  INNER JOIN related_derivations
    ON related_derivations.id = derivation_inputs.derivation_id
)
SELECT builds.uuid
FROM builds
INNER JOIN related_derivations
  ON related_derivations.id = builds.derivation_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:derivation_id (db-find-derivation-id db derivation))

       (let ((result (sqlite-map
                      (match-lambda
                        (#(uuid) uuid))
                      statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-insert-build-tags
                (datastore <sqlite-datastore>)
                build-uuid
                tags)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((insert-tag-statement
            (sqlite-prepare
             db
             "
INSERT INTO tags (\"key\", \"value\") VALUES (:tagkey, :tagvalue)"
             #:cache? #t))
           (find-tag-statement
            (sqlite-prepare
             db
             "
SELECT id FROM tags WHERE key = :tag_key AND value = :tag_value"
             #:cache? #t))
           (build-tags-statement
            (sqlite-prepare
             db
             "
INSERT INTO build_tags (build_id, tag_id) VALUES (:build_id, :tag_id)"
             #:cache? #t)))

       (define (tag->id key value)
         (sqlite-bind-arguments find-tag-statement
                                #:tag_key key
                                #:tag_value value)
         (match (sqlite-step-and-reset find-tag-statement)
           (#(id) id)
           (#f
            (sqlite-bind-arguments insert-tag-statement
                                   #:tagkey key
                                   #:tagvalue value)
            (sqlite-step-and-reset insert-tag-statement)
            (last-insert-rowid db))))

       (for-each
        (match-lambda
          ((key . value)
           (sqlite-bind-arguments build-tags-statement
                                  #:build_id (db-find-build-id db build-uuid)
                                  #:tag_id (tag->id key value))
           (sqlite-step-and-reset build-tags-statement)))
        (if (vector? tags)
            (vector->list tags)
            tags)))))
  #t)

(define-method (datastore-cancel-build
                (datastore <sqlite-datastore>)
                uuid)
  (datastore-call-with-transaction
   datastore
   (lambda (db)
     (let ((statement (sqlite-prepare
                       db
                       "
UPDATE builds SET canceled = 1 WHERE uuid = :uuid"
                       #:cache? #t)))
       (sqlite-bind-arguments
        statement
        #:uuid uuid)

       (sqlite-step-and-reset statement))

     (let ((statement (sqlite-prepare
                       db
                       "
DELETE FROM unprocessed_builds_with_derived_priorities
  WHERE build_id = (
    SELECT id FROM builds WHERE uuid = :uuid
)"
                       #:cache? #t)))
       (sqlite-bind-arguments
        statement
        #:uuid uuid)

       (sqlite-step-and-reset statement))))
  #t)

(define (db-get-build-priority db build-id)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT priority FROM builds WHERE id = :build_id"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:build_id build-id)

    (vector-ref (sqlite-step-and-reset statement)
                0)))

(define (get-derived-priority db build-id)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT max(dependent_unprocessed_builds_with_derived_priorities.derived_priority)
FROM builds
INNER JOIN derivation_outputs
  ON builds.derivation_id = derivation_outputs.derivation_id
INNER JOIN derivation_outputs AS all_derivation_outputs
  ON all_derivation_outputs.output_id = derivation_outputs.output_id
INNER JOIN derivation_inputs
  ON derivation_inputs.derivation_output_id = all_derivation_outputs.id
INNER JOIN builds AS dependent_builds
  ON dependent_builds.processed = 0
 AND dependent_builds.canceled = 0
 AND dependent_builds.derivation_id = derivation_inputs.derivation_id
INNER JOIN unprocessed_builds_with_derived_priorities
  AS dependent_unprocessed_builds_with_derived_priorities
  ON dependent_builds.id =
     dependent_unprocessed_builds_with_derived_priorities.build_id
WHERE builds.id = :build_id"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:build_id build-id)

    (match (sqlite-step-and-reset statement)
      (#(#f)
       (db-get-build-priority db build-id))
      (#(derived-priority)
       derived-priority))))

(define (update-unprocessed-builds-with-higher-derived-priorities
         db
         build-id
         derived-priority)
  ;; Recursively find builds for all missing outputs that this build takes
  ;; as inputs. These builds should have a derived priority of at least the
  ;; derived priority of this build
  (let ((find-builds-statement
         (sqlite-prepare
          db
          "
WITH RECURSIVE relevant_builds (id) AS (
  VALUES (:build_id)
UNION
  SELECT builds.id
  FROM relevant_builds
  INNER JOIN builds AS relevant_builds_full
    ON relevant_builds.id = relevant_builds_full.id
  INNER JOIN derivation_inputs
    ON relevant_builds_full.derivation_id = derivation_inputs.derivation_id
  INNER JOIN derivation_outputs
    ON derivation_inputs.derivation_output_id = derivation_outputs.id
  INNER JOIN unbuilt_outputs
    ON unbuilt_outputs.output_id = derivation_outputs.output_id
  INNER JOIN derivation_outputs AS all_derivation_outputs
    ON all_derivation_outputs.output_id = unbuilt_outputs.output_id
  INNER JOIN builds
    ON builds.processed = 0
   AND builds.derivation_id = all_derivation_outputs.derivation_id
)
SELECT build_id
FROM relevant_builds
INNER JOIN unprocessed_builds_with_derived_priorities
  ON relevant_builds.id = unprocessed_builds_with_derived_priorities.build_id
WHERE unprocessed_builds_with_derived_priorities.derived_priority <
      :derived_priority"
          #:cache? #t))

        (update-derived-priority-statement
         (sqlite-prepare
          db
          "
UPDATE unprocessed_builds_with_derived_priorities
SET derived_priority = :derived_priority
WHERE build_id = :build_id"
          #:cache? #t)))

    (sqlite-bind-arguments
     find-builds-statement
     #:build_id build-id
     #:derived_priority derived-priority)

    (let ((builds-to-update
           (call-with-delay-logging
            (lambda ()
              (sqlite-map (lambda (row)
                            (vector-ref row 0))
                          find-builds-statement)))))
      (sqlite-reset find-builds-statement)

      (map (lambda (id)
             (sqlite-bind-arguments update-derived-priority-statement
                                    #:build_id id
                                    #:derived_priority derived-priority)
             (sqlite-step-and-reset
              update-derived-priority-statement))
           builds-to-update))))

(define (datastore-update-unprocessed-builds-with-lower-derived-priorities
         datastore
         uuid
         explicit-priority-lower-bound)
  (define builds-to-consider
    (call-with-worker-thread/delay-logging
     (slot-ref datastore 'worker-reader-thread-channel)
     (lambda (db)
       ;; Recursively find builds for all missing outputs that this build
       ;; takes as inputs. The order is important here, since we want to
       ;; compute the new derived priorities, starting with the root of the
       ;; graph (the build-id build) and working down, as each level is
       ;; dependent on the derived priorities of the level above.
       (let* ((build-id
               (db-find-build-id db uuid))
              (priority-lower-bound
               (or explicit-priority-lower-bound
                   (get-derived-priority db build-id)))
              (statement
               (sqlite-prepare
                db
                "
WITH RECURSIVE relevant_builds (id, level) AS (
  VALUES (:build_id, 0)
UNION
  SELECT builds.id, level+1
  FROM relevant_builds
  INNER JOIN builds AS relevant_builds_full
    ON relevant_builds.id = relevant_builds_full.id
  INNER JOIN derivation_inputs
    ON relevant_builds_full.derivation_id = derivation_inputs.derivation_id
  INNER JOIN derivation_outputs
    ON derivation_inputs.derivation_output_id = derivation_outputs.id
  INNER JOIN unbuilt_outputs
    ON unbuilt_outputs.output_id = derivation_outputs.output_id
  INNER JOIN derivation_outputs AS all_derivation_outputs
    ON all_derivation_outputs.output_id = unbuilt_outputs.output_id
  INNER JOIN builds
    ON builds.processed = 0
   AND builds.canceled = 0
   AND builds.derivation_id = all_derivation_outputs.derivation_id
)
SELECT build_id, MAX(level)
FROM relevant_builds
INNER JOIN unprocessed_builds_with_derived_priorities
  ON relevant_builds.id = unprocessed_builds_with_derived_priorities.build_id
WHERE unprocessed_builds_with_derived_priorities.derived_priority > :priority
GROUP BY build_id
ORDER BY level ASC"
                #:cache? #t)))

         (sqlite-bind-arguments
          statement
          #:build_id build-id
          #:priority priority-lower-bound)

         (let ((builds-to-consider
                (call-with-delay-logging
                 (lambda ()
                   (sqlite-map (lambda (row)
                                 (vector-ref row 0))
                               statement)))))
           (sqlite-reset statement)

           builds-to-consider)))))

  (for-each
   (lambda (build-id)
     (datastore-call-with-transaction
      datastore
      (lambda (db)
        (let ((statement
               (sqlite-prepare
                db
                "
UPDATE unprocessed_builds_with_derived_priorities
SET derived_priority = :derived_priority
WHERE build_id = :build_id"
                #:cache? #t))

              (new-derived-priority
               (get-derived-priority db build-id)))

          (sqlite-bind-arguments statement
                                 #:build_id build-id
                                 #:derived_priority new-derived-priority)
          (sqlite-step-and-reset statement)))))
   builds-to-consider))


(define-method (datastore-update-build-priority
                (datastore <sqlite-datastore>)
                .
                rest)

  (define (db-get-unprocessed-builds-entry db build-id)
    (let ((statement
           (sqlite-prepare
            db
            "
SELECT derived_priority, all_inputs_built
FROM unprocessed_builds_with_derived_priorities
WHERE build_id = :build_id"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:build_id build-id)

      (match (sqlite-step-and-reset statement)
        (#(derived-priority all-inputs-built)
         (values derived-priority
                 (= 1 all-inputs-built))))))

  (define (db-update-build-priority db build-id new-priority)
    (let ((statement
           (sqlite-prepare
            db
            "
UPDATE builds
SET priority = :priority
WHERE id = :build_id"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:priority new-priority
       #:build_id build-id)

      (sqlite-step-and-reset statement))
    #t)

  (define (db-update-build-derived-priority db build-id new-derived-priority)
    (let ((statement
           (sqlite-prepare
            db
            "
UPDATE unprocessed_builds_with_derived_priorities
SET derived_priority = :derived_priority
WHERE build_id = :build_id"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:derived_priority new-derived-priority
       #:build_id build-id)

      (sqlite-step-and-reset statement))
    #t)

  (apply
   (lambda* (uuid new-priority
                  #:key skip-updating-other-build-derived-priorities)
     (let ((build-id
            old-priority
            (call-with-worker-thread
             (slot-ref datastore 'worker-reader-thread-channel)
             (lambda (db)
               (let ((build-id
                      (db-find-build-id db uuid)))
                 (values
                  build-id
                  (db-get-build-priority db build-id)))))))

       (unless (eq? old-priority new-priority)
         (datastore-call-with-transaction
          datastore
          (lambda (db)
            (db-update-build-priority db
                                      build-id
                                      new-priority)

            (let ((old-derived-priority
                   all-inputs-built?
                   (db-get-unprocessed-builds-entry
                    db
                    build-id))
                  (new-derived-priority
                   (max new-priority
                        (get-derived-priority db build-id))))

              (unless (eq? old-derived-priority
                           new-derived-priority)
                (db-update-build-derived-priority db
                                                  build-id
                                                  new-derived-priority))

              (unless (or all-inputs-built?
                          skip-updating-other-build-derived-priorities)
                (when (> new-derived-priority
                         old-derived-priority)
                  (update-unprocessed-builds-with-higher-derived-priorities
                   db
                   build-id
                   new-derived-priority))))))

         ;; Potentially reduce the derived priorities of builds feeding in to
         ;; this build. It isn't ideal that this happens outside of the above
         ;; transaction, but since there could be lots of builds to update the
         ;; priority of, and a new derived priority has to be calculated for
         ;; each one, it's better to handle it afterwards.
         (when (< new-priority
                  old-priority)
           (datastore-update-unprocessed-builds-with-lower-derived-priorities
            datastore
            uuid
            ;; This acts as a lower bound, since builds feeding in to this
            ;; build can't have a derived priority less than new-priority
            new-priority))))
     #t)
   rest))

(define-method (datastore-remove-build-from-allocation-plan
                (datastore <sqlite-datastore>)
                uuid)
  (define (update-build-allocation-plan-metrics)
    (let ((allocation-plan-metric
           (metrics-registry-fetch-metric
            (slot-ref datastore 'metrics-registry)
            "build_allocation_plan_total")))
      (for-each
       (match-lambda
         ((agent-id . count)
          (metric-set allocation-plan-metric
                      count
                      #:label-values
                      `((agent_id . ,agent-id)))))
       (datastore-count-build-allocation-plan-entries datastore))))

  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement (sqlite-prepare
                       db
                       "
DELETE FROM build_allocation_plan WHERE build_id = :build_id"
                       #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db uuid))

       (sqlite-step-and-reset statement)

       (unless (= 0 (changes-count db))
         (update-build-allocation-plan-metrics)))))
  #t)

(define-method (datastore-count-build-results
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT agent_id, result, count FROM build_results_counts"
             #:cache? #t)))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(agent_id result count)
                  (cons (list agent_id result) count)))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-insert-build-result
                (datastore <sqlite-datastore>)
                build-uuid agent-id result failure-reason)
  (define (increment-count db)
    (let ((statement
           (sqlite-prepare
            db
            "
UPDATE build_results_counts
SET count = count + 1
WHERE agent_id = :agent_id
  AND result = :result
RETURNING count"
            #:cache? #t))
          (insert-statement
           (sqlite-prepare
            db
            "
INSERT INTO build_results_counts (agent_id, result, count)
VALUES (:agent_id, :result, 1)"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:agent_id agent-id
       #:result result)

      (match (sqlite-step-and-reset statement)
        (#(count) #t)
        (#f
         (sqlite-bind-arguments
          insert-statement
          #:agent_id agent-id
          #:result result)

         (sqlite-step-and-reset insert-statement)

         #t))))

  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
INSERT INTO build_results (
  build_id, agent_id, result, failure_reason
) VALUES (:build_id, :agent_id, :result, :failure_reason)"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid)
        #:agent_id agent-id
        #:result result
        #:failure_reason failure-reason)

       (sqlite-step-and-reset statement))

     (increment-count db)))
  #t)

(define-method (datastore-update-unprocessed-builds-for-build-success
                (datastore <sqlite-datastore>)
                build-uuid)
  (define (all-inputs-built? db build-id)
    (let ((statement
           (sqlite-prepare
            db
            "
SELECT 1
FROM derivation_inputs
INNER JOIN derivation_outputs
  ON derivation_inputs.derivation_output_id = derivation_outputs.id
INNER JOIN unbuilt_outputs
  ON unbuilt_outputs.output_id = derivation_outputs.output_id
INNER JOIN builds
  ON builds.derivation_id = derivation_inputs.derivation_id
WHERE builds.id = :build_id
LIMIT 1"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:build_id build-id)

      (match (sqlite-step-and-reset statement)
        (#f #t)
        (#(1) #f))))

  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((builds-statement
            (sqlite-prepare
             db
             "
SELECT DISTINCT unprocessed_builds.id
FROM builds
INNER JOIN derivation_outputs
  ON builds.derivation_id = derivation_outputs.derivation_id
INNER JOIN derivation_outputs AS all_derivation_outputs
  ON all_derivation_outputs.output_id = derivation_outputs.output_id
INNER JOIN derivation_inputs
  ON derivation_inputs.derivation_output_id = all_derivation_outputs.id
INNER JOIN builds AS unprocessed_builds
  ON unprocessed_builds.processed = 0
 AND unprocessed_builds.derivation_id = derivation_inputs.derivation_id
INNER JOIN unprocessed_builds_with_derived_priorities
  ON unprocessed_builds_with_derived_priorities.build_id = unprocessed_builds.id
 AND unprocessed_builds_with_derived_priorities.all_inputs_built = 0
WHERE builds.id = :build_id"
             #:cache? #t))

           (update-statement
            (sqlite-prepare
             db
             "
UPDATE unprocessed_builds_with_derived_priorities
SET all_inputs_built = 1
WHERE build_id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments builds-statement
                              #:build_id (db-find-build-id db build-uuid))

       (sqlite-fold
        (lambda (row result)
          (match row
            (#(build-id)
             (when (all-inputs-built? db build-id)
               (sqlite-bind-arguments update-statement
                                      #:build_id build-id)
               (sqlite-step-and-reset update-statement))))
          #f)
        #f
        builds-statement)
       (sqlite-reset builds-statement)))))

(define-method (datastore-remove-build-allocation
                (datastore <sqlite-datastore>)
                build-uuid agent-id)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
DELETE FROM allocated_builds
 WHERE build_id = :build_id
   AND agent_id = :agent_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid)
        #:agent_id agent-id)

       (sqlite-step-and-reset statement))))
  #t)

(define-method (datastore-mark-build-as-processed
                (datastore <sqlite-datastore>)
                build-uuid end-time)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
UPDATE builds
SET processed = 1, end_time = :end_time
WHERE id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:end_time (or end-time "")
        #:build_id (db-find-build-id db build-uuid))

       (sqlite-step-and-reset statement))

     (let ((statement (sqlite-prepare
                       db
                       "
DELETE FROM unprocessed_builds_with_derived_priorities
  WHERE build_id = (
    SELECT id FROM builds WHERE uuid = :uuid
)"
                       #:cache? #t)))
       (sqlite-bind-arguments
        statement
        #:uuid build-uuid)

       (sqlite-step-and-reset statement))))
  #t)

(define-method (datastore-delete-relevant-outputs-from-unbuilt-outputs
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
DELETE FROM unbuilt_outputs
WHERE output_id IN (
  SELECT derivation_outputs.output_id
  FROM derivation_outputs
  INNER JOIN builds
    ON builds.derivation_id = derivation_outputs.derivation_id
  WHERE builds.id = :build_id
)"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (sqlite-step-and-reset statement)
       #t))))

(define-method (datastore-store-output-metadata
                (datastore <sqlite-datastore>)
                build-uuid
                output-metadata)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (define (name->output-id name)
       (let ((statement
              (sqlite-prepare
               db
               "
SELECT derivation_outputs.id
FROM derivation_outputs
INNER JOIN builds
  ON builds.derivation_id = derivation_outputs.derivation_id
WHERE builds.uuid = :build_uuid AND derivation_outputs.name = :name"
               #:cache? #t)))

         (sqlite-bind-arguments
          statement
          #:build_uuid build-uuid
          #:name name)

         (match (sqlite-step-and-reset statement)
           (#(id) id))))

     (sqlite-exec
      db
      (string-append
       "
INSERT INTO output_metadata (build_id, derivation_output_id,
                             hash, size, store_references)
VALUES "
       (string-join
        (map (lambda (output)
               (simple-format
                #f "('~A', ~A, '~A', ~A, '~A')"
                (db-find-build-id db build-uuid)
                (name->output-id (assoc-ref output "name"))
                (assoc-ref output "hash")
                (assoc-ref output "size")
                (string-join
                 (vector->list (assoc-ref output "references"))
                 " ")))
             output-metadata)
        ", ")))
     #t)))

(define-method (datastore-store-build-start
                (datastore <sqlite-datastore>)
                build-uuid
                agent-id)

  (define (handle-inserting-unprocessed-hook-event db build-uuid agent-id)
    (insert-unprocessed-hook-event
     db
     "build-started"
     (list build-uuid agent-id)))

  (datastore-call-with-transaction
   datastore
   (lambda (db)
     (sqlite-exec
      db
      (string-append
       "
INSERT INTO build_starts (
  build_id, agent_id, start_time
) VALUES ("
       (number->string (db-find-build-id db build-uuid)) ", '"
       agent-id "', "
       "datetime('now')"
       ")"))

     (handle-inserting-unprocessed-hook-event db build-uuid agent-id)))

  #t)

(define-method (datastore-find-build-starts
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT start_time, agent_id
FROM build_starts
WHERE build_id = :build_id
ORDER BY start_time DESC"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(start_time agent_id)
                  `((start-time . ,(match (strptime "%F %T" start_time)
                                             ((parts . _) parts)))
                    (agent-id   . ,agent_id))))
               statement)))
         (sqlite-reset statement)

         result)))))

(define (insert-setup-failure-and-remove-allocation
         db
         build-id
         agent-id
         failure-reason)
  (sqlite-exec
   db
   (string-append
    "
DELETE FROM allocated_builds WHERE build_id = '"
    (number->string build-id)
    "' AND agent_id = '"
    agent-id
    "'"))

  (sqlite-exec
   db
   (string-append
    "
INSERT INTO setup_failures (
  build_id, agent_id, failure_reason
) VALUES ('"
    (number->string build-id) "', '"
    agent-id "', '"
    failure-reason
    "')"))

  (last-insert-rowid db))

(define-method (datastore-store-setup-failure/missing-inputs
                (datastore <sqlite-datastore>)
                build-uuid
                agent-id
                missing-inputs)
  (define (insert-missing-inputs db setup-failure-id missing-inputs)
    (sqlite-exec
     db
     (string-append
      "
INSERT INTO setup_failure_missing_inputs (
  setup_failure_id, missing_input_store_path
) VALUES "
      (string-join
       (map (lambda (missing-input)
              (simple-format #f "(~A, '~A')"
                             setup-failure-id missing-input))
            missing-inputs)
       ", "))))

  (define (handle-inserting-unprocessed-hook-event db build-uuid missing-inputs)
    (insert-unprocessed-hook-event
     db
     "build-missing-inputs"
     (list build-uuid
           missing-inputs)))

  (datastore-call-with-transaction
   datastore
   (lambda (db)
     (let* ((build-id (db-find-build-id db build-uuid))
            (setup-failure-id
             (insert-setup-failure-and-remove-allocation db
                                                         build-id
                                                         agent-id
                                                         "missing_inputs")))
       (insert-missing-inputs db setup-failure-id missing-inputs))
     ;; TODO This logic should be part of the coordinator, but it's here to be
     ;; inside the transaction
     (handle-inserting-unprocessed-hook-event db build-uuid missing-inputs))
   #:duration-metric-name "store_setup_failure_missing_inputs")

  (metric-increment
   (metrics-registry-fetch-metric (slot-ref datastore 'metrics-registry)
                                  "setup_failures_total")
   #:label-values `((agent_id . ,agent-id)
                    (reason   . "missing_inputs")))
  #t)

(define-method (datastore-list-setup-failure-missing-inputs
                (datastore <sqlite-datastore>)
                setup-failure-id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT missing_input_store_path
FROM setup_failure_missing_inputs
WHERE setup_failure_id = :id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:id setup-failure-id)

       (let ((result (sqlite-map
                      (match-lambda
                        (#(missing-input) missing-input))
                      statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-store-setup-failure
                (datastore <sqlite-datastore>)
                build-uuid
                agent-id
                failure-reason)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (insert-setup-failure-and-remove-allocation db
                                                 (db-find-build-id db build-uuid)
                                                 agent-id
                                                 failure-reason)

     (metric-increment
      (metrics-registry-fetch-metric (slot-ref datastore 'metrics-registry)
                                     "setup_failures_total")
      #:label-values `((agent_id . ,agent-id)
                       (reason   . ,failure-reason)))))
  #t)


(define-method (datastore-count-builds
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT system_id, count
FROM builds_counts"
             #:cache? #t)))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(system-id count)
                  (cons (db-system-id->system db system-id) count)))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-for-each-build
                (datastore <sqlite-datastore>)
                proc)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT uuid FROM builds ORDER BY id"
             #:cache? #t)))

       (let loop ((row (sqlite-step statement)))
         (match row
           (#(uuid)
            (proc uuid)

            (loop (sqlite-step statement)))
           (#f
            (sqlite-reset statement)

            #t)))))))

(define (db-find-build-id db uuid)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT id FROM builds WHERE uuid = :uuid"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:uuid uuid)

    (match (sqlite-step-and-reset statement)
      (#f #f)
      (#(id) id))))

(define-method (datastore-find-build
                (datastore <sqlite-datastore>)
                uuid)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT uuid, derivations.name, priority, processed, canceled, created_at, end_time
FROM builds
INNER JOIN derivations
  ON derivations.id = builds.derivation_id
WHERE uuid = :uuid"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:uuid uuid)

       (match (sqlite-step-and-reset statement)
         (#(uuid derivation_name priority processed canceled
                 created_at end_time)
          `((uuid            . ,uuid)
            (derivation-name . ,derivation_name)
            (priority        . ,priority)
            (processed       . ,(cond
                                 ((= 0 processed) #f)
                                 ((= 1 processed) #t)
                                 (else
                                  (error "unknown processed value"))))
            (canceled        . ,(cond
                                 ((= 0 canceled) #f)
                                 ((= 1 canceled) #t)
                                 (else
                                  (error "unknown canceled value"))))
            (created-at      . ,(if (string? created_at)
                                    (match (strptime "%F %T" created_at)
                                      ((parts . _) parts))
                                    #f))
            (end-time        . ,(if (string? end_time)
                                    (match (strptime "%F %T" end_time)
                                      ((parts . _) parts))
                                    #f)))))))))

(define-method (datastore-list-builds
                (datastore <sqlite-datastore>)
                .
                rest)
  (define* (list-builds #:key
                        (tags '())
                        (not-tags '())
                        (systems '())
                        (not-systems '())
                        (processed 'unset)
                        (canceled 'unset)
                        (priority-> 'unset)
                        (priority-< 'unset)
                        (after-id #f)
                        (limit 1000)
                        ;; other-builds-dependent or no-dependent-builds
                        (relationship 'unset))
    (call-with-worker-thread
     (slot-ref datastore 'worker-reader-thread-channel)
     (lambda (db)
       (define tag->expression
         (let ((statement
                (sqlite-prepare
                 db
                 "
SELECT id FROM tags WHERE key = :key AND value = :value"
                 #:cache? #t))
               (key-statement
                (sqlite-prepare
                 db
                 "
SELECT id FROM tags WHERE key = :key"
                 #:cache? #t)))
           (lambda (tag not?)
             (match tag
               ((key . value)
                (sqlite-bind-arguments statement
                                       #:key key
                                       #:value value)

                (match (sqlite-step-and-reset statement)
                  (#(id)
                   (simple-format
                    #f "tag_string ~A '%,~A,%'"
                    (if not?
                        "NOT LIKE"
                        "LIKE")
                    id))
                  (#f #f)))
               (key
                (sqlite-bind-arguments key-statement
                                       #:key key)

                (let* ((tag-ids (sqlite-map
                                 (match-lambda
                                   (#(id) id))
                                 key-statement))
                       (result
                        (string-append
                         "("
                         (string-join
                          (map (lambda (id)
                                 (simple-format
                                  #f "tag_string ~A '%,~A,%'"
                                  (if not?
                                      "NOT LIKE"
                                      "LIKE")
                                  id))
                               tag-ids)
                          (if not? " AND " " OR "))
                         ")")))
                  (sqlite-reset key-statement)

                  result))))))

       (let ((tag-expressions
              (map (lambda (tag)
                     (tag->expression tag #f))
                   tags))
             (not-tag-expressions
              (filter-map (lambda (tag)
                            (tag->expression tag #t))
                          not-tags)))

         ;; If one of the requested tags doesn't exist, nothing can be tagged to
         ;; it, so just return nothing
         (if (memq #f tag-expressions)
             '()
             (let* ((where-needed?
                     (or (not (null? tag-expressions))
                         (not (null? not-tag-expressions))
                         (not (null? systems))
                         (not (null? not-systems))
                         (not (eq? priority-> 'unset))
                         (not (eq? priority-< 'unset))
                         (not (eq? processed 'unset))
                         (not (eq? canceled 'unset))
                         (not (eq? relationship 'unset))
                         after-id))
                    (query
                     (string-append
                      "
SELECT uuid, derivations.name, priority, processed, canceled, created_at, end_time
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
LEFT JOIN (
  SELECT build_id, (',' || group_concat(tag_id) || ',') AS tag_string
  FROM build_tags
  GROUP BY build_id
) AS all_build_tags
  ON builds.id = all_build_tags.build_id
"
                      (if where-needed?
                          (string-append
                           "WHERE\n"
                           (string-join
                            (append
                             (let ((all-tag-expressions
                                    (append tag-expressions
                                            not-tag-expressions)))
                               (if (null? all-tag-expressions)
                                   '()
                                   all-tag-expressions))
                             (if (null? systems)
                                 '()
                                 (list
                                  (string-append
                                   "("
                                   (string-join
                                    (map (lambda (system)
                                           (simple-format
                                            #f
                                            "derivations.system_id = ~A"
                                            (db-system->system-id db system)))
                                         systems)
                                    " OR ")
                                   ")")))
                             (map (lambda (system)
                                    (simple-format
                                     #f "derivations.system_id != ~A"
                                     (db-system->system-id db system)))
                                  not-systems)
                             (if (number? priority->)
                                 (list
                                  (simple-format #f "priority > ~A" priority->))
                                 '())
                             (if (number? priority-<)
                                 (list
                                  (simple-format #f "priority < ~A" priority-<))
                                 '())
                             (cond
                              ((eq? processed #t) '("processed = 1"))
                              ((eq? processed #f) '("processed = 0"))
                              (else '()))
                             (cond
                              ((eq? canceled #t) '("canceled = 1"))
                              ((eq? canceled #f) '("canceled = 0"))
                              (else '()))
                             (if after-id
                                 '("uuid > :after_id")
                                 '())
                             (if (not (eq? relationship 'unset))
                                 (list
                                  (string-append
                                   (match relationship
                                     ('other-builds-dependent "EXISTS")
                                     ('no-dependent-builds "NOT EXISTS"))
                                   " (
SELECT 1
FROM derivation_outputs
INNER JOIN outputs ON derivation_outputs.output_id = outputs.id
INNER JOIN derivation_outputs AS all_derivation_outputs
  ON outputs.id = all_derivation_outputs.output_id
INNER JOIN derivation_inputs
  ON derivation_inputs.derivation_output_id = all_derivation_outputs.id
INNER JOIN derivations AS dependent_derivations
  ON dependent_derivations.id = derivation_inputs.derivation_id
INNER JOIN builds AS dependent_builds
  ON dependent_builds.derivation_id = dependent_derivations.id
 AND dependent_builds.processed = 0
 AND dependent_builds.canceled = 0
WHERE derivation_outputs.derivation_id = builds.derivation_id)"))
                                 '()))
                            " AND ")
                           "\n")
                          "")
                      "ORDER BY uuid ASC\n"
                      (if limit
                          (string-append "LIMIT " (number->string limit) "\n")
                          "")))
                    (statement
                     (sqlite-prepare db query #:cache? #f)))

               (when after-id
                 (sqlite-bind-arguments
                  statement
                  #:after_id after-id))

               (let ((result
                      (sqlite-map
                       (match-lambda
                         (#(uuid derivation_name priority processed canceled
                                 created_at end_time)
                          `((uuid            . ,uuid)
                            (derivation-name . ,derivation_name)
                            (priority        . ,priority)
                            (processed       . ,(cond
                                                 ((= 0 processed) #f)
                                                 ((= 1 processed) #t)
                                                 (else
                                                  (error
                                                   "unknown processed value"))))
                            (canceled        . ,(cond
                                                 ((= 0 canceled) #f)
                                                 ((= 1 canceled) #t)
                                                 (else
                                                  (error "unknown canceled value"))))
                            (created-at      . ,(if (string? created_at)
                                                    (match (strptime "%F %T"
                                                                     created_at)
                                                      ((parts . _) parts)
                                                      (#f
                                                       (error
                                                        (simple-format
                                                         #f
                                                         "error parsing created_at ~A (~A)"
                                                         created_at
                                                         uuid))))
                                                    #f))
                            (end-time        . ,(if (string? end_time)
                                                    (match (strptime "%F %T"
                                                                     end_time)
                                                      ((parts . _) parts)
                                                      (#f
                                                       (error
                                                        (simple-format
                                                         #f
                                                         "error parsing end_time ~A (~A)"
                                                         end_time
                                                         uuid))))
                                                    #f)))))
                       statement)))
                 (sqlite-finalize statement)

                 result)))))))

  (apply list-builds rest))

(define-method (datastore-fetch-build-tags
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT key, value
FROM tags
INNER JOIN build_tags
  ON tags.id = build_tags.tag_id
WHERE build_tags.build_id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (let ((result
              (list->vector
               (sqlite-fold
                (lambda (row result)
                  (match row
                    (#(key value)
                     `((,key . ,value)
                       ,@result))))
                '()
                statement))))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-find-build-result
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT agent_id, result, failure_reason
FROM build_results
WHERE build_id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (match (sqlite-step-and-reset statement)
         (#(agent_id result failure_reason)
          `((agent_id       . ,agent_id)
            (result         . ,result)
            (failure_reason . ,failure_reason)))
         (#f #f))))))

(define-method (datastore-find-build-derivation-system
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT systems.system
FROM builds
INNER JOIN derivations ON builds.derivation_id = derivations.id
INNER JOIN systems ON derivations.system_id = systems.id
WHERE builds.id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (match (sqlite-step-and-reset statement)
         (#(system) system)
         (#f
          (error
           (simple-format
            #f
            "no system for build ~A"
            build-uuid))))))))

(define-method (datastore-list-builds-for-output
                (datastore <sqlite-datastore>)
                output)
  (call-with-time-tracking
   datastore
   "list_builds_for_output"
   (lambda ()
     (call-with-worker-thread
      (slot-ref datastore 'worker-reader-thread-channel)
      (lambda (db)
        (let ((statement
               (sqlite-prepare
                db
                "
SELECT uuid, derivations.name, priority, processed, canceled, result
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
INNER JOIN derivation_outputs
  ON builds.derivation_id = derivation_outputs.derivation_id
LEFT JOIN build_results
  ON builds.id = build_results.build_id
WHERE derivation_outputs.output_id = :output_id"
                #:cache? #t)))

          (sqlite-bind-arguments
           statement
           #:output_id (db-output->output-id db output))

          (let ((result
                 (sqlite-map
                  (match-lambda
                    (#(uuid derivation priority processed canceled result)
                     `((uuid            . ,uuid)
                       (derivation      . ,derivation)
                       (priority        . ,priority)
                       (processed       . ,(cond
                                            ((= 0 processed) #f)
                                            ((= 1 processed) #t)
                                            (else (error "unknown value"))))
                       (canceled        . ,(cond
                                            ((= 0 canceled) #f)
                                            ((= 1 canceled) #t)
                                            (else (error "unknown value"))))
                       (result          . ,result))))
                  statement)))
            (sqlite-reset statement)

            result)))))))

(define-method (datastore-list-builds-for-output-and-system
                (datastore <sqlite-datastore>)
                .
                rest)
  (apply
   (lambda* (output system #:key include-canceled?)
     (call-with-worker-thread
      (slot-ref datastore 'worker-reader-thread-channel)
      (lambda (db)
        (let ((statement
               (sqlite-prepare
                db
                (string-append
                 "
SELECT uuid, derivations.name
FROM builds
INNER JOIN derivation_outputs
  ON builds.derivation_id = derivation_outputs.derivation_id
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
WHERE derivation_outputs.output_id = :output_id
  AND derivations.system_id = :system_id"
                 (if include-canceled?
                     ""
                     "
  AND builds.canceled = 0"))
                #:cache? #t)))

          (sqlite-bind-arguments
           statement
           #:output_id (db-output->output-id db output)
           #:system_id (db-system->system-id db system))

          (let ((result
                 (sqlite-map
                  (match-lambda
                    (#(uuid derivation)
                     `((uuid            . ,uuid)
                       (derivation      . ,derivation))))
                  statement)))
            (sqlite-reset statement)

            result)))))
   rest))

(define-method (datastore-count-builds-for-derivation
                (datastore <sqlite-datastore>)
                .
                rest)
  (apply
   (lambda* (derivation #:key (include-canceled? #t))
     (call-with-worker-thread
      (slot-ref datastore 'worker-reader-thread-channel)
      (lambda (db)
        (let ((statement
               (sqlite-prepare
                db
                (string-append
                 "
SELECT COUNT(*)
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
WHERE derivations.name = :derivation"
                 (if include-canceled?
                     ""
                     "
  AND canceled = 0"))
                #:cache? #t)))

          (sqlite-bind-arguments
           statement
           #:derivation derivation)

          (match (sqlite-step-and-reset statement)
            (#(x) x))))))
   rest))

(define-method (datastore-update
                (datastore <sqlite-datastore>))
  (run-sqitch
   (slot-ref datastore 'database-file))

  #t)

(define-method (datastore-count-setup-failures
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT agent_id, failure_reason, COUNT(*)
FROM setup_failures
GROUP BY agent_id, failure_reason"
             #:cache? #t)))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(agent_id failure_reason count)
                  (cons (list agent_id failure_reason) count)))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-list-setup-failures-for-build
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT id, agent_id, failure_reason
FROM setup_failures
WHERE build_id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (let ((result (sqlite-map
                      (match-lambda
                        (#(id agent-id failure-reason)
                         `((id  . ,id)
                           (agent-id . ,agent-id)
                           (failure-reason . ,failure-reason))))
                      statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-fetch-setup-failures
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT setup_failures.id, builds.uuid, agent_id, failure_reason
FROM setup_failures
INNER JOIN builds
  ON builds.id = setup_failures.build_id
WHERE builds.processed = 0
  AND builds.canceled = 0
  AND builds.id NOT IN (
    SELECT build_id FROM allocated_builds
  )"
             #:cache? #t)))

       (let ((result (sqlite-fold
                      (lambda (row result)
                        (match row
                          (#(id build-id agent-id failure-reason)
                           (let ((failures-for-build-id
                                  (or (hash-ref result build-id)
                                      '())))
                             (hash-set!
                              result
                              build-id
                              (cons `((id  . ,id)
                                      (agent-id . ,agent-id)
                                      (failure-reason . ,failure-reason))
                                    failures-for-build-id)))))
                        result)
                      (make-hash-table)
                      statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-list-processed-builds
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT uuid, derivations.name, priority
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
WHERE processed = 1"
             #:cache? #t)))

       (let ((builds (sqlite-map
                      (match-lambda
                        (#(uuid derivation_name priority)
                         `((uuid            . ,uuid)
                           (derivation-name . ,derivation_name)
                           (priority        . ,priority))))
                      statement)))
         (sqlite-reset statement)

         builds)))))

(define-method (datastore-list-unprocessed-builds
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT uuid, derivations.name, priority
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
WHERE processed = 0
  AND canceled = 0
  AND (
    deferred_until IS NULL
    OR
    deferred_until < datetime('now')
  )
  AND builds.id NOT IN (
    SELECT build_id FROM allocated_builds
  )
ORDER BY priority DESC"
             #:cache? #t)))

       (let ((builds (sqlite-map
                      (match-lambda
                        (#(uuid derivation_name priority)
                         `((uuid            . ,uuid)
                           (derivation-name . ,derivation_name)
                           (priority        . ,priority))))
                      statement)))
         (sqlite-reset statement)

         builds)))))

(define-method (datastore-find-first-unallocated-deferred-build
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT uuid, derivations.name, priority, created_at, deferred_until
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
WHERE processed = 0
  AND canceled = 0
  AND deferred_until IS NOT NULL
  AND builds.id NOT IN (SELECT build_id FROM build_allocation_plan)
ORDER BY deferred_until ASC
LIMIT 1"
             #:cache? #t)))

       (match (sqlite-step-and-reset statement)
         (#(uuid derivation_name priority created_at deferred_until)
          `((uuid            . ,uuid)
            (derivation-name . ,derivation_name)
            (priority        . ,priority)
            (created-at      . ,(if (string? created_at)
                                    (string->date created_at
                                                  "~Y-~m-~d ~H:~M:~S")
                                    #f))
            (deferred-until . ,(if (string? deferred_until)
                                   (string->date deferred_until
                                                 "~Y-~m-~d ~H:~M:~S")
                                   #f))))
         (#f #f))))))

(define-method (datastore-fetch-prioritised-unprocessed-builds
                (datastore <sqlite-datastore>))
  (define (fetch-prioritised-unprocessed-builds db)
    (let ((statement
           (sqlite-prepare
            db
            "
SELECT builds.uuid
FROM unprocessed_builds_with_derived_priorities
INNER JOIN builds ON build_id = builds.id
WHERE all_inputs_built = 1
  AND NOT EXISTS (
    SELECT 1
    FROM allocated_builds
    WHERE unprocessed_builds_with_derived_priorities.build_id =
          allocated_builds.build_id
) ORDER BY derived_priority ASC"
            #:cache? #t)))

      (let ((result (sqlite-fold
                     (lambda (row result)
                       (cons (vector-ref row 0)
                             result))
                     '()
                     statement)))
        (sqlite-reset statement)

        result)))

  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   fetch-prioritised-unprocessed-builds))

(define-method (datastore-insert-unprocessed-hook-event
                (datastore <sqlite-datastore>)
                event
                arguments)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (insert-unprocessed-hook-event db
                                    event
                                    arguments))))

(define (insert-unprocessed-hook-event
         db
         event
         arguments)
  (let ((statement
         (sqlite-prepare
          db
          "
INSERT INTO unprocessed_hook_events (event, arguments)
VALUES (:event, :arguments)"
          #:cache? #t)))

    (sqlite-bind-arguments statement
                           #:event event
                           #:arguments
                           (call-with-output-string
                             (lambda (port)
                               (write arguments port))))

    (sqlite-step-and-reset statement))
  #t)

(define-method (datastore-count-unprocessed-hook-events
                (datastore <sqlite-datastore>))
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT event, COUNT(*) FROM unprocessed_hook_events GROUP BY event"
             #:cache? #t)))

       (let ((counts (sqlite-map
                      (match-lambda
                        (#(event count)
                         `((event . ,event)
                           (count . ,count))))
                      statement)))
         (sqlite-reset statement)

         counts)))))

(define-method (datastore-list-unprocessed-hook-events
                (datastore <sqlite-datastore>)
                event
                limit)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT id, event, arguments
FROM unprocessed_hook_events
WHERE event = :event
LIMIT :limit"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:event (symbol->string event)
        #:limit limit)

       (let ((events (sqlite-map
                      (match-lambda
                        (#(id event arguments)
                         (list id
                               (string->symbol event)
                               (call-with-input-string arguments
                                 (lambda (port)
                                   (read port))))))
                      statement)))
         (sqlite-reset statement)

         events)))))

(define-method (datastore-find-unprocessed-hook-event
                (datastore <sqlite-datastore>)
                id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT event, arguments
FROM unprocessed_hook_events
WHERE id = :id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:id id)

       (match (sqlite-step-and-reset statement)
         (#f #f)
         (#(event arguments)
          (list (string->symbol event)
                (call-with-input-string arguments
                  (lambda (port)
                    (read port))))))))))

(define-method (datastore-delete-unprocessed-hook-event
                (datastore <sqlite-datastore>)
                id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
DELETE FROM unprocessed_hook_events WHERE id = :id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:id id)

       (sqlite-step-and-reset statement))))
  #t)

(define-method (datastore-count-build-allocation-plan-entries
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT agent_id, COUNT(*)
FROM build_allocation_plan
GROUP BY agent_id"
             #:cache? #t)))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(agent_id count)
                  (cons agent_id count)))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-replace-build-allocation-plan
                (datastore <sqlite-datastore>)
                planned-builds)
  (define (clear-current-plan db)
    (sqlite-exec
     db
     "DELETE FROM build_allocation_plan"))

  (define insert-sql
    (call-with-worker-thread
     (slot-ref datastore 'worker-reader-thread-channel)
     (lambda (db)
       (string-append
        "
INSERT INTO build_allocation_plan (build_id, agent_id, ordering) VALUES "
        (string-join
         (map (match-lambda
                ((build-uuid agent-id ordering)
                 (simple-format
                  #f
                  "('~A', '~A', ~A)"
                  (db-find-build-id db build-uuid)
                  agent-id
                  ordering)))
              planned-builds)
         ", ")
        ";"))))

  (define (insert-new-plan db planned-builds)
    (sqlite-exec
     db
     insert-sql))

  (datastore-call-with-transaction
   datastore
   (lambda (db)
     (clear-current-plan db)
     (unless (null? planned-builds)
       (insert-new-plan db planned-builds)))
   #:duration-metric-name "replace_build_allocation_plan")

  (let* ((agent-ids
          (map (lambda (agent-details)
                 (assq-ref agent-details 'uuid))
               (datastore-list-agents datastore)))
         (counts-by-agent
          (make-vector (length agent-ids) 0)))
    (for-each
     (match-lambda
       ((_ agent-id _)
        (let ((index (list-index (lambda (list-agent-id)
                                   (string=? agent-id list-agent-id))
                                 agent-ids)))
          (vector-set! counts-by-agent
                       index
                       (+ (vector-ref counts-by-agent
                                      index)
                          1)))))
     planned-builds)

    (let ((metric
           (metrics-registry-fetch-metric (slot-ref datastore 'metrics-registry)
                                          "build_allocation_plan_total")))
      (for-each
       (lambda (index agent-id)
         (metric-set metric
                     (vector-ref counts-by-agent index)
                     #:label-values
                     `((agent_id . ,agent-id))))
       (iota (length agent-ids))
       agent-ids)))
  #t)

(define-method (datastore-count-allocated-builds
                (datastore <sqlite-datastore>))
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT agent_id, COUNT(*) FROM allocated_builds GROUP BY agent_id"
             #:cache? #t)))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(agent_id count)
                  (cons agent_id count)))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-agent-requested-systems
                (datastore <sqlite-datastore>)
                agent-id)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT system_id
FROM build_allocation_agent_requested_systems
WHERE agent_id = :agent_id
ORDER BY system_id ASC"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:agent_id agent-id)

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(system-id)
                  (db-system-id->system db system-id)))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-update-agent-requested-systems
                (datastore <sqlite-datastore>)
                agent-id
                systems)
  (define update-not-needed?
    (equal? (sort systems
                  string<?)
            (datastore-agent-requested-systems
             datastore
             agent-id)))

  (if update-not-needed?
      #f
      (datastore-call-with-transaction
       datastore
       (lambda (db)
         (sqlite-exec
          db
          (simple-format
           #f
           "
DELETE FROM build_allocation_agent_requested_systems
WHERE agent_id = '~A'"
           agent-id))

         (sqlite-exec
          db
          (string-append
           "
INSERT INTO build_allocation_agent_requested_systems (agent_id, system_id) VALUES "
           (string-join
            (map (lambda (system)
                   (simple-format
                    #f
                    "('~A', ~A)"
                    agent-id
                    (or
                     (db-system->system-id db system)
                     (db-insert-system db system))))
                 systems)
            ", ")
           ";"))

         #t))))

(define-method (datastore-fetch-build-to-allocate
                (datastore <sqlite-datastore>)
                agent-id)
  (datastore-call-with-transaction
   datastore
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             ;; This needs to guard against the plan being out of date
             "
SELECT builds.uuid, derivations.id, derivations.name,
       unprocessed_builds_with_derived_priorities.derived_priority
FROM builds
INNER JOIN build_allocation_plan
  ON builds.id = build_allocation_plan.build_id
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
INNER JOIN build_allocation_agent_requested_systems
  ON build_allocation_agent_requested_systems.agent_id = :agent_id
 AND build_allocation_agent_requested_systems.system_id = derivations.system_id
LEFT JOIN unprocessed_builds_with_derived_priorities
  ON unprocessed_builds_with_derived_priorities.build_id = builds.id
WHERE build_allocation_plan.agent_id = :agent_id
  AND builds.processed = 0
  AND builds.canceled = 0
  AND builds.id NOT IN (SELECT build_id FROM allocated_builds)
ORDER BY build_allocation_plan.ordering ASC"
             #:cache? #t))
           (output-conflicts-statement
            (sqlite-prepare
             db
             "
SELECT 1
FROM derivation_outputs AS build_derivation_outputs
INNER JOIN allocated_builds
  ON allocated_builds.agent_id = :agent_id
INNER JOIN builds AS allocated_build_details
  ON allocated_build_details.id = allocated_builds.build_id
INNER JOIN derivation_outputs AS allocated_builds_derivation_outputs
  ON allocated_build_details.derivation_id =
     allocated_builds_derivation_outputs.derivation_id
WHERE build_derivation_outputs.derivation_id = :derivation_id
  AND build_derivation_outputs.output_id =
      allocated_builds_derivation_outputs.output_id"
             #:cache? #t)))

       (define (get-build-to-allocate)
         (let loop ((build-details (sqlite-step statement)))
           (match build-details
             (#f #f)
             (#(uuid derivation-id derivation-name derived_priority)

              (sqlite-bind-arguments output-conflicts-statement
                                     #:agent_id agent-id
                                     #:derivation_id derivation-id)

              (if (eq? (sqlite-step-and-reset output-conflicts-statement)
                       #f)
                  `((uuid . ,uuid)
                    (derivation_name . ,derivation-name)
                    (derived_priority . ,derived_priority))
                  (loop (sqlite-step statement)))))))

       (sqlite-bind-arguments
        statement
        #:agent_id agent-id)

       (let ((result (get-build-to-allocate)))
         (sqlite-reset statement)

         result)))

   #:readonly? #t
   #:duration-metric-name "fetch_builds_to_allocate"))

(define-method (datastore-insert-to-allocated-builds
                (datastore <sqlite-datastore>)
                agent-id
                build-uuids)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (sqlite-exec
      db
      (string-append
       "
INSERT INTO allocated_builds (build_id, agent_id) VALUES "
       (string-join
        (map (lambda (build-uuid)
               (simple-format
                #f
                "(~A, '~A')"
                (db-find-build-id db build-uuid)
                agent-id))
             build-uuids)
        ", ")
       ";")))))

(define-method (datastore-remove-builds-from-plan
                (datastore <sqlite-datastore>)
                build-uuids)
  (call-with-worker-thread
   (slot-ref datastore 'worker-writer-thread-channel)
   (lambda (db)
     (sqlite-exec
      db
      (string-append
       "
DELETE FROM build_allocation_plan
WHERE build_id IN ("
       (string-join
        (map (lambda (build-uuid)
               (number->string (db-find-build-id db build-uuid)))
             build-uuids)
        ", ")
       ")")))))

(define-method (datastore-list-allocation-plan-builds
                (datastore <sqlite-datastore>)
                .
                rest)
  (apply
   (lambda* (agent-id #:key limit)
     (call-with-worker-thread
      (slot-ref datastore 'worker-reader-thread-channel)
      (lambda (db)
        (let ((statement
               (sqlite-prepare
                db
                (string-append
                 "
SELECT builds.uuid, derivations.name, systems.system,
       builds.priority,
       unprocessed_builds_with_derived_priorities.derived_priority
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
INNER JOIN systems
  ON derivations.system_id = systems.id
INNER JOIN build_allocation_plan
  ON builds.id = build_allocation_plan.build_id
LEFT JOIN unprocessed_builds_with_derived_priorities
  ON builds.id = unprocessed_builds_with_derived_priorities.build_id
WHERE build_allocation_plan.agent_id = :agent_id
  AND builds.processed = 0
  AND builds.id NOT IN (SELECT build_id FROM allocated_builds)
ORDER BY build_allocation_plan.ordering ASC"
                 (if limit
                     "
LIMIT :limit"
                     ""))
                #:cache? #t)))

          (apply sqlite-bind-arguments
                 statement
                 #:agent_id agent-id
                 (if limit
                     (list #:limit limit)
                     '()))

          (let ((builds (sqlite-map
                         (match-lambda
                           (#(uuid derivation_name system
                                   priority derived_priority)
                            `((uuid             . ,uuid)
                              (derivation_name  . ,derivation_name)
                              (system           . ,system)
                              (priority         . ,priority)
                              (derived_priority . ,derived_priority)
                              (tags . ,(vector-map
                                        (lambda (_ tag)
                                          (match tag
                                            ((key . value)
                                             `((key   . ,key)
                                               (value . ,value)))))
                                        (datastore-fetch-build-tags
                                         datastore
                                         uuid))))))
                         statement)))
            (sqlite-reset statement)

            builds)))))
   rest))

(define-method (datastore-list-agent-builds
                (datastore <sqlite-datastore>)
                agent-id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT builds.uuid, derivations.name,
       unprocessed_builds_with_derived_priorities.derived_priority,
       builds.canceled
FROM builds
INNER JOIN derivations
  ON builds.derivation_id = derivations.id
INNER JOIN allocated_builds
  ON builds.id = allocated_builds.build_id
LEFT JOIN unprocessed_builds_with_derived_priorities
  ON unprocessed_builds_with_derived_priorities.build_id = builds.id
WHERE allocated_builds.agent_id = :agent_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:agent_id agent-id)

       (let ((builds (sqlite-map
                      (match-lambda
                        (#(uuid derivation_name derived_priority canceled)
                         `((uuid             . ,uuid)
                           (derivation_name  . ,derivation_name)
                           (derived_priority . ,derived_priority)
                           (canceled         . ,canceled))))
                      statement)))
         (sqlite-reset statement)

         builds)))))

(define-method (datastore-agent-for-build
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT agent_id
FROM allocated_builds
WHERE allocated_builds.build_id = :build_id
UNION SELECT agent_id
FROM build_results
WHERE build_results.build_id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (match (sqlite-step-and-reset statement)
         (#(agent-id) agent-id)
         (#f #f))))))

(define* (db-open database
                  #:key (write? #t))
  (define flags
    (cons
     (if write?
         SQLITE_OPEN_READWRITE
         SQLITE_OPEN_READONLY)
     (list SQLITE_OPEN_NOMUTEX)))

  (unless (file-exists? database)
    (run-sqitch database))

  (sqlite-open database (apply logior flags)))

(define (run-sqitch database-file)
  (let ((command
         (list (%config 'sqitch)
               "deploy"
               "--db-client" (%config 'sqitch-sqlite)
               ;; if sqitch.conf exists (which it should when developing),
               ;; just use the current directory as the chdir value. Otherwise
               ;; use the directory which should contain the right files after
               ;; installation.
               "--chdir" (if (file-exists? "sqitch.conf")
                             (getcwd)
                             (string-append (dirname (%config 'sqitch-plan))
                                            "/sqlite"))
               "--plan-file" (%config 'sqitch-plan)
               "--registry" (string-append
                             (canonicalize-path
                              (dirname database-file))
                             "/"
                             (basename
                              (if (string-suffix? ".db" database-file)
                                  (string-drop-right database-file 3)
                                  database-file))
                             "_sqitch_registry.db")
               (string-append "db:sqlite:" database-file))))

    (simple-format #t "running command: ~A\n"
                   (string-join command))
    (unless (zero? (apply system* command))
      (simple-format
       (current-error-port)
       "error: sqitch command failed\n")
      (exit 1))))

(define (changes-count db)
  (let ((statement
         (sqlite-prepare
          db
          "SELECT changes();"
          #:cache? #t)))

    (vector-ref (sqlite-step-and-reset statement)
                0)))

(define (last-insert-rowid db)
  (let ((statement
         (sqlite-prepare
          db
          "SELECT last_insert_rowid();"
          #:cache? #t)))

    (vector-ref (sqlite-step-and-reset statement)
                0)))

(define (select-derivation-outputs db derivation-name)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT name, id FROM derivation_outputs WHERE derivation_id = :derivation_id"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:derivation_id (db-find-derivation-id db derivation-name))

    (let ((outputs (sqlite-map
                    (match-lambda
                      (#(name output-id)
                       (cons name output-id)))
                    statement)))
      (sqlite-reset statement)

      outputs)))

(define (db-find-derivation-id db name)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT id FROM derivations WHERE name = :name"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:name name)

    (match (sqlite-step-and-reset statement)
      (#f #f)
      (#(id) id))))

(define (db-find-derivation db name)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT systems.system, fixed_output
FROM derivations
INNER JOIN systems
  ON systems.id = derivations.system_id
WHERE name = :name"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:name name)

    (match (sqlite-step-and-reset statement)
      (#f #f)
      (#(system fixed_output)
       `((system        . ,system)
         (fixed-output? . ,(cond
                            ((= fixed_output 0) #f)
                            ((= fixed_output 1) #t)
                            (else fixed_output))))))))

(define-method (datastore-find-derivation
                (datastore <sqlite-datastore>)
                name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (db-find-derivation db name))))

(define-method (datastore-find-derivation-outputs
                (datastore <sqlite-datastore>)
                derivation-name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT name, outputs.output
FROM derivation_outputs
INNER JOIN outputs ON derivation_outputs.output_id = outputs.id
WHERE derivation_id = :derivation_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:derivation_id (db-find-derivation-id db derivation-name))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(name output)
                  `((name   . ,name)
                    (output . ,output))))
               statement)))
         (sqlite-reset statement)

         (if (null? result)
             #f
             result))))))

(define-method (datastore-find-derivation-output-details
                (datastore <sqlite-datastore>)
                derivation-name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT outputs.output,
       derivation_outputs.name,
       derivation_output_details.hash_algorithm,
       derivation_output_details.hash,
       derivation_output_details.recursive
FROM derivation_outputs
INNER JOIN outputs
  ON derivation_outputs.output_id = outputs.id
INNER JOIN derivation_output_details
  ON derivation_output_details.derivation_output_id = derivation_outputs.id
WHERE derivation_id = :derivation_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:derivation_id (db-find-derivation-id db derivation-name))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(output name hash-algorithm hash recursive)
                  `((output . ,output)
                    (name   . ,name)
                    (hash-algorithm . ,hash-algorithm)
                    (hash . ,hash)
                    (recursive? . ,(= 1 recursive)))))
               statement)))
         (sqlite-reset statement)

         (if (null? result)
             #f
             result))))))

(define-method (datastore-list-unbuilt-derivation-outputs
                (datastore <sqlite-datastore>)
                derivation-name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT derivation_outputs.name, outputs.output
FROM derivation_outputs
INNER JOIN outputs
  ON derivation_outputs.output_id = outputs.id
INNER JOIN unbuilt_outputs
  ON derivation_outputs.output_id = unbuilt_outputs.output_id
WHERE derivation_id = :derivation_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:derivation_id (db-find-derivation-id db derivation-name))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(name output)
                  `((name   . ,name)
                    (output . ,output))))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-list-build-outputs
                (datastore <sqlite-datastore>)
                build-uuid)
  (call-with-worker-thread/delay-logging
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT name, outputs.output, hash, size, store_references
FROM builds
INNER JOIN derivation_outputs
  ON builds.derivation_id = derivation_outputs.derivation_id
INNER JOIN outputs
  ON derivation_outputs.output_id = outputs.id
LEFT JOIN output_metadata
  ON output_metadata.derivation_output_id = derivation_outputs.id
 AND output_metadata.build_id = builds.id
WHERE builds.id = :build_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:build_id (db-find-build-id db build-uuid))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(name output hash size store_references)
                  `((name       . ,name)
                    (output     . ,output)
                    (hash       . ,hash)
                    (size       . ,size)
                    (references . ,(and store_references
                                        (list->vector
                                         (string-split store_references
                                                       #\space)))))))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-find-derivation-system
                (datastore <sqlite-datastore>)
                derivation-name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT systems.system
FROM derivations
INNER JOIN systems ON systems.id = derivations.system_id
WHERE name = :name"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:name derivation-name)

       (match (sqlite-step-and-reset statement)
         (#f #f)
         (#(system) system))))))

(define-method (datastore-find-derivation-inputs
                (datastore <sqlite-datastore>)
                derivation-name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT output_derivations.name,
       derivation_outputs.name,
       outputs.output
FROM derivations
INNER JOIN derivation_inputs
  ON derivation_inputs.derivation_id = derivations.id
INNER JOIN derivation_outputs
  ON derivation_inputs.derivation_output_id = derivation_outputs.id
INNER JOIN derivations AS output_derivations
  ON output_derivations.id = derivation_outputs.derivation_id
INNER JOIN outputs
  ON derivation_outputs.output_id = outputs.id
WHERE derivations.id = :derivation_id"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:derivation_id (db-find-derivation-id db derivation-name))

       (let ((result
              (sqlite-map
               (match-lambda
                 (#(derivation output-name output)
                  `((derivation  . ,derivation)
                    (output_name . ,output-name)
                    (output      . ,output))))
               statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-find-derivation-for-output
                (datastore <sqlite-datastore>)
                start-derivation-name
                output)
  (call-with-worker-thread
   (slot-ref datastore 'worker-reader-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
WITH RECURSIVE related_derivations(id) AS (
  VALUES(:derivation_id)
UNION
  SELECT derivation_outputs.derivation_id
  FROM derivation_outputs
  INNER JOIN derivation_inputs
    ON derivation_outputs.id = derivation_inputs.derivation_output_id
  INNER JOIN related_derivations
    ON related_derivations.id = derivation_inputs.derivation_id
)
SELECT derivations.name
FROM related_derivations
INNER JOIN derivations
  ON derivations.id = related_derivations.id
INNER JOIN derivation_outputs
  ON related_derivations.id = derivation_outputs.derivation_id
WHERE derivation_outputs.output_id = :output_id
"
             #:cache? #t)))

       (sqlite-bind-arguments
        statement
        #:derivation_id (db-find-derivation-id db start-derivation-name)
        #:output_id (db-output->output-id db output))

       (match (sqlite-step-and-reset statement)
         (#f #f)
         (#(derivation) derivation))))))

(define (db-insert-system db system)
  (let ((statement
         (sqlite-prepare
          db
          "
INSERT INTO systems (system) VALUES (:system)"
          #:cache? #t)))
    (sqlite-bind-arguments
     statement
     #:system system)

    (sqlite-step-and-reset statement)
    (last-insert-rowid db)))

(define (db-system->system-id db system)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT id FROM systems WHERE system = :system"
          #:cache? #t)))
    (sqlite-bind-arguments
     statement
     #:system system)

    (match (sqlite-step-and-reset statement)
      (#f #f)
      (#(id) id))))

(define (db-system-id->system db system-id)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT system FROM systems WHERE id = :id"
          #:cache? #t)))
    (sqlite-bind-arguments
     statement
     #:id system-id)

    (match (sqlite-step-and-reset statement)
      (#f #f)
      (#(id) id))))

(define (insert-derivation-and-return-outputs db derivation)
  (define derivation-name
    (derivation-file-name derivation))

  (define (maybe-fix-fixed-output-field derivation-details)
    (let ((fixed-output? (fixed-output-derivation? derivation)))
      (unless (equal? (assq-ref derivation-details 'fixed-output?)
                      fixed-output?)
        (let ((statement (sqlite-prepare
                          db
                          "
UPDATE derivations SET fixed_output = :fixed_output WHERE name = :name"
                          #:cache? #t)))
          (sqlite-bind-arguments statement
                                 #:name derivation-name
                                 #:fixed_output (if fixed-output? 1 0))
          (sqlite-step-and-reset statement)))))

  (define (insert-derivation)
    (let ((statement
           (sqlite-prepare
            db
            "
INSERT INTO derivations (name, system_id, fixed_output)
  VALUES (:name, :system_id, :fixed_output)"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:name derivation-name
       #:system_id
       (let ((system (derivation-system derivation)))
         (or
          (db-system->system-id db system)
          (db-insert-system db system)))
       #:fixed_output (if (fixed-output-derivation? derivation) 1 0))

      (sqlite-step-and-reset statement)

      (last-insert-rowid db)))


  (or (hash-ref %derivation-outputs-cache
                derivation-name)
      (let ((derivation-details
             (db-find-derivation db derivation-name)))
        (if derivation-details
            (begin
              (maybe-fix-fixed-output-field derivation-details)

              (let ((derivation-outputs
                     (select-derivation-outputs db derivation-name)))

                (hash-set! %derivation-outputs-cache
                           derivation-name
                           derivation-outputs)

                derivation-outputs))
            (let ((derivation-id (insert-derivation)))
              (call-with-delay-logging
               insert-derivation-inputs
               #:args (list
                       db
                       derivation-id
                       (derivation-inputs derivation)))

              (let ((derivation-outputs
                     (call-with-delay-logging
                      insert-derivation-outputs
                      #:args (list
                              db
                              derivation-id
                              (derivation-outputs derivation)))))
                (hash-set! %derivation-outputs-cache
                           derivation-name
                           derivation-outputs)

                derivation-outputs))))))

(define (insert-derivation-inputs db derivation-id derivation-inputs)
  (unless (null? derivation-inputs)
    (let ((derivation-output-ids
           (append-map
            (lambda (derivation-input)
              (let ((output-ids-by-name
                     (call-with-delay-logging
                      insert-derivation-and-return-outputs
                      #:args (list
                              db
                              (derivation-input-derivation derivation-input)))))
                (map
                 (lambda (output-name)
                   (assoc-ref output-ids-by-name output-name))
                 (derivation-input-sub-derivations derivation-input))))
            derivation-inputs)))
      (let ((statement
             (sqlite-prepare
              db
              "
INSERT INTO derivation_inputs (derivation_id, derivation_output_id)
VALUES (:derivation_id, :derivation_output_id)"
              #:cache? #t)))
        (for-each (lambda (derivation-output-id)
                    (sqlite-bind-arguments
                     statement
                     #:derivation_id derivation-id
                     #:derivation_output_id derivation-output-id)
                    (sqlite-step-and-reset statement))
                  derivation-output-ids)))))

(define (db-insert-output db output)
  (let ((statement
         (sqlite-prepare
          db
          "
INSERT INTO outputs (output) VALUES (:output)"
          #:cache? #t)))

    (sqlite-bind-arguments statement #:output output)

    (sqlite-step-and-reset statement)

    (last-insert-rowid db)))

(define (db-output->output-id db output)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT id FROM outputs WHERE output = :output"
          #:cache? #t)))

    (sqlite-bind-arguments statement #:output output)

    (match (sqlite-step-and-reset statement)
      (#f #f)
      (#(id) id))))

(define (insert-derivation-outputs db derivation-id derivation-outputs)
  (define output-has-successful-build?
    (let ((statement
           (sqlite-prepare
            db
            "
SELECT build_results.result
FROM derivation_outputs
INNER JOIN builds
  ON builds.derivation_id = derivation_outputs.derivation_id
INNER JOIN build_results
  ON builds.id = build_results.build_id
WHERE derivation_outputs.output_id = :output_id"
            #:cache? #t)))

      (lambda (output-id)
        (sqlite-bind-arguments statement
                               #:output_id output-id)

        (let* ((build-results
                (sqlite-map (match-lambda
                              (#(result) result))
                            statement))
               (result (if (member "success" build-results)
                           #t
                           #f)))
          (sqlite-reset statement)

          result))))

  (define insert-into-unbuilt-outputs
    (let ((statement
           (sqlite-prepare
            db
            "
INSERT OR IGNORE INTO unbuilt_outputs (output_id) VALUES (:output_id)"
            #:cache? #t)))
      (lambda (output-id)
        (sqlite-bind-arguments statement
                               #:output_id output-id)

        (sqlite-step-and-reset statement)
        #t)))

  (let ((derivation-outputs-with-ids
         (map (match-lambda
                ((name . derivation-output)
                 (let ((output
                        (derivation-output-path derivation-output)))
                   (cons name
                         (or
                          (db-output->output-id db output)
                          (db-insert-output db output))))))
              derivation-outputs))
        (statement
         (sqlite-prepare
          db
          "
INSERT INTO derivation_outputs (derivation_id, name, output_id)
  VALUES (:derivation_id, :name, :output_id)"
          #:cache? #t))

        (output-details-statement
         (sqlite-prepare
          db
          "
INSERT INTO derivation_output_details
  (derivation_output_id, hash_algorithm, hash, recursive)
VALUES (:derivation_output_id, :hash_algorithm, :hash, :recursive)"
          #:cache? #t)))

    (let ((result
           (map (match-lambda*
                  (((name . output-id) derivation-output)
                   (sqlite-bind-arguments
                    statement
                    #:derivation_id derivation-id
                    #:name name
                    #:output_id output-id)
                   (sqlite-step-and-reset statement)

                   (let ((derivation-output-id
                          (last-insert-rowid db)))
                     (sqlite-bind-arguments
                      output-details-statement
                      #:derivation_output_id derivation-output-id
                      #:hash_algorithm (and=> (derivation-output-hash-algo
                                               derivation-output)
                                              symbol->string)
                      #:hash (and=> (derivation-output-hash derivation-output)
                                    bytevector->base16-string)
                      #:recursive
                      (if (derivation-output-recursive? derivation-output)
                          1
                          0))

                     (sqlite-step-and-reset output-details-statement)

                     (cons name derivation-output-id))))
                derivation-outputs-with-ids
                (map cdr derivation-outputs))))

      (for-each (lambda (output-id)
                  (unless (output-has-successful-build? output-id)
                    (insert-into-unbuilt-outputs output-id)))
                (map cdr derivation-outputs-with-ids))

      result)))

(define-method (datastore-insert-build
                (datastore <sqlite-datastore>)
                .
                rest)
  (define (insert-build db drv-name uuid priority defer-until)
    (let ((statement
           (sqlite-prepare
            db
            "
INSERT INTO builds (uuid, derivation_id, priority, created_at, deferred_until)
VALUES (:uuid, :derivation_id, :priority, datetime('now'), :deferred_until)
RETURNING id"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:uuid uuid
       #:derivation_id (call-with-delay-logging
                        db-find-derivation-id
                        #:args (list db drv-name))
       #:priority priority
       #:deferred_until
       (and=> defer-until
              (lambda (date)
                (date->string date "~1 ~3"))))

      (match (sqlite-step-and-reset statement)
        (#(id) id))))

  (define (increment-builds-counts db system-id)
    (let ((statement
           (sqlite-prepare
            db
            "
UPDATE builds_counts
SET count = count + 1
WHERE system_id = :system_id
RETURNING count"
            #:cache? #t))
          (insert-statement
           (sqlite-prepare
            db
            "
INSERT INTO builds_counts (system_id, count) VALUES (:system_id, 1)"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:system_id system-id)

      (match (sqlite-step-and-reset statement)
        (#(count) #t)
        (#f
         (sqlite-bind-arguments
          insert-statement
          #:system_id system-id)

         (sqlite-step-and-reset insert-statement)

         #t))))

  (define (all-inputs-built? db build-id)
    (let ((statement
           (sqlite-prepare
            db
            "
SELECT 1
FROM derivation_inputs
INNER JOIN derivation_outputs
  ON derivation_inputs.derivation_output_id = derivation_outputs.id
INNER JOIN unbuilt_outputs
  ON unbuilt_outputs.output_id = derivation_outputs.output_id
INNER JOIN builds
  ON builds.derivation_id = derivation_inputs.derivation_id
WHERE builds.id = :build_id
LIMIT 1"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:build_id build-id)

      (match (sqlite-step-and-reset statement)
        (#f #t)
        (#(1) #f))))

  (define (insert-unprocessed-builds-with-derived-priorities-entry
           db
           build-id
           derived-priority
           all-inputs-built?)
    (let ((statement
           (sqlite-prepare
            db
            "
INSERT INTO unprocessed_builds_with_derived_priorities (build_id, derived_priority, all_inputs_built)
VALUES (:build_id, :derived_priority, :all_inputs_built)"
            #:cache? #t)))

      (sqlite-bind-arguments
       statement
       #:build_id build-id
       #:derived_priority derived-priority
       #:all_inputs_built (if all-inputs-built? 1 0))

      (sqlite-step-and-reset statement)))

  (apply
   (lambda* (uuid drv-name priority defer-until
                  #:key skip-updating-other-build-derived-priorities)
     (define system-id
       (call-with-worker-thread
        (slot-ref datastore 'worker-reader-thread-channel)
        (lambda (db)
          (db-system->system-id
           db
           (datastore-find-derivation-system datastore drv-name)))))

     (call-with-worker-thread/delay-logging
      (slot-ref datastore 'worker-writer-thread-channel)
      (lambda (db)
        (let* ((build-id          (insert-build db drv-name uuid priority
                                                defer-until))
               (derived-priority  (max (or (call-with-delay-logging
                                            get-derived-priority
                                            #:args (list db build-id))
                                           priority)
                                       priority))
               (all-inputs-built? (all-inputs-built? db build-id)))

          (increment-builds-counts db system-id)

          (insert-unprocessed-builds-with-derived-priorities-entry
           db
           build-id
           derived-priority
           all-inputs-built?)

          (unless (or all-inputs-built?
                      skip-updating-other-build-derived-priorities)
            (call-with-delay-logging
             update-unprocessed-builds-with-higher-derived-priorities
             #:args
             (list db
                   build-id
                   derived-priority)))))))
   rest)
  #t)

(define (insert-agent db uuid name description)
  (let ((statement
         (sqlite-prepare
          db
          "
INSERT INTO agents (id, name, description)
VALUES (:id, :name, :description)"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:id uuid
     #:name name
     #:description description)

    (sqlite-step-and-reset statement)))

(define (insert-agent-password db uuid password)
  (let ((statement
         (sqlite-prepare
          db
          "
INSERT INTO agent_passwords (agent_id, password)
VALUES (:agent_id, :password)"
          #:cache? #t)))

    (sqlite-bind-arguments
     statement
     #:agent_id uuid
     #:password password)

    (sqlite-step-and-reset statement)))