1 Commits

Author SHA1 Message Date
95d0537c69 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>
2026-08-12 15:17:54 -07:00
6 changed files with 418 additions and 285 deletions

View File

@@ -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)))

View File

@@ -1,5 +1,6 @@
(ns auto-ap.jobs.sales-summaries (ns auto-ap.jobs.sales-summaries
(:require [auto-ap.datomic :refer [conn]] (:require [auto-ap.datomic :refer [conn]]
[auto-ap.datomic.sales-summaries :as d-ss]
[auto-ap.jobs.core :refer [execute]] [auto-ap.jobs.core :refer [execute]]
[auto-ap.logging :as alog] [auto-ap.logging :as alog]
[auto-ap.time :as atime] [auto-ap.time :as atime]
@@ -8,6 +9,7 @@
[clj-time.periodic :as per] [clj-time.periodic :as per]
[clojure.string :as str] [clojure.string :as str]
[com.brunobonacci.mulog :as mu] [com.brunobonacci.mulog :as mu]
[config.core :refer [env]]
[datomic.api :as dc])) [datomic.api :as dc]))
(defn mark-dirty [client start end] (defn mark-dirty [client start end]
@@ -39,27 +41,100 @@
(dc/db conn) (dc/db conn)
number))) number)))
(defn delete-all [] (defn delete-all []
@(dc/transact-async conn @(dc/transact-async conn
(->> (->>
(dc/q '[:find ?ss (dc/q '[:find ?ss
:where [?ss :sales-summary/date]] :where [?ss :sales-summary/date]]
(dc/db conn)) (dc/db conn))
(map (fn [[ ss]] (map (fn [[ss]]
[:db/retractEntity 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] (defn dirty-sales-summaries [c]
(let [client-id (dc/entid (dc/db conn) c)] (let [client-id (dc/entid (dc/db conn) c)]
(->> (dc/index-pull (dc/db conn) (->> (dc/index-pull (dc/db conn)
{:index :avet {: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]]}) :start [:sales-summary/client+dirty [client-id true]]})
(filter (fn [sales-summary] (filter (fn [sales-summary]
(= client-id (:db/id (:sales-summary/client 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] (defn- get-fee [c date]
(- (or (ffirst (dc/q '[:find ?f (- (or (ffirst (dc/q '[:find ?f
:in $ ?client ?d :in $ ?client ?d
@@ -192,8 +267,6 @@
:ledger-mapped/amount v :ledger-mapped/amount v
:ledger-mapped/ledger-side :ledger-side/credit})))) :ledger-mapped/ledger-side :ledger-side/credit}))))
(defn get-fees [c date] (defn get-fees [c date]
(when-let [fee (get-fee c date)] (when-let [fee (get-fee c date)]
{:db/id (str (java.util.UUID/randomUUID)) {:db/id (str (java.util.UUID/randomUUID))
@@ -287,7 +360,8 @@
(alog/info ::updating) (alog/info ::updating)
(let [manual-items (->> existing-summary (let [manual-items (->> existing-summary
:sales-summary/items :sales-summary/items
(filter :sales-summary-item/manual?)) (filter :sales-summary-item/manual?)
(map d-ss/<-pulled-item))
calculated-items (->> calculated-items (->>
(get-sales c date) (get-sales c date)
(concat (get-payment-items c date)) (concat (get-payment-items c date))
@@ -315,7 +389,6 @@
@(dc/transact conn [[:upsert-sales-summary result]])) @(dc/transact conn [[:upsert-sales-summary result]]))
@(dc/transact conn [{:db/id id :sales-summary/dirty false}])))))) @(dc/transact conn [{:db/id id :sales-summary/dirty false}]))))))
(defn reset-summaries [] (defn reset-summaries []
@(dc/transact conn (->> (dc/q '[:find ?sos @(dc/transact conn (->> (dc/q '[:find ?sos
:in $ :in $
@@ -324,9 +397,6 @@
(map (fn [[sos]] (map (fn [[sos]]
[:db/retractEntity sos]))))) [:db/retractEntity sos])))))
(comment (comment
(auto-ap.datomic/transact-schema conn) (auto-ap.datomic/transact-schema conn)
@@ -336,26 +406,19 @@
(dirty-sales-summaries [:client/code "NGWH"]) (dirty-sales-summaries [:client/code "NGWH"])
(apply mark-dirty [:client/code "NGWH"] (last-n-days 5)) (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/summary->journal-entry (dc/db conn) 17592314245819)
(iol-ion.tx.upsert-sales-summary-ledger/upsert-sales-summary (dc/db conn) {:db/id 17592314241429}) (iol-ion.tx.upsert-sales-summary-ledger/upsert-sales-summary (dc/db conn) {:db/id 17592314241429})
(mark-all-dirty 5) (mark-all-dirty 5)
(delete-all) (delete-all)
(sales-summaries-v2) (sales-summaries-v2)
1 1
(dc/q '[:find (pull ?sos [* {:sales-summary/sales-items [*]}]) (dc/q '[:find (pull ?sos [* {:sales-summary/sales-items [*]}])
:in $ :in $
:where [?sos :sales-summary/client [:client/code "NGHW"]] :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} @(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}]) {: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 [& _] (defn -main [& _]
(execute "sales-summaries" sales-summaries-v2)) (execute "sales-summaries" #(refresh-sales-summaries (days-arg (:args env)))))

View File

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

View File

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

View File

@@ -4,6 +4,7 @@
:refer [apply-pagination apply-sort-3 conn merge-query pull-many :refer [apply-pagination apply-sort-3 conn merge-query pull-many
query2]] query2]]
[auto-ap.datomic.accounts :as d-accounts] [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.graphql.utils :refer [extract-client-ids]]
[auto-ap.query-params :refer [wrap-copy-qp-pqp]] [auto-ap.query-params :refer [wrap-copy-qp-pqp]]
[auto-ap.client-routes :as client-routes] [auto-ap.client-routes :as client-routes]
@@ -116,18 +117,6 @@
(defn sort-items [ss] (defn sort-items [ss]
(sort-by (juxt :ledger-mapped/ledger-side :sales-summary-item/sort-order :sales-summary-item/category) 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] (defn truncate [s max-len]
(if (> (count s) max-len) (if (> (count s) max-len)
(str (subs s 0 (- max-len 3)) "...") (str (subs s 0 (- max-len 3)) "...")

View File

@@ -386,6 +386,20 @@ module "close_auto_invoices_job" {
cpu = 512 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" { module "yodlee2_accounts_job" {
count = var.enable_schedules ? 1 : 0 count = var.enable_schedules ? 1 : 0
source = "./background-job/" source = "./background-job/"