From 9aae86ada1418058b54bd2b3b0a05f98ef04ba63 Mon Sep 17 00:00:00 2001
From: eltrac <hi@eltr.ac>
Date: Thu, 15 Oct 2026 15:02:50 +0800
Subject: processor: concurrently starts multiple pipeline loops

---
 resources/config.edn                       |   3 +-
 src/weepinbell/processor/facade.clj        | 100 ++++++++++++++++++-----------
 test/weepinbell/processor_starter_test.clj |   3 +-
 3 files changed, 67 insertions(+), 39 deletions(-)

diff --git a/resources/config.edn b/resources/config.edn
index e5076eb..3c6ea6e 100644
--- a/resources/config.edn
+++ b/resources/config.edn
@@ -8,4 +8,5 @@
            :on-update []
            :on-delete []}
  :webhook-request-fn nil
- :feed-length 8}
+ :feed-length 8
+ :pipeline-concurrency 5}
diff --git a/src/weepinbell/processor/facade.clj b/src/weepinbell/processor/facade.clj
index fa52364..29febc2 100644
--- a/src/weepinbell/processor/facade.clj
+++ b/src/weepinbell/processor/facade.clj
@@ -16,53 +16,79 @@
   (s/keys :req-un [::on-receive ::on-create ::on-delete ::on-update]))
 (s/def ::storage (partial s/valid? :weepinbell.storage.spec/vtable))
 (s/def ::config
-  (s/keys :req-un [::base-url ::allowed-target-domains]))
+  (s/keys :req-un [::base-url ::allowed-target-domains ::pipeline-concurrency]))
 (s/def ::processor-meta
   (s/keys :req-un [::wmc ::callbacks ::storage ::config]))
 
-(defn exec-pipeline [callbacks storage config wm]
-  (let [{:keys [s t]} wm
-        id (digest (str s t))]
-    (try
-      ((:on-receive callbacks) {:webmention {:id id :source s :target t}})
-      (-> {:id id
-           :wm wm
-           :ok? true}
-          (check-storage storage)
-          verify
-          (clean storage callbacks)
-          (wrap config)
-          (write storage callbacks))
-      (catch Exception e
-        (timbre/error "unexpceted exception in pipeline: " (.getMessage e))
-        (timbre/debug e)))))
+(defn pipe
+  "Takes variable number of functions and form a pipeline.
+   Essentially a -> macro that returns a function, or a reversed comp."
+  [& functions]
+  (apply comp (reverse functions)))
+
+(defn assemble-pipeline
+  "Makes a proceesor pipeline. Returns a function that accepts webmention."
+  [{:keys [callbacks storage config]}]
+  (pipe
+   ;; initializer function
+   ;;; turns {:s ... :t ...} to pipeline map that the following functions
+   ;;; pass around; calls :on-receive hook.
+   (fn [wm]
+     (let [{:keys [s t]} wm
+           id (digest (str s t))]
+       ((:on-receive callbacks) {:webmention {:id id :source s :target t}})
+       {:id id
+        :wm wm
+        :ok? true}))
+   ;; pipeline steps
+   (partial check-storage storage)    ;; check if webmention exists
+   verify                             ;; verify if target is mentioned
+   (partial clean storage callbacks)  ;; delete old unverified webmention
+   (partial wrap config)              ;; wrap metadata into webmention
+   (partial write storage callbacks)));; write to storage
+
+(defn exec-pipeline
+  "Basically wraps (pipeline wm) call inside a try block.
+   catches exceptions and errors so they don't paralyze pipelines."
+  [pipeline wm]
+  (try
+    (pipeline wm)
+    (catch Throwable e
+      (timbre/error "unexpceted exception in pipeline: " (.getMessage e))
+      (timbre/debug e))))
+
+(defn dispatch-pipeline-loop
+  "Start a loop that takes from wmc and do pipeline when data come in."
+  [n wmc stopped pipeline]
+  (go-loop [wm (<! wmc)]
+    (let [{:keys [stop?]} wm]
+      (if (or stop? (nil? wm))
+        (do ;; handle stop signal
+          (put! stopped true)
+          (timbre/debug (str "#" n) "processor shuts down."))
+        (do ;; process found webmention
+          (timbre/info (str "#" n) "webmention found in queue, processing.")
+          (exec-pipeline pipeline wm)
+          (recur (<! wmc)))))))
 
 (defn start-processor
+  "Starts webmention processing pipelines. Returns a stopper function."
   [metadata]
   {:pre [(s/valid? ::processor-meta metadata)]
    :post [fn?]}
-  (let [{:keys [wmc callbacks storage config]} metadata
-        wmc-closed (chan)
-        close-wmc (fn []
-                    (close! wmc)
-                    (close! wmc-closed))]
-    ;; start processing loop
-    (go-loop [wm (<! wmc)]
-      (let [{:keys [stop?]} wm]
-        (if stop?
-          (do
-            (close-wmc)
-            (timbre/warn "webmention queue is cleared. processor shuts down."))
-          ;; process found webmention
-          (do
-            (timbre/info "webmention found in queue, processing.")
-            (exec-pipeline callbacks storage config wm)
-            (recur (<! wmc))))))
+  ;; wmc is webmention queue; stopped is channel where processors 
+  ;; put signals into, to notify that they're shut down
+  (let [wmc (:wmc metadata), stopped (chan)
+        concurrency (get-in metadata [:config :pipeline-concurrency])]
+    ;; dispatch pipelines
+    (doseq [n (range concurrency)
+            :let [pipeline (assemble-pipeline metadata)]]
+      (dispatch-pipeline-loop n wmc stopped pipeline))
     ;; return a function that stops the loop 
     (fn []
-      (put! wmc {:stop? true})
+      (dotimes [_ concurrency] (put! wmc {:stop? true}))
       ;; after putting the stop signal, processor might not
       ;; stop immediately. wait for it to process what's left 
       ;; inside webmention queue and close it.
-      (timbre/info "waiting for processor to shut down")
-      (<!! wmc-closed))))
+      (timbre/info "waiting for queue to clear, and processor to shut down")
+      (dotimes [_ concurrency] (<!! stopped)))))
diff --git a/test/weepinbell/processor_starter_test.clj b/test/weepinbell/processor_starter_test.clj
index f1ec273..8b604a5 100644
--- a/test/weepinbell/processor_starter_test.clj
+++ b/test/weepinbell/processor_starter_test.clj
@@ -16,7 +16,8 @@
                          :update (fn [& _])
                          :get-by-domain (fn [& _])}
                :config {:base-url "https://test.com"
-                        :allowed-target-domains ["test.com"]}}]
+                        :allowed-target-domains ["test.com"]
+                        :pipeline-concurrency 1}}]
     (start-processor pmeta)))
 
 (deftest start-processor-test
-- 
cgit v1.2.3