add database appender

Author: tamizhvendan

clojure/wheel/.joker

@@ -1 +1,2 @@
-{:known-macros [mount.core/defstate]}
+{:known-macros [mount.core/defstate
+                toucan.models/defmodel]}

clojure/wheel/project.clj

@@ -10,7 +10,8 @@
                  [org.postgresql/postgresql "42.2.6"]
                  [org.flywaydb/flyway-core "5.2.4"]
                  [com.taoensso/timbre "4.10.0"]
-                 [cheshire "5.9.0"]]
+                 [cheshire "5.9.0"]
+                 [toucan "1.14.0"]]
   :main ^:skip-aot wheel.core
   :target-path "target/%s"
   :profiles {:uberjar {:aot :all}

clojure/wheel/resources/db/migration/V201910021105__create_event.sql

@@ -0,0 +1,17 @@
+CREATE TYPE event_level AS ENUM (
+  'info', 'debug',
+  'error', 'warn',
+  'fatal');
+
+CREATE TYPE channel_name AS ENUM (
+  'tata-cliq', 'amazon', 'flipkart');
+
+CREATE TABLE event (
+  id UUID PRIMARY KEY,
+  parent_id UUID REFERENCES event(id),
+  level event_level NOT NULL,
+  name TEXT NOT NULL,
+  channel_id TEXT NOT NULL,
+  channel_name channel_name NOT NULL,
+  timestamp TIMESTAMP WITH TIME ZONE NOT NULL
+);

clojure/wheel/src/wheel/infra/database.clj

@@ -1,8 +1,11 @@
 (ns wheel.infra.database
   (:require [wheel.infra.config :as config]
             [mount.core :as mount]
-            [hikari-cp.core :as hikari])
-  (:import [org.flywaydb.core Flyway]))
+            [hikari-cp.core :as hikari]
+            [toucan.db :as db]
+            [toucan.models :as models])
+  (:import [org.flywaydb.core Flyway]
+           [org.postgresql.util PGobject]))
 
 (defn- make-datasource []
   (hikari/make-datasource (config/database)))
@@ -18,7 +21,27 @@
       load
       migrate))
 
+(defn- pg-object-fn [pg-enum-type]
+  (fn [value]
+    (doto (PGobject.)
+      (.setType pg-enum-type)
+      (.setValue (name value)))))
+
+(defn- configure-toucan []
+  (db/set-default-db-connection! {:datasource datasource})
+  (models/set-root-namespace! 'wheel.model)
+  (db/set-default-automatically-convert-dashes-and-underscores! true)
+  (models/add-type! :event-level
+                    :in (pg-object-fn "event_level")
+                    :out keyword)
+  (models/add-type! :channel-name
+                    :in (pg-object-fn "channel_name")
+                    :out keyword))
+
+(mount/defstate toucan
+  :start (configure-toucan))
+
 (comment
-  (mount/start)
+  (user/reset)
   (migrate)
   (mount/stop))

clojure/wheel/src/wheel/infra/log.clj

@@ -2,14 +2,16 @@
   (:require [taoensso.timbre :as timbre]
             [cheshire.core :as json]
             [clojure.spec.alpha :as s]
-            [wheel.middleware.event :as event]))
+            [wheel.middleware.event :as event]
+            [wheel.infra.log-appender.database :as database]))
 
 (defn- json-output [{:keys [msg_]}]
   (let [event (read-string (force msg_))]
     (json/generate-string event)))
 
 (defn init []
-  (timbre/merge-config! {:output-fn json-output}))
+  (timbre/merge-config! {:output-fn json-output
+                         :appenders {:database database/appender}}))
 
 (defn write! [{:keys [level]
                :as event}]
@@ -33,12 +35,11 @@
   (write! {:level :info :name :foo})
   (s/check-asserts false)
   (s/assert ::event/event {:level :info :name :foo})
-  (write! {:name :ranging/succeeded
+  (write! {:name :deranging/succeeded
            :type :domain
            :level :info
-           :channel-id 1
-           :timestamp "2019-10-01T21:56+05:30"
-           :id #uuid "8e70f4df-fc3d-4474-8ee9-33f4c87bf934"
-           :parent-id #uuid "20c84be2-7ec2-4127-8d39-a9cb4065a9ce"
+           :channel-id "UB"
+           :timestamp "2019-10-04T15:56+05:30"
+           :id (java.util.UUID/randomUUID)
            :channel-name :tata-cliq})
   )

clojure/wheel/src/wheel/infra/log_appender/database.clj

@@ -0,0 +1,12 @@
+(ns wheel.infra.log-appender.database
+  (:require [wheel.model.event :as event]))
+
+(defn- append-to-db [{:keys [msg_]}]
+  (let [evnt (read-string (force msg_))]
+    (when (= :domain (:type evnt))
+      (event/create! evnt))))
+
+(def appender {:enabled? true
+               :output-fn :inherit
+               :async? true
+               :fn append-to-db})

clojure/wheel/src/wheel/middleware/event.clj

@@ -26,6 +26,9 @@
 
 (s/def ::event (s/multi-spec event-type :type))
 
+(defn domain? [event]
+  (and (s/valid? ::event event) (= :domain (:type event))))
+
 (comment
   (s/valid?
    ::event

clojure/wheel/src/wheel/model/event.clj

@@ -0,0 +1,34 @@
+(ns wheel.model.event
+  (:require [toucan.models :as models]
+            [clojure.spec.alpha :as s]
+            [toucan.db :as db]
+            [wheel.middleware.event :as event])
+  (:import [java.time OffsetDateTime]
+           [java.time.format DateTimeFormatter]))
+
+(models/defmodel Event :event
+  models/IModel
+  (types [_]
+         {:name :keyword
+          :channel-name :channel-name
+          :level :event-level}))
+
+(defn- timestamp->offset-date-time [timestamp]
+  (OffsetDateTime/parse timestamp DateTimeFormatter/ISO_OFFSET_DATE_TIME))
+
+(defn create! [new-event]
+  {:pre [(s/assert ::event/event new-event)
+         (s/assert event/domain? new-event)]}
+  (as-> new-event evt
+    (update evt :timestamp timestamp->offset-date-time)
+    (dissoc evt :type)
+    (db/insert! Event evt)))
+
+(comment
+ (create! {:name :ranging/succeeded
+           :type :domain
+           :channel-id "UA"
+           :level :info
+           :timestamp "2019-10-01T12:30+05:30"
+           :id (java.util.UUID/randomUUID)
+           :channel-name :tata-cliq}))