<commit_before>(ns re-frame-async-flow-fx.core
  (:require [re-frame.core]
            [clojure.set :as set]))


(defn startable-tasks
  "Given a set of 'now-seen-events' and a set of already started tasks
  return the list of tasks which should now be started."
  [rules now-seen-events started-tasks]
  (->> (remove (comp started-tasks :id) rules)
       (filterv (fn [task]
                  (condp = (:when task)
                    :seen-all-of  (empty? (set/difference (:events task)  now-seen-events))
                    :seen-any-of  (seq (set/intersection now-seen-events (:events task))))))))


(defn make-flow-event-handler
  [{:keys [id db-path rules first-dispatch]}]
  (let [all-events  (apply set/union (map :events rules))       ;; all of the events refered to in the spec
        new-state (fn [db seen started]
                    (assoc-in db db-path {:seen-events seen :started-tasks started}))


        ;;  rules'  XXX make all :dispatch a list PLUS make :done   [id :done]   then use rules' below
        ;;
        ]

    ;; return an event handler which will manage the flow
    ;; This event handler will receive 3 kinds of events
    ;;   (dispatch [:id :steup])
    ;;   (dispatch [:id :done])
    ;;   (dispatch [:id [:a :forwarded :event :vector])
    ;;
    (fn flow-event-hander
      [{:keys [db]} event-v]
        (condp = event-v

              ;; Setup the flow coordinator:
              ;;  1. Initialise the state  (seen-events and started-tasks)
              ;;  2. dispatch the first event, to kick start
              ;;  3. arrange for the events to get forwarded to this handler
              :setup {:db (new-state db #{} #{})
                      :dispatch first-dispatch
                      :event-forwarder {:register id
                                        :events   all-events
                                        :to       id}}

              ;; Teardown the flow coordinator:
              ;;  1. remove this event handler
              ;;  2. remove state
              ;;  3. stop the events forwarder
              :done (do
                      (re-frame.core/clear-event-handler! id)
                      {:db (dissoc db db-path)
                       :event-forwarder {:unregister id}})

              ;;
              ;; A new event has been forwarded to this handler. What does it mean?
              ;;  1. does this new event mean we need to dispatch another
              ;;  2. remember that this new event has happened
              (let [[event-id forwarded-event] event-v
                    _   (assert (= id event-id))
                    _   (assert (vector? forwarded-event))
                    forwarded-event-id     (first forwarded-event)
                    {:keys [seen-events started-tasks]} (get-in db db-path)
                    new-seen-events    (conj seen-events forwarded-event-id)
                    ready-tasks    (startable-tasks rules new-seen-events started-tasks)
                    ready-task-ids (->> ready-tasks (map :dispatch) set)
                    new-started-tasks (set/union started-tasks ready-task-ids)]
                {:db       (new-state db new-seen-events new-started-tasks)
                 :dispatch (concat (map :dispatch ready-tasks))})))))

(defn register-fx!
  []
  (re-frame.core/def-fx
    :aync-flow
    (fn [{:as flow :keys [id]}]
      ;; XXX check that flow is valid.
      (re-frame.core/def-event-fx
        id                           ;; add debug middleware???  XXXX
        (make-flow-event-handler flow))
      (re-frame.core/dispatch [id :steup]))))

<commit_msg>WIP on code<commit_after>(ns re-frame-async-flow-fx.core
  (:require [re-frame.core]
            [clojure.set :as set]))

(def default-id  :async/flow)

(defn all-events-seen?
  [required-events seen-events]
  (empty? (set/difference required-events seen-events)))


(defn any-events-seen?
  [required-events seen-events]
  (seq (set/intersection seen-events required-events)))

(def when->fn {:all-events-seen all-events-seen? :any-events-seen any-events-seen?})


(defn newly-startable-tasks
  "Given the accumulated set of seen events and the set of already started tasks,
  return the list of tasks which should now be started.
  In effect, give that the list of events seen has changed, what is the consequence."
  [rules now-seen-events started-tasks]
  (->> (remove (comp started-tasks :id) rules)
       (filterv (fn [task] ((:when task) (:events task) now-seen-events)))


(defn make-flow-event-handler
  [{:keys [id db-path rules first-dispatch]}]
  (let [id          (or id default-id)

        ;; Subject to db-path, state is either stored in app-db or in a local atom
        ;; Two pieces of state are maintained:
        ;;  - the set of seen events
        ;;  - the set of started tasks
        local-store (atom {})
        set-state   (if db-path
                      (fn [db seen started]
                        (assoc-in db db-path {:seen-events seen :started-tasks started}))
                      (fn [db seen started]
                        (reset! local-store {:seen-events seen :started-tasks started})
                        db))
        get-state   (if db-path
                      (fn [db] (get-in db db-path))
                      (fn [_]  @local-store))

        ;; Tweak supplied rules:
        ;; - replace `:when` keyword value with a function
        ;; - ensure that `:dispatch` is always a list and turn  `:done` into the right thing
        ;; - ensure that :events is a set
        rules   (-> rules
                    (map (fn [rule]
                           (update rule :when #(or (when->fn %1) %1))))
                    (map (fn [rule]
                           (update rule :dispatch (cond )))))

        all-events  (apply set/union (map :events rules))       ;; all of the events refered to in the spec
        ]

    ;; return an event handler which will manage the flow
    ;; This event handler will receive 3 kinds of events
    ;;   (dispatch [:id :setup])
    ;;   (dispatch [:id :done])
    ;;   (dispatch [:id [:a :forwarded :event :vector])
    ;;
    (fn flow-event-hander
      [{:keys [db]} event-v]
        (condp = event-v

              ;; Setup the flow coordinator:
              ;;  1. Initialise the state  (seen-events and started-tasks)
              ;;  2. dispatch the first event, to kick start
              ;;  3. arrange for the events to get forwarded to this handler
              :setup {:db (set-state db #{} #{})
                      :dispatch first-dispatch
                      :event-forwarder {:register id
                                        :events   all-events
                                        :to       id}}

              ;; Teardown the flow coordinator:
              ;;  1. remove this event handler
              ;;  2. remove state
              ;;  3. stop the events forwarder
              :done {:db (dissoc db db-path)
                     :event-forwarder {:unregister id}
                     :deregister-event-handler id})          ;; XXX write this effects handler

              ;; A new event has been forwarded to this handler. What does it mean?
              ;;  1. does this new event mean we need to dispatch another
              ;;  2. remember that this new event has happened
              (let [[_ [forwarded-event-id & args]] event-v
                    {:keys [seen-events started-tasks]} (get-state db)
                    new-seen-events    (conj seen-events forwarded-event-id)
                    ready-tasks        (newly-startable-tasks rules new-seen-events started-tasks)
                    ready-task-ids     (->> ready-tasks (map :dispatch) set)
                    new-started-tasks  (set/union started-tasks ready-task-ids)]
                {:db       (set-state db new-seen-events new-started-tasks)
                 :dispatch (concat (map :dispatch ready-tasks))})))))

(re-frame.core/def-fx
  :aync-flow
  (fn [{:as flow :keys [id]}]
    (re-frame.core/def-event-fx
      (or id default-id)                                 ;; add debug middleware???  XXXX
      (make-flow-event-handler flow))
    (re-frame.core/dispatch [id :steup])))


{'fraction_non_alphanumeric': 0.11402439024390244, 'fraction_numerical': 0.0024390243902439024, 'mean_word_length': 1.6201277955271565, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '232460', 'n_tokens_mistral': 2369, 'n_tokens_neox': 2210, 'n_words': 756}