feat: Defer command when In-Reply-To absent, retry on thread arrival

Author: bzg

resources/bark-schema.edn

@@ -44,6 +44,16 @@
  :attachment/data             {:db/valueType :db.type/string}
  :email/in-reply-to           {:db/valueType :db.type/string}
  :email/references            {:db/valueType :db.type/string}
+ ;; Ancestors-mids: persisted union of References + In-Reply-To,
+ ;; one entry per ancestor message-id.  Indexed multi-valued so the
+ ;; pending-thread retry can locate emails sharing a thread in O(hits).
+ :email/ancestor-mids         {:db/valueType   :db.type/string
+                               :db/cardinality :db.cardinality/many}
+ ;; Set when an email is ingested while its In-Reply-To target is
+ ;; absent from the DB.  Phase 3/4 of process-email! is skipped on
+ ;; pending emails; they are retried when a related email arrives or
+ ;; when a TTL job forces the threading.
+ :email/pending-thread?       {:db/valueType :db.type/boolean}
  :email/headers-edn           {:db/valueType :db.type/string}
  :email/ingested-at           {:db/valueType :db.type/instant}
  :email/digested-at           {:db/valueType :db.type/instant}

src/bark/digest.clj

@@ -25,7 +25,8 @@
 
 (def email-pull-pattern
   '[:db/id :email/id :email/source :email/subject :email/message-id
-    :email/in-reply-to :email/references
+    :email/in-reply-to :email/references :email/ancestor-mids
+    :email/pending-thread?
     :email/author-address :email/author-name
     :email/from-address :email/from-name
     :email/date-sent :email/ingested-at
@@ -472,17 +473,68 @@
       (log/info "Auto-created single-member series for"
                 (count patches) "patch attachments"))))
 
+;; ---------------------------------------------------------------------------
+;; Pending-thread retry (out-of-order delivery rescue)
+;; ---------------------------------------------------------------------------
+
+(defn- in-reply-to-resolved?
+  "True when the email's In-Reply-To target is nil (root) or already
+  stored in the DB.  When false, the email is held as pending — its
+  threading and commands are deferred until the missing ancestor (or
+  any thread sibling) arrives."
+  [db email]
+  (let [irt (:email/in-reply-to email)]
+    (or (nil? irt)
+        (some? (d/entid db [:email/message-id irt])))))
+
+(declare process-email!)
+
+(defn- retry-pending-in-shared-thread!
+  "After processing `email` normally, retry any pending email that
+  shares at least one ancestor mid with it (or that has `email`'s own
+  mid in its ancestors).  The recursive `process-email!` call retracts
+  the pending flag if its In-Reply-To is now resolvable."
+  [conn email source-map sources]
+  (let [own-mid   (:email/message-id email)
+        ancestors (cond-> (set (:email/ancestor-mids email))
+                    own-mid (conj own-mid))]
+    (when (seq ancestors)
+      (let [pendings (d/q '[:find [?e ...] :in $ [?mid ...]
+                            :where
+                            [?e :email/pending-thread? true]
+                            [?e :email/ancestor-mids ?mid]]
+                          (d/db conn) (vec ancestors))]
+        (doseq [pending-eid pendings]
+          (let [pending-email (d/pull (d/db conn) email-pull-pattern pending-eid)]
+            (log/info "Retrying pending email" (:email/message-id pending-email)
+                      "(triggered by" own-mid ")")
+            (process-email! conn source-map sources pending-email)))))))
+
 ;; ---------------------------------------------------------------------------
 ;; Single-email processing — orchestrator
 ;; ---------------------------------------------------------------------------
 
 (defn process-email!
   "Process a single email: classify source, detect report, thread,
-  apply commands, manage series. Called after store-email! succeeds."
-  [conn source-map sources email]
-  (let [message-id (:email/message-id email)
-        eid        (:db/id email)
-        from-addr  (:email/author-address email)
+  apply commands, manage series. Called after store-email! succeeds.
+
+  When the email's In-Reply-To points to a message-id absent from
+  the DB, Phase 3 (threading + commands) and Phase 4 (post-creation
+  hooks) are skipped and the email is flagged
+  `:email/pending-thread? true`.  A later arrival sharing the same
+  thread triggers the retry via `retry-pending-in-shared-thread!`,
+  or the TTL flush forces processing after N days.
+
+  Opts:
+    :force-thread? — bypass the pending-thread guard, threading the
+                     email with whatever ancestors currently exist in
+                     the DB.  Used by the TTL flush."
+  ([conn source-map sources email] (process-email! conn source-map sources email {}))
+  ([conn source-map sources email {:keys [force-thread?]}]
+  (let [message-id   (:email/message-id email)
+        eid          (:db/id email)
+        from-addr    (:email/author-address email)
+        was-pending? (:email/pending-thread? email)
         [source-name email delivery] (resolve-email-source! conn email sources)]
     (if-not source-name
       (log/debug "No matching source for" message-id "— skipping")
@@ -499,23 +551,75 @@
               [report-eid report-info]
               (maybe-create-report! conn eid message-id email from-addr
                                     source-name source-cfg via-channel? rroles)
+              db           (d/db conn)]
+
+          (if (or force-thread? (in-reply-to-resolved? db email))
+            ;; --- Normal path: threading, commands, post-creation hooks ---
+            (do
+              (when was-pending?
+                (d/transact! conn [[:db/retract eid :email/pending-thread? true]])
+                (log/info "Cleared pending flag on" message-id))
+              (let [parent-eids  (find-reports-for-email email db)
+                    nearest-eids (find-nearest-report email db)
+                    ;; Recover the existing report-eid on retry so Phase 4
+                    ;; hooks (link-related, close-superseded-thread, …) can
+                    ;; run for pending emails that created a report on first
+                    ;; pass but were skipped past Phase 3/4.
+                    report-eid   (or report-eid
+                                     (when (and was-pending?
+                                                (report-exists? db message-id))
+                                       (d/entid db [:report/message-id message-id])))]
+
+                (when (and (seq parent-eids) via-channel?)
+                  (thread-and-apply-commands! conn eid email from-addr source-name rroles
+                                              source-map delivery parent-eids nearest-eids
+                                              (some? report-eid)))
+
+                ;; Phase 4: post-creation hooks (plan is pure, execution is effectful)
+                (when report-eid
+                  (let [patches (detect/build-patch-entities email)
+                        plan    (post-creation-plan report-info nearest-eids parent-eids patches)]
+                    (run-post-creation-hooks! conn report-eid eid email from-addr report-info
+                                              parent-eids nearest-eids patches plan)))
+
+                ;; Mark email as fully digested so future re-fetches can skip it.
+                (d/transact! conn [{:db/id eid :email/digested-at (Date.)}])
+
+                ;; Out-of-order rescue: this email may have unblocked pending
+                ;; siblings/descendants.  Run AFTER the digested-at write so
+                ;; the recursive call sees a consistent state.
+                (retry-pending-in-shared-thread! conn email source-map sources)))
+
+            ;; --- Pending path: defer threading and commands ---
+            (do (d/transact! conn [{:db/id eid
+                                    :email/pending-thread? true
+                                    :email/digested-at (Date.)}])
+                (log/info "Pending:" message-id "— in-reply-to"
+                          (:email/in-reply-to email) "absent from DB")))))))))
+
+;; ---------------------------------------------------------------------------
+;; TTL flush — force-process pending emails older than max-age-days
+;; ---------------------------------------------------------------------------
 
-              ;; Phase 3: threading and commands
-              db           (d/db conn)
-              parent-eids  (find-reports-for-email email db)
-              nearest-eids (find-nearest-report email db)]
-
-          (when (and (seq parent-eids) via-channel?)
-            (thread-and-apply-commands! conn eid email from-addr source-name rroles
-                                        source-map delivery parent-eids nearest-eids
-                                        (some? report-eid)))
-
-          ;; Phase 4: post-creation hooks (plan is pure, execution is effectful)
-          (when report-eid
-            (let [patches (detect/build-patch-entities email)
-                  plan    (post-creation-plan report-info nearest-eids parent-eids patches)]
-              (run-post-creation-hooks! conn report-eid eid email from-addr report-info
-                                        parent-eids nearest-eids patches plan)))
-
-          ;; Mark email as fully digested so future re-fetches can skip it.
-          (d/transact! conn [{:db/id eid :email/digested-at (Date.)}]))))))
+(defn flush-stale-pending!
+  "Force-process pending emails older than `max-age-days`. The pending
+  flag is retracted and threading runs against whatever ancestors
+  currently exist in the DB. Returns the count of flushed emails."
+  [conn source-map sources max-age-days]
+  (let [cutoff   (Date. (- (System/currentTimeMillis)
+                           (* max-age-days 24 60 60 1000)))
+        pendings (d/q '[:find [?e ...] :in $ ?cutoff
+                        :where
+                        [?e :email/pending-thread? true]
+                        [?e :email/ingested-at ?ts]
+                        [(.before ^java.util.Date ?ts ^java.util.Date ?cutoff)]]
+                      (d/db conn) cutoff)]
+    (when (seq pendings)
+      (log/info "Flushing" (count pendings)
+                "stale pending email(s) older than" max-age-days "day(s)"))
+    (doseq [eid pendings]
+      (d/transact! conn [[:db/retract eid :email/pending-thread? true]])
+      (let [email (d/pull (d/db conn) email-pull-pattern eid)]
+        (log/info "TTL-flush" (:email/message-id email))
+        (process-email! conn source-map sources email {:force-thread? true})))
+    (count pendings)))

src/bark/ingest.clj

@@ -175,6 +175,14 @@
                       (parse-message-ids (if (vector? v)
                                            (str/join " " (keep identity v))
                                            v)))
+        ancestor-mids (let [refs-vec (when references
+                                       (re-seq #"<[^>]+>" references))
+                            base     (vec refs-vec)]
+                        (->> (cond-> base
+                               (and in-reply-to (not (some #{in-reply-to} base)))
+                               (conj in-reply-to))
+                             distinct
+                             vec))
         attachments (mapv (fn [att]
                             (let [filename (or (:filename att) "unnamed")
                                   is-patch (re-find #"(?i)\.(patch|diff)$" filename)
@@ -231,6 +239,7 @@
       (seq attachments)           (assoc :email/attachments attachments)
       in-reply-to                 (assoc :email/in-reply-to in-reply-to)
       references                  (assoc :email/references references)
+      (seq ancestor-mids)         (assoc :email/ancestor-mids ancestor-mids)
       headers-edn                 (assoc :email/headers-edn headers-edn)))))
 
 ;; ---------------------------------------------------------------------------

src/bark/main.clj

@@ -417,18 +417,25 @@
 
 (def ^:private one-day-ms (* 24 60 60 1000))
 
+;; Pending-thread emails are flushed (force-threaded) after this many
+;; days, so that genuine orphans (In-Reply-To target never arrives)
+;; eventually leave the queue rather than accumulating forever.
+(def ^:private pending-flush-max-age-days 7)
+
 (defn- maybe-expire!
-  "Run expire-reports! if at least one day has elapsed since `last-ms`.
-  Returns the updated timestamp on success, or `last-ms` on failure
-  so that the next cycle retries."
-  [db-conn source-map last-ms]
+  "Run expire-reports! and flush stale pending emails if at least one
+  day has elapsed since `last-ms`.  Returns the updated timestamp on
+  success, or `last-ms` on failure so that the next cycle retries."
+  [db-conn source-map sources last-ms]
   (let [now (System/currentTimeMillis)]
     (if (> (- now last-ms) one-day-ms)
       (try
         (expire/expire-reports! db-conn source-map)
+        (digest/flush-stale-pending! db-conn source-map sources
+                                     pending-flush-max-age-days)
         now
         (catch Exception e
-          (log/error e "Expire failed:" (blog/exception-msg e))
+          (log/error e "Expire/flush failed:" (blog/exception-msg e))
           last-ms))
       last-ms)))
 
@@ -518,7 +525,7 @@
                   (try
                     (log/info "Mailbox connected, folder:" folder)
                     (catch-up-fetch! src db-conn folder fetch-opts source-map sources ingest-opts mailbox-type)
-                    (let [ts (maybe-expire! db-conn source-map last-expire-ms)]
+                    (let [ts (maybe-expire! db-conn source-map sources last-expire-ms)]
                       (when-not (shutting-down?)
                         (start-watch! src db-conn folder source-map sources ingest-opts))
                       ts)
@@ -548,6 +555,8 @@
     (try
       (catch-up-fetch! src db-conn folder fetch-opts source-map sources ingest-opts mailbox-type)
       (expire/expire-reports! db-conn source-map)
+      (digest/flush-stale-pending! db-conn source-map sources
+                                   pending-flush-max-age-days)
       (finally
         (close-mailbox! src)))))