         [meta features] (json-reader/features-from-stream stream)]
     [meta features]))
 
+(defn- inserted-features [extracts dataset collection extract-type features]
+  (doseq [f features]
+    (let [max-row-id (if (empty? @extracts)
+                       0
+                       (reduce max (map first @extracts)))]
+          (swap! extracts assoc (inc max-row-id) f))))
+
+
+(defn- filter-collection [item-to-filter collection] 
+  (some #(= % item-to-filter) collection))
+
+(defn- deleted-versions [extracts dataset collection extract-type versions-to-delete]
+  (let [rows-to-delete (map first (filter #(filter-collection (:_version (second %)) versions-to-delete) @extracts))]
+    (doseq [row-id rows-to-delete]
+      (swap! extracts dissoc row-id))))
+
+;(defn- deleted-versions [extracts dataset collection extract-type versions-to-delete]
+;  (println "do nothing"))
+
+
 (defn process-feature-permutation [meta feature-permutation]
   (println "  " (map #(into [] (map :action %1)) feature-permutation))
-  (apply merge-with (fn [a b] (if (seq? a) (conj a b) (+ a b)))
-         (map (fn [features]
+  (let [extracts (atom {})]
+    (with-bindings
+      {#'e/*process-insert-extract* (partial inserted-features extracts)
+       #'e/*process-delete-extract* (partial deleted-versions extracts)
+       #'e/*initialized-collection?* (constantly true)}
+      {:stats (apply merge-with (fn [a b] (if (seq? a) (conj a b) (+ a b)))
+            (map (fn [features]
                 (let [persistence (config/persistence)
                       projectors (conj (config/projectors persistence) (config/timeline persistence))
                       processor (processor/create meta "regression-set" persistence projectors)]
                   (dorun (processor/consume processor features))
+                  (e/fill-extract "regression-set" nil)
+                  (e/flush-changelog "regression-set")
                   (:statistics (processor/shutdown processor))))
-              feature-permutation)))
+              feature-permutation)) 
+      :extracts extracts})))
 
 (defn replay []
   (println "  Replaying")

  (is (= n (count (query-geoserver collection)))))


(defpermutatedtest new-feature "col-2_id-b_new" stats
  (is (= 1 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 1 :features 1})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 0}
                                 :timeline-changelog {:n-new 1}})
  (test-timeline->extract 1)
  (test-geoserver "col-2" 1))

(defpermutatedtest new-change-feature "col-2_id-b_new-change" stats
  (is (= 2 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 2 :features 1})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 1}
                                 :timeline-changelog {:n-new 1 :n-change 1}})
  (test-timeline->extract 2)
  (test-geoserver "col-2" 1))

(defpermutatedtest new-change-close-feature "col-2_id-b_new-change-close" stats
  (is (= 3 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 3 :features 1})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 1}
                                 :timeline-changelog {:n-new 1 :n-change 1 :n-close 1}})
  (test-timeline->extract 2)
  (test-geoserver "col-2" 0))

(defpermutatedtest new-change-close_with_attributes-feature "col-2_id-b_new-change-close_with_attributes" stats
  (is (= 4 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 4 :features 1})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 2}
                                 :timeline-changelog {:n-new 1 :n-change 2 :n-close 1}})
  (test-timeline->extract 3)
  (test-geoserver "col-2" 0))

(defpermutatedtest new-change-change-delete-feature "col-1_id-a_new-change-change-delete" stats
  (is (= 4 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-1" "id-a" {:events 4 :features 1})
  (test-timeline "col-1" "id-a" {:timeline-current {:n 0}
                                 :timeline {:n 0}
                                 :timeline-changelog {:n-new 1 :n-change 2 :n-delete 3}})
  (test-timeline->extract 0)
  (test-geoserver "col-1" 0))

(defpermutatedtest new-change-change-delete-new-change-feature "col-1_id-a_new-change-change-delete-new-change" stats
  (is (= 6 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-1" "id-a" {:events 6 :features 1})
  (test-timeline "col-1" "id-a" {:timeline-current {:n 1}
                                 :timeline {:n 1}
                                 :timeline-changelog {:n-new 2 :n-change 3 :n-delete 3}})
  (test-timeline->extract 2)
  (test-geoserver "col-1" 1))

(defpermutatedtest new-feature-with-nested-null-geom "col-2_id-b_new_with_nested_null_geom" stats
  (is (= 2 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 1 :features 1})
  (test-persistence "col-2$nested" {:events 1 :features 1})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 0}
                                 :timeline-changelog {:n-new 1 :n-change 1}})
  (test-timeline->extract 1)
  (test-geoserver "col-2" 1))

(defpermutatedtest new-feature-with-nested-crappy-geom "col-2_id-b_new_with_nested_crappy_geom" stats
  (is (= 2 (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 1 :features 1})
  (test-persistence "col-2$nested" {:events 1 :features 1})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 0}
                                 :timeline-changelog {:n-new 1 :n-change 1}})
  (test-timeline->extract 1)
  (test-geoserver "col-2" 1))

(defregressiontest double-nested-new-delete-new-change-feature "col-2_id-b_double_nested_new_delete_new_change" stats
  (is (= (+ 3 3 3 5) (:n-processed stats)))
  (is (= 0 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 4 :features 1})
  (test-persistence "col-2$nestedserie" {:events 5 :features 3})
  (test-persistence "col-2$nestedserie$label" {:events 5 :features 3})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 1}
                                 :timeline-changelog {:n-new 2 :n-change 7 :n-delete 3 :n-close 2}})
  (test-timeline->extract 2)
  (test-geoserver "col-2" 1)
  (test-geoserver "col-2$nestedserie$label" 1))

(defpermutatedtest invalid-nested-new-feature "col-2_id-b_invalid_nested_new" stats
  (is (= 2 (:n-processed stats)))
  (is (= 2 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 0 :features 0})
  (test-persistence "col-2$nested" {:events 0 :features 0})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 0}
                                 :timeline {:n 0}
                                 :timeline-changelog {}})
  (test-timeline->extract 0)
  (test-geoserver "col-2" 0)
  (test-geoserver "col-2$nested" 0))

(defpermutatedtest double-nested-new-invalid-change-feature "col-2_id-b_double_nested_new_invalid_change" stats
  (is (= (+ 3 1 2 2) (:n-processed stats)))
  (is (= 5 (:n-errored stats)))
  (test-persistence "col-2" "id-b" {:events 1 :features 1})
  (test-persistence "col-2$nestedserie" {:events 1 :features 1})
  (test-persistence "col-2$nestedserie$label" {:events 1 :features 1})
  (test-timeline "col-2" "id-b" {:timeline-current {:n 1}
                                 :timeline {:n 0}
                                 :timeline-changelog {:n-new 1 :n-change 2}})
  (test-timeline->extract 1)
  (test-geoserver "col-2" 1)
  (test-geoserver "col-2$nestedserie" 0)
  (test-geoserver "col-2$nestedserie$label" 1))

  (with-test-out
    (println ">>"(-> m :var meta :name))))

(defmacro defregressiontest* [name permutated? file results-var & body]
  `(deftest ~name
     (with-bindings
       {#'dc/*persistence-schema-prefix* :featured_regression
