updates to prevent in-progress orders

src/clj/user.clj

@@ -1,7 +1,8 @@
 (ns user
   (:require [auto-ap.datomic :refer [uri]]
-
+            [config.core :refer [env]]
             [auto-ap.utils :refer [by]]
+            [clojure.core.async :as async]
             #_[auto-ap.ledger :as l]
             [mount.core :as mount]
             [auto-ap.server ]
@@ -10,7 +11,8 @@
             [clj-time.coerce :as c]
             [clj-time.core :as t]
             [clojure.java.io :as io]
-            [clojure.string :as str])
+            [clojure.string :as str]
+            [amazonica.aws.s3 :as s3])
   (:import [org.apache.commons.io.input BOMInputStream]))
 
 (defn mark-until-date [client end]
@@ -444,3 +446,29 @@
                        (for [r data]
                          ((apply juxt columns) r )))))
 
+
+(defn find-queries [words]
+  (let [obj (s3/list-objects-v2 :bucket-name (:data-bucket env)
+                                :prefix (str "queries/"))]
+    (let [concurrent 30
+          output-chan (async/chan)]
+      (async/pipeline-blocking concurrent
+                               output-chan
+                               (comp 
+                                (map #(do
+                                        (println "looking up " (:key %))
+                                        [(:key %)
+                                         (str (slurp  (:object-content (s3/get-object
+                                                                        :bucket-name (:data-bucket env)
+                                                                        :key (:key %)))))]))
+                                
+                                (filter #(->> words
+                                              (every? (fn [w] (str/includes? (second %) w)))))
+                                (map first)
+                                (map #(str/replace % #"queries/" ""))
+                                )
+                               (async/to-chan (:object-summaries obj))
+                               true
+                               (fn [e]
+                                 (println "failed " e)))
+      (async/<!! (async/into [] output-chan)))))