 (ns samsara.qanal.state-machine
   (:require [safely.core :refer [sleeper]]))
+
+
+
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
+;;                                                                            ;;
+;;        ---==| S T A T E   M A C H I N E   U T I L I T I E S |==----        ;;
+;;                                                                            ;;
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 
 ;;
 ;; This is the implementation of the state machine described
 ;; at doc/state-machine.md
 ;;
 
-;;
-;; Available states
-;;
-;; :extract/init
-;; :extract/check
-;; :extract/poll
-;; :retry
-;; :transform/parse
-;; :transform/validate
-;; :transform/transform
-;; :load/init
-;; :load/bulk-load
-;;
-
 ;; sample state
-{:config {:topic "topic1" :partition 2}
- :fns {:init-consumer (fn [cfg] :a-consemer)}
+{
+ ;;
+ ;; the configuration of what this state machine
+ ;; has to consume and where all the data
+ ;; has to be loaded it is part of the `:config` element
+ ;;
+ :config {;; this is the connect string for the kafka
+          ;; brokers, a comma-separated list
+          ;; of host and ports
+          :brokers ""
+
+          ;; this is the topic and partition to consume
+          :topic "topic1" :partition 2
+
+          ;; the elasticsearch endpoint where
+          ;; to load the data
+          :elasticsearch ""
+          }
+
+ ;; TODO: signatures and doc
+ :fns {:init-consumer (fn [cfg] :new-consumer)
+       :check (fn [c cfg] {["topic1" 2] 2345})
+       :poll (fn [c offset] [1 2 3 4])
+       :parse    1
+       :validate 1
+       :tranform 1
+       :init-load 1
+       :bulk-load 1}
 
  :state :retry
 

(defmethod transition :extract
  [s]
  (with-state-machine s
    :on-error   (dispatch-error
                 :errors/kafka-unavailable    retry-error
                 :errors/topic-not-found      retry-error
                 :errors/partition-not-found  retry-error
                 :errors/network-error        retry-error

                 ;; TODO: err: 4,5
                 ;;:errors/message-too-big      send-to-errors
                 ;;:errors/offset-not-available reset-offset
                 )
    :on-success (fn [s]
                  (if (empty? (get-in s [:data :batch]))
                    s
                    (clean-move-to :transform s)))

    (fn [s]
      (-> s
          extract-init-consumer

     (dispatch-error

       #{:errors/network-error
         :errors/kafka-unavailable}   retry-action

       :errors/message-to-big         send-to-errors

       :errors/offset-not-available   reset-offset

       #(= % :errors/topic-not-found) retry-action
       :errors/partition-not-found    retry-action

      (fn [s e]
         (throw (make-error :errors/uncategorized \"Unknown error\" {} e))))
