681 lines
37 KiB
Clojure
681 lines
37 KiB
Clojure
(ns auto-ap.ledger
|
|
(:require
|
|
[auto-ap.datomic
|
|
:refer [audit-transact
|
|
audit-transact-batch
|
|
conn
|
|
pull-id
|
|
pull-ref
|
|
remove-nils]]
|
|
[auto-ap.utils :refer [by dollars-0? dollars=]]
|
|
[clj-time.coerce :as c]
|
|
[clj-time.core :as t]
|
|
[clojure.tools.logging :as log]
|
|
[com.brunobonacci.mulog :as mu]
|
|
[com.unbounce.dogstatsd.core :as statsd]
|
|
[datomic.client.api :as dc]
|
|
[iol-ion.tx :refer [upsert-ledger]]))
|
|
|
|
(defn datums->impacted-entity [db [e changes]]
|
|
(let [entity (dc/pull db '[{:invoice/_expense-accounts [:db/id] :transaction/_accounts [:db/id]}] e)
|
|
namespaces (->> changes
|
|
(map #(:db/ident (dc/pull db '[:db/ident] (:a %))))
|
|
(map namespace)
|
|
set)]
|
|
(cond (namespaces "invoice" ) [[:invoice e]]
|
|
(namespaces "invoice-expense-account" ) [[:invoice (:db/id (:invoice/_expense-accounts entity))]]
|
|
(namespaces "transaction-account" ) [[:transaction (:db/id (:transaction/_accounts entity))]]
|
|
(namespaces "transaction" ) [[:transaction e]]
|
|
#_#_(namespaces "expected-deposit" ) [[:expected-deposit e]]
|
|
:else nil)))
|
|
|
|
(defmulti entity-change->ledger (fn [_ [type]]
|
|
type))
|
|
|
|
(defmethod entity-change->ledger :invoice
|
|
[db [_ id]]
|
|
(when id
|
|
(let [entity (dc/pull db ['* {:invoice/vendor '[*]
|
|
:invoice/payment '[*]
|
|
:invoice/status '[:db/ident]
|
|
:invoice/import-status '[:db/ident]}] id)
|
|
credit-invoice? (< (:invoice/total entity 0.0) 0.0)]
|
|
(when-not (or
|
|
(not (:invoice/total entity))
|
|
(= true (:invoice/exclude-from-ledger entity))
|
|
(= :import-status/pending (:db/ident (:invoice/import-status entity)))
|
|
(= :invoice-status/voided (:db/ident (:invoice/status entity)))
|
|
(dollars-0? (:invoice/total entity)))
|
|
|
|
(remove-nils
|
|
{:journal-entry/source "invoice"
|
|
:journal-entry/client (:db/id (:invoice/client entity))
|
|
:journal-entry/date (:invoice/date entity)
|
|
:journal-entry/original-entity (:db/id entity)
|
|
:journal-entry/vendor (:db/id (:invoice/vendor entity))
|
|
:journal-entry/amount (Math/abs (:invoice/total entity))
|
|
|
|
:journal-entry/line-items (into [(cond-> {:db/id (str (:db/id entity) "-" 0)
|
|
:journal-entry-line/account :account/accounts-payable
|
|
:journal-entry-line/location "A"
|
|
}
|
|
credit-invoice? (assoc :journal-entry-line/debit (Math/abs (:invoice/total entity)))
|
|
(not credit-invoice?) (assoc :journal-entry-line/credit (Math/abs (:invoice/total entity))))]
|
|
(map-indexed (fn [i ea]
|
|
(cond->
|
|
{:db/id (str (:db/id entity) "-" (inc i))
|
|
:journal-entry-line/account (:db/id (:invoice-expense-account/account ea))
|
|
:journal-entry-line/location (or (:invoice-expense-account/location ea) "HQ")
|
|
}
|
|
credit-invoice? (assoc :journal-entry-line/credit (Math/abs (:invoice-expense-account/amount ea)))
|
|
(not credit-invoice?) (assoc :journal-entry-line/debit (Math/abs (:invoice-expense-account/amount ea)))))
|
|
(:invoice/expense-accounts entity)))
|
|
:journal-entry/cleared (and (< (:invoice/outstanding-balance entity) 0.01)
|
|
(every? #(= :payment-status/cleared (:payment/status %)) (:invoice/payments entity))
|
|
)})))))
|
|
|
|
(defmethod entity-change->ledger :transaction
|
|
[db [_ id]]
|
|
(when id
|
|
(let [entity (dc/pull db ['* {:transaction/vendor '[*]
|
|
:transaction/client '[*]
|
|
:transaction/approval-status '[*]
|
|
:transaction/bank-account '[* {:bank-account/type [:db/ident]}]
|
|
:transaction/accounts '[*
|
|
{:transaction-account/account [*]}] }] id)
|
|
decreasing? (< (or (:transaction/amount entity) 0.0) 0.0)
|
|
credit-from-bank? decreasing?
|
|
debit-from-bank? (not decreasing?)]
|
|
(when (and (not (= :transaction-approval-status/excluded (:db/ident (:transaction/approval-status entity))))
|
|
(not (= :transaction-approval-status/suppressed (:db/ident (:transaction/approval-status entity))))
|
|
(:transaction/amount entity)
|
|
(not (dollars-0? (:transaction/amount entity))))
|
|
(remove-nils
|
|
{:journal-entry/source "transaction"
|
|
:journal-entry/client (:db/id (:transaction/client entity))
|
|
:journal-entry/date (:transaction/date entity)
|
|
:journal-entry/original-entity (:db/id entity)
|
|
:journal-entry/alternate-description (:transaction/description-original entity)
|
|
:journal-entry/vendor (:db/id (:transaction/vendor entity))
|
|
:journal-entry/amount (Math/abs (:transaction/amount entity))
|
|
:journal-entry/cleared-against (:transaction/cleared-against entity)
|
|
|
|
:journal-entry/line-items (into [(remove-nils {:journal-entry-line/account (:db/id (:transaction/bank-account entity))
|
|
:db/id (str (:db/id entity) "-" 0)
|
|
:journal-entry-line/location "A"
|
|
:journal-entry-line/credit (when credit-from-bank?
|
|
(Math/abs (:transaction/amount entity)))
|
|
:journal-entry-line/debit (when debit-from-bank?
|
|
(Math/abs (:transaction/amount entity)))})
|
|
]
|
|
(map-indexed
|
|
(fn [i a]
|
|
(remove-nils{
|
|
:db/id (str (:db/id entity) "-" (inc i))
|
|
:journal-entry-line/account (:db/id (:transaction-account/account a))
|
|
:journal-entry-line/location (:transaction-account/location a)
|
|
:journal-entry-line/debit (when credit-from-bank?
|
|
(Math/abs (:transaction-account/amount a)))
|
|
:journal-entry-line/credit (when debit-from-bank?
|
|
(Math/abs (:transaction-account/amount a)))}))
|
|
(if (seq (:transaction/accounts entity))
|
|
(:transaction/accounts entity)
|
|
[{:transaction-account/amount (:transaction/amount entity)}])))
|
|
|
|
:journal-entry/cleared true})))))
|
|
|
|
#_(defmethod entity-change->ledger :expected-deposit
|
|
[db [type id]]
|
|
(let [{:expected-deposit/keys [total client date]} (d/pull db '[:expected-deposit/total :expected-deposit/client :expected-deposit/date] id)]
|
|
#:journal-entry
|
|
{:source "expected-deposit"
|
|
:original-entity id
|
|
|
|
:client client
|
|
:date date
|
|
:amount total
|
|
:vendor :vendor/ccp-square
|
|
:line-items [#:journal-entry-line
|
|
{:credit total
|
|
:location "A"
|
|
:account :account/receipts-split}
|
|
#:journal-entry-line
|
|
{:debit total
|
|
:location "A"
|
|
:account :account/ccp}]}))
|
|
|
|
(defmethod entity-change->ledger :invoice-expense-account
|
|
[_ _]
|
|
nil
|
|
)
|
|
|
|
(defmethod entity-change->ledger nil
|
|
[_ _]
|
|
nil)
|
|
|
|
(defn reconcile-ledger
|
|
([] (reconcile-ledger (-> (t/now)
|
|
(t/plus (t/months -6))
|
|
(c/to-date))))
|
|
([start-date]
|
|
(let [txes-missing-ledger-entries (->> (dc/q {:query {:find ['?t ]
|
|
:in ['$ '?sd]
|
|
:where [
|
|
'[?t :transaction/date ?d]
|
|
'[(>= ?d ?sd)]
|
|
'(not [_ :journal-entry/original-entity ?t])
|
|
'(not [?t :transaction/amount 0.0])
|
|
'(not [?t :transaction/approval-status :transaction-approval-status/excluded])
|
|
'(not [?t :transaction/approval-status :transaction-approval-status/suppressed])
|
|
]}
|
|
:args [(dc/db conn) start-date]})
|
|
(map first)
|
|
(mapv #(entity-change->ledger (dc/db conn) [:transaction %])))
|
|
|
|
|
|
invoices-missing-ledger-entries (->> (dc/q {:query {:find ['?t ]
|
|
:in ['$ '?sd]
|
|
:where ['[?t :invoice/date ?d]
|
|
'[(>= ?d ?sd)]
|
|
'(not [_ :journal-entry/original-entity ?t])
|
|
'[?t :invoice/total ?amt]
|
|
'[(not= 0.0 ?amt)]
|
|
'(not [?t :invoice/status :invoice-status/voided])
|
|
'(not [?t :invoice/import-status :import-status/pending])
|
|
'(not [?t :invoice/exclude-from-ledger true])
|
|
]}
|
|
:args [(dc/db conn) start-date]})
|
|
(map first)
|
|
(mapv #(entity-change->ledger (dc/db conn) [:invoice %])))
|
|
repairs (vec (concat txes-missing-ledger-entries invoices-missing-ledger-entries))]
|
|
(when (seq repairs)
|
|
(log/info (take 3 repairs))
|
|
(log/warn "repairing " (count txes-missing-ledger-entries) " missing transactions, " (count invoices-missing-ledger-entries) " missing invoices that were missing ledger entries")
|
|
(dc/transact conn {:tx-data repairs})))))
|
|
|
|
|
|
(defn touch-transaction [e]
|
|
(when-let [change (entity-change->ledger (dc/db conn)
|
|
[:transaction e])]
|
|
(dc/transact conn {:tx-data [{:db/id "datomic.tx"
|
|
:db/doc "touching transaction to update ledger"}
|
|
`(upsert-ledger ~change)]})))
|
|
|
|
(defn touch-invoice [e]
|
|
(when-let [change (entity-change->ledger (dc/db conn)
|
|
[:invoice e])]
|
|
(dc/transact conn [{:db/id "datomic.tx"
|
|
:db/doc "touching invoice to update ledger"}
|
|
`(upsert-ledger ~change)])))
|
|
|
|
(defn lazy-tx-range
|
|
([start end xf] (lazy-tx-range start end xf 0))
|
|
([start end xf o]
|
|
(let [next-results (dc/tx-range conn {:start start
|
|
:end end
|
|
:offset o
|
|
:limit 200})]
|
|
(lazy-seq
|
|
(if (seq next-results)
|
|
(concat (sequence (comp (mapcat :data)
|
|
xf) next-results) (lazy-tx-range start
|
|
end
|
|
xf
|
|
(+ (count next-results)
|
|
o)))
|
|
next-results)))))
|
|
|
|
|
|
|
|
(defn recently-changed-entities [start end]
|
|
(set (map (fn [d]
|
|
(:e d))
|
|
(mapcat :data (dc/tx-range conn {:start start
|
|
:end end})))))
|
|
|
|
(defn entities-since-last-ledger-entry []
|
|
(count (dc/tx-range conn {:start (c/to-date (t/plus (t/now) (t/days -5)))
|
|
:end (c/to-date (t/now))})))
|
|
|
|
(defn mismatched-transactions
|
|
([]
|
|
(mismatched-transactions (c/to-date (t/minus (t/now) (t/days 7)))
|
|
(c/to-date (t/minus (t/now) (t/hours 1)))) )
|
|
([changed-between-start changed-between-end]
|
|
(let [entities-to-consider (recently-changed-entities
|
|
changed-between-start
|
|
changed-between-end)
|
|
_ (log/info "checking" (count entities-to-consider) "transactions looking for mismatches between" changed-between-start changed-between-end)
|
|
jel-accounts (reduce
|
|
(fn [acc [e lia]]
|
|
(update acc e (fnil conj #{} ) lia))
|
|
{}
|
|
(dc/q '[:find ?e ?lia
|
|
:in $ [?e ...]
|
|
:where
|
|
[?je :journal-entry/original-entity ?e]
|
|
[?e :transaction/date]
|
|
[?je :journal-entry/line-items ?li]
|
|
[?li :journal-entry-line/account ?lia]
|
|
[?lia :account/name]]
|
|
(dc/db conn)
|
|
entities-to-consider))
|
|
transaction-accounts (reduce
|
|
(fn [acc [e lia]]
|
|
(update acc e (fnil conj #{} ) lia))
|
|
{}
|
|
(dc/q '[:find ?e ?lia
|
|
:in $ [?e ...]
|
|
:where
|
|
[?e :transaction/date ?d]
|
|
[?e :transaction/accounts ?li]
|
|
(not [?e :transaction/approval-status :transaction-approval-status/excluded])
|
|
(not [?e :transaction/approval-status :transaction-approval-status/suppressed])
|
|
[?li :transaction-account/account ?lia]
|
|
[?lia :account/name]
|
|
[?e :transaction/amount ?amt]
|
|
[(not= ?amt 0.0)]]
|
|
(dc/db conn)
|
|
entities-to-consider))]
|
|
(->> transaction-accounts
|
|
(filter
|
|
(fn [[e accounts]] (not= accounts (get jel-accounts e))))))))
|
|
|
|
(defn unbalanced-transactions
|
|
([] (unbalanced-transactions (c/to-date (t/minus (t/now) (t/days 7)))
|
|
(c/to-date (t/minus (t/now) (t/hours 1)))))
|
|
([changed-between-start changed-between-end]
|
|
(let [entities-to-consider (recently-changed-entities changed-between-start changed-between-end)]
|
|
(log/info "checking" (count entities-to-consider) "transaction journal entries looking for mismatches between" changed-between-start changed-between-end)
|
|
(->> (dc/q '[:find ?je ?a (sum ?debit) (sum ?credit)
|
|
:with ?jel
|
|
:in $ [?je ...]
|
|
:where [?je :journal-entry/amount ?a]
|
|
[?je :journal-entry/line-items ?jel]
|
|
[(get-else $ ?jel :journal-entry-line/debit 0.0) ?debit]
|
|
[(get-else $ ?jel :journal-entry-line/credit 0.0) ?credit]
|
|
]
|
|
(dc/db conn)
|
|
entities-to-consider)
|
|
(filter (fn [[_ a d c]]
|
|
(or (not (dollars= a d))
|
|
(not (dollars= a c)))))
|
|
(map first)
|
|
(map (fn [je]
|
|
(pull-ref (dc/db conn) :journal-entry/original-entity je)))))))
|
|
|
|
|
|
|
|
(defn unbalanced-invoices
|
|
([] (unbalanced-invoices (c/to-date (t/minus (t/now) (t/days 7)))
|
|
(c/to-date (t/minus (t/now) (t/hours 1)))))
|
|
([changed-between-start changed-between-end]
|
|
(let [entities-to-consider (recently-changed-entities
|
|
changed-between-start
|
|
changed-between-end)]
|
|
(log/info "checking" (count entities-to-consider) "invoice journal entries looking for mismatches between" changed-between-start changed-between-end)
|
|
(->> (dc/q '[:find ?je ?a (sum ?debit) (sum ?credit)
|
|
:with ?jel
|
|
:in $ [?je ...]
|
|
:where [?je :journal-entry/amount ?a]
|
|
[?je :journal-entry/original-entity ?i]
|
|
[?i :invoice/date]
|
|
[?je :journal-entry/line-items ?jel]
|
|
[(get-else $ ?jel :journal-entry-line/debit 0.0) ?debit]
|
|
[(get-else $ ?jel :journal-entry-line/credit 0.0) ?credit]]
|
|
(dc/db conn)
|
|
entities-to-consider)
|
|
(filter (fn [[_ a d c]]
|
|
(or (not (dollars= a d))
|
|
(not (dollars= a c)))))
|
|
(map first)
|
|
(map (fn [je]
|
|
(pull-ref (dc/db conn) :journal-entry/original-entity je)))))))
|
|
|
|
(defn mismatched-invoices
|
|
([]
|
|
(mismatched-invoices (c/to-date (t/minus (t/now) (t/days 7)))
|
|
(c/to-date (t/minus (t/now) (t/hours 1)))) )
|
|
|
|
([changed-between-start changed-between-end]
|
|
(let [entities-to-consider (recently-changed-entities changed-between-start changed-between-end)
|
|
_ (log/info (count entities-to-consider) "invoices have changed between" changed-between-start "and" changed-between-end)
|
|
jel-accounts (reduce
|
|
(fn [acc [e lia]]
|
|
(update acc e (fnil conj #{} ) lia))
|
|
{}
|
|
(dc/q '[:find ?e ?lia
|
|
:in $ [?e ...]
|
|
:where
|
|
[?je :journal-entry/original-entity ?e]
|
|
[?e :invoice/date]
|
|
[?je :journal-entry/line-items ?li]
|
|
[?li :journal-entry-line/account ?lia]
|
|
(not [?lia :account/numeric-code 21000])
|
|
[?lia :account/name]]
|
|
(dc/db conn)
|
|
entities-to-consider))
|
|
invoice-accounts (reduce
|
|
(fn [acc [e lia]]
|
|
(update acc e (fnil conj #{} ) lia))
|
|
{}
|
|
(dc/q '[:find ?e ?lia
|
|
:in $ [?e ...]
|
|
:where
|
|
[?e :invoice/expense-accounts ?li]
|
|
(not [?e :invoice/total 0.0])
|
|
[?li :invoice-expense-account/account ?lia]
|
|
[?lia :account/name]
|
|
(not [?lia :account/numeric-code 21000])
|
|
(not [?e :invoice/status :invoice-status/voided])
|
|
(not [?e :invoice/exclude-from-ledger true])
|
|
[?e :invoice/import-status :import-status/imported]]
|
|
(dc/db conn)
|
|
entities-to-consider))
|
|
]
|
|
(filter
|
|
(fn [[e accounts]]
|
|
(not= accounts (get jel-accounts e)))
|
|
invoice-accounts))))
|
|
|
|
(defn touch-broken-ledger []
|
|
(statsd/event {:title "Reconciling Ledger"
|
|
:text "This process looks for unbalance ledger entries, or missing ledger entries"
|
|
:priority :low}
|
|
nil)
|
|
(log/info "Attempting to fix transactions that are in the ledger but are wrong")
|
|
(let [mismatched-ts (mismatched-transactions)]
|
|
(if (seq mismatched-ts)
|
|
(do
|
|
(log/warn (count mismatched-ts) " transactions exist but don't match ledger " (pr-str (take 10 mismatched-ts) ))
|
|
(doseq [[m] mismatched-ts]
|
|
(touch-transaction m))
|
|
(statsd/gauge "data.mismatched_transactions" (count (mismatched-transactions))))
|
|
(statsd/gauge "data.mismatched_transactions" 0.0)))
|
|
(log/info "Attempting to fix transactions that are in the ledger but debits/credits don't add up")
|
|
(let [unbalanced-ts (unbalanced-transactions)]
|
|
(if (seq unbalanced-ts)
|
|
(do
|
|
(log/warn (count unbalanced-ts) " transactions exist but don't have matching debits/credits (" (pr-str (take 10 unbalanced-ts) ) ")")
|
|
(doseq [m unbalanced-ts]
|
|
(touch-transaction m))
|
|
(statsd/gauge "data.unbalanced_transactions" (count (unbalanced-transactions))))
|
|
(statsd/gauge "data.unbalanced_transactions" 0.0)))
|
|
(log/info "Finished fixing transactions that are in the ledger but are wrong")
|
|
(let [mismatched-is (mismatched-invoices)]
|
|
(if (seq mismatched-is)
|
|
(do
|
|
(log/warn (count mismatched-is) " invoice exist but don't match ledger ")
|
|
(doseq [[m] mismatched-is]
|
|
(touch-invoice m))
|
|
(statsd/gauge "data.mismatched_invoices" (count (mismatched-invoices))))
|
|
(statsd/gauge "data.mismatched_invoices" 0.0)))
|
|
(log/info "Attempting to fix transactions that are in the ledger but debits/credits don't add up")
|
|
(let [unbalanced-is (unbalanced-invoices)]
|
|
(if (seq unbalanced-is)
|
|
(do
|
|
(log/warn (count unbalanced-is) " invoices exist but don't have matching debits/credits ")
|
|
(doseq [m unbalanced-is]
|
|
(touch-invoice m))
|
|
(statsd/gauge "data.unbalanced_invoices" (count (unbalanced-invoices))))
|
|
(statsd/gauge "data.unbalanced_invoices" 0.0)))
|
|
|
|
(log/info "Finish fixing invoices that are in the ledger but are wrong")
|
|
(statsd/event {:title "Finished Reconciling Ledger"
|
|
:text "This process looks for unbalance ledger entries, or missing ledger entries"
|
|
:priority :low}
|
|
nil))
|
|
|
|
(defn transact-with-ledger [transaction id]
|
|
(let [db (dc/db conn)
|
|
tx (audit-transact transaction id)
|
|
affected-entities (->> (:tx-data tx)
|
|
(map (fn [x]
|
|
{:e (:e x)
|
|
:a (:a x)
|
|
:v (:v x)
|
|
:added (:added x)}))
|
|
(group-by :e)
|
|
(mapcat #(datums->impacted-entity db %))
|
|
(set))
|
|
ledger-txs (->> affected-entities
|
|
(map #(entity-change->ledger (:db-after tx) %))
|
|
(filter seq)
|
|
(map (fn [l]
|
|
`(upsert-ledger ~l))))]
|
|
(when (seq ledger-txs)
|
|
(audit-transact ledger-txs id))
|
|
tx))
|
|
|
|
(defn transact-batch-with-ledger [txes id]
|
|
(let [batch-id (.toString (java.util.UUID/randomUUID))]
|
|
(reduce
|
|
(fn [full-tx batch]
|
|
(let [batch (conj (vec batch) {:db/id "datomic.tx"
|
|
:audit/batch batch-id})
|
|
_ (log/info "transacting batch " batch-id " " (count batch))
|
|
tx-result (transact-with-ledger batch id)]
|
|
|
|
(cond-> full-tx
|
|
(:tx-data full-tx) (update :tx-data #(into % (:tx-data tx-result)))
|
|
(not (:tx-data full-tx)) (assoc :tx-data (vec (:tx-data tx-result)))
|
|
(not (:db-before full-tx)) (assoc :db-before (:db-before tx-result))
|
|
true (assoc :db-after (:db-after tx-result))
|
|
true (update :tempids merge (:tempids tx-result)))))
|
|
|
|
{}
|
|
(partition-all 50 txes))))
|
|
|
|
(defn build-account-lookup [client-id]
|
|
(let [accounts (by :db/id (map first (dc/q {:query {:find ['(pull ?e [:db/id :account/name
|
|
:account/numeric-code
|
|
{:account/type [:db/ident]
|
|
:account/client-overrides [:account-client-override/client :account-client-override/name]}
|
|
])]
|
|
:in ['$]
|
|
:where ['[?e :account/name]]}
|
|
:args [(dc/db conn )]})))
|
|
|
|
bank-accounts (by :db/id (map first (dc/q {:query {:find ['(pull ?e [:db/id :bank-account/name :bank-account/numeric-code {:bank-account/type [:db/ident]}])]
|
|
:in ['$]
|
|
:where ['[?e :bank-account/name]]}
|
|
:args [(dc/db conn)]})))
|
|
overrides-by-client (->> accounts
|
|
vals
|
|
(mapcat (fn [a]
|
|
(map (fn [o]
|
|
[[(:db/id a) (:db/id (:account-client-override/client o))]
|
|
(:account-client-override/name o)])
|
|
(:account/client-overrides a))
|
|
) )
|
|
(into {} ))]
|
|
(fn [a]
|
|
{:name (or (:bank-account/name (bank-accounts a))
|
|
(overrides-by-client [a client-id])
|
|
(:account/name (accounts a)))
|
|
:account_type (or (:db/ident (:account/type (accounts a)))
|
|
({:bank-account-type/check :account-type/asset
|
|
:bank-account-type/cash :account-type/asset
|
|
:bank-account-type/credit :account-type/liability}
|
|
(:db/ident (:bank-account/type (bank-accounts a)))))
|
|
:numeric_code (or (:account/numeric-code (accounts a))
|
|
(:bank-account/numeric-code (bank-accounts a)))
|
|
:client_id client-id})))
|
|
|
|
(defn reset-client+account+location+date
|
|
([] (reset-client+account+location+date (map first (dc/q '[:find ?c :where [?c :client/code]] (dc/db conn)))))
|
|
([clients]
|
|
(doseq [client clients
|
|
:let [_ (mu/log ::reseting-index-for :client client)]
|
|
batch
|
|
(->> (dc/qseq '[:find (pull ?je [:journal-entry/date :journal-entry/client {:journal-entry/line-items [:journal-entry-line/account :journal-entry-line/location :db/id]}])
|
|
:in $ ?c
|
|
:where [?je :journal-entry/client ?c]]
|
|
(dc/db conn)
|
|
client
|
|
|
|
)
|
|
(map first)
|
|
(mapcat (fn [je]
|
|
(map (fn [jel]
|
|
{:db/id (:db/id jel)
|
|
:journal-entry-line/client+account+location+date
|
|
[(-> je :journal-entry/client :db/id)
|
|
(-> jel :journal-entry-line/account :db/id)
|
|
|
|
(-> jel :journal-entry-line/location)
|
|
|
|
(-> je :journal-entry/date)]})
|
|
(:journal-entry/line-items je))))
|
|
(partition-all 500)
|
|
)]
|
|
(mu/log ::batch-completed)
|
|
(dc/transact conn {:tx-data batch}))))
|
|
|
|
(defn find-mismatch-index []
|
|
(reduce + 0
|
|
(for [c (map first (dc/q '[:find ?c :where [?c :client/code]] (dc/db conn)))
|
|
:let [_ (println "searching for" c)
|
|
a (->> (dc/index-pull (dc/db conn)
|
|
{:index :avet
|
|
:selector [:db/id :journal-entry-line/location :journal-entry-line/account :journal-entry-line/client+account+location+date {:journal-entry/_line-items [:journal-entry/date :journal-entry/client]}]
|
|
:start [:journal-entry-line/client+account+location+date [c]]})
|
|
(take-while (fn [result]
|
|
(= c (first (:journal-entry-line/client+account+location+date result)))
|
|
))
|
|
(filter (fn [{index :journal-entry-line/client+account+location+date :as result}]
|
|
(not= index
|
|
[(-> result :journal-entry/_line-items :journal-entry/client :db/id)
|
|
(-> result :journal-entry-line/account :db/id)
|
|
(-> result :journal-entry-line/location)
|
|
(-> result :journal-entry/_line-items :journal-entry/date)]))))]]
|
|
(do (println (count a))
|
|
(count a)))))
|
|
|
|
|
|
|
|
(defn accounts-needing-rebuild [ db client]
|
|
(let [client (pull-id db client)]
|
|
(->> (dc/qseq '[:find ?c ?a ?l (min ?d)
|
|
:in $ ?c
|
|
:where
|
|
[?jel :journal-entry-line/dirty true]
|
|
[?jel :journal-entry-line/account ?a]
|
|
[?jel :journal-entry-line/location ?l]
|
|
[?je :journal-entry/line-items ?jel]
|
|
[?je :journal-entry/client ?c]
|
|
[?je :journal-entry/date ?d]]
|
|
db
|
|
client)
|
|
(map (fn [[client account location starting-at ]]
|
|
{:client client
|
|
:account account
|
|
:starting-at starting-at
|
|
:location location})))))
|
|
|
|
(defn find-running-balance-start [{:keys [client account location starting-at]} db ]
|
|
(let [client (pull-id db client)
|
|
account (pull-id db account)]
|
|
(or
|
|
(->> (dc/index-pull db
|
|
{:index :avet
|
|
:selector [:db/id :journal-entry-line/running-balance :journal-entry-line/client+account+location+date]
|
|
:start [:journal-entry-line/client+account+location+date [client account location starting-at]]
|
|
:reverse true
|
|
:limit 500})
|
|
(take-while (fn [result]
|
|
(= [client
|
|
account
|
|
location]
|
|
(take 3 (:journal-entry-line/client+account+location+date result)))))
|
|
(take 5)
|
|
(drop-while (fn [{[_ _ _ date] :journal-entry-line/client+account+location+date}]
|
|
(>= (compare date starting-at) 0)))
|
|
first
|
|
:journal-entry-line/running-balance)
|
|
0.0)))
|
|
|
|
(defn get-dirty-entries [{:keys [client account location starting-at]} db ]
|
|
(let [client (pull-id db client)
|
|
account (pull-id db account)]
|
|
(into []
|
|
(comp
|
|
(map (fn [i]
|
|
(dc/index-pull db
|
|
{:index :avet
|
|
:selector [:db/id :journal-entry-line/debit :journal-entry-line/credit :journal-entry-line/client+account+location+date]
|
|
:start [:journal-entry-line/client+account+location+date
|
|
[client account location starting-at]]
|
|
:offset (* i 1000)
|
|
:limit 1000}
|
|
)))
|
|
(take-while identity)
|
|
(mapcat identity)
|
|
(take-while (fn [{[result-client result-account result-location] :journal-entry-line/client+account+location+date}]
|
|
#_(println
|
|
[[ client result-client]
|
|
[ account result-account]
|
|
[ location result-location]])
|
|
(and
|
|
(= client result-client)
|
|
(= account result-account)
|
|
(= location result-location))))
|
|
(map (fn [result]
|
|
[(:db/id result) (:journal-entry-line/debit result 0.0) (:journal-entry-line/credit result 0.0) ])))
|
|
(range 100))))
|
|
|
|
(defn compute-running-balance [account-needing-refresh]
|
|
(mu/trace ::compute
|
|
[:dirty-count (count (:dirty-entries account-needing-refresh))]
|
|
(second
|
|
(reduce
|
|
(fn [[running-balance rows] [id debit credit] ]
|
|
(let [new-running-balance (+ running-balance
|
|
(if (#{:account-type/asset
|
|
:account-type/dividend
|
|
:account-type/expense} (:account-type account-needing-refresh))
|
|
(- debit credit)
|
|
(- credit debit)))]
|
|
[new-running-balance
|
|
(conj rows
|
|
{:db/id id
|
|
:journal-entry-line/running-balance new-running-balance
|
|
:journal-entry-line/dirty false})]))
|
|
|
|
[(:build-from account-needing-refresh) []]
|
|
(:dirty-entries account-needing-refresh)))))
|
|
|
|
|
|
(defn refresh-running-balance-cache
|
|
([] (refresh-running-balance-cache (shuffle (map first
|
|
(dc/q '[:find (pull ?c [:client/code :db/id])
|
|
:where [?c :client/code]]
|
|
(dc/db conn))))))
|
|
([clients]
|
|
(doseq [c clients]
|
|
(mu/trace ::building-running-balance
|
|
[:client c]
|
|
(mu/with-context {:client c}
|
|
(let [db (dc/db conn)
|
|
accounts-needing-rebuild (accounts-needing-rebuild db (:db/id c))]
|
|
(when (seq accounts-needing-rebuild)
|
|
(mu/log ::found-accounts-needing-rebuild
|
|
:accounts (count accounts-needing-rebuild))
|
|
(audit-transact-batch
|
|
(->> accounts-needing-rebuild
|
|
(mapcat (fn [account-needing-rebuild]
|
|
(mu/with-context {:account account-needing-rebuild}
|
|
(-> account-needing-rebuild
|
|
(assoc :build-from (find-running-balance-start account-needing-rebuild db))
|
|
(assoc :dirty-entries (get-dirty-entries account-needing-rebuild db))
|
|
(assoc :account-type (:account_type ((build-account-lookup (:client account-needing-rebuild)) (:account account-needing-rebuild))))
|
|
(compute-running-balance))))))
|
|
{:user/name "running-balance-cache"}))))))))
|
|
|
|
|
|
;; TODO only enable once IOL is set up in clod
|
|
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
|
|
#_(mount/defstate running-balance-cache-worker
|
|
:start (scheduler/every (* 15 60 (+ 500 (rand-int 500))) (heartbeat refresh-running-balance-cache "running-balance-cache"))
|
|
:stop (scheduler/stop running-balance-cache-worker))
|