finish augment

Author: yeehaa123

functions/src/app/augment/index.cljs

@@ -8,14 +8,18 @@
             [shared.protocols.actionable :as ac])
   (:require-macros [cljs.core.async.macros :refer [go]]))
 
-(def adapters {:db {:table-names {:bookmarks (.. js/process -env -bookmarksTable)
-                                  :resources (.. js/process -env -resourcesTable)
-                                  :courses   (.. js/process -env -coursesTable)}}})
+(def adapters {:db     {:table-names {:bookmarks (.. js/process -env -bookmarksTable)
+                                      :resources (.. js/process -env -resourcesTable)
+                                      :courses   (.. js/process -env -coursesTable)}}
+               :stream {:stream-names {:courses (.. js/process -env -coursesStream)}}})
 
 (defn augment [& args]
   (go
     (let [{:keys [event] :as service} (apply service/create specs mappings adapters args)
           {:keys [added] :as payload} (cv/to-db-events event)
           {:keys [found errors]}      (async/<! (qa/fetch service added))
-          res                         (async/<! (ac/perform service [:transform found]))]
+          payloads                    (async/<! (ac/perform service [:transform found]))
+          ops-chans                   (async/merge (map #(ac/perform service [:put %])
+                                                          (vals payloads)))
+          res                         (async/<! (async/into [] ops-chans))]
       (service/done service res))))

functions/src/app/augment/mappings.cljs

@@ -5,19 +5,17 @@
             [shared.protocols.queryable :as qa]
             [shared.protocols.loggable :as log]
             [cljs.core.async :as async]
-            [clojure.string :as str])
+            [clojure.string :as str]
+            [shared.protocols.convertible :as cv])
   (:require-macros [cljs.core.async.macros :refer [go]]))
 
 (defn mappings []
 
-  (defn to-course-query [{:keys [offcourse-id] :as bm}]
-    (let [[repo curator course-id revision checkpoint-id] (str/split offcourse-id "::")]
-      {:course-id (str repo "::" curator "::" course-id)
-       :revision (int revision)}))
-
   (defmethod fetch :resources [{:keys [db]} bookmarks]
     (go
-      (let [courses-query   (map to-course-query bookmarks)
+      (let [courses-query   (->> bookmarks
+                                 (map :offcourse-id)
+                                 (map cv/to-query))
             courses-res     (async/<! (qa/fetch db courses-query))
             courses         (:found courses-res)
             resources-res   (async/<! (qa/fetch db bookmarks))
@@ -31,7 +29,9 @@
     (go
       (let [bookmarks-res   (async/<! (qa/fetch db resources))
             bookmarks       (:found bookmarks-res)
-            courses-query   (map to-course-query bookmarks)
+            courses-query   (->> bookmarks
+                                 (map :offcourse-id)
+                                 (map cv/to-query))
             courses-res     (async/<! (qa/fetch db courses-query))
             courses         (:found courses-res)
             errors          (mapcat :errors [courses-res bookmarks-res])]
@@ -44,13 +44,13 @@
           tags-data   (map #(identity [(:tags %1) (:resource-url %1)]) resources)
           ;; move impl logic to course model
           courses     (map #(impl/augment-course %1 tags-data) courses)]
-      (go courses)))
+      (go {:courses courses})))
 
   (defmethod perform [:put :nothing] [_ _]
     (go {:error :no-payload}))
 
   (defmethod perform [:put :errors] [_ [_ errors]]
     (go errors))
 
-  (defmethod perform :default [{:keys [bucket]} action]
-    (ac/perform bucket action)))
+  (defmethod perform :default [{:keys [stream]} action]
+    (ac/perform stream action)))

functions/src/app/augment/specs.cljs

@@ -10,10 +10,17 @@
 
 (defn specs []
 
-  (defmethod action-spec :transform [_]
-    (spec/tuple :offcourse/actions (spec/or :motherload ::motherload)))
-
   (spec/def :offcourse/query (spec/or :bookmarks (spec/coll-of :offcourse/resource)
                                       :resources (spec/coll-of :offcourse/bookmark)
                                       :courses (spec/coll-of :query/course-by-id)
-                                      :errors (spec/coll-of :offcourse/error))))
+                                      :errors (spec/coll-of :offcourse/error)))
+
+  (defmethod action-spec :transform [_]
+    (spec/tuple :offcourse/actions (spec/or :motherload ::motherload)))
+
+  (spec/def :offcourse/payload (spec/or :courses ::courses
+                                        :error   (spec/coll-of :offcourse/error)
+                                        :nothing nil?))
+
+  (defmethod action-spec :put [_]
+    (spec/tuple :offcourse/actions :offcourse/payload)))

serverless.yml

@@ -47,6 +47,7 @@ functions:
   augment:
     handler: functions/index.augment
     environment:
+      coursesStream: "courses-${opt:stage}"
       bookmarksTable: "bookmarks-${opt:stage}"
       resourcesTable: "resources-${opt:stage}"
       coursesTable: "courses-${opt:stage}"