                 (flatten))))))
 
 
-(defn new-conv-actor [init-fn & opts]
-  "Create a new conv actor which responds to snd. Messages should look like [<t> [...]], where <t> is either :votes or
-  :moderation, depending on what kind of batch is being sent/processed. Implements deref for state retrieval."
-  (let [msgbox (chan Long/MAX_VALUE) ; we want this to be as big as possible, since backpressure doesn't really work
-        conv (atom (init-fn)) ; keep track of conv state in atom so it can be dereffed
-        ca (ConvActor. msgbox conv)
-        err-handler (build-update-error-handler msgbox ca)]
-    ; Initialize the go loop which watches the mailbox, and runs updates all all pending messages when they
-    ; arrive
-    (go-loop []
-      (try
-        (let [first-msg (<! msgbox) ; do this so we park efficiently
-              msgs      (take-all! msgbox)
-              msgs      (concat [first-msg] msgs)
-              {votes :votes mods :moderation}
-                        (split-batches msgs)]
-          (when mods
-            (swap! conv conv/mod-update mods))
-          (swap! conv update-fn (or votes []) err-handler))
-        (catch Exception e
-          (log/error "Excpetion not handler by err-handler:" e)
-          (.printStackTrace e)))
-      (recur))
-    ca))
-
-
+(defrecord ConversationManager [config mongo metrics conversations]
+  component/Lifecycle
+  (start [component]
+    (let [conversations (atom {})]
+      (assoc component :conversations conversations)))
+  (stop [component]
+    ;; Close all our message channels for good measure
+    (doseq [[zid {:keys [message-chan]}] @conversations]
+      (async/close! message-chan))
+    ;; Not sure, but we might want this for GC
+    (reset! conversations nil)
+    (assoc component :conversations nil)))
+
+
+(defn queue-message-batch!
+  [{:keys [conversations] :as conv-man} message-type zid message-batch recompute]
+  (if-let [{:keys [conv message-chan]} (get @conversations zid)]
+    ;; Then we already have a go loop running for this
+    (>!! message-chan {:message-type message-type :message-batch message-batch})
+    ;; Then we need to initialize the conversation and set up the conversation channel and go routine
+    (let [conv (load-or-init zid :recompute recompute) ;; XXX recompute should be env var?
+          ;; XXX Need to set up message chan buffer as a env var
+          message-chan (chan 100000)]
+      (swap! conversations assoc zid {:conv conv :message-chan message-chan})
+      ;; However, we don't use the conversations atom conv state, but keep track of it explicitly in the loop,
+      ;; ensuring that we don't have race conditions for conv state. The state kept in the atom is basically
+      ;; just a convenience.
+      (go-loop [conv conv]
+        (let [first-msg (<! message-chan)
+              msgs (concat [first-msg] (take-all! message-chan))
+              {:keys [votes moderation]} (split-batches msgs)
+              error-handler (build-update-error-handler message-chan conv)
+              conv (-> conv
+                       (pc/?> moderation (conv/mod-update moderation))
+                       (pc/?> votes (update-fn votes error-handler)))]
+          (swap! conversations assoc-in [zid :conv] conv)
+          (recur conv))))))
+
+

        :group-votes})))


(defn mongo-upsert-results
  "Perform upsert of new results on mongo collection name"
  [mongo config collection-name new-results]
  (mc/update
    ; Bleg; not the cleanest separation here; should at least try to make this something rebindable
    (:db mongo)
    (db/mongo-db (env/env :mongolab-uri))
    collection-name
    {:zid (or (:zid new-results) ; covering our bases for strings or keywords due to cheshire hack
              (get new-results "zid"))}
    new-results
    {:upsert true}))


(defn mongo-insert-results
  "Perform insert to mongo collection by zid"
  [collection-name object]
  (mc/insert
    (db/mongo-db (env/env :mongolab-uri))
    collection-name
    object))


(defn handle-profile-data
  "For now, just log profile data. Eventually want to send to influxDB and graphite."
  [conv & {:keys [recompute n-votes finish-time] :as extra-data}]
  (if-let [prof-atom (:profile-data conv)]
    (let [prof @prof-atom
          tot (apply + (map second prof))

      ; the conv-actor update?
      (try
        (log/info "Preparing to re-queue votes for failed conversation update for" zid-str)
        ;; XXX Should we check if the queue got close due to error or manager close, and log?
        (async/go (async/>! queue [:votes votes]))
        (catch Exception qe
          (log/error "Unable to re-queue votes after conversation update failed for" zid-str)
