path: root/bffe/server.scm

;;; Build Farm Front-End
;;;
;;; Copyright © 2023 Christopher Baines <mail@cbaines.net>
;;;
;;; This program is free software: you can redistribute it and/or
;;; modify it under the terms of the GNU Affero General Public License
;;; as published by the Free Software Foundation, either version 3 of
;;; the License, or (at your option) any later version.
;;;
;;; This program is distributed in the hope that it will be useful,
;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;;; Affero General Public License for more details.
;;;
;;; You should have received a copy of the GNU Affero General Public
;;; License along with this program.  If not, see
;;; <http://www.gnu.org/licenses/>.

(define-module (bffe server)
  #:use-module (srfi srfi-1)
  #:use-module (srfi srfi-11)
  #:use-module (srfi srfi-19)
  #:use-module (srfi srfi-71)
  #:use-module (ice-9 vlist)
  #:use-module (ice-9 threads)
  #:use-module (ice-9 match)
  #:use-module (ice-9 atomic)
  #:use-module (ice-9 textual-ports)
  #:use-module (rnrs bytevectors)
  #:use-module (web http)
  #:use-module (web client)
  #:use-module (web request)
  #:use-module (web response)
  #:use-module (web uri)
  #:use-module (json)
  #:use-module (prometheus)
  #:use-module (system repl error-handling)
  #:use-module (fibers)
  #:use-module (fibers channels)
  #:use-module (fibers conditions)
  #:use-module (fibers web server)
  #:use-module (guix store)
  #:use-module ((guix build utils) #:select (dump-port))
  #:use-module (guix-data-service web util)
  #:use-module ((guix-build-coordinator utils)
                #:select (with-time-logging get-gc-metrics-updater
                           call-with-delay-logging))
  #:use-module ((guix-build-coordinator utils fibers)
                #:select (run-server/patched call-with-sigint))
  #:use-module (guix-build-coordinator client-communication)
  #:use-module (bffe view util)
  #:use-module (bffe view home)
  #:use-module (bffe view activity)
  #:export (start-bffe-web-server))

(define (make-state-channel)
  (let ((channel (make-channel)))
    (call-with-new-thread
     (lambda ()
       (let loop ((last-fetch-time #f)
                  (state #f))
         (let ((reply-channel (get-message channel))
               (state-age-seconds
                (and last-fetch-time
                     (- (time-second (current-time))
                        last-fetch-time))))
           (if (or (not state-age-seconds)
                   (> state-age-seconds
                      120))
               (let ((response
                      body
                      (http-get
                       (string->uri "http://localhost:8746/state"))))
                 (let ((state
                        (json-string->scm
                         (utf8->string body))))
                   (put-message reply-channel
                                state)
                   (loop (time-second (current-time))
                         state)))
               (begin
                 (put-message
                  reply-channel
                  state)
                 (loop last-fetch-time
                       state)))))))

    channel))

(define (get-state state-channel)
  (let ((reply-channel (make-channel)))
    (put-message state-channel
                 reply-channel)

    (get-message reply-channel)))

(define (make-events-channel initial-state-id)
  (let* ((submission-channel (make-channel))
         (listener-channels-box (make-atomic-box vlist-null))

         (buffer-size 10000)
         (event-buffer (make-vector buffer-size))

         (current-state-id-and-event-buffer-index-box
          (make-atomic-box (cons 0 -1))))

    (define (get-state-id)
      (match (atomic-box-ref
              current-state-id-and-event-buffer-index-box)
        ((current-state-id . event-buffer-index)
         current-state-id)))

    (define (spawn-fiber-for-listener callback
                                      after-state-id
                                      submission-channel
                                      listener-channel
                                      listening-finished-channel)
      (spawn-fiber
       (lambda ()
         (let loop ((last-sent-state-id after-state-id))
           (let ((new-state-id
                  (with-exception-handler
                      (lambda (exn)
                        (simple-format
                         (current-error-port)
                         "exception listening for events: ~A\n" exn)
                        (put-message submission-channel
                                     (list 'remove-listener
                                           listener-channel))

                        (put-message listening-finished-channel #t)

                        #f)
                    (lambda ()
                      (with-throw-handler #t
                        (lambda ()
                          (match (atomic-box-ref
                                  current-state-id-and-event-buffer-index-box)
                            ((current-state-id . event-buffer-index)
                             (let ((event-count-to-send
                                    (- current-state-id last-sent-state-id)))
                               (for-each
                                (lambda (index index-state-id)
                                  (match (vector-ref event-buffer index)
                                    ((event-state-id event-name data)

                                     (when (not (= event-state-id index-state-id))
                                       (error "listener behind"))

                                     (callback event-state-id event-name data))))
                                (map (lambda (i)
                                       (modulo i buffer-size))
                                     (iota event-count-to-send
                                           (- event-buffer-index
                                              (- event-count-to-send 1))))
                                (iota event-count-to-send
                                      (+ 1 last-sent-state-id))))

                             current-state-id)))
                        (lambda (key . args)
                          (peek "ERROR" key args)
                          (if (and
                               (eq? key 'system-error)
                               (match args
                                 (("fport_write" "~A" ("Broken pipe") rest ...)
                                  #t)
                                 (_ #f)))
                              #f
                              (backtrace)))))
                    #:unwind? #t)))

             (unless (eq? #f new-state-id)
               (get-message listener-channel)
               (loop new-state-id)))))
       #:parallel? #t))

    (define (store-event id event-name data)
      (match (atomic-box-ref
              current-state-id-and-event-buffer-index-box)
        ((current-state-id . event-buffer-index)
         (let ((new-event-index
                (modulo
                 (+ 1 event-buffer-index)
                 buffer-size)))

           (atomic-box-set! current-state-id-and-event-buffer-index-box
                            (cons id new-event-index))

           (vector-set! event-buffer
                        new-event-index
                        (list id event-name data))))))

    (spawn-fiber
     (lambda ()
       (let* ((response
               remote-port
               (http-get
                (string->uri "http://localhost:8746/events")
                #:headers
                `((last-event-id
                   . ,(number->string
                       (- initial-state-id
                          ;; Get some earlier events as well, as clients that
                          ;; are reconnecting might want them
                          100))))
                #:streaming? #t)))
         (setvbuf remote-port 'none)

         (let loop ((line (get-line remote-port)))
           (or (eof-object? line)
               ;; TODO: It would be good to replace this with a more general way
               ;; of processing the event stream that handles
               (begin
                 (when (string-prefix? "id: " line)
                   (let ((id
                          (string->number (string-drop line 4)))
                         (event-name
                          (string-drop
                           (get-line remote-port)
                           (string-length
                            "event: ")))
                         (data
                          (json-string->scm
                           (string-drop
                            (get-line remote-port)
                            (string-length "data: ")))))
                     (put-message
                      submission-channel
                      (list id event-name data))))

                 (loop (get-line remote-port)))))))
     #:parallel? #t)

    (spawn-fiber
     (lambda ()
       (while #t
         (match (get-message submission-channel)
           (('new-listener callback requested-after-state-id
                           listening-finished-channel)

            (let ((listener-channel (make-channel))
                  (after-state-id
                   (match (atomic-box-ref
                           current-state-id-and-event-buffer-index-box)
                     ((current-state-id . event-buffer-index)
                      (if (and requested-after-state-id
                               (<= requested-after-state-id
                                   current-state-id)
                               (match (vector-ref
                                       event-buffer
                                       (modulo
                                        (+ event-buffer-index
                                           (- requested-after-state-id
                                              current-state-id))
                                        buffer-size))
                                 ((? unspecified? _) #f)
                                 ((event-state-id event-name data)
                                  (= event-state-id
                                     requested-after-state-id))))
                          requested-after-state-id
                          current-state-id)))))
              (atomic-box-set!
               listener-channels-box
               (vhash-consq listener-channel
                            #t
                            (atomic-box-ref listener-channels-box)))

              (spawn-fiber-for-listener callback
                                        after-state-id
                                        submission-channel
                                        listener-channel
                                        listening-finished-channel)))

           (('remove-listener listener-channel)
            (atomic-box-set!
             listener-channels-box
             (vhash-delq listener-channel
                         (atomic-box-ref listener-channels-box))))

           ((id event-name data)
            (store-event id event-name data)

            (vhash-fold
             (lambda (listener-channel val res)
               (spawn-fiber
                (lambda ()
                  (put-message listener-channel #t))
                #:parallel? #t)
               #f)
             #f
             (atomic-box-ref listener-channels-box)))))))

    (values submission-channel
            get-state-id)))

(define* (listen-for-events events-channel callback
                            #:key after-state-id)
  (let ((listening-finished-channel
         (make-channel)))
    (put-message
     events-channel
     (list 'new-listener
           callback
           after-state-id
           listening-finished-channel))
    ;; This is designed to be useful in the controller, so this procedure
    ;; shouldn't exit until callback has finished being called
    (get-message listening-finished-channel)))

(define* (make-controller assets-directory metrics-registry
                          events-channel state-channel
                          title
                          template-directory)

  (define handle-static-assets
    (if (string-prefix? (%store-prefix)
                        assets-directory)
        (static-asset-from-store-renderer assets-directory)
        (static-asset-from-directory-renderer assets-directory)))

  (define gc-metrics-updater!
    (get-gc-metrics-updater metrics-registry))

  (define home-template-content
    (call-with-input-file
        (string-append
         template-directory "/home.html")
      get-string-all))

  (lambda (request
           method-and-path-components
           mime-types
           body)

    (define path
      (uri-path (request-uri request)))

    (match method-and-path-components
      (('GET)
       (render-html
        #:sxml (home title home-template-content)))
      (('GET "activity")
       (render-html
        #:sxml (activity title (get-state state-channel))))
      (('GET "state")
       (render-json
        (get-state state-channel)))
      (('GET "events")
       (let ((headers (request-headers request))
             (query-parameters
              (let lp ((lst
                        (or (and=> (uri-query (request-uri request))
                                   (lambda (query)
                                     (map uri-decode
                                          (string-split
                                           query
                                           (char-set #\& #\=)))))
                            '())))
                (match lst
                  ((key value . rest)
                   (cons (cons key value) (lp rest)))
                  (("") '())
                  (() '())))))

         (list (build-response
                #:code 200
                #:headers '((content-type . (text/event-stream))))
               (lambda (port)
                 (listen-for-events
                  events-channel
                  (lambda (id event-name data)
                    (display
                     (simple-format #f "id: ~A\nevent: ~A\ndata: ~A\n\n"
                                    id
                                    event-name
                                    (scm->json-string data))
                     port)
                    (force-output port)
                    ;; TODO because the chunked output port doesn't call
                    ;; force-output on the underlying port, do that here. We
                    ;; want this event to be sent now, rather than when some
                    ;; buffer fills up.
                    (force-output (request-port request)))
                  #:after-state-id
                  (with-exception-handler
                      (lambda (exn)
                        (simple-format
                         (current-error-port)
                         "error: when processing Last-Event-ID header value: ~A\n"
                         (assoc-ref headers 'last-event-id))
                        #f)
                    (lambda ()
                      (or (and=> (assoc-ref headers 'last-event-id)
                                 string->number)
                          (and=> (assoc-ref query-parameters "last_event_id")
                                 string->number)))
                    #:unwind? #t))))))

      (('GET "assets" rest ...)
       (or (handle-static-assets (string-join rest "/")
                                 (request-headers request))
           (list (build-response #:code 404)
                 (string-append
                  "Resource not found: "
                  (uri->string
                   (request-uri request))))))
      (('GET "metrics")
       (gc-metrics-updater!)
       (list (build-response
              #:code 200
              #:headers '((content-type . (text/plain))
                          (vary . (accept))))
             (lambda (port)
               (write-metrics metrics-registry port))))
      ((method path ...)
       (render-html
        #:sxml (general-not-found
                "Page not found"
                "")
        #:code 404)))))

(define (handler request body controller)
  (display
   (format #f "~a ~a\n"
           (request-method request)
           (uri-path (request-uri request))))

  (call-with-error-handling
   (lambda ()
     (let-values (((request-components mime-types)
                   (request->path-components-and-mime-type request)))
       (controller request
                   (cons (request-method request)
                         request-components)
                   mime-types
                   body)))
   #:on-error 'backtrace
   #:post-error (lambda args
                  (render-html #:sxml (error-page args)
                               #:code 500))))

(define* (start-bffe-web-server port host assets-directory
                                metrics-registry
                                #:key (controller-args '()))
  (define state-channel
    (make-state-channel))

  (call-with-error-handling
   (lambda ()
     (let ((finished? (make-condition)))
       (call-with-sigint
        (lambda ()
          (run-fibers
           (lambda ()
             (let* ((initial-state-id
                     (assoc-ref
                      (get-state state-channel)
                      "state_id"))
                    (events-channel
                     get-state-id
                     (make-events-channel
                      initial-state-id))
                    (controller
                     (apply make-controller assets-directory
                            metrics-registry
                            events-channel state-channel
                            controller-args)))

               ;; Wait until the events channel catches up
               (while (< (get-state-id) initial-state-id)
                 (sleep 10))

               (simple-format #t "Starting the server\n")
               (run-server/patched (lambda (request body)
                                     (apply values
                                            (handler request body controller)))
                                   #:host host
                                   #:port port))

             (wait finished?))))
        finished?)))
   #:on-error 'backtrace
   #:post-error (lambda (key . args)
                  (when (eq? key 'system-error)
                    (match args
                      (("bind" "~A" ("Address already in use") _)
                       (simple-format
                        (current-error-port)
                        "\n
error: bffe could not start, as it could not bind to port ~A

Check if it's already running, or whether another process is using that
port. Also, the port used can be changed by passing the --port option.\n"
                        port)))))))