Add kv store and worker queue

.gitignore

@@ -1,3 +1,4 @@
-/db/data/dev
+/db/data
 .nrepl-port
 .cpcache
+/kv/data

db/README.org

@@ -27,7 +27,7 @@ docker build -t db .
 #+END_SRC
 
 #+BEGIN_SRC sh :tangle run.sh :tangle-mode (identity #o755)
-#!/usr/bin/env bash
+#!/usr/bin/env sh
 set -euo pipefail
 
 PGDATA=${PGDATA:-"$(pwd)/data/dev"}

kv/Dockerfile

@@ -0,0 +1,3 @@
+FROM redis
+COPY redis.conf /usr/local/etc/redis/redis.conf
+CMD [ "redis-server", "/usr/local/etc/redis/redis.conf" ]

kv/README.org

@@ -0,0 +1,20 @@
+#+TITLE: Persistent KV Store
+
+There's really nothing to build right now.
+
+#+BEGIN_SRC sh :tangle run.sh :tangle-mode (identity #o755)
+#!/usr/bin/env sh
+set -euo pipefail
+
+KVDATA=${KVDATA:-"$(pwd)/data/dev"}
+
+docker run \
+    -p 127.0.0.1:6379:6379 \
+    -v $KVDATA:/data \
+    --name darklimericks-kv \
+    redis --appendonly yes
+#+END_SRC
+
+#+BEGIN_SRC sh :tangle start.sh :tangle-mode (identity #o755)
+docker start darklimericks-kv
+#+END_SRC

kv/redis.conf

@@ -0,0 +1 @@
+requirepass dev

kv/run.sh

@@ -0,0 +1,10 @@
+#!/usr/bin/env sh
+set -euo pipefail
+
+KVDATA=${KVDATA:-"$(pwd)/data/dev"}
+
+docker run \
+    -p 127.0.0.1:6379:6379 \
+    -v $KVDATA:/data \
+    --name darklimericks-kv \
+    redis --appendonly yes

kv/start.sh

@@ -0,0 +1 @@
+docker start darklimericks-kv

web/deps.edn

@@ -5,6 +5,7 @@
    :sha "d492e97259c013ba401c5238842cd3445839d020"}
   hiccup {:mvn/version "1.0.5"}
   com.taoensso/timbre {:mvn/version "5.1.0"}
+  com.taoensso/carmine {:mvn/version "3.0.1"}
   http-kit {:mvn/version "2.5.0"}
   integrant {:mvn/version "0.8.0"}
   integrant/repl {:mvn/version "0.3.2"}

web/dev/user.clj

@@ -2,6 +2,8 @@
   (:require [clojure.tools.deps.alpha.repl :refer [add-lib]]
             [clojure.tools.namespace.repl :refer [set-refresh-dirs]]
             [integrant.repl :as repl]
+            [taoensso.carmine :as car]
+            [taoensso.carmine.message-queue :as car-mq]
             [integrant.repl.state :as state]
             [migratus.core :as migratus]
             [com.darklimericks.server.handlers :as handlers]
@@ -59,11 +61,25 @@
 (comment
   (init)
   (auto-reset)
+  (-> state/system :worker/limerick-gen)
+  (car/wcar
+   (-> state/system :database.kv/connection)
+   (car-mq/enqueue "limericks" '((A 9) (A 9) (B 5) (B 5) (A 9))))
+
+  (car/wcar
+   (-> state/system :database.kv/connection)
+   (car/ping)
+   (car/set "foo" "bar")
+   (car/set "baz" "buzz")
+   (car/get "baz"))
   (limericks/get-artist-and-album-for-new-limerick (-> state/system :database.sql/connection))
+  (repeatedly
+   50
+   (fn []
      (let [handler (handlers/limerick-generation-post-handler
                     (-> state/system :database.sql/connection)
                     (-> state/system :app/cache))]
-    (handler {:params {:scheme "A9 A9 B5 B5 A9" #_'((A 9) (A 9) (B 5) (B 5) (A 9))}}))
+       (handler {:params {:scheme "A9 A9 B5 B5 A9" #_'((A 9) (A 9) (B 5) (B 5) (A 9))}}))))
 
   (reitit/match-by-path
    (-> state/system :app/router)

web/resources/server/config.edn

@@ -1,6 +1,10 @@
 {:app/logging {:min-level :debug}
  :app/server {:port 8000 :handler #ig/ref :app/handler}
  :app/cache {}
+ :worker/limerick-gen {:kv #ig/ref :database.kv/connection
+                       :db #ig/ref :database.sql/connection}
+ :database.kv/connection {:pool {}
+                          :spec {:uri ""}}
  :database.sql/connection {}
  :app/router {:db #ig/ref :database.sql/connection :cache #ig/ref :app/cache}
  :app/handler {:router #ig/ref :app/router}}

web/src/com/darklimericks/server/limericks.clj

@@ -4,7 +4,7 @@
             [reitit.core :as reitit]
             [com.darklimericks.db.artists :as artists]
             [com.darklimericks.db.albums :as albums]
-            [com.darklimericks.db.limericks :as limericks]
+            [com.darklimericks.db.limericks :as db.limericks]
             [com.darklimericks.linguistics.core :as linguistics]
             [com.owoga.prhyme.limerick :as limerick]
             [com.owoga.prhyme.data.dictionary :as dict]
@@ -60,7 +60,7 @@
 (defn get-artist-and-album-for-new-limerick [db]
   (let [artist (artists/most-recent-artist db)
         albums (albums/artist-albums db (:artist/id artist))
-        limericks (limericks/album-limericks db (:album/id (first albums)))]
+        limericks (db.limericks/album-limericks db (:album/id (first albums)))]
     (cond
       (< (count limericks) 10)
       [(:artist/id artist) (:album/id (first albums))]
@@ -102,3 +102,15 @@
           (println "Exception" e)
           (swap! db update-in [:tasks task-id] assoc :status :failed)
           (swap! db assoc-in [:tasks task-id] {:text (.getMessage e)}))))))
+
+(defn generate-limerick-worker [db scheme]
+  (let [limerick (limerick/rhyme-from-scheme
+               dict/prhyme-dict
+               darklyrics-markov-2
+               scheme)
+        [artist-id album-id] (get-artist-and-album-for-new-limerick db)]
+    (db.limericks/insert-limerick
+     db
+     (get-limerick-name limerick)
+     (string/join "\n" limerick)
+     album-id)))

web/src/com/darklimericks/server/system.clj

@@ -1,20 +1,42 @@
 (ns com.darklimericks.server.system
   (:require [integrant.core :as ig]
-            [clojure.java.io :as io]
             [taoensso.timbre :as timbre]
             [org.httpkit.server :as kit]
             [next.jdbc :as jdbc]
             [environ.core :refer [env]]
+            [taoensso.carmine.message-queue :as car-mq]
             [reitit.http :as http]
             [reitit.ring :as ring]
             [reitit.coercion.spec]
-            [reitit.ring.coercion :as coercion]
             [reitit.interceptor.sieppari :as sieppari]
             [reitit.http.interceptors.parameters :refer [parameters-interceptor]]
             [com.darklimericks.server.handlers :as handlers]
+            [com.darklimericks.server.limericks :as limericks]
             [com.darklimericks.server.interceptors :as interceptors]
             [com.darklimericks.server.db :as db]))
 
+(defmethod ig/init-key :worker/limerick-gen [_ {:keys [db kv]}]
+  (car-mq/worker
+   kv
+   "limericks"
+   {:handler
+    (fn [{:keys [message attempt]}]
+      (timbre/info "Received" message)
+      (limericks/generate-limerick-worker db message))}))
+
+(defmethod ig/prep-key :database.kv/connection [_ config]
+  (let [pass (or (env :redis-pass) "dev")]
+    (assoc-in
+     config
+     [:spec :uri]
+     (format "redis://localhost:6379/"))))
+
+;; Point of this? We aren't and maintaining a connection here.
+;; I guess carmine does that behind the scenes with it's "pool"
+;; option. But it's nice to have all boundary config in one place.
+(defmethod ig/init-key :database.kv/connection [_ config]
+  config)
+
 (defmethod ig/prep-key :database.sql/connection [_ _]
   {:jdbcUrl (str "jdbc:postgresql://localhost:5432/?user="
                  (or (env :postgres-user)