 ;; Catalog replacement
 
 (defn replace-catalog*
-  [{:keys [certname version id received payload]} db]
+  [{:keys [certname version id received payload]} start-time db]
   (let [{producer-timestamp :producer_timestamp :as catalog} payload]
     (jdbc/with-transacted-connection' db :repeatable-read
       (scf-storage/maybe-activate-node! certname producer-timestamp)
       (scf-storage/replace-catalog! catalog received))
-    (log/info (trs "[{0}-{1}] ''{2}'' command processed for {3}"
+    (log/info (trs "[{0}-{1}] [{2} ms] ''{3}'' command processed for {4}"
                    (str id)
                    (str (tcoerce/to-long received))
+                   (in-millis (interval start-time (now)))
                    (command-names :replace-catalog)
                    certname))))
 
-(defn replace-catalog [{:keys [payload received version] :as command} db]
+(defn replace-catalog [{:keys [payload received version] :as command} start-time db]
   (let [validated-payload (upon-error-throw-fatality
                            (cat/parse-catalog payload version received))]
     (-> command
         (assoc :payload validated-payload)
-        (replace-catalog* db))))
+        (replace-catalog* start-time db))))
 
 ;; Fact replacement
 
 (defn replace-facts*
-  [{:keys [payload id received] :as command} db]
+  [{:keys [payload id received] :as command} start-time db]
   (let [{:keys [certname values] :as fact-data} payload
         producer-timestamp (:producer_timestamp fact-data)]
     (jdbc/with-transacted-connection' db :repeatable-read
       (scf-storage/maybe-activate-node! certname producer-timestamp)
       (scf-storage/replace-facts! fact-data))
-    (log/info (trs "[{0}-{1}] ''{2}'' command processed for {3}"
+    (log/info (trs "[{0}-{1}] [{2} ms] ''{3}'' command processed for {4}"
                    (str id)
                    (str (tcoerce/to-long received))
+                   (in-millis (interval start-time (now)))
                    (command-names :replace-facts)
                    certname))))
 
-(defn replace-facts [{:keys [payload version received] :as command} db]
+(defn replace-facts [{:keys [payload version received] :as command} start-time db]
   (replace-facts* (upon-error-throw-fatality
-                   (assoc command :payload (fact/normalize-facts version received payload)))
+                    (assoc command :payload (fact/normalize-facts version received payload)))
+                  start-time
                   db))
 
 ;; Node deactivation

      deactivate-node-wire-v2->wire-3))

(defn deactivate-node*
  [{:keys [id received payload]} db]
  (let [certname (:certname payload)
        producer-timestamp (to-timestamp (:producer_timestamp payload (now)))]
    (jdbc/with-transacted-connection db
      (when-not (scf-storage/certname-exists? certname)
        (scf-storage/add-certname! certname))
      (scf-storage/deactivate-node! certname producer-timestamp))
    (log/info (trs "[{0}-{1}] ''{2}'' command processed for {3}"
                   (str id)
                   (str (tcoerce/to-long received))
                   (command-names :deactivate-node)
                   certname))))

(defn deactivate-node [{:keys [payload version] :as command} db]
  (-> command
      (assoc :payload (case version
                        1 (deactivate-node-wire-v1->wire-3 payload)
                        2 (deactivate-node-wire-v2->wire-3 payload)
                        payload))
      (deactivate-node* db)))

;; Report submission

(defn store-report*
  [{:keys [payload id received]} db]
  (let [{:keys [certname puppet_version] :as report} payload
        producer-timestamp (to-timestamp (:producer_timestamp payload (now)))]
    (jdbc/with-transacted-connection db
      (scf-storage/maybe-activate-node! certname producer-timestamp)
      (scf-storage/add-report! report received))
    (log/info (trs "[{0}-{1}] ''{2}'' puppet v{3} command processed for {4}"
                   (str id)
                   (str (tcoerce/to-long received))
                   (command-names :store-report)
                   puppet_version
                   certname))))

(defn store-report [{:keys [payload version received] :as command} db]
  (let [validated-payload (upon-error-throw-fatality
                           (s/validate report/report-wireformat-schema
                                       (case version

            [schema.core :as s]
            [puppetlabs.puppetdb.config :as conf]
            [puppetlabs.puppetdb.time :refer [to-timestamp]]
            [clj-time.core :as time :refer [now interval in-millis]]
            [clj-time.format :as fmt-time]
            [clojure.set :as set]
            [clojure.core.async :as async]
