(ns auto-ap.ledger (:require [datomic.api :as d] [yang.scheduler :as scheduler] [mount.core :as mount] [auto-ap.datomic.accounts :as a] [auto-ap.datomic :refer [uri remove-nils conn]] [clojure.spec.alpha :as s] [clojure.tools.logging :as log] [auto-ap.logging :refer [info-event]] [unilog.context :as lc])) (defn datums->impacted-entity [db [e changes]] (let [entity (d/pull db '[* {:invoice/_expense-accounts [:db/id] :transaction/_accounts [:db/id]}] e) namespaces (->> changes (map :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]] :else nil))) (defn infer-entity [_ [_ changes]] (let [namespaces (->> changes (map :a) (map namespace) set)] (cond (namespaces "invoice" ) :invoice (namespaces "invoice-expense-account" ) :invoice-expense-account (namespaces "transaction-account" ) :transaction-account :else nil))) (defmulti entity-change->ledger (fn [_ [type]] type)) (defmethod entity-change->ledger :invoice [db [type id]] (let [entity (d/pull db ['* {:invoice/vendor '[*] :invoice/payment '[*] :invoice/status '[:db/ident] :invoice/import-status '[:db/ident]}] id)] (when-not (or (= true (:invoice/exclude-from-ledger entity)) (= :import-status/pending (:db/ident (:invoice/import-status entity))) (= :invoice-status/voided (:db/ident (:invoice/status 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 (:invoice/total entity) :journal-entry/line-items (into [{:journal-entry-line/account (:db/id (a/get-account-by-numeric-code-and-sets 21000 ["default"])) :journal-entry-line/location "A" :journal-entry-line/credit (:invoice/total entity)}] (map (fn [ea] (remove-nils {:journal-entry-line/account (:db/id (:invoice-expense-account/account ea)) :journal-entry-line/location (or (:invoice-expense-account/location ea) "HQ") ;; TODO? :journal-entry-line/debit (: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 [type id]] (let [entity (d/pull db ['* {:transaction/vendor '[*] :transaction/client '[*] :transaction/approval-status '[*] :transaction/bank-account '[* {:bank-account/type [:db/ident]}] :transaction/accounts '[* {:transaction-account/account [*]}] }] id) bank-account-type (-> entity :transaction/bank-account :bank-account/type :db/ident) decreasing? (< (:transaction/amount entity) 0.0) credit-from-bank? decreasing? debit-from-bank? (not decreasing?)] (when-not (= :transaction-approval-status/excluded (:db/ident (:transaction/approval-status 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)) :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 (fn [a] (remove-nils{: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 :invoice-expense-account [db [entity changes]] nil ) (defmethod entity-change->ledger nil [db [entity changes]] nil) (defn ledger-entries->transaction [entries] (into [[:replace-general-ledger (:journal-entry/original-entity (first entries))]] entries)) (mount/defstate tx-report-queue :start (d/tx-report-queue conn) :stop (d/remove-tx-report-queue conn)) (defn process-one [] (lc/with-context {:source "process-txes"} (try (let [transaction (.take tx-report-queue) _ (log/info "Converting tranasction to ledger") db (:db-after transaction) affected-entities (->> (:tx-data transaction) (map (fn [^datomic.db.Datum x] {:e (:e x) :a (d/ident db (:a x)) :v (:v x) :added (:added x)})) (group-by :e) (mapcat #(datums->impacted-entity db %)) (set)) _ (info-event (str "Found " (count affected-entities) " affected entities") {:affected-entities (count affected-entities)}) d-txs (->> affected-entities (map #(entity-change->ledger db %)) (filter seq)) retractions (map (fn [[_ e]] [:db/retractEntity [:journal-entry/original-entity e]]) affected-entities)] (when (seq retractions) @(d/transact conn retractions)) (doseq [d-tx d-txs] @(d/transact conn [d-tx])) (log/info "Succesfully process transaction")) (catch Exception e (log/error e))))) (mount/defstate process-txes-worker :start (scheduler/run-fun process-one 1) :stop (-> process-txes-worker :running? (reset! false))) (defn reconcile-ledger [] (lc/with-context {:source "reconcile-ledger"} (try (log/info "Attempting to reconcile the ledger") (let [txes-missing-ledger-entries (->> (d/query {:query {:find ['?t ] :in ['$] :where ['[?t :transaction/date] '(not [?t :transaction/approval-status :transaction-approval-status/excluded]) '(not-join [?t] [?e :journal-entry/original-entity ?t])]} :args [(d/db conn)]}) (map first) (mapv #(entity-change->ledger (d/db conn) [:transaction %]))) invoices-missing-ledger-entries (->> (d/query {:query {:find ['?t ] :in ['$] :where ['[?t :invoice/date] '(not [?t :invoice/status :invoice-status/voided]) '(not [?t :invoice/import-status :import-status/pending]) '(not [?t :invoice/exclude-from-ledger true]) '(not-join [?t] [?e :journal-entry/original-entity ?t])]} :args [(d/db conn)]}) (map first) (mapv #(entity-change->ledger (d/db conn) [:invoice %]))) repairs (vec (concat txes-missing-ledger-entries invoices-missing-ledger-entries))] (when (seq repairs) (log/warn "repairing " (count txes-missing-ledger-entries) " missing transactions, " (count invoices-missing-ledger-entries) " missing invoices that were missing ledger entries") @(d/transact conn repairs)) (log/info "Finished reconciling ledger")) (catch Exception e (log/error e))))) (mount/defstate reconciliation-frequency :start 60000) (mount/defstate ledger-reconciliation-worker :start (scheduler/every reconciliation-frequency reconcile-ledger) :stop (scheduler/stop ledger-reconciliation-worker))