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 (ice-9 match)
  #:use-module (ice-9 threads)
  #:use-module (sqlite3)
  #:use-module (guix derivations)
  #:use-module (guix-build-coordinator utils)
  #:use-module (guix-build-coordinator config)
  #:use-module (guix-build-coordinator datastore abstract)
  #:export (sqlite-datastore
            datastore-update
            datastore-store-derivation
            datastore-store-build
            datastore-find-build
            datastore-store-build-result
            datastore-new-agent
            datastore-list-agents
            datastore-find-agent
            datastore-fetch-setup-failures
            datastore-store-setup-failure
            datastore-store-setup-failure/missing-inputs
            datastore-find-derivation-outputs
            datastore-new-agent-password
            datastore-agent-password-exists?
            datastore-list-unprocessed-builds
            datastore-list-agent-builds
            datastore-agent-for-build
            datastore-replace-build-allocation-plan
            datastore-allocate-builds-to-agent
            datastore-list-allocation-plan-builds))

(define-class <sqlite-datastore> (<abstract-datastore>)
  database-file
  worker-thread-channel)

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

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

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

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

    (slot-set!
     datastore
     'worker-thread-channel
     (make-worker-thread-channel (lambda ()
                                   (list (db-open database-file)))
                                 #:parallelism
                                 (min (current-processor-count) 4)))

    datastore))

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

       (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-new-agent
                (datastore <sqlite-datastore>)
                uuid
                description)
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (insert-agent db uuid description)))
  #t)

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

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

         agents)))))

(define-method (datastore-new-agent-password
                (datastore <sqlite-datastore>)
                agent-uuid
                password)
  (call-with-worker-thread
   (slot-ref datastore 'worker-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-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT 1 FROM agent_passwords \
WHERE agent_id = :agent_id AND password = :password")))

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

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

         result)))))

(define-method (datastore-store-derivation
                (datastore <sqlite-datastore>)
                derivation)
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (sqlite-exec db "BEGIN TRANSACTION;")
     (insert-derivation-and-return-outputs db derivation)
     (sqlite-exec db "COMMIT TRANSACTION;")))
  #t)

(define-method (datastore-store-build
                (datastore <sqlite-datastore>)
                derivation-name
                uuid
                priority)
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (sqlite-exec db "BEGIN TRANSACTION;")
     (insert-build db uuid derivation-name priority)
     (sqlite-exec db "COMMIT TRANSACTION;")))
  #t)

(define-method (datastore-store-build-result
                (datastore <sqlite-datastore>)
                build-id
                agent-id
                result
                failure-reason)
  (define (insert-build-result db build-id agent-id result failure-reason)
    (sqlite-exec
     db
     (string-append
      "
INSERT INTO build_results (
  build_id, agent_id, result, failure_reason
) VALUES ('"
      build-id "', '"
      agent-id "', '"
      result "', "
      (if failure-reason
          (string-append "'" failure-reason "'")
          "NULL")
      ")")))

  (define (remove-build-allocation db build-id agent-id)
    (sqlite-exec
     db
     (string-append
      "
DELETE FROM allocated_builds WHERE build_id = '"
      build-id
      "' AND agent_id = '"
      agent-id
      "'")))

  (define (mark-build-as-processed db build-id)
    (sqlite-exec
     db
     (string-append
      "
UPDATE builds SET processed = 1 WHERE uuid = '" build-id "'")))

  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (sqlite-exec db "BEGIN TRANSACTION;")
     (insert-build-result db build-id agent-id result failure-reason)
     (remove-build-allocation db build-id agent-id)
     (mark-build-as-processed db build-id)
     (sqlite-exec db "COMMIT TRANSACTION;")))
  #t)

(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 = '"
    build-id
    "' AND agent_id = '"
    agent-id
    "'"))

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

  (last-insert-rowid db))

(define-method (datastore-store-setup-failure/missing-inputs
                (datastore <sqlite-datastore>)
                build-id
                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)
       ", "))))

  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (sqlite-exec db "BEGIN TRANSACTION;")
     (let ((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))
     (sqlite-exec db "COMMIT TRANSACTION;")))
  #t)

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

(define-method (datastore-find-build
                (datastore <sqlite-datastore>)
                uuid)
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT uuid, derivation_name, priority, processed
FROM builds
WHERE uuid = :uuid")))

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

       (let ((result
              (match (sqlite-step statement)
                (#(uuid derivation_name priority processed)
                 `((uuid            . ,uuid)
                   (derivation-name . ,derivation_name)
                   (priority        . ,priority)
                   (processed       . ,processed))))))
         (sqlite-reset statement)

         result)))))

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

  #t)

(define-method (datastore-fetch-setup-failures
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT id, build_id, agent_id, failure_reason FROM setup_failures")))

       (let ((result (sqlite-fold
                      (lambda (row result)
                        (match row
                          (#(id build-id agent-id failure-reason)
                           (let ((failures-for-build-id
                                  (or (assoc-ref result build-id)
                                      '())))
                             (alist-cons
                              build-id
                              (cons `((id  . ,id)
                                      (agent-id . ,agent-id)
                                      (failure-reason . ,failure-reason))
                                    failures-for-build-id)
                              (alist-delete build-id result))))))
                      '()
                      statement)))
         (sqlite-reset statement)

         result)))))

(define-method (datastore-list-unprocessed-builds
                (datastore <sqlite-datastore>))
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT uuid, derivation_name, priority FROM builds WHERE processed = 0")))

       (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-replace-build-allocation-plan
                (datastore <sqlite-datastore>)
                planned-builds)
  (define (clear-current-plan db)
    (sqlite-exec
     db
     "DELETE FROM build_allocation_plan"))

  (define (insert-new-plan db planned-builds)
    (sqlite-exec
     db
     (string-append
      "
INSERT INTO build_allocation_plan (build_id, agent_id, ordering) VALUES "
      (string-join
       (map (match-lambda
              ((build-id agent-id ordering)
               (simple-format
                #f
                "('~A', '~A', ~A)"
                build-id
                agent-id
                ordering)))
            planned-builds)
       ", ")
      ";")))

  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (sqlite-exec db "BEGIN TRANSACTION;")
     (clear-current-plan db)
     (unless (null? planned-builds)
       (insert-new-plan db planned-builds))
     (sqlite-exec db "COMMIT TRANSACTION;")))
  #t)

(define-method (datastore-allocate-builds-to-agent
                (datastore <sqlite-datastore>)
                agent-id
                build-ids)
  (define (insert-to-allocated-builds db agent-id build-ids)
    (sqlite-exec
     db
     (string-append
      "
INSERT INTO allocated_builds (build_id, agent_id) VALUES "
      (string-join
       (map (lambda (build-id)
              (simple-format
               #f
               "('~A', '~A')"
               build-id
               agent-id))
            build-ids)
       ", ")
      ";")))

  (define (remove-builds-from-plan db build-ids)
    (sqlite-exec
     db
     (string-append
      "
DELETE FROM build_allocation_plan
WHERE build_id IN ("
      (string-join
       (map (lambda (build-id)
              (string-append "'" build-id "'"))
            build-ids)
       ", ")
      ")")))

  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (sqlite-exec db "BEGIN TRANSACTION;")
     (insert-to-allocated-builds db agent-id build-ids)
     (remove-builds-from-plan db build-ids)
     (sqlite-exec db "COMMIT TRANSACTION;")))
  #t)

(define-method (datastore-list-allocation-plan-builds
                (datastore <sqlite-datastore>)
                agent-id
                limit)
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT builds.uuid, builds.derivation_name
FROM builds
INNER JOIN build_allocation_plan
  ON builds.uuid = build_allocation_plan.build_id
WHERE build_allocation_plan.agent_id = :agent_id
LIMIT :limit")))

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

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

         builds)))))

(define-method (datastore-list-agent-builds
                (datastore <sqlite-datastore>)
                agent-id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT builds.uuid, builds.derivation_name, builds.priority
FROM builds
INNER JOIN allocated_builds
  ON builds.uuid = allocated_builds.build_id
WHERE allocated_builds.agent_id = :agent_id")))

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

       (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-agent-for-build
                (datastore <sqlite-datastore>)
                build-id)
  (call-with-worker-thread
   (slot-ref datastore 'worker-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")))

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

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

         result)))))

(define (db-open database)
  (define flags
    (list SQLITE_OPEN_READWRITE
          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)
               "--chdir" (dirname (dirname (%config 'sqitch-plan)))
               "--plan-file" (%config 'sqitch-plan)
               (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();")))
    (let ((count
           (vector-ref (sqlite-step statement)
                       0)))

      (sqlite-reset statement)

      count)))

(define (last-insert-rowid db)
  (let ((statement
         (sqlite-prepare
          db
          "SELECT last_insert_rowid();")))
    (let ((id
           (vector-ref (sqlite-step statement)
                       0)))

      (sqlite-reset statement)

      id)))

(define (select-derivation-outputs db derivation-name)
  (let ((statement
         (sqlite-prepare
          db
          "
SELECT name, id FROM derivation_outputs WHERE derivation_name = :derivation_name")))

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

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

      outputs)))

(define-method (datastore-find-derivation-outputs
                (datastore <sqlite-datastore>)
                derivation-name)
  (call-with-worker-thread
   (slot-ref datastore 'worker-thread-channel)
   (lambda (db)
     (let ((statement
            (sqlite-prepare
             db
             "
SELECT name, output
FROM derivation_outputs
WHERE derivation_name = :derivation_name")))

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

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

         result)))))

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

  (define (insert-derivation)
    (let ((statement
           (sqlite-prepare
            db
            "
INSERT OR IGNORE INTO derivations (name, system) VALUES (:name, :system)")))

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

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

      (changes-count db)))

  (let ((changes (insert-derivation)))
    (unless (eq? changes 0)
      (insert-derivation-inputs
       db
       derivation-name
       (derivation-inputs derivation))

      (insert-derivation-outputs
       db
       derivation-name
       (derivation-outputs derivation)))

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

(define (insert-derivation-inputs db derivation-name derivation-inputs)
  (unless (null? derivation-inputs)
    (let ((derivation-output-ids
           (append-map
            (lambda (derivation-input)
              (let ((output-ids-by-name
                     (insert-derivation-and-return-outputs
                      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)))
      (sqlite-exec
       db
       (string-append
        "
INSERT INTO derivation_inputs (derivation_name, derivation_output_id) VALUES "
        (string-join
         (map (lambda (derivation-output-id)
                (simple-format
                 #f
                 "('~A', ~A)"
                 derivation-name
                 derivation-output-id))
              derivation-output-ids)
         ", ")
        ";")))))

(define (insert-derivation-outputs db derivation-name derivation-outputs)
  (sqlite-exec
   db
   (string-append
    "
INSERT INTO derivation_outputs (derivation_name, name, output) VALUES "
    (string-join
     (map (match-lambda
            ((name . derivation-output)
             (simple-format
              #f
              "('~A', '~A', '~A')"
              derivation-name
              name
              (derivation-output-path derivation-output))))
          derivation-outputs)
     ", ")
    ";")))

(define (insert-build db uuid derivation-name priority)
  (let ((statement
         (sqlite-prepare
          db
          "
INSERT INTO builds (uuid, derivation_name, priority)
VALUES (:uuid, :derivation_name, :priority)")))

    (sqlite-bind-arguments
     statement
     #:uuid uuid
     #:derivation_name derivation-name
     #:priority priority)

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

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

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

    (sqlite-step statement)
    (sqlite-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)")))

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

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