       (doseq [[a b] hierarchy]
         (connect-to! (verts a) (verts b) signal-forward nil))
       verts)))
+
+(defn signal-triple
+  [vertex _]
+  [(:id vertex) @vertex])
+
+(defn collect-index
+  [spo]
+  (simple-collect
+   (fn [val incoming]
+     (transduce
+      (map (fn [[id t]] [id (nth t spo)]))
+      (completing (fn [acc [id x]] (update acc x (fnil conj #{}) id)))
+      val incoming))))
+
+(defn index-vertex
+  [g spo]
+  (add-vertex! g {::val {}
+                  ::collect-fn (collect-index spo)
+                  ::score-signal-fn (constantly 1)}))
+
+(def s-idx (index-vertex g 0))
+(def p-idx (index-vertex g 1))
+(def o-idx (index-vertex g 2))
+
+(defn add-triple
+  [g t]
+  (let [v (add-vertex! g {::val t})]
+    (connect-to! v s-idx signal-triple nil)
+    (connect-to! v p-idx signal-triple nil)
+    (connect-to! v o-idx signal-triple nil)
+    (signal! v)
+    v))
+
+(defn signal-select
+  [vertex [idx sel]]
+  [idx (if sel (@vertex sel [nil]) (->> @vertex vals (eduction (mapcat identity))))])
+
+(def collect-select
+  (simple-collect
+   (fn [val incoming]
+     (reduce (fn [acc [idx res]] (update acc idx (fnil into #{}) res)) val incoming))))
+
+(def aggregate-select
+  (simple-collect
+   (fn [val incoming]
+     (let [res (vals (last incoming))]
+       ;;(info :res res)
+       (when (and (seq res) (every? #(not= #{nil} %) res))
+         (->> res
+              (map #(disj % nil))
+              (set)
+              (sort-by count)
+              ;;(#(do (info :sorted %) %))
+              (reduce set/intersection)
+              (map #(deref (vertex-for-id g %)))))))))
+
+;; TODO figure out way how to at to a running context
+(defn register-query
+  [g s p o]
+  (let [acc (add-vertex! g {::val {} ::collect-fn collect-select})
+        res (add-vertex! g {::val nil ::collect-fn aggregate-select})]
+    (connect-to! s-idx acc signal-select [0 s])
+    (connect-to! p-idx acc signal-select [1 p])
+    (connect-to! o-idx acc signal-select [2 o])
+    (connect-to! acc res signal-forward nil)
+    ;;[acc res]
+    res))
+
+#_(defn select
+  [g s p o]
+  (let [acc (add-vertex! g {::val {} ::collect-fn collect-select})
+        ctx (async-context
+             {:graph     g
+              :processor eager-vertex-processor})]
+    (connect-to! s-idx acc signal-select [0 s])
+    (connect-to! p-idx acc signal-select [1 p])
+    (connect-to! o-idx acc signal-select [2 o])
+    (info @(execute! ctx))
+    (disconnect-vertex! s-idx acc)
+    (disconnect-vertex! p-idx acc)
+    (disconnect-vertex! o-idx acc)
+    (remove-vertex! g acc)
+    (let [res (vals @acc)]
+      (info :res res)
+      (when-not (some #(= #{:void} %) res)
+        (->> res
+             (map #(disj % :void))
+             (set)
+             (sort-by count)
+             (#(do (info :sorted %) %))
+             (reduce set/intersection)
+             (map #(deref (vertex-for-id g %))))))))
+
+(def triples
+  (mapv
+   #(add-triple g %)
+   '[[toxi author fabric]
+     [fabric type project]
+     [toxi type person]]))
+
+(def ctx (async-context {:graph g :processor eager-vertex-processor :timeout nil}))
+(def toxi (register-query g 'toxi nil nil))
+(def types (register-query g nil 'type nil))
+(def projects (register-query g nil 'type 'project))
+
+(execute! ctx)
+
+;;(warn :free-ram (.freeMemory (Runtime/getRuntime)))
+
+;; TODO add TripleGraph w/ separate sets of triple, index, rule & query vertices

  [] (Graph. (atom {:vertices {} :next-id 0})))

(defn eager-vertex-processor
  [{:keys [id state] :as vertex} ctx]
  (let [{:keys [collect-thresh signal-thresh]} @ctx
        in   (input-channel vertex)]
    (go-loop []
      (let [[src-id sig] (<! in)]
        (if sig

    _)
  (close-input! [_]
    (if @in
      (do #_(debug (str id " closing..."))
          (close! @in)
          (reset! in nil))
      (warn (str id " already closed")))
