notid/integreat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

much more background process tracking

Browse Source
This commit is contained in:
Bryce Covert
2022-06-22 10:43:37 -07:00
parent 480066b202
commit 1db8d7a52c
15 changed files with 379 additions and 486 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/clj/auto_ap/background/invoices.clj
View File

@@ -1,18 +1,15 @@
(ns auto-ap.background.invoices
(:require [auto-ap.datomic.invoices :as d-invoices]
[auto-ap.datomic :refer [uri conn]]
[datomic.api :as d]
(:require
[auto-ap.datomic :refer [conn]]
[auto-ap.time :as time]
[auto-ap.utils :refer [heartbeat]]
[clj-time.coerce :as coerce]
[clojure.tools.logging :as log]
[datomic.api :as d]
[mount.core :as mount]
[yang.scheduler :as scheduler]
[unilog.context :as lc]
[clojure.tools.logging :as log]))
[yang.scheduler :as scheduler]))
(defn close-auto-invoices []
(lc/with-context {:source "close-auto-invoices"}
(try
(let [invoices-to-close (d/query {:query {:find ['?e]
:in ['$ '?today]
:where ['[?e :invoice/scheduled-payment ?d]
@@ -28,10 +25,8 @@
:invoice/status :invoice-status/paid}))
(d/transact conn)
deref)
(log/info "Closed " (count invoices-to-close) "scheduled invoices"))
(catch Exception e
(log/error e)))))
(log/info "Closed " (count invoices-to-close) "scheduled invoices")))
(mount/defstate close-auto-invoices-worker
:start (scheduler/every 60000 close-auto-invoices)
:start (scheduler/every 60000 (heartbeat close-auto-invoices "close-auto-invoices"))
:stop (scheduler/stop close-auto-invoices-worker))

23
src/clj/auto_ap/background/mail.clj
View File

@@ -1,23 +1,25 @@
(ns auto-ap.background.mail
(:require [amazonica.aws.s3 :as s3]
[amazonica.aws.sqs :as sqs]
(:require
[amazonica.aws.s3 :as s3]
[amazonica.aws.simpleemail :as ses]
[amazonica.aws.sqs :as sqs]
[auto-ap.parse :as parse]
[auto-ap.routes.invoices :as invoices]
[auto-ap.utils :refer [heartbeat]]
[clojure-mail.message :as message]
[clojure.data.json :as json]
[clojure.java.io :as io]
[clojure.string :as str]
[config.core :refer [env]]
[clojure.tools.logging :as log]
[unilog.context :as lc]
[config.core :refer [env]]
[mount.core :as mount]
[unilog.context :as lc]
[yang.scheduler :as scheduler])
(:import (java.util Properties UUID)
(:import
(java.util Properties UUID)
(javax.mail Session)
(javax.mail.internet MimeMessage)))
(defn send-email-about-failed-message [mail-bucket mail-key]
(let [target-key (str "failed-emails/" mail-key ".eml")
target-url (str "http://" (:data-bucket env)
@@ -33,9 +35,6 @@
(defn process-sqs []
(lc/with-context {:source "import-uploaded-invoices"}
(try
(log/info "Fetching messages from sqs...")
(doseq [message (:messages (sqs/receive-message {:queue-url (:invoice-import-queue-url env)
:wait-time-seconds 5
@@ -77,10 +76,8 @@
(send-email-about-failed-message (-> r :s3 :bucket :name) (-> r :s3 :object :key)))
(finally
(io/delete-file filename)))))))
(sqs/delete-message (assoc message :queue-url (:invoice-import-queue-url env) )))
(catch Exception e
(log/error e)))))
(sqs/delete-message (assoc message :queue-url (:invoice-import-queue-url env) ))))
(mount/defstate import-invoices
:start (scheduler/every (* 60 5000) process-sqs)
:start (scheduler/every (* 60 5000) (heartbeat process-sqs "import-uploaded-invoices"))
:stop (scheduler/stop import-invoices))

32
src/clj/auto_ap/background/requests.clj
View File

@@ -1,23 +1,19 @@
(ns auto-ap.background.requests
(:require
[amazonica.aws.sqs :as sqs]
[config.core :refer [env]]
[mount.core :as mount]
[yang.scheduler :as scheduler]
[auto-ap.yodlee.core2 :as client2]
[clojure.tools.logging :as log]
[auto-ap.import.intuit :as i]
[auto-ap.import.plaid :as p]
[unilog.context :as lc]
[auto-ap.import.yodlee :as y]
[auto-ap.import.yodlee2 :as y2]
))
[auto-ap.utils :refer [heartbeat]]
[clojure.tools.logging :as log]
[config.core :refer [env]]
[mount.core :as mount]
[yang.scheduler :as scheduler]))
(def queue-url (:requests-queue-url env))
(defn process-1 []
(lc/with-context {:source "Request poller"}
(log/info "Checking SQS...")
(let [[{:keys [message-id receipt-handle body]}] (:messages (sqs/receive-message {:queue-url queue-url
:wait-time-seconds 1
:count 1}))]
@@ -28,35 +24,21 @@
(log/infof "processing message %s with body %s" message-id body )
(cond
(= ":intuit" body)
(try
(i/import-intuit)
(catch Exception e
(log/error e)))
(= ":yodlee" body)
(try
(y/import-yodlee)
(catch Exception e
(log/error e)))
(= ":yodlee2" body)
(try
(client2/upsert-accounts)
(y2/import-yodlee2)
(catch Exception e
(log/error e)))
(= ":plaid" body)
(try
(p/import-plaid)
(catch Exception e
(log/error e))))
))))
(p/import-plaid)))))
(defn fake-message []
(sqs/send-message {:queue-url (:requests-queue-url env)
:message-body ":intuit"} ))
(mount/defstate request-listener
:start (scheduler/every (* 1000 30) process-1)
:start (scheduler/every (* 1000 30) (heartbeat process-1 "request-poller"))
:stop (scheduler/stop request-listener))

8
src/clj/auto_ap/background/sysco.clj
View File

@@ -6,6 +6,7 @@
[auto-ap.datomic.invoices :refer [code-invoice]]
[auto-ap.parse :as parse]
[auto-ap.time :as t]
[auto-ap.utils :refer [heartbeat]]
[clj-time.coerce :as coerce]
[clojure.data.csv :as csv]
[clojure.java.io :as io]
@@ -15,8 +16,6 @@
[config.core :refer [env]]
[datomic.api :as d]
[mount.core :as mount]
[unilog.context :as lc]
#_{:clj-kondo/ignore [:unused-namespace]}
[yang.scheduler :as scheduler])
(:import
(java.util UUID)))
@@ -111,7 +110,6 @@
(defn import-sysco []
(lc/with-context {:source "sysco-importer"}
(let [sysco-vendor (get-sysco-vendor)
clients (d-clients/get-all)
keys (->> (s3/list-objects-v2 {:bucket-name bucket-name
@@ -156,9 +154,9 @@
(mark-key k))
(statsd/event {:title "Sysco import ended"
:text "Sysco completed"
:priority :low} nil))))
:priority :low} nil)))
(mount/defstate sysco-invoice-importer
:start (scheduler/every (* 1000 60 60) import-sysco)
:start (scheduler/every (* 1000 60 60) (heartbeat import-sysco "sysco-importer"))
:stop (scheduler/stop sysco-invoice-importer))

21
src/clj/auto_ap/background/vendor.clj
View File

@@ -1,18 +1,12 @@
(ns auto-ap.background.vendor
(:require [auto-ap.datomic.invoices :as d-invoices]
[auto-ap.datomic :refer [uri conn]]
(:require
[auto-ap.datomic :refer [conn]]
[auto-ap.utils :refer [heartbeat]]
[datomic.api :as d]
[auto-ap.time :as time]
[clj-time.coerce :as coerce]
[mount.core :as mount]
[yang.scheduler :as scheduler]
[unilog.context :as lc]
[clojure.tools.logging :as log]))
[yang.scheduler :as scheduler]))
(defn refresh-vendor-usages []
(lc/with-context {:source "refreshing vendor-usages"}
(try
(->> {:query {:find ['?v '?c '(count ?e)]
:in ['$]
:where ['[?v :vendor/name]
@@ -34,11 +28,8 @@
:key (str v "-" c)
:count cnt}))
(d/transact conn)
deref)
(log/info "updated vendor usage")
(catch Exception e
(log/error e)))))
deref))
(mount/defstate refresh-vendor-usages-worker
:start (scheduler/every (* 60 60 1000) refresh-vendor-usages)
:start (scheduler/every (* 60 60 1000) (heartbeat refresh-vendor-usages "vendor-usages"))
:stop (scheduler/stop refresh-vendor-usages))

41
src/clj/auto_ap/graphql/clients.clj
View File

@@ -1,24 +1,26 @@
(ns auto-ap.graphql.clients
(:require [auto-ap.datomic :refer [audit-transact conn remove-nils]]
(:require
[amazonica.aws.s3 :as s3]
[auto-ap.datomic :refer [audit-transact conn remove-nils]]
[auto-ap.datomic.clients :as d-clients]
[auto-ap.graphql.utils :refer [->graphql assert-admin can-see-client? is-admin?]]
[auto-ap.utils :refer [by]]
[auto-ap.yodlee.core :refer [in-memory-cache]]
[auto-ap.graphql.utils
:refer [->graphql assert-admin can-see-client? is-admin?]]
[auto-ap.routes.queries :as q]
[auto-ap.square.core :as square]
[com.walmartlabs.lacinia.util :refer [attach-resolvers]]
[auto-ap.utils :refer [by heartbeat]]
[auto-ap.yodlee.core :refer [in-memory-cache]]
[clj-time.coerce :as coerce]
[clojure.string :as str]
[unilog.context :as lc]
[clojure.tools.logging :as log]
[datomic.api :as d]
[clojure.java.io :as io]
[amazonica.aws.s3 :as s3]
#_{:clj-kondo/ignore [:unused-namespace]}
[yang.scheduler :as scheduler]
[mount.core :as mount])
(:import [org.apache.commons.codec.binary Base64]
java.util.UUID))
[clojure.string :as str]
[clojure.tools.logging :as log]
[com.walmartlabs.lacinia.util :refer [attach-resolvers]]
[datomic.api :as d]
[mount.core :as mount]
[unilog.context :as lc]
[yang.scheduler :as scheduler])
(:import
(java.util UUID)
(org.apache.commons.codec.binary Base64)))
(defn assert-client-code-is-unique [code]
(when (seq (d/query {:query {:find '[?id]
@@ -245,15 +247,10 @@
(map first)))))
(defn refresh-current-balance []
(lc/with-context {:source "current-balance-cache"}
(try
(log/info "Refreshing running balance cache")
(build-current-balance (bank-accounts-needing-refresh))
(catch Exception e
(log/error e)))))
(build-current-balance (bank-accounts-needing-refresh)))
(mount/defstate current-balance-worker
:start (scheduler/every (* 17 60 1000) refresh-current-balance)
:start (scheduler/every (* 17 60 1000) (heartbeat refresh-current-balance "current-balance-cache"))
:stop (scheduler/stop current-balance-worker))
(defn get-client [context _ _]

17
src/clj/auto_ap/graphql/ledger.clj
View File

@@ -8,16 +8,16 @@
[auto-ap.graphql.utils
:refer [->graphql <-graphql assert-admin assert-can-see-client result->page]]
[auto-ap.parse.util :as parse]
[auto-ap.utils :refer [by dollars=]]
[auto-ap.pdf.ledger :refer [print-pnl print-balance-sheet]]
[auto-ap.pdf.ledger :refer [print-balance-sheet print-pnl]]
[auto-ap.utils :refer [by dollars= heartbeat]]
[clj-time.coerce :as coerce]
[clj-time.core :as t]
[clojure.tools.logging :as log]
[com.walmartlabs.lacinia.util :refer [attach-resolvers]]
[datomic.api :as d]
[mount.core :as mount]
[unilog.context :as lc]
[yang.scheduler :as scheduler]
[clj-time.core :as t]))
[yang.scheduler :as scheduler]))
(mount/defstate running-balance-cache
:start (atom {}))
@@ -525,16 +525,11 @@
(defn refresh-running-balance-cache []
(lc/with-context {:source "running-balance-cache"}
(try
(log/info "Refreshing running balance cache")
(build-running-balance-cache)
(catch Exception e
(log/error e)))))
(build-running-balance-cache))
(mount/defstate running-balance-cache-worker
:start (scheduler/every (* 15 60 1000) refresh-running-balance-cache)
:start (scheduler/every (* 15 60 1000) (heartbeat refresh-running-balance-cache "running-balance-cache"))
:stop (scheduler/stop running-balance-cache-worker))

15
src/clj/auto_ap/import/intuit.clj
View File

@@ -2,18 +2,18 @@
(:require
[auto-ap.datomic :refer [conn]]
[auto-ap.import.transactions :as t]
[auto-ap.time :as atime]
[auto-ap.intuit.core :as i]
[auto-ap.utils :refer [allow-once]]
[auto-ap.time :as atime]
[auto-ap.utils :refer [allow-once heartbeat]]
[clj-time.coerce :as coerce]
[clj-time.core :as time]
[clojure.string :as str]
[clojure.tools.logging :as log]
[com.unbounce.dogstatsd.core :as statsd]
[datomic.api :as d]
[mount.core :as mount]
[unilog.context :as lc]
[yang.scheduler :as scheduler]
[clojure.string :as str]
[clojure.tools.logging :as log]))
[yang.scheduler :as scheduler]))
(defn get-intuit-bank-accounts [db]
(d/q '[:find ?external-id ?ba ?c
@@ -47,7 +47,6 @@
(t/apply-synthetic-ids)))
(defn import-intuit []
(lc/with-context {:source "Import intuit transactions"}
(statsd/event {:title "Intuit import started"
:text "Starting"
:priority :low}
@@ -74,7 +73,7 @@
:text (str e)
:alert-type :warning
:priority :normal}
nil))))))
nil)))))
(def upsert-transactions (allow-once upsert-transactions))
@@ -88,7 +87,7 @@
bank-accounts))))
(mount/defstate import-worker
:start (scheduler/every (* 1000 60 60 24) import-intuit)
:start (scheduler/every (* 1000 60 60 24) (heartbeat import-intuit "import-intuit"))
:stop (scheduler/stop import-worker))
(mount/defstate account-worker

12
src/clj/auto_ap/import/plaid.clj
View File

@@ -1,17 +1,15 @@
(ns auto-ap.import.plaid
(:require
[auto-ap.datomic :refer [conn]]
[auto-ap.plaid.core :as p]
[auto-ap.utils :refer [allow-once by]]
[auto-ap.import.transactions :as t]
[auto-ap.plaid.core :as p]
[auto-ap.utils :refer [allow-once by heartbeat]]
[clj-time.coerce :as coerce]
[clj-time.core :as time]
[clojure.tools.logging :as log]
[datomic.api :as d]
[mount.core :as mount]
[unilog.context :as lc]
[yang.scheduler :as scheduler]
[clj-time.coerce :as coerce]))
[yang.scheduler :as scheduler]))
(defn get-plaid-accounts [db]
(-> (d/q '[:find ?ba ?c ?external-id ?t
@@ -59,7 +57,7 @@
(def import-plaid (allow-once import-plaid))
(mount/defstate import-worker
:start (scheduler/every (* 1000 60 60 3) import-plaid)
:start (scheduler/every (* 1000 60 60 3) (heartbeat import-plaid "import-plaid"))
:stop (scheduler/stop import-worker))

7
src/clj/auto_ap/import/yodlee.clj
View File

@@ -3,7 +3,7 @@
[auto-ap.datomic :refer [conn]]
[auto-ap.import.transactions :as t]
[auto-ap.time :as atime]
[auto-ap.utils :refer [allow-once]]
[auto-ap.utils :refer [allow-once heartbeat]]
[auto-ap.yodlee.core :as client]
[clj-time.coerce :as coerce]
[clojure.string :as str]
@@ -45,7 +45,6 @@
:status status}))
(defn import-yodlee []
(lc/with-context {:source "Import yodlee transactions"}
(statsd/event {:title "Yodlee import started"
:text "Starting"
:priority :low}
@@ -78,10 +77,10 @@
:text (str e)
:alert-type :warning
:priority :normal}
nil))))))
nil)))))
(def import-yodlee (allow-once import-yodlee))
(mount/defstate import-worker
:start (scheduler/every (* 1000 60 60 8) import-yodlee)
:start (scheduler/every (* 1000 60 60 8) (heartbeat import-yodlee "import-yodlee"))
:stop (scheduler/stop import-worker))

6
src/clj/auto_ap/import/yodlee2.clj
View File

@@ -3,7 +3,7 @@
[auto-ap.datomic :refer [conn]]
[auto-ap.import.transactions :as t]
[auto-ap.import.yodlee :as y]
[auto-ap.utils :refer [allow-once]]
[auto-ap.utils :refer [allow-once heartbeat]]
[auto-ap.yodlee.core2 :as client2]
[com.unbounce.dogstatsd.core :as statsd]
[datomic.api :as d]
@@ -53,9 +53,9 @@
(mount/defstate import-worker
:start (scheduler/every (* 1000 60 60 4) import-yodlee2)
:start (scheduler/every (* 1000 60 60 4) (heartbeat import-yodlee2 "import-yodlee"))
:stop (scheduler/stop import-worker))
(mount/defstate account-worker
:start (scheduler/every (* 5 60 1000) client2/upsert-accounts)
:start (scheduler/every (* 5 60 1000) (heartbeat client2/upsert-accounts "upsert-yodlee2-accounts"))
:stop (scheduler/stop account-worker))

16
src/clj/auto_ap/ledger.clj
View File

@@ -196,9 +196,6 @@
: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]
@@ -228,15 +225,12 @@
(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")
@(d/transact conn repairs))
(log/info "Finished reconciling ledger"))
(catch Exception e
(log/error e)))))
@(d/transact conn repairs))))
(mount/defstate reconciliation-frequency :start (* 1000 60 60))
(mount/defstate ledger-reconciliation-worker
:start (scheduler/every reconciliation-frequency reconcile-ledger)
:start (scheduler/every reconciliation-frequency (heartbeat reconcile-ledger "reconcile-ledger"))
:stop (scheduler/stop ledger-reconciliation-worker))
@@ -371,12 +365,10 @@
invoice-accounts)))
(defn touch-broken-ledger []
(lc/with-context {:source "touch-broken-ledger"}
(statsd/event {:title "Reconciling Ledger"
:text "This process looks for unbalance ledger entries, or missing ledger entries"
:priority :low}
nil)
(try
(log/info "Attempting to fix transactions that are in the ledger but are wrong")
(let [mismatched-ts (mismatched-transactions)]
(if (seq mismatched-ts)
@@ -418,9 +410,7 @@
(statsd/event {:title "Finished Reconciling Ledger"
:text "This process looks for unbalance ledger entries, or missing ledger entries"
:priority :low}
nil)
(catch Exception e
(log/error e)))))
nil))
(mount/defstate touch-broken-ledger-worker
:start (scheduler/every reconciliation-frequency (heartbeat touch-broken-ledger "touch-broken-ledger"))

59
src/clj/auto_ap/plaid/import.clj
View File

@@ -1,59 +0,0 @@
(ns auto-ap.plaid.import
(:require
[auto-ap.datomic :refer [conn]]
[auto-ap.plaid.core :as p]
[auto-ap.utils :refer [allow-once]]
[auto-ap.yodlee.import :as y]
[clj-time.core :as time]
[clojure.tools.logging :as log]
[datomic.api :as d]
[mount.core :as mount]
[unilog.context :as lc]
[yang.scheduler :as scheduler]
[clj-time.coerce :as coerce]))
(defn get-plaid-accounts [db]
(-> (d/q '[:find ?ba ?c ?external-id ?t
:in $
:where
[?c :client/bank-accounts ?ba]
[?ba :bank-account/plaid-account ?pa]
[?pa :plaid-account/external-id ?external-id]
[?pi :plaid-item/accounts ?pa]
[?pi :plaid-item/access-token ?t]]
db )))
(defn plaid->transaction [t]
#:transaction {:description-original (:name t)
:raw-id (:transaction_id t)
:id (digest/sha-256 (:transaction_id t))
:amount (double (:amount t))
:date (coerce/to-date (auto-ap.time/parse (:date t) auto-ap.time/iso-date))
:status "POSTED"})
(defn import-plaid []
(lc/with-context {:source "Import plaid transactions"}
(let [import-batch (y/start-import-batch :import-source/plaid "Automated plaid user")
end (auto-ap.time/local-now)
start (time/plus end (time/days -30))]
(try
(doseq [[bank-account-id client-id external-id access-token] (get-plaid-accounts (d/db conn))
transaction (:transactions (p/get-transactions access-token external-id start end))]
(when (not (:pending transaction))
(y/import-transaction! import-batch (assoc (plaid->transaction transaction)
:transaction/bank-account bank-account-id
:transaction/client client-id))))
(y/finish! import-batch)
(catch Exception e
(y/fail! import-batch e))))))
(def import-plaid (allow-once import-plaid))
(mount/defstate import-worker
:start (scheduler/every (* 1000 60 60 3) import-plaid)
:stop (scheduler/stop import-worker))

3
src/clj/auto_ap/square/core.clj
View File

@@ -491,8 +491,7 @@
(defn upsert-all []
(doseq [client (get-square-clients)
:when (seq (filter :square-location/client-location (:client/square-locations client)))]
(lc/with-context {:source "Square loading"
:client (:client/code client)}
(lc/with-context {:client (:client/code client)}
(upsert-locations client)
(log/info "Loading Orders")
(upsert client)

16
src/cljc/auto_ap/utils.cljc
View File

@@ -1,7 +1,9 @@
(ns auto-ap.utils
#?@
(:clj
[(:require [com.unbounce.dogstatsd.core :as statsd])]))
[(:require [com.unbounce.dogstatsd.core :as statsd]
[clojure.tools.logging :as log]
[unilog.context :as lc])]))
(defn by
([f xs]
@@ -87,9 +89,19 @@
(defn heartbeat [f id]
(fn []
#?(:clj (do
(lc/with-context {:source id}
(try
(log/info "Starting background process " id)
(f)
(log/info "Completed background process " id)
(statsd/service-check {:name (str id)
:status :ok}
nil))
nil)
(catch Exception e
(log/error e)
(statsd/service-check {:name (str id)
:status :critical}
nil)))))
:cljs (do (println "Heartbeat for " id)
(f)))))