<commit_before>(ns ^:no-doc onyx.windowing.window-compile
  (:require [onyx.windowing.window-extensions :as w]
            [onyx.windowing.aggregation :as a]
            #?(:clj [taoensso.timbre :refer [info error warn trace fatal] :as timbre])
            #?(:clj [onyx.static.validation :as validation])
            #?(:clj [onyx.schema :refer [TriggerState Trigger Window Event WindowState]])
            #?(:clj [schema.core :as s])
            [onyx.peer.window-state :as ws]
            [onyx.types :refer [map->TriggerState]]
            [onyx.state.memory]
            [onyx.static.util :as u]
            [onyx.peer.grouping :as g]))

(defn filter-windows [windows task]
  (filter #(= (:window/task %) task) windows))

(defn resolve-window-init [window calls]
  (if-not (:aggregation/init calls)
    (let [init (:window/init window)]
      (when-not init
        (throw (ex-info "No :window/init supplied, this is required for this aggregation" {:window window})))
      (constantly (:window/init window)))
    (:aggregation/init calls)))

(defn resolve-var [v]
  #?(:clj (var-get v))
  #?(:cljs v))

(defn filter-ns-key-map [m ns-str]
  (->> m
       (filter (fn [[k _]] (= ns-str (namespace k))))
       (map (fn [[k v]] [(keyword (name k)) v]))
       (into {})))

(defn resolve-trigger [indices {:keys [trigger/sync trigger/emit trigger/refinement trigger/on trigger/id trigger/window-id] :as trigger}]
  (let [refinement-calls (resolve-var (u/kw->fn refinement))
        trigger-calls (resolve-var (u/kw->fn on))]
    #?(:clj (validation/validate-refinement-calls refinement-calls))
    #?(:clj (validation/validate-trigger-calls trigger-calls))
    (let [f-init-state (:trigger/init-state trigger-calls)
          f-init-locals (:trigger/init-locals trigger-calls)
          sync-fn (if sync (u/kw->fn sync))
          emit-fn (if emit (u/kw->fn emit))
          locals (f-init-locals trigger)]
      (-> trigger
          (filter-ns-key-map "trigger")
          (into locals)
          (assoc :trigger trigger)
          (assoc :idx (or (get indices [id window-id]) (throw (ex-info "Could not find state index for window id." {}))))
          (assoc :id (:trigger/id trigger))
          (assoc :sync-fn sync-fn)
          (assoc :emit-fn emit-fn)
          (assoc :init-state f-init-state)
          (assoc :next-trigger-state (:trigger/next-state trigger-calls))
          (assoc :trigger-fire? (:trigger/trigger-fire? trigger-calls))
          (assoc :create-state-update (:refinement/create-state-update refinement-calls))
          (assoc :apply-state-update (:refinement/apply-state-update refinement-calls))
          map->TriggerState))))

(defn resolve-window-extension [window]
  (-> window
      (filter-ns-key-map "window")
      ((w/windowing-builder window))
      (assoc :window window)))

(defn build-window-executor [{:keys [window/id] :as window} all-triggers state-store indices task-map]
  (let [agg (:window/aggregation window)
        agg-var (if (sequential? agg) (first agg) agg)
        calls (resolve-var (u/kw->fn agg-var))
        _ (validation/validate-state-aggregation-calls calls)
        init-fn (resolve-window-init window calls)
        triggers (->> all-triggers 
                      (filter #(= (:window/id window) (:trigger/window-id %)))
                      (map (partial resolve-trigger indices))
                      (map (juxt :idx identity))
                      (into {}))]
    (ws/map->WindowExecutor
     {:id id 
      :idx (get indices id)
      :window-extension (resolve-window-extension window)
      :triggers triggers
      :emitted (atom [])
      :window window
      :state-store state-store
      :init-fn init-fn
      :create-state-update (:aggregation/create-state-update calls)
      :super-agg-fn (:aggregation/super-aggregation-fn calls)
      :apply-state-update (:aggregation/apply-state-update calls)})))<commit_msg>Fix missing reader conditional.<commit_after>(ns ^:no-doc onyx.windowing.window-compile
  (:require [onyx.windowing.window-extensions :as w]
            [onyx.windowing.aggregation :as a]
            #?(:clj [taoensso.timbre :refer [info error warn trace fatal] :as timbre])
            #?(:clj [onyx.static.validation :as validation])
            #?(:clj [onyx.schema :refer [TriggerState Trigger Window Event WindowState]])
            #?(:clj [schema.core :as s])
            [onyx.peer.window-state :as ws]
            [onyx.types :refer [map->TriggerState]]
            [onyx.state.memory]
            [onyx.static.util :as u]
            [onyx.peer.grouping :as g]))

(defn filter-windows [windows task]
  (filter #(= (:window/task %) task) windows))

(defn resolve-window-init [window calls]
  (if-not (:aggregation/init calls)
    (let [init (:window/init window)]
      (when-not init
        (throw (ex-info "No :window/init supplied, this is required for this aggregation" {:window window})))
      (constantly (:window/init window)))
    (:aggregation/init calls)))

(defn resolve-var [v]
  #?(:clj (var-get v))
  #?(:cljs v))

(defn filter-ns-key-map [m ns-str]
  (->> m
       (filter (fn [[k _]] (= ns-str (namespace k))))
       (map (fn [[k v]] [(keyword (name k)) v]))
       (into {})))

(defn resolve-trigger [indices {:keys [trigger/sync trigger/emit trigger/refinement trigger/on trigger/id trigger/window-id] :as trigger}]
  (let [refinement-calls (resolve-var (u/kw->fn refinement))
        trigger-calls (resolve-var (u/kw->fn on))]
    #?(:clj (validation/validate-refinement-calls refinement-calls))
    #?(:clj (validation/validate-trigger-calls trigger-calls))
    (let [f-init-state (:trigger/init-state trigger-calls)
          f-init-locals (:trigger/init-locals trigger-calls)
          sync-fn (if sync (u/kw->fn sync))
          emit-fn (if emit (u/kw->fn emit))
          locals (f-init-locals trigger)]
      (-> trigger
          (filter-ns-key-map "trigger")
          (into locals)
          (assoc :trigger trigger)
          (assoc :idx (or (get indices [id window-id]) (throw (ex-info "Could not find state index for window id." {}))))
          (assoc :id (:trigger/id trigger))
          (assoc :sync-fn sync-fn)
          (assoc :emit-fn emit-fn)
          (assoc :init-state f-init-state)
          (assoc :next-trigger-state (:trigger/next-state trigger-calls))
          (assoc :trigger-fire? (:trigger/trigger-fire? trigger-calls))
          (assoc :create-state-update (:refinement/create-state-update refinement-calls))
          (assoc :apply-state-update (:refinement/apply-state-update refinement-calls))
          map->TriggerState))))

(defn resolve-window-extension [window]
  (-> window
      (filter-ns-key-map "window")
      ((w/windowing-builder window))
      (assoc :window window)))

(defn build-window-executor [{:keys [window/id] :as window} all-triggers state-store indices task-map]
  (let [agg (:window/aggregation window)
        agg-var (if (sequential? agg) (first agg) agg)
        calls (resolve-var (u/kw->fn agg-var))
        #?@(:clj (_ (validation/validate-state-aggregation-calls calls)))
        init-fn (resolve-window-init window calls)
        triggers (->> all-triggers 
                      (filter #(= (:window/id window) (:trigger/window-id %)))
                      (map (partial resolve-trigger indices))
                      (map (juxt :idx identity))
                      (into {}))]
    (ws/map->WindowExecutor
     {:id id 
      :idx (get indices id)
      :window-extension (resolve-window-extension window)
      :triggers triggers
      :emitted (atom [])
      :window window
      :state-store state-store
      :init-fn init-fn
      :create-state-update (:aggregation/create-state-update calls)
      :super-agg-fn (:aggregation/super-aggregation-fn calls)
      :apply-state-update (:aggregation/apply-state-update calls)})))
{'fraction_non_alphanumeric': 0.1547983665135273, 'fraction_numerical': 0.0, 'mean_word_length': 2.5333633904418393, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1198960', 'n_tokens_mistral': 2519, 'n_tokens_neox': 2469, 'n_words': 611}