qkbox: toot: Add Streaming API (HTTP).

1 files changed, 158 insertions, 0 deletions

diff --git a/qkbox/toot.scm b/qkbox/toot.scm
index ad6b742..70c97b8 100644
--- a/qkbox/toot.scm
+++ b/qkbox/toot.scm
@@ -660,3 +660,161 @@
   (request
    'GET
    (format #f "/api/v1/accounts/~a" id)))
+
+(define (streaming-health?)
+  (receive (res body)
+      (/api/v1/streaming/health)
+    (case (response-code res)
+      ((200)
+       (let ((result (utf8->string (get-bytevector-all body))))
+         (close-input-port body)
+         (string=? "OK" result)))
+      (else
+       #f))))
+
+(define (/api/v1/streaming/health)
+  (raw-request 'GET "/api/v1/streaming/health" #:streaming? #t))
+
+(define (utf8-read-line in)
+  (call/cc
+   (lambda (k)
+     (utf8->string
+      (call-with-bytevector-output-port
+       (lambda (out)
+         (let loop ((b (get-u8 in)))
+           (cond
+            ((eof-object? b)
+             (k b))
+            ((or (= b 10))
+             'done)
+            (else
+             (put-u8 out b)
+             (loop (get-u8 in)))))))))))
+
+(define (read-streaming port)
+  (let ((line (utf8-read-line port)))
+    (cond
+     ((eof-object? line)
+      (values (eof-object) ""))
+     (else
+      (let* ((i (string-index line #\:)))
+        (if (or (not i)
+                (<= i 1))
+            (read-streaming port)
+            (values
+             (string-trim-both (substring line 0 i))
+             (string-trim-both
+              (substring line (+ i 1))))))))))
+
+(define (streaming-user handler)
+  (streaming /api/v1/streaming/user handler))
+
+(define* (streaming-public handler #:key only-media?)
+  (streaming (if only-media?
+                 /api/v1/streaming/public
+                 /api/v1/streaming/public?only_media=true)
+             handler))
+
+(define* (streaming-local handler #:key only-media?)
+  (streaming (if only-media?
+                 /api/v1/streaming/local
+                 /api/v1/streaming/local?only_media=true)
+             handler))
+
+(define (streaming-hashtag hashtag handler)
+  (streaming (lambda ()
+               (/api/v1/streaming/hashtag?tag=:hashtag hashtag))
+             handler))
+
+(define (streaming-local-hashtag hashtag handler)
+  (streaming (lambda ()
+               (/api/v1/streaming/hashtag/local?tag=:hashtag hashtag))
+             handler))
+
+(define (streaming-list list-id handler)
+  (streaming (lambda ()
+               (/api/v1/streaming/list?list=:list_id list-id))
+             handler))
+
+(define* (streaming-direct handler)
+  (streaming /api/v1/streaming/direct handler))
+
+(define (streaming streamer handler)
+  (receive (res body)
+      (streamer)
+    (case (response-code res)
+      ((200)
+       (dynamic-wind
+         (lambda () 'ok)
+         (lambda ()
+           (let loop ((event #f))
+             (receive (type data)
+                 (read-streaming body)
+               (cond
+                ((eof-object? type) 'end)
+                ((string=? type "event")
+                 (loop (string->symbol data)))
+                ((string=? type "data")
+                 (case event
+                   ((update)
+                    (handler event
+                             (make-status
+                              (json-string->scm data))))
+                   ((delete)
+                    (handler event data))
+                   ((notification)
+                    (handler event
+                             (make-notification
+                              (json-string->scm data))))
+                   (else
+                    (handler event data)))
+                 (loop #f))
+                (else
+                 (format #t "[DEBUG] ~a: ~a" event data)
+                 (loop #f))))))
+         (lambda ()
+           (close-input-port body))))
+      (else #f))))
+
+(define (/api/v1/streaming/user)
+  (raw-request 'GET "/api/v1/streaming/user"
+               #:streaming? #t
+               #:authorization? #t))
+
+(define (/api/v1/streaming/public)
+  (raw-request 'GET "/api/v1/streaming/public"
+               #:streaming? #t))
+
+(define (/api/v1/streaming/public?only_media=true)
+  (raw-request 'GET "/api/v1/streaming/public?only_media=true"
+               #:streaming? #t))
+
+(define (/api/v1/streaming/local)
+  (raw-request 'GET "/api/v1/streaming/local"
+               #:streaming? #t))
+
+(define (/api/v1/streaming/local?only_media=true)
+  (raw-request 'GET "/api/v1/streaming/local?only_media=true"
+               #:streaming? #t))
+
+(define (/api/v1/streaming/hashtag?tag=:hashtag hashtag)
+  (raw-request 'GET
+               (format #f "/api/v1/streaming/hashtag?tag=~a" hashtag)
+               #:streaming? #t))
+
+(define (/api/v1/streaming/hashtag/local?tag=:hashtag hashtag)
+  (raw-request 'GET
+               (format #f "/api/v1/streaming/hashtag/local?tag=~a" hashtag)
+               #:streaming? #t))
+
+(define (/api/v1/streaming/list?list=:list_id hashtag)
+  (raw-request 'GET
+               (format #f "/api/v1/streaming/list?list=~a" hashtag)
+               #:streaming? #t
+               #:authorization? #t))
+
+(define (/api/v1/streaming/direct)
+  (raw-request 'GET
+               "/api/v1/streaming/direct"
+               #:streaming? #t
+               #:authorization? #t))