(defpackage :fwoar.websocket-sink
   (:use :cl )
   (:export ))
 (in-package :fwoar.websocket-sink)
 

 (defparameter *obj-style*
   :hash-table)
 (defparameter *event-table*
   (make-array 100
               :adjustable t
               :fill-pointer 0))
 (defparameter *event-queue*
   (lparallel.queue:make-queue))

 
 (defun parse-fa-body (ht)
   (prog1 ht
     (ecase *obj-style*

       (:hash-table
        (let ((body (gethash "body" ht)))
          (setf (gethash "body" ht)
                (or (when (not (equal body ""))
                      (handler-case
                          (yason:parse body
                                       :object-as *obj-style*)
                        (error ())))
                    body))))
       #+(or)
       (:alist (pushnew (assoc "body" ht)
                        (yason:parse (gethash "body" ht)
                                     :object-as *obj-style*))))))

 
 (defparameter *handlers* (make-hash-table :test 'equal))
 (defgeneric handle-message (path message)
   (:method :around ((path string) message)
     (alexandria:if-let ((handler (gethash path *handlers*)))
       (handle-message handler message)
       (call-next-method)))
   (:method (path message)
     message))
 
 (defun parse (string)
   (destructuring-bind (path result)
       (coerce (fwoar.string-utils:split #\space string
                                         :count 1)
               'list)
     (format t "~&(PATH ~a) (RESULT ~a)~%"
             path result)
     (list path
           (handle-message path
                           (parse-fa-body
                            (yason:parse result
                                         :object-as *obj-style*))))))
 
 (defun get-and-process-one-message (socket)
   (let* ((raw-reply (pzmq:recv-string socket :encoding :utf-8)))
     (format t "~&~a~%" raw-reply)
     (destructuring-bind (path payload . r) (parse raw-reply)
       (when r
         (format t "~&extra stuff? PATH: ~a~%~4t~a~%~4t~a~2%"
                 path
                 payload
                 r)) 
       payload)))
 
 (defmacro with-hub ((symbol port subscription) &body body)
   (alexandria:once-only (port subscription)
     `(pzmq:with-socket ,symbol (:sub :subscribe ,subscription)
        (pzmq:connect ,symbol ,port)
        ,@body)))
 
 (defvar *server*)
 (defvar *ws-server*)
 
 (defvar listeners ())

 (defun update-listeners (new-listeners)
   (setf listeners new-listeners))
 

 (defun main (&optional (wait t) (subscription "/") (port "tcp://127.0.0.1:5557"))
   (clack:clackup (lambda (env)
                    (let ((server (websocket-driver:make-server env)))
                      (event-emitter:on :open server
                                        (lambda ()
                                          (push server listeners)))
                      (event-emitter:on :close server
                                        (lambda (&key code reason)
                                          (declare (ignore code reason))
                                          (setf listeners

                                                (remove server listeners))
                                          (update-listeners listeners)))

                      (lambda (responder)
                        (declare (ignore responder))
                        (websocket-driver:start-connection server))))
                  :port 5009)
   (if wait
       (pzmq-process port subscription)
       (bt:make-thread (lambda () (pzmq-process port subscription))
                       :name "websocket-handler")))
 
 (defun handle (reply)
   (mapcar (serapeum:op (handler-case
                            (websocket-driver:send _1
                                                   (With-output-to-string (s)
                                                     (yason:encode reply
                                                                   s)))
                          (error ()
                            (format t "~&failed to push ~s to ~s~%" reply _1))))
           listeners))
 
 (defun pzmq-process (port subscription)
   (with-hub (pzmq port subscription)
     (loop for raw-reply = (pzmq:recv-string pzmq :encoding :utf-8)
           for (_ reply) = (parse raw-reply)
           do (format t "~&REPLY: ~s~%" reply)
           do (handle reply))))