;; require kafka-utils and datascript
(require '[kafka-utils.core :as ku]
         '[datascript.core :as d])

;; Get a core.async channel of all the records in a topic.
;; For the serdes, I'd recommend using franzy, but this isn't required
(def channel
    (ku/read-from-beginning "backblaze_smart" ["localhost:9092"]
                            (franzy.deserializers/string-deserializer)
                            (franzy.json.deserializers/json-deserializer {:key-fn true})))

;; Create a datascript db
(def conn (d/create-conn))

;; Sink the topic data into the datascript db
(ku/sink-to-db! channel db-conn)