feat(sales-summaries): refresh on a schedule, skipping accepted days

sales-summaries-v2 recomputes every dirty summary, but nothing set the dirty
flag on a schedule: mark-all-dirty was only ever called by hand from the
comment block, and the job was registered in neither server.clj's
INTEGREAT_JOB dispatch nor terraform/deploy.tf, so -main was dead code that
could never run in production. Summaries were only recalculated when someone
remembered to do it in the REPL, and POS data keeps arriving after a business
day closes, so a summary computed once on the day was routinely wrong and
stayed wrong.

Add a daily job that marks the trailing 7 days dirty and recomputes them,
leaving finished work alone. "Finished" is the condition the app already calls
Balanced -- debits equal credits and every line is mapped to an account. Since
that is derived rather than stored, a summary that later falls out of balance
is picked up again on the next run.

Extract the Balanced predicate into auto-ap.datomic.sales-summaries so the
grid's pill and the job share one definition, rather than a background job
requiring an SSR namespace. total-debits/total-credits resolve the ledger side
from either a plain keyword or the {:db/ident ...} map a pull returns, and
accepted? requires every item to declare a side: un-normalized pulled items
otherwise sum to 0.0 on both sides, read as balanced, and get skipped
silently and permanently.

Also fix sales-summaries-v2 destroying user-entered line items. It filtered
for :sales-summary-item/manual? to preserve them, but dirty-sales-summaries'
index-pull selector never fetched :sales-summary/items, so manual-items was
always empty. Because items is a component attribute upserted via
[:reset-rels ...], every recompute deleted the hand-entered lines -- often the
very lines that make a summary balance. Harmless while nothing ran on a
schedule; destructive the moment this does.

Register the job in the admin Background Jobs dropdown too, with a days
field: schedules are prod-only, so the admin page is the only way to run it
on staging.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/clj/auto_ap/datomic/sales_summaries.clj

@@ -0,0 +1,55 @@
+(ns auto-ap.datomic.sales-summaries
+  (:require
+   [iol-ion.query :refer [dollars=]]))
+
+(defn- ledger-side
+  "The ledger side of an item as a keyword, whether it arrived as a plain keyword (a
+  transaction map, or a pull using `:xform iol-ion.query/ident`) or as the `{:db/ident ...}`
+  map a plain `pull` returns. Resolving both shapes here matters: items whose side does not
+  compare equal are counted on neither side, which would leave a summary looking balanced at
+  zero and therefore silently accepted."
+  [item]
+  (let [side (:ledger-mapped/ledger-side item)]
+    (if (map? side) (:db/ident side) side)))
+
+(defn- side-total [side items]
+  (->> items
+       (filter #(= side (ledger-side %)))
+       (map #(:ledger-mapped/amount % 0.0))
+       (reduce + 0.0)))
+
+(defn total-debits [items]
+  (side-total :ledger-side/debit items))
+
+(defn total-credits [items]
+  (side-total :ledger-side/credit items))
+
+(defn fully-mapped? [items]
+  (every? :ledger-mapped/account items))
+
+(defn fully-sided?
+  "Every item says which side of the ledger it belongs on. Guards `accepted?` against
+  reading a collection of sideless items as balanced at zero."
+  [items]
+  (every? #(#{:ledger-side/debit :ledger-side/credit} (ledger-side %)) items))
+
+(defn balanced? [items]
+  (dollars= (total-debits items) (total-credits items)))
+
+(defn accepted?
+  "True once a summary is finished: every line is mapped to an account and debits equal
+  credits. This is the same condition the sales summaries grid renders as \"Balanced\", and
+  the condition the scheduled refresh treats as \"leave this alone\"."
+  [items]
+  (boolean (and (seq items)
+                (fully-mapped? items)
+                (fully-sided? items)
+                (balanced? items))))
+
+(defn <-pulled-item
+  "Flattens the ref values on a pulled sales summary item back to the scalars a transaction
+  expects. `accepted?` reads either shape, so this is only needed on the write path."
+  [item]
+  (cond-> item
+    (map? (:ledger-mapped/ledger-side item)) (update :ledger-mapped/ledger-side :db/ident)
+    (map? (:ledger-mapped/account item))     (update :ledger-mapped/account :db/id)))

src/clj/auto_ap/jobs/sales_summaries.clj

@@ -1,5 +1,6 @@
 (ns auto-ap.jobs.sales-summaries
   (:require [auto-ap.datomic :refer [conn]]
+            [auto-ap.datomic.sales-summaries :as d-ss]
             [auto-ap.jobs.core :refer [execute]]
             [auto-ap.logging :as alog]
             [auto-ap.time :as atime]
@@ -8,6 +9,7 @@
             [clj-time.periodic :as per]
             [clojure.string :as str]
             [com.brunobonacci.mulog :as mu]
+            [config.core :refer [env]]
             [datomic.api :as dc]))
 
 (defn mark-dirty [client start end]
@@ -39,27 +41,100 @@
                 (dc/db conn)
                 number)))
 
-
 (defn delete-all []
   @(dc/transact-async conn
                       (->>
                        (dc/q '[:find ?ss
                                :where [?ss :sales-summary/date]]
                              (dc/db conn))
-                (map (fn [[ ss]]
+                       (map (fn [[ss]]
                               [:db/retractEntity ss])))))
 
-
+(def item-read
+  "Enough of a summary item to both evaluate `d-ss/accepted?` and transact the item back
+  unchanged. `:db/id` matters: `:sales-summary/items` is a component attribute upserted via
+  `[:reset-rels ...]`, so an item re-transacted without its id is deleted and recreated."
+  '[:db/id
+    :sales-summary-item/category
+    :sales-summary-item/sort-order
+    :sales-summary-item/manual?
+    :ledger-mapped/amount
+    {:ledger-mapped/ledger-side [:db/ident]}
+    {:ledger-mapped/account [:db/id]}])
 
 (defn dirty-sales-summaries [c]
   (let [client-id (dc/entid (dc/db conn) c)]
     (->> (dc/index-pull  (dc/db conn)
                          {:index :avet
-                          :selector '[:sales-summary/date :sales-summary/client :db/id]
+                          :selector (conj '[:sales-summary/date :sales-summary/client :db/id]
+                                          {:sales-summary/items item-read})
                           :start [:sales-summary/client+dirty [client-id true]]})
          (filter (fn [sales-summary]
                    (= client-id (:db/id (:sales-summary/client sales-summary))))))))
 
+(def default-days
+  "How far back the scheduled refresh looks for summaries that still need recomputing."
+  7)
+
+(defn trailing-window
+  "`[start end)` covering the last `days` business days, ending with today. `end` is
+  exclusive, matching both `periodic-seq`'s 3-arity and the grid's date filters, so
+  `(trailing-window 7)` is day -6 through today inclusive."
+  [days]
+  [(.toDateMidnight (atime/localize (time/minus (time/now) (time/days (dec days)))))
+   (.toDateMidnight (atime/localize (time/plus (time/now) (time/days 1))))])
+
+(defn accepted-client+dates
+  "Set of `[client-id date]` pairs in `[start end)` whose summary is already accepted, and
+  so should be left alone rather than re-marked. Accepted means balanced with every line
+  mapped to an account — the condition the grid renders as \"Balanced\"."
+  [db start end]
+  (->> (dc/q '[:find (pull ?ss selector)
+               :in $ ?start ?end selector
+               :where
+               [?ss :sales-summary/date ?d]
+               [(>= ?d ?start)]
+               [(< ?d ?end)]]
+             db
+             (c/to-date start)
+             (c/to-date end)
+             (conj '[:sales-summary/date {:sales-summary/client [:db/id]}]
+                   {:sales-summary/items item-read}))
+       (map first)
+       (filter #(d-ss/accepted? (map d-ss/<-pulled-item (:sales-summary/items %))))
+       (map (juxt (comp :db/id :sales-summary/client) :sales-summary/date))
+       set))
+
+(defn mark-stale-dirty
+  "Marks every client/day in the trailing `days` window dirty so `sales-summaries-v2` will
+  recompute it, skipping days whose summary is already accepted. Because accepted is derived
+  rather than stored, a summary that later falls out of balance is picked up again on the
+  next run. Returns the number of client/days marked."
+  [days]
+  (let [db (dc/db conn)
+        [start end] (trailing-window days)
+        accepted (accepted-client+dates db start end)
+        clients (map first (dc/q '[:find ?c
+                                   :in $
+                                   :where [_ :sales-order/client ?c]]
+                                 db))
+        dates (map c/to-date (per/periodic-seq start end (time/days 1)))
+        tx-data (for [client clients
+                      date dates
+                      :when (not (accepted [client date]))]
+                  {:sales-summary/client client
+                   :sales-summary/date date
+                   :sales-summary/dirty true
+                   :sales-summary/client+date [client date]})]
+    (alog/info ::marking-dirty
+               :days days
+               :client-count (count clients)
+               :accepted-count (count accepted)
+               :marked (count tx-data))
+    (doseq [batch (partition-all 500 tx-data)]
+      @(dc/transact conn batch))
+    (count tx-data)))
+
 (defn- get-fee [c date]
   (- (or (ffirst (dc/q '[:find ?f
                          :in $ ?client ?d
@@ -192,8 +267,6 @@
            :ledger-mapped/amount v
            :ledger-mapped/ledger-side :ledger-side/credit}))))
 
-
-
 (defn get-fees [c date]
   (when-let [fee (get-fee c date)]
     {:db/id (str (java.util.UUID/randomUUID))
@@ -287,7 +360,8 @@
       (alog/info ::updating)
       (let [manual-items (->> existing-summary
                               :sales-summary/items
-                               (filter :sales-summary-item/manual?))
+                              (filter :sales-summary-item/manual?)
+                              (map d-ss/<-pulled-item))
             calculated-items (->>
                               (get-sales c date)
                               (concat (get-payment-items c date))
@@ -315,7 +389,6 @@
             @(dc/transact conn [[:upsert-sales-summary result]]))
           @(dc/transact conn [{:db/id id :sales-summary/dirty false}]))))))
 
-
 (defn reset-summaries []
   @(dc/transact conn (->> (dc/q '[:find ?sos
                                   :in $
@@ -324,9 +397,6 @@
                           (map (fn [[sos]]
                                  [:db/retractEntity sos])))))
 
-
-
-
 (comment
   (auto-ap.datomic/transact-schema conn)
 
@@ -336,26 +406,19 @@
 
   (dirty-sales-summaries [:client/code "NGWH"])
 
-
   (apply mark-dirty [:client/code "NGWH"] (last-n-days 5))
 
   (iol-ion.tx.upsert-sales-summary-ledger/summary->journal-entry (dc/db conn) 17592314245819)
 
-
   (iol-ion.tx.upsert-sales-summary-ledger/upsert-sales-summary (dc/db conn) {:db/id 17592314241429})
 
-
-
   (mark-all-dirty 5)
   (delete-all)
 
-
   (sales-summaries-v2)
 
-  
   1
 
-
   (dc/q '[:find (pull ?sos [* {:sales-summary/sales-items [*]}])
           :in $
           :where [?sos :sales-summary/client [:client/code "NGHW"]]
@@ -386,15 +449,24 @@
   @(dc/transact conn [{:db/id :sales-summary/total-tax :db/ident :sales-summary/total-tax-legacy}
                       {:db/id :sales-summary/total-tip :db/ident :sales-summary/total-tip-legacy}])
 
-  (auto-ap.datomic/transact-schema conn)
-
-
-
-
-  )
+  (auto-ap.datomic/transact-schema conn))
 
+(defn days-arg
+  "Trailing-window size from the job's `args`, e.g. `{:days 30}` set as a container override
+  from the admin Background Jobs page for an ad-hoc wider backfill. Values arrive as EDN but
+  may still be strings, so coerce defensively the way load-historical-sales does."
+  [args]
+  (let [days (:days args)]
+    (cond-> (or days default-days)
+      (string? days) (#(Long/parseLong %)))))
 
+(defn refresh-sales-summaries
+  "Marks the trailing `days` window dirty, skipping accepted summaries, then recomputes
+  everything left dirty."
+  ([] (refresh-sales-summaries default-days))
+  ([days]
+   (mark-stale-dirty days)
+   (sales-summaries-v2)))
 
 (defn -main [& _]
-  (execute "sales-summaries" sales-summaries-v2))
-  
+  (execute "sales-summaries" #(refresh-sales-summaries (days-arg (:args env)))))

src/clj/auto_ap/server.clj

@@ -13,6 +13,7 @@
    [auto-ap.jobs.load-historical-sales :as job-load-historical-sales]
    [auto-ap.jobs.plaid :as job-plaid]
    [auto-ap.jobs.register-invoice-import :as job-register-invoice-import]
+   [auto-ap.jobs.sales-summaries :as job-sales-summaries]
    [auto-ap.jobs.square :as job-square]
    [auto-ap.jobs.sysco :as job-sysco]
    [auto-ap.jobs.vendor-usages :as job-vendor-usages]
@@ -33,7 +34,6 @@
   (.addShutdownHook (Runtime/getRuntime)
                     (Thread. f)))
 
-
 (defn gzip-handler []
   (let [gz (GzipHandler.)]
     (doto gz
@@ -126,6 +126,9 @@
           (= job "close-auto-invoices")
           (job-close-auto-invoices/-main)
 
+          (= job "sales-summaries")
+          (job-sales-summaries/-main)
+
           (= job "ezcater-upsert")
           (job-ezcater-upsert/-main)
 

src/clj/auto_ap/ssr/admin/background_jobs.clj

@@ -36,7 +36,6 @@
    (sort-by :created-at)
    reverse))
 
-
 (defn is-background-job?
   "This function checks whether a given task is a background job.
    It does this by checking the environment of the task's container definitions for an environment variable
@@ -60,7 +59,7 @@
 (defn job-exited-successfully? [task]
   (if (= 0 (->> task
                 :containers
-                (filter (comp #{"integreat-app" } :name))
+                (filter (comp #{"integreat-app"} :name))
                 (first)
                 :exit-code))
     true
@@ -85,7 +84,7 @@
                   (filter is-background-job?)
                   (map ecs-task->job))]
     [jobs (count jobs)]))
-(def query-schema (mc/schema [:map ]))
+(def query-schema (mc/schema [:map]))
 
 (def grid-page
   (helper/build {:id                 "job-table"
@@ -107,8 +106,7 @@
                  :entity-name "Job"
                  :query-schema query-schema
                  :route       :admin-job-table
-                 :headers     [
-                               {:key      "start"
+                 :headers     [{:key      "start"
                                 :name     "Start"
                                 :render   #(some-> % :start-date (atime/unparse-local atime/standard-time))}
                                {:key      "end"
@@ -150,7 +148,7 @@
                          :network-configuration      {:aws-vpc-configuration {:subnets          ["subnet-5e675761" "subnet-8519fde2" "subnet-89bab8d4"]
                                                                               :security-groups  ["sg-004e5855310c453a3" "sg-02d167406b1082698"]
                                                                               :assign-public-ip AssignPublicIp/ENABLED}}}
-                  args (assoc-in [:overrides :container-overrides ] [{:name "integreat-app" :environment [{:name "args" :value (pr-str args)}]}]))))
+                  args (assoc-in [:overrides :container-overrides] [{:name "integreat-app" :environment [{:name "args" :value (pr-str args)}]}]))))
 
 (defn job-start [{:keys [form-params]}]
   (if (not (get (currently-running-jobs) (:name form-params)))
@@ -174,20 +172,18 @@
                                        [:pre.text-xs.mr-1 "s3://data.prod.app.integreatconsult.com/bulk-import/"]
                                        (com/text-input {:placeholder "ledger-data.csv"
                                                         :name        (fc/field-name)
-                                              :value (fc/field-value)} )]))]
+                                                        :value (fc/field-value)})]))]
               (= "register-invoice-import" name)
-              [
-               (fc/with-field :invoice-url
+              [(fc/with-field :invoice-url
                  (com/validated-field {:label  "Url"
                                        :errors (fc/field-errors)}
                                       [:div.flex.place-items-center.gap-2
                                        [:pre.text-xs.mr-1 "s3://data.prod.app.integreatconsult.com/bulk-import/"]
                                        (com/text-input {:placeholder "invoice-data.csv"
                                                         :name        (fc/field-name)
-                                                          :value       (fc/field-value)} )]))]
+                                                        :value       (fc/field-value)})]))]
               (= "load-historical-sales" name)
-              [
-               (fc/with-field :client
+              [(fc/with-field :client
                  (com/validated-field {:label  "Client"
                                        :errors (fc/field-errors)}
                                       (com/typeahead {:name        (fc/field-name)
@@ -200,13 +196,17 @@
                                        :errors (fc/field-errors)}
                                       (com/text-input {:placeholder "60"
                                                        :name        (fc/field-name)
-                                                       :value       (fc/field-value)} )))]
-              :else nil))
+                                                       :value       (fc/field-value)})))]
+              (= "sales-summaries" name)
+              [(fc/with-field :days
+                 (com/validated-field {:label  "Days to refresh"
+                                       :errors (fc/field-errors)}
+                                      (com/text-input {:placeholder "7"
+                                                       :name        (fc/field-name)
+                                                       :value       (fc/field-value)})))]
+              :else nil)))
 
-  
-  )
-
-(defn subform [{{:keys [name]} :query-params }]
+(defn subform [{{:keys [name]} :query-params}]
   (html-response
    (fc/start-form {} nil
                   (subform* {:name name}))))
@@ -239,13 +239,14 @@
                                                                                       ["register-invoice-import" "Register Invoice Import "]
                                                                                       ["ezcater-upsert" "Upsert recent ezcater orders"]
                                                                                       ["load-historical-sales" "Load Historical Square Sales"]
+                                                                                      ["sales-summaries" "Refresh Sales Summaries"]
                                                                                       ["export-backup" "Export Backup"]]
                                                                           :hx-get    (bidi/path-for ssr-routes/only-routes
                                                                                                     :admin-job-subform)
                                                                           :hx-target "#sub-form"
                                                                           :hx-swap   "innerHTML"})))
 
-             [:div#sub-form (subform* {:name (fc/with-field :name (fc/field-value))}) ]]
+                                      [:div#sub-form (subform* {:name (fc/with-field :name (fc/field-value))})]]
                                      [:div
 
                                       (com/form-errors {:errors (:errors fc/*form-errors*)})
@@ -256,8 +257,7 @@
                              [:ledger-url {:optional true} [:string {:min 1}]]
                              [:invoice-url {:optional true} [:string {:min 1}]]
                              [:client {:optional true} entity-id]
-                             [:days {:optional true} [:int {:min 1 :max 120}]]
-                             ]))
+                             [:days {:optional true} [:int {:min 1 :max 120}]]]))
 
 (def key->handler
   (apply-middleware-to-all-handlers

src/clj/auto_ap/ssr/pos/sales_summaries.clj

@@ -4,6 +4,7 @@
     :refer [apply-pagination apply-sort-3 conn merge-query pull-many
             query2]]
    [auto-ap.datomic.accounts :as d-accounts]
+   [auto-ap.datomic.sales-summaries :refer [total-credits total-debits]]
    [auto-ap.graphql.utils :refer [extract-client-ids]]
    [auto-ap.query-params :refer [wrap-copy-qp-pqp]]
    [auto-ap.client-routes :as client-routes]
@@ -116,18 +117,6 @@
 (defn sort-items [ss]
   (sort-by (juxt :ledger-mapped/ledger-side :sales-summary-item/sort-order :sales-summary-item/category) ss))
 
-(defn total-debits [items]
-  (->> items
-       (filter #(= :ledger-side/debit (:ledger-mapped/ledger-side %)))
-       (map #(:ledger-mapped/amount % 0.0))
-       (reduce + 0.0)))
-
-(defn total-credits [items]
-  (->> items
-       (filter #(= :ledger-side/credit (:ledger-mapped/ledger-side %)))
-       (map #(:ledger-mapped/amount % 0.0))
-       (reduce + 0.0)))
-
 (defn truncate [s max-len]
   (if (> (count s) max-len)
     (str (subs s 0 (- max-len 3)) "...")

terraform/deploy.tf

@@ -386,6 +386,20 @@ module "close_auto_invoices_job" {
        cpu = 512
 }
 
+module "sales_summaries_job" {
+       count = var.enable_schedules ? 1 : 0
+       source = "./background-job/"
+       ecs_cluster = var.ecs_cluster
+       task_role_arn = var.task_role_arn
+       stage = var.stage
+       schedule = "rate(1 day)"
+       job_name = "sales-summaries"
+       execution_role_arn = var.execution_role_arn
+       use_schedule = true
+       memory = 4096
+       cpu = 2048
+}
+
 module "yodlee2_accounts_job" {
        count = var.enable_schedules ? 1 : 0
        source = "./background-job/"