   (kv/restore (or kvstore (kv/make-in-memory-kvstore)) txlog-events))
 
 
+(defn- normalize-kvstore-events [events]
+  (map (fn [event]
+         (if (map? event)
+           event
+           {:timestamp 0 :eventName kv/EVENT-STATE-UPDATED
+            :sourceId (first event) :version (second event)
+            :value (last event)}))
+       events))
+
+
 
 (defn global-kv-restore
   "Takes a bunch of tx-log events and restores them in the global-stores"

         (partition-filter-wrapper input-partitions))))


(defn- tx-log->formatter
  "Return a function which turns the tx-log into a a triplet of

   [topic partition-key json-data]"
  [kvstore-topic]
  (fn [txlog]
    (->> txlog
         (map (fn [[k ver v]]
                [kvstore-topic k (to-json [k ver v])])))))

;; TODO: change txlog format look like events
;; TODO: support dispatch-fn and  format fn
(defn topic-process-wrapper [component stream-id pipeline]
  (let [{:keys [local-store state-topic
                output-topic-partition-fn output-topic ]
         } (stream-config component stream-id)
        output-collector (:output-collector component)
        tx-log->messages (tx-log->formatter state-topic)
        out-dispatch (fn [event] [output-topic
                                 (output-topic-partition-fn event)
                                 (to-json event)])]

  ([{:keys [component stream-id events]}]
   (let [kv-store (:local-store (stream-config component stream-id))]
     (var-set kv-store
              (kv-restore (var-get kv-store)
                          (normalize-kvstore-events events))))))


;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
