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

Makes logging unified

Browse Source
This commit is contained in:
Bryce
2023-10-30 12:35:18 -07:00
parent f0a7c378f7
commit 930b900849
44 changed files with 485 additions and 555 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/clj/auto_ap/background/metrics.clj
View File

@@ -2,9 +2,8 @@
(:require [com.unbounce.dogstatsd.core :as statsd]
[mount.core :as mount]
[clj-http.client :as http]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[config.core :refer [env]]
[unilog.context :as lc]
[com.brunobonacci.mulog :as mu]))
(defn get-container-data []
@@ -16,7 +15,8 @@
:body)]
result)
(catch Exception e
(log/warn "cannot find container tags" e)
(alog/error ::no-container-tags
:error e)
{})))
(mount/defstate container-data
@@ -34,8 +34,6 @@
(defn set-logging-context []
(when (seq container-data)
(lc/push-context "container" (:DockerId container-data))
(lc/push-context "ip" (-> container-data :Networks first :IPv4Addresses first))
(mu/start-publisher! {:type :console-json
:transform (fn [events]
@@ -56,8 +54,7 @@
(defn stop-logging-context []
(when (seq container-data)
(lc/pull-context "container")
(lc/pull-context "ip")))
))
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
(mount/defstate logging-context

1
src/clj/auto_ap/datomic.clj
View File

@@ -11,7 +11,6 @@
[iol-ion.tx.upsert-transaction]
[com.github.ivarref.gen-fn :refer [gen-fn! datomic-fn]]
[auto-ap.utils :refer [default-pagination-size by]]
[clojure.tools.logging :as log]
[clojure.edn :as edn]
[config.core :refer [env]]
[datomic.api :as dc]

2
src/clj/auto_ap/datomic/accounts.clj
View File

@@ -9,7 +9,6 @@
pull-many
query2]]
[clojure.string :as str]
[clojure.tools.logging :as log]
[datomic.api :as dc]))
(defn <-datomic [a]
@@ -132,7 +131,6 @@
accounts))
(defn get-graphql [args]
(log/info "ARGS" args)
(let [db (dc/db conn)
{ids-to-retrieve :ids matching-count :count} (raw-graphql-ids db args)]
[(->> (graphql-results ids-to-retrieve db args))

58
src/clj/auto_ap/graphql.clj
View File

@@ -17,7 +17,7 @@
[auto-ap.graphql.utils :refer [attach-tracing-resolvers]]
[auto-ap.graphql.vendors :as gq-vendors]
[auto-ap.graphql.yodlee-merchants :as ym]
[auto-ap.logging :as alog :refer [error-event info-event warn-event]]
[auto-ap.logging :as alog]
[auto-ap.time :as time]
[clj-time.coerce :as coerce]
[clj-time.core :as t]
@@ -29,7 +29,6 @@
[com.walmartlabs.lacinia.parser :as p]
[com.walmartlabs.lacinia.schema :as schema]
[datomic.api :as dc]
[unilog.context :as lc]
[yang.time :refer [time-it]])
(:import
(clojure.lang IPersistentMap)))
@@ -807,36 +806,31 @@
([id q v]
(statsd/increment "query.graphql.count" {:tags #{(str "query:" (query-name q))}})
(statsd/time! [(str "query.graphql.time" ) {:tags #{(str "query:" (query-name q))}}]
(mu/with-context {:query-string q :user id}
(lc/with-context {:query q}
(alog/info ::executing-query
:query-name (query-name q))
(try
(let [[result time] (time-it (simplify (execute schema q (dissoc v
:clients) {:id id
:clients (:clients v)})))]
(info-event "Query completed"
{:time (:time time)
:errors (seq (:errors result))})
(when (seq (:errors result))
(throw (ex-info "GraphQL error" {:result result})))
result)
(mu/with-context {:query-name (query-name q) :user id :query q}
(mu/trace ::executing-query
[]
(try
(let [[result time] (time-it (simplify (execute schema q (dissoc v
:clients) {:id id
:clients (:clients v)
:log-context (or (mu/local-context) {})})))]
(when (seq (:errors result))
(throw (ex-info "GraphQL error" {:result result})))
result)
(catch Exception e
(if-let [v (or (:validation-error (ex-data e))
(:validation-error (ex-data (.getCause e))))]
(do
(alog/warn ::query-validation
:exception e)
(warn-event "validation error" {:validation-error v
:data (ex-data e)})
(throw e)
#_{:errors [{:message v}]})
(do
(error-event "query error" {:error e})
(alog/error ::query-error
:exception e)
(catch Exception e
(if-let [v (or (:validation-error (ex-data e))
(:validation-error (ex-data (.getCause e))))]
(do
(alog/warn ::query-validation
:exception e)
(throw e)
#_{:errors [{:message v}]})
(do
(alog/error ::query-error
:exception e)
(throw e))))))))))
(throw e))))))))))

5
src/clj/auto_ap/graphql/checks.clj
View File

@@ -8,6 +8,7 @@
[auto-ap.datomic.clients :as d-clients]
[auto-ap.datomic.invoices :as d-invoices]
[auto-ap.datomic.transactions :as d-transactions]
[auto-ap.logging :as alog]
[auto-ap.datomic.vendors :as d-vendors]
[auto-ap.graphql.utils
:refer [->graphql
@@ -29,7 +30,6 @@
[clojure.java.io :as io]
[clojure.set :as set]
[clojure.string :as str]
[clojure.tools.logging :as log]
[com.brunobonacci.mulog :as mu]
[com.walmartlabs.lacinia.util :refer [attach-resolvers]]
[config.core :refer [env]]
@@ -430,6 +430,7 @@
nil)}))
(defn get-payment-page [context args _]
(alog/info ::TEST)
(let [[payments checks-count] (d-checks/get-graphql (-> args
:filters
(<-graphql)
@@ -570,7 +571,7 @@
:ids)
specific-ids (d-checks/filter-ids (:ids args))
all-ids (into (set ids) specific-ids)]
(log/info "Voiding " (count all-ids) args)
(alog/info ::voiding-payments :count (count all-ids))
(void-payments-internal all-ids (:id context))
{:message (str "Succesfully voided " (count all-ids))}))

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

@@ -18,12 +18,10 @@
[clojure.java.io :as io]
[clojure.set :as set]
[clojure.string :as str]
[clojure.tools.logging :as log]
[com.brunobonacci.mulog :as mu]
[datomic.api :as dc]
[iol-ion.tx :refer [random-tempid]]
[mount.core :as mount]
[unilog.context :as lc]
[yang.scheduler :as scheduler]
[auto-ap.solr :as solr])
(:import
@@ -151,7 +149,6 @@
_ (mu/log ::upserting :up updated-entity)
_ (assert-no-shared-transaction-sources client-code [[:upsert-entity updated-entity]])
_ (log/info "upserting client" updated-entity)
result (audit-transact [[:upsert-entity updated-entity]] (:id context))]
(when (:square_auth_token edit_client)
@@ -182,7 +179,7 @@
(defn refresh-all-current-balance []
(lc/with-context {:source "current-balance-refresh"}
(mu/with-context {:source "current-balance-refresh"}
(let [db (dc/db conn)
clients (dc/q '[:find (pull ?c [:db/id :client/code {:client/bank-accounts [:db/id :bank-account/code]}])
:where [?c :client/code]]

9
src/clj/auto_ap/graphql/invoices.clj
View File

@@ -6,6 +6,7 @@
[auto-ap.datomic.invoices :as d-invoices]
[auto-ap.datomic.vendors :as d-vendors]
[auto-ap.graphql.checks :as gq-checks]
[auto-ap.logging :as alog]
[auto-ap.graphql.utils
:as u
:refer [<-graphql
@@ -20,7 +21,6 @@
[auto-ap.utils :refer [dollars=]]
[clj-time.coerce :as coerce]
[clj-time.core :as time]
[clojure.tools.logging :as log]
[com.brunobonacci.mulog :as mu]
[datomic.api :as dc]
[auto-ap.solr :as solr]))
@@ -293,10 +293,10 @@
all-ids)
(map first))
]
(log/info "Voiding " (count voidable-cash-payments) "cash payments first")
(alog/info ::void-payments :count (count voidable-cash-payments))
(gq-checks/void-payments-internal voidable-cash-payments (:id context))
(log/info "Voiding " (count all-ids) args)
(alog/info ::voiding-invoices :count (count all-ids))
(audit-transact
(->> all-ids
(dc/q '[:find (pull ?i [:db/id :invoice/date {:invoice/expense-accounts [:db/id]}])
@@ -452,7 +452,6 @@
all-ids (all-ids-not-locked (get-ids-matching-filters args))
invoices (pull-many (dc/db conn) '[:db/id :invoice/total] (vec all-ids))
account-total (reduce + 0 (map (fn [x] (:percentage x)) (:accounts args)))]
(log/info "client is" locations)
(when
(not (dollars= 1.0 account-total))
(let [error (str "Account total (" account-total ") does not reach 100%")]
@@ -467,7 +466,7 @@
(:location a))))
(let [err (str "Account " name " uses location " (:location a) ", but doesn't belong to the client.")]
(throw (ex-info err {:validation-error err}) ))))
(log/info "Bulk coding " (count all-ids) args)
(alog/info ::bulk-code :count (count all-ids))
(audit-transact-batch
(map (fn [i]
[:upsert-invoice {:db/id (:db/id i)

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

@@ -27,7 +27,6 @@
[clj-time.coerce :as coerce]
[clj-time.core :as t]
[clojure.data.csv :as csv]
[clojure.tools.logging :as log]
[com.brunobonacci.mulog :as mu]
[datomic.api :as dc]
[iol-ion.tx :refer [random-tempid]])
@@ -180,7 +179,7 @@
(map (fn [client-id]
[client-id (build-account-lookup client-id)]))
(into {}))]
(log/info "Running balance sheet with " args)
(alog/info ::balance-sheet :params args)
(cond-> {:balance-sheet-accounts (mapcat
#(roll-up-until (lookup-account %) (all-ledger-entries %) end-date )
@@ -357,7 +356,7 @@
(if (> (count all-ids) 1000)
{:message (str "You can only delete 1000 ledger entries at a time.")}
(do
(log/info "Deleting " (count all-ids) args)
(alog/info ::deleting :args args)
(audit-transact-batch
(map (fn [i]
[:db/retractEntity i])
@@ -547,8 +546,9 @@
)
(map first)
(map (fn [je] [:db/retractEntity je])))]
(log/info "manual ledger import has " (count success) " new rows")
(log/info errors)
(alog/info ::manual-import
:errors (count errors)
:sample (take 3 errors))
(mu/trace ::retraction-tx
@@ -558,7 +558,6 @@
[:count (count ignore-retraction)]
(when (seq ignore-retraction)
(audit-transact-batch ignore-retraction (:id context))))
#_(log/info (map :tx success))
(let [invalidated
(mu/trace ::success-tx
[:count (count success)]

22
src/clj/auto_ap/graphql/utils.clj
View File

@@ -8,9 +8,9 @@
[iol-ion.query :refer [entid]]
[clojure.walk :as walk]
[com.walmartlabs.lacinia.util :refer [attach-resolvers]]
[clojure.tools.logging :as log]
[com.brunobonacci.mulog :as mu]
[clojure.set :as set]))
[clojure.set :as set]
[auto-ap.logging :as alog]))
(defn snake->kebab [s]
@@ -56,7 +56,7 @@
(defn assert-admin [id]
(when-not (= "admin" (:user/role id))
(log/warn "user " id " not an admin!")
(alog/warn ::unauthorized :user id :role "admin")
(throw-unauthorized)))
(defn assert-present
@@ -74,19 +74,19 @@
(defn assert-power-user [id]
(when-not (#{"power-user" "admin"} (:user/role id))
(log/warn "user " id " not an power-user!")
(alog/warn ::unauthorized :user id :role "power-user")
(throw-unauthorized)))
(defn can-see-client? [identity client]
(when (not client)
(log/warn "WARNING - permission checking for null client"))
(alog/error ::checking-for-null-client :id identity))
(or (= "admin" (:user/role identity))
((set (map :db/id (:user/clients identity))) (:db/id client))
((set (map :db/id (:user/clients identity))) client)))
(defn assert-can-see-client [identity client]
(when-not (can-see-client? identity client)
(log/warn "IDENTITY " identity " can not see company " client)
(alog/warn ::unauthorized :id identity :client client)
(throw-unauthorized)))
(defn limited-clients [id]
@@ -149,10 +149,12 @@
(defn trace-query [key f]
(fn trace [a b c]
(mu/with-context {:query key
:mutation (boolean (= "mutation"
(namespace key)))
:user (:id a)}
(mu/with-context (merge
(:log-context a {})
{:query key
:mutation (boolean (= "mutation"
(namespace key)))
:user (:id a)})
(mu/trace (keyword "graphql" (name key))
[]
(f a b c)))))

27
src/clj/auto_ap/handler.clj
View File

@@ -36,7 +36,6 @@
[ring.middleware.session :refer [wrap-session]]
[ring.middleware.session.cookie :refer [cookie-store]]
[ring.util.response :as response]
[unilog.context :as lc]
[clojure.set :as set]))
(when (:aws-access-key-id env)
@@ -122,20 +121,24 @@
(defn wrap-logging [handler]
(fn [request]
(mu/with-context {:uri (:uri request)
:query (:uri request)
:request-method (:request-method request)
:user (:identity request)
:user-role (:user/role (:identity request))
:user-name (:user/name (:identity request))
:query-params (:query-params request)}
(mu/with-context (cond-> {:uri (:uri request)
:route (:handler (bidi.bidi/match-route all-routes
(:uri request)
:request-method (:request-method request)))
:query (:uri request)
:request-method (:request-method request)
:user (dissoc (:identity request)
:gz-clients)
:user-role (:user/role (:identity request))
:user-name (:user/name (:identity request))}
(not= "/api/graphql" (:uri request))
(assoc :query-params (:query-params request)))
(mu/trace ::http-request-trace
[]
(lc/with-context {:uri (:uri request)
(mu/with-context {:uri (:uri request)
:source "request"
:user-role (:user/role (:identity request))
:user-name (:user/name (:identity request))
:query-params (:query-params request)}
:user-name (:user/name (:identity request))}
(when-not (str/includes? (:uri request) "health-check")
@@ -218,7 +221,7 @@
(pull-many (dc/db conn)
d-clients/full-read))]
(lc/with-context {:clients (map :client/code clients)}
(mu/with-context {:clients (map :client/code clients)}
(handler (assoc request
:clients clients
:client (when (= 1 (count clients))

4
src/clj/auto_ap/import/common.clj
View File

@@ -1,7 +1,7 @@
(ns auto-ap.import.common
(:require
[auto-ap.datomic :refer [conn pull-ref random-tempid]]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[datomic.api :as dc]))
(defn bank-account->integration-id [bank-account]
@@ -25,5 +25,5 @@
:integration-status/state :integration-state/failed
:integration-status/last-attempt (java.util.Date.)
:integration-status/message (.getMessage e)}}])
(log/warn e)
(alog/warn ::integration-failed :error e)
nil)))

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

@@ -4,15 +4,13 @@
[auto-ap.import.common :refer [wrap-integration]]
[auto-ap.import.transactions :as t]
[auto-ap.intuit.core :as i]
[auto-ap.logging :as alog]
[auto-ap.time :as atime]
[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 dc]
[mount.core :as mount]
[yang.scheduler :as scheduler]))
[datomic.api :as dc]))
(defn get-intuit-bank-accounts [db]
(dc/q '[:find ?external-id ?ba ?c
@@ -28,7 +26,7 @@
(try
(Integer/parseInt (:Num transaction))
(catch NumberFormatException e
(log/warn "Got an invalid check number " e)
(alog/warn ::invalid-check-number :check-number (:Num transaction) :error e)
nil)))]
(cond-> {:transaction/description-original (:Memo/Description transaction)
:transaction/amount (Double/parseDouble (:Amount transaction))

60
src/clj/auto_ap/import/manual.clj
View File

@@ -5,8 +5,7 @@
[auto-ap.import.transactions :as t]
[clj-time.coerce :as coerce]
[clojure.data.csv :as csv]
[datomic.api :as dc]
[unilog.context :as lc]))
[datomic.api :as dc]))
@@ -33,33 +32,32 @@
(c/assoc-or-error :transaction/amount #(c/parse-amount transaction))))
(defn import-batch [transactions user]
(lc/with-context {:source "Manual import transactions"}
(let [bank-account-code->client (into {}
(dc/q '[:find ?bac ?c
:in $
:where
[?c :client/bank-accounts ?ba]
[?ba :bank-account/code ?bac]]
(dc/db conn)))
bank-account-code->bank-account (into {}
(dc/q '[:find ?bac ?ba
:in $
:where [?ba :bank-account/code ?bac]]
(dc/db conn)))
import-batch (t/start-import-batch :import-source/manual user)
transactions (->> transactions
(map (fn [t]
(manual->transaction t bank-account-code->bank-account bank-account-code->client)))
(t/apply-synthetic-ids ))]
(try
(doseq [transaction transactions]
(when-not (seq (:errors transaction))
(t/import-transaction! import-batch transaction)))
(let [bank-account-code->client (into {}
(dc/q '[:find ?bac ?c
:in $
:where
[?c :client/bank-accounts ?ba]
[?ba :bank-account/code ?bac]]
(dc/db conn)))
bank-account-code->bank-account (into {}
(dc/q '[:find ?bac ?ba
:in $
:where [?ba :bank-account/code ?bac]]
(dc/db conn)))
import-batch (t/start-import-batch :import-source/manual user)
transactions (->> transactions
(map (fn [t]
(manual->transaction t bank-account-code->bank-account bank-account-code->client)))
(t/apply-synthetic-ids ))]
(try
(doseq [transaction transactions]
(when-not (seq (:errors transaction))
(t/import-transaction! import-batch transaction)))
(t/finish! import-batch)
(assoc (t/get-stats import-batch)
:failed-validation (count (filter :errors transactions))
:sample-error (first (first (map :errors (filter :errors transactions)))))
(catch Exception e
(t/fail! import-batch e)
(t/get-stats import-batch))))))
(t/finish! import-batch)
(assoc (t/get-stats import-batch)
:failed-validation (count (filter :errors transactions))
:sample-error (first (first (map :errors (filter :errors transactions)))))
(catch Exception e
(t/fail! import-batch e)
(t/get-stats import-batch)))))

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

@@ -14,7 +14,6 @@
[digest :as di]
[manifold.deferred :as de]
[manifold.executor :as ex]
[unilog.context :as lc]
[clojure.string :as str]))
(defn get-plaid-accounts [db]
@@ -73,32 +72,31 @@
"name" (:plaid-merchant/name result)}))))
(defn import-plaid-int []
(lc/with-context {:source "Import plaid transactions"}
(let [import-batch (t/start-import-batch :import-source/plaid "Automated plaid user")
end (atime/local-now)
start (time/plus end (time/days -30))
plaid-merchant->vendor-id (build-plaid-merchant->vendor-id)]
(try
(doseq [[bank-account-id client-id external-id access-token] (get-plaid-accounts (dc/db conn))
:let [transaction-result (wrap-integration #(p/get-transactions access-token external-id start end)
bank-account-id)
accounts-by-id (by :account_id (:accounts transaction-result))]
transaction (:transactions transaction-result)]
(when (not (:pending transaction))
(t/import-transaction! import-batch (assoc (plaid->transaction (assoc transaction
:account
(accounts-by-id (:account_id transaction)))
plaid-merchant->vendor-id)
(let [import-batch (t/start-import-batch :import-source/plaid "Automated plaid user")
end (atime/local-now)
start (time/plus end (time/days -30))
plaid-merchant->vendor-id (build-plaid-merchant->vendor-id)]
(try
(doseq [[bank-account-id client-id external-id access-token] (get-plaid-accounts (dc/db conn))
:let [transaction-result (wrap-integration #(p/get-transactions access-token external-id start end)
bank-account-id)
accounts-by-id (by :account_id (:accounts transaction-result))]
transaction (:transactions transaction-result)]
(when (not (:pending transaction))
(t/import-transaction! import-batch (assoc (plaid->transaction (assoc transaction
:account
(accounts-by-id (:account_id transaction)))
plaid-merchant->vendor-id)
:transaction/bank-account bank-account-id
:transaction/client client-id))))
(try
(rebuild-search-index)
(catch Exception e
(alog/error ::cant-index-plaid
:error e)))
(t/finish! import-batch)
(try
(rebuild-search-index)
(catch Exception e
(t/fail! import-batch e))))))
(alog/error ::cant-index-plaid
:error e)))
(t/finish! import-batch)
(catch Exception e
(t/fail! import-batch e)))))
(def import-plaid (allow-once import-plaid-int))

41
src/clj/auto_ap/import/transactions.clj
View File

@@ -12,7 +12,7 @@
[clj-time.coerce :as coerce]
[clj-time.core :as t]
[clojure.core.cache :as cache]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[datomic.api :as dc]
[digest :as di]))
@@ -30,11 +30,11 @@
(defn transaction->existing-payment [_ check-number client-id bank-account-id amount id]
(log/info "Searching for a matching check for "
{:client-id client-id
(alog/info ::searching
:client-id client-id
:check-number check-number
:bank-account-id bank-account-id
:amount amount})
:amount amount)
(cond (not (and client-id bank-account-id))
nil
@@ -57,7 +57,6 @@
(rough-match client-id bank-account-id amount)))
(defn match-transaction-to-unfulfilled-autopayments [amount client-id]
(log/info "trying to find uncleared autopay invoices")
(let [candidate-invoices-vendor-groups (->> (dc/q {:find ['?vendor-id '?e '?total '?sd]
:in ['$ '?client-id]
:where ['[?e :invoice/client ?client-id]
@@ -77,12 +76,15 @@
(+ acc amount)) 0.0 consideration)
(- amount))]
consideration)]
(log/info "Found " (count considerations) "considerations for transaction of" amount)
considerations
))
(alog/info ::unfulfilled-autoapayment-considerations
:count (count considerations)
:amount amount)
considerations))
(defn match-transaction-to-unpaid-invoices [amount client-id]
(log/info "trying to find unpaid invoices for " client-id amount)
(alog/info ::searching-unpaid-invoice
:client-id client-id
:amount amount)
(let [candidate-invoices-vendor-groups (->> (dc/q {:find ['?vendor-id '?e '?outstanding-balance '?d]
:in ['$ '?client-id]
:where ['[?e :invoice/client ?client-id]
@@ -102,7 +104,10 @@
(+ acc amount)) 0.0 consideration)
(- amount))]
consideration)]
(log/info "Found " (count considerations) "unpaid invoice considerations for transaction of" amount)
(alog/info ::unpaid-invoice-considerations-found
:client-id client-id
:amount amount
:count (count considerations))
considerations))
(defn match-transaction-to-single-unfulfilled-autopayments [amount client-id]
@@ -112,7 +117,9 @@
[])))
(defn add-new-payment [transaction [[vendor] :as invoice-payments] bank-account-id client-id]
(log/info "Adding a new payment for transaction " (:transaction/id transaction) " and invoices " invoice-payments)
(alog/info ::adding-payment
:transaction-id (:transaction/id transaction)
:invoices (count invoice-payments))
(let [payment-id (random-tempid)]
(-> [[:upsert-transaction
(assoc transaction
@@ -279,7 +286,6 @@
(defn get-existing [bank-account]
(log/info "looking up bank account data for" bank-account)
(into {}
(dc/q '[:find ?tid ?as2
:in $ ?ba
@@ -314,7 +320,8 @@
:import-batch/status :import-status/started
:import-batch/user-name user}])) "import-batch")
rule-applying-function (rm/rule-applying-fn (tr/get-all))]
(log/info "Importing transactions from " source)
(alog/info ::starting-transaction-import
:source source)
(reify ImportBatch
(import-transaction! [_ transaction]
@@ -344,15 +351,17 @@
@stats)
(fail! [_ error]
(log/errorf "Couldn't complete import %d with error." import-id)
(log/error error)
(alog/error ::cant-complete-import
:import-id import-id
:error error)
@(dc/transact-async conn [(merge {:db/id import-id
:import-batch/status :import-status/completed
:import-batch/error-message (str error)}
@stats)]))
(finish! [_]
(log/infof "Finishing import batch %d for %s with stats %s " import-id (name source) (pr-str @stats))
(alog/info ::finished :import-id import-id :source source :stats (pr-str @stats))
@(dc/transact conn [(merge {:db/id import-id
:import-batch/status :import-status/completed}

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

@@ -10,8 +10,7 @@
[clojure.string :as str]
[com.unbounce.dogstatsd.core :as statsd]
[datomic.api :as dc]
[digest :as di]
[unilog.context :as lc]))
[digest :as di]))
#_{:clj-kondo/ignore [:unresolved-var]}
(defn yodlee->transaction [transaction use-date-instead-of-post-date?]
@@ -45,43 +44,42 @@
:status status}))
(defn import-yodlee2-int []
(lc/with-context {:source "Import yodlee2 transactions"}
(statsd/event {:title "Yodlee2 import started"
:text "Starting"
:priority :low}
nil)
(let [import-batch (t/start-import-batch :import-source/yodlee2 "Automated yodlee2 user")]
(try
(let [account-lookup (dc/q '[:find ?ya ?ba ?cd ?ud
:in $
:where
[?ba :bank-account/yodlee-account ?y]
[(get-else $ ?ba :bank-account/use-date-instead-of-post-date? false) ?ud]
[?c :client/bank-accounts ?ba]
[?c :client/code ?cd]
[?y :yodlee-account/id ?ya]
]
(dc/db conn))]
(doseq [[yodlee-account bank-account client-code use-date-instead-of-post-date?] account-lookup
transaction (wrap-integration #(client2/get-specific-transactions client-code yodlee-account)
bank-account)]
(t/import-transaction! import-batch (assoc (yodlee->transaction transaction use-date-instead-of-post-date?)
:transaction/bank-account bank-account
:transaction/client [:client/code client-code])))
(statsd/event {:title "Yodlee2 import started"
:text "Starting"
:priority :low}
nil)
(let [import-batch (t/start-import-batch :import-source/yodlee2 "Automated yodlee2 user")]
(try
(let [account-lookup (dc/q '[:find ?ya ?ba ?cd ?ud
:in $
:where
[?ba :bank-account/yodlee-account ?y]
[(get-else $ ?ba :bank-account/use-date-instead-of-post-date? false) ?ud]
[?c :client/bank-accounts ?ba]
[?c :client/code ?cd]
[?y :yodlee-account/id ?ya]
]
(dc/db conn))]
(doseq [[yodlee-account bank-account client-code use-date-instead-of-post-date?] account-lookup
transaction (wrap-integration #(client2/get-specific-transactions client-code yodlee-account)
bank-account)]
(t/import-transaction! import-batch (assoc (yodlee->transaction transaction use-date-instead-of-post-date?)
:transaction/bank-account bank-account
:transaction/client [:client/code client-code])))
(t/finish! import-batch))
(statsd/event {:title "Yodlee2 import Finished"
:text (pr-str (t/get-stats import-batch))
:priority :low}
nil)
(t/finish! import-batch))
(statsd/event {:title "Yodlee2 import Finished"
:text (pr-str (t/get-stats import-batch))
:priority :low}
nil)
(catch Exception e
(t/fail! import-batch e)
(statsd/event {:title "Yodlee2 import failed"
:text (str e)
:alert-type :warning
:priority :normal}
nil))))))
(catch Exception e
(t/fail! import-batch e)
(statsd/event {:title "Yodlee2 import failed"
:text (str e)
:alert-type :warning
:priority :normal}
nil)))))

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

@@ -5,7 +5,6 @@
[clojure.core.memoize :as m]
[clojure.java.io :as io]
[clojure.string :as str]
[clojure.tools.logging :as log]
[config.core :as cfg :refer [env]])
(:import
(org.apache.commons.codec.binary Base64)))
@@ -87,7 +86,7 @@
"refresh_token" refresh-token}
:as :json}))]
(set-access-token (:access_token response))
(set-refresh-token (doto (:refresh_token response) log/info))
(set-refresh-token (:refresh_token response))
(:access_token response)))
(def prod-base-headers {"Accept" "application/json"

4
src/clj/auto_ap/jobs/bulk_journal_import.clj
View File

@@ -7,7 +7,7 @@
[clojure.data.csv :as csv]
[clojure.java.io :as io]
[clojure.string :as str]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[config.core :refer [env]]))
(defn line->id [{:keys [source id client-code]}]
@@ -67,7 +67,7 @@
(defn bulk-journal-import [args]
(let [{:keys [ledger-url]} args
_ (log/info "importing ledger from" ledger-url)
_ (alog/info ::importing :url ledger-url)
csv-stream (s3->csv ledger-url)
import-rows (csv->graphql-rows csv-stream)]
(import-ledger {:id {:user/name "Bulk-import" :user/role "admin"}}

17
src/clj/auto_ap/jobs/close_auto_invoices.clj
View File

@@ -5,7 +5,7 @@
[auto-ap.jobs.core :refer [execute]]
[auto-ap.time :as time]
[clj-time.coerce :as coerce]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[datomic.api :as dc]))
(defn close-auto-invoices []
@@ -15,15 +15,16 @@
'[?e :invoice/status :invoice-status/unpaid]
'[(<= ?d ?today)]]}
(dc/db conn) (coerce/to-date (time/local-now)))]
(log/info "Closing " (count invoices-to-close) "scheduled invoices")
(alog/info ::closing :count (count invoices-to-close))
@(dc/transact conn (some->> invoices-to-close
seq
seq
(mapv (fn [[i]] {:db/id i
:invoice/outstanding-balance 0.0
:invoice/status :invoice-status/paid}))
))
(log/info "Closed " (count invoices-to-close) "scheduled invoices")))
(mapv (fn [[i]] {:db/id i
:invoice/outstanding-balance 0.0
:invoice/status :invoice-status/paid}))
))
(alog/info ::closed :count (count invoices-to-close))))
(defn -main [& _]

36
src/clj/auto_ap/jobs/core.clj
View File

@@ -2,26 +2,26 @@
(:require [auto-ap.utils :refer [heartbeat]]
[mount.core :as mount]
[auto-ap.datomic :refer [conn ]]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[nrepl.server :refer [start-server]]
[auto-ap.background.metrics :refer [metrics-setup container-tags container-data logging-context]]
[unilog.context :as lc]
[com.brunobonacci.mulog :as mu]))
(defn execute [name f]
(try
(lc/with-context {:background-job name}
(mu/with-context {:background-job name
:service name}
(mount/start (mount/only #{#'conn #'metrics-setup #'container-tags #'logging-context #'container-data }))
(start-server :port 9000 :bind "0.0.0.0" #_#_:handler (cider-nrepl-handler))
((heartbeat f name))
(log/info "Stopping " name)
(Thread/sleep 15000)
(mount/stop)))
(catch Exception e
(log/error "ERROR" e)
(println e)
(throw e))
(finally
(System/exit 0))))
(mu/with-context {:background-job name
:source name
:service name}
(mu/trace ::execute-background-job
[]
(try
(mount/start (mount/only #{#'conn #'metrics-setup #'container-tags #'logging-context #'container-data }))
(start-server :port 9000 :bind "0.0.0.0" #_#_:handler (cider-nrepl-handler))
((heartbeat f name))
(alog/info ::stopping :job name)
(Thread/sleep 15000)
(mount/stop)
(catch Exception e
(alog/error ::job-error :error e)
(throw e))
(finally
(System/exit 0))))))

32
src/clj/auto_ap/jobs/load_historical_sales.clj
View File

@@ -5,18 +5,16 @@
[auto-ap.jobs.core :refer [execute]]
[auto-ap.square.core :as square]
[auto-ap.square.core3 :as square3]
[auto-ap.time :as atime]
[clj-time.coerce :as coerce]
[clj-time.core :as time]
[clj-time.periodic :as per]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[config.core :refer [env]]
[datomic.api :as dc]
[unilog.context :as lc]))
[datomic.api :as dc]))
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
(defn historical-load-sales [client days]
(log/info "loading old approach")
(alog/info ::old-sales-loading)
(let [client (dc/pull (dc/db auto-ap.datomic/conn)
square/square-read
client)]
@@ -24,12 +22,11 @@
:when (:square-location/client-location square-location)]
(println "orders")
(lc/with-context {:source "Historical loading data"}
(doseq [d (per/periodic-seq (time/plus (time/today) (time/days (- days)))
(time/today)
(time/days 1))]
(println d)
(square/upsert client square-location (coerce/to-date-time d) (coerce/to-date-time (time/plus d (time/days 1))))))
(doseq [d (per/periodic-seq (time/plus (time/today) (time/days (- days)))
(time/today)
(time/days 1))]
(println d)
(square/upsert client square-location (coerce/to-date-time d) (coerce/to-date-time (time/plus d (time/days 1)))))
(println "refunds")
(square/upsert-refunds client square-location)
@@ -48,7 +45,7 @@
(defn historical-load-sales2 [client days]
(log/info "loading new approach")
(alog/info ::new-sales-loading)
(let [client (dc/pull (dc/db auto-ap.datomic/conn)
square/square-read
client)
@@ -57,12 +54,11 @@
:when (:square-location/client-location square-location)]
(println "orders")
(lc/with-context {:source "Historical loading data"}
(doseq [d (per/periodic-seq (time/plus (time/today) (time/days (- days)))
(time/today)
(time/days 1))]
(println d)
@(square3/upsert client square-location (coerce/to-date-time d) (coerce/to-date-time (time/plus d (time/days 1))))))
(doseq [d (per/periodic-seq (time/plus (time/today) (time/days (- days)))
(time/today)
(time/days 1))]
(println d)
@(square3/upsert client square-location (coerce/to-date-time d) (coerce/to-date-time (time/plus d (time/days 1)))))
(println "refunds")
@(square3/upsert-refunds client square-location)

15
src/clj/auto_ap/jobs/register_invoice_import.clj
View File

@@ -7,10 +7,10 @@
[auto-ap.time :as atime]
[auto-ap.utils :refer [dollars=]]
[clj-time.coerce :as coerce]
[auto-ap.logging :as alog]
[clojure.data.csv :as csv]
[clojure.java.io :as io]
[clojure.string :as str]
[clojure.tools.logging :as log]
[config.core :refer [env]]
[datomic.api :as dc])
(:import
@@ -26,8 +26,10 @@
io/reader
csv/read-csv))
(catch Exception e
(log/error (str "Could not read the file " url ". Are you sure you uploaded it?"))
(log/error e)
(alog/error
:file-not-found
:error e
:url url)
(throw e))))
(defn register-invoice-import* [data]
@@ -138,9 +140,12 @@
(defn register-invoice-import [args]
(let [{:keys [invoice-url]} args
data (s3->csv invoice-url)]
(log/info "contains " (count data) " rows")
(alog/info ::rows
:count (count data))
(doseq [n (partition-all 50 (register-invoice-import* data))]
(log/info "transacting" n)
(alog/info ::transacting
:count (count n)
:sample (take 2 n))
(audit-transact n {:user/name "register-invoice-import"
:user/role "admin"}))))

27
src/clj/auto_ap/jobs/sysco.clj
View File

@@ -12,8 +12,8 @@
[clojure.data.csv :as csv]
[clojure.java.io :as io]
[com.brunobonacci.mulog :as mu]
[auto-ap.logging :as alog]
[clojure.string :as str]
[clojure.tools.logging :as log]
[com.unbounce.dogstatsd.core :as statsd]
[config.core :refer [env]]
[datomic.api :as dc]
@@ -77,7 +77,9 @@
date (t/parse
(header-row "InvoiceDate")
"yyMMdd")]
(log/infof "Importing %s for %s" (header-row "InvoiceNumber") (header-row "CustomerName"))
(alog/info ::importing
:invoice-number (header-row "InvoiceNumber")
:customer-name (header-row "CustomerName"))
(cond-> #:invoice {:invoice-number (header-row "InvoiceNumber")
:db/id (random-tempid)
@@ -124,7 +126,9 @@
(map :key))]
(log/infof "Found %d sysco invoice to import: %s" (count keys) (pr-str keys))
(alog/info ::importing-sysco
:count (count keys)
:keys (pr-str keys))
(let [transaction (->> keys
(mapcat (fn [k]
@@ -141,15 +145,16 @@
(extract-invoice-details sysco-vendor)
(assoc :invoice/source-url invoice-url))]])
(catch Exception e
(log/error (str "Cannot load file " k) e)
(log/info
(s3/copy-object {:source-bucket-name (:data-bucket env)
:destination-bucket-name (:data-bucket env)
:source-key k
:destination-key (str "sysco/error/"
(.getName (io/file k)))}))
(alog/error ::cant-load-file
:file k
:error e e)
(s3/copy-object {:source-bucket-name (:data-bucket env)
:destination-bucket-name (:data-bucket env)
:source-key k
:destination-key (str "sysco/error/"
(.getName (io/file k)))})
[])))))
result (audit-transact transaction {:user/name "sysco importer" :user/role "admin"})])
result (audit-transact transaction {:user/name "sysco importer" :user/role "admin"})])
(doseq [k keys]
(mark-key k))))

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

@@ -1,20 +1,6 @@
(ns auto-ap.logging
(:require [clojure.tools.logging :as log]
[unilog.context :as lc]
[com.brunobonacci.mulog :as mu]))
(:require [com.brunobonacci.mulog :as mu]))
(defn info-event [message context]
(lc/with-context context
(log/info message)))
(defn warn-event [message context]
(lc/with-context context
(log/warn message)))
(defn error-event [message context]
(lc/with-context context
(log/warn message)))
(defmacro with-context-as [ctx s & body]
`(mu/with-context ~ctx

7
src/clj/auto_ap/parse.clj
View File

@@ -8,7 +8,7 @@
[clojure.java.shell :as sh]
[clojure.set :as set]
[clojure.string :as str]
[clojure.tools.logging :as log]))
[auto-ap.logging :as alog]))
(def last-text (atom nil))
@@ -18,7 +18,8 @@
(defn extract-template
([text template]
(log/info "Template was determined to be" template)
(alog/info ::template-determined
:template (str template))
(if (:multi template)
(mapcat
@@ -41,7 +42,7 @@
(assoc result k (try
(u/parse-value value-parser parser-params value)
(catch Exception e
(log/warn e))))))
(alog/warn ::cant-parse-value :error e :raw value))))))
{:vendor-code (:vendor template)
:text text
:full-text full-text}))])))

6
src/clj/auto_ap/parse/csv.clj
View File

@@ -3,11 +3,11 @@
[clojure.data.csv :as csv]
[clojure.java.io :as io]
[clojure.string :as str]
[clojure.tools.logging :as log]))
[auto-ap.logging :as alog]))
(defn determine
[[header]]
(log/info "Importing with header" header)
(alog/info ::importing :header header)
(let [csv-type (cond (str/includes? (second header) "Customer's PO No.")
:mama-lus
@@ -32,7 +32,7 @@
:else
nil)]
(log/info "csv type was determined to be" csv-type)
(alog/info ::csv-type-determined :type csv-type)
csv-type))
(defmulti parse-csv

4
src/clj/auto_ap/parse/util.clj
View File

@@ -3,7 +3,7 @@
[clj-time.core :as time]
[clj-time.format :as f]
[clojure.string :as str]
[clojure.tools.logging :as log]))
[auto-ap.logging :as alog]))
(defmulti parse-value (fn [method _ _]
method))
@@ -46,7 +46,7 @@
(reduced (time/from-time-zone (f/parse (f/formatter format) value)
(time/time-zone-for-id "America/Los_Angeles")))
(catch Exception e
(log/warn e)
(alog/warn ::cant-parse-date :error e :raw-value (str value))
nil)))
nil
format)

4
src/clj/auto_ap/pdf/ledger.clj
View File

@@ -9,7 +9,7 @@
[clj-pdf.core :as pdf]
[clojure.java.io :as io]
[clojure.string :as str]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[config.core :refer [env]]
[datomic.api :as dc])
(:import
@@ -279,7 +279,7 @@
clients (pull-many (dc/db conn) '[:client/code :client/name :db/id] (:client-ids args))
report (l-reports/journal-detail-report args data (by :db/id :client/code clients))
output-stream (ByteArrayOutputStream.)]
(log/info report)
(alog/info ::make-detail-report)
(pdf/pdf
(-> [{:left-margin 10 :right-margin 10 :top-margin 15 :bottom-margin 15
:size :letter

7
src/clj/auto_ap/plaid/core.clj
View File

@@ -2,7 +2,7 @@
(:require
[clj-http.client :as client]
[clojure.data.json :as json]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[config.core :as cfg :refer [env]]
[auto-ap.time :as atime]))
@@ -62,7 +62,10 @@
:body))
(defn get-transactions [access-token account-id start end]
(log/infof "looking up transactions from %s to %s for %s" start end account-id)
(alog/info ::searching
:start (str start)
:end (str end)
:acct (str account-id))
(-> (client/post (str base-url "/transactions/get")
{:as :json
:headers {"Content-Type" "application/json"}

5
src/clj/auto_ap/routes/auth.clj
View File

@@ -4,7 +4,7 @@
[buddy.sign.jwt :as jwt]
[clj-http.client :as http]
[clj-time.core :as time]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[config.core :refer [env]]
[com.brunobonacci.mulog :as mu]
[clojure.java.io :as io]
@@ -99,7 +99,8 @@
{:status 401
:body "Couldn't authenticate"}))
(catch Exception e
(log/warn e)
(alog/warn ::cant-authenticate
:error e)
{:status 401
:body (str "Couldn't authenticate " (.toString e))})))

6
src/clj/auto_ap/routes/exports.clj
View File

@@ -16,7 +16,6 @@
[clj-time.core :as time]
[clojure.data.csv :as csv]
[clojure.edn :refer [read-string]]
[clojure.tools.logging :as log]
[com.unbounce.dogstatsd.core :as statsd]
[config.core :refer [env]]
[datomic.api :as dc]
@@ -277,7 +276,8 @@
(let [start-date (or (some-> (query-params "start-date")
(atime/parse atime/iso-date))
(time/plus (time/now) (time/days -120)))]
(log/info "exporting for " (query-params "client-code") "starting" start-date)
(alog/info ::exporting-ledger
:params query-params)
(assert-admin identity)
(statsd/time! [(str "export.time") {:tags #{(client-tag query-params)
"export:ledger2"}}]
@@ -458,7 +458,7 @@
(defn export-raw [{:keys [query-params identity]}]
(assert-admin identity)
(log/info "Executing raw query " (get query-params "query" ))
(alog/info ::executing-query :q (get query-params "query" ))
(statsd/time! [(str "export.time") {:tags #{"export:raw"}}]
{:body
(into (list) (apply dc/q (read-string (get query-params "query" )) (into [(dc/db conn)] (read-string (get query-params "args" "[]")))))}))

13
src/clj/auto_ap/routes/graphql.clj
View File

@@ -4,12 +4,11 @@
[auto-ap.datomic.clients :as d-clients]
[auto-ap.graphql :as ql]
[auto-ap.graphql.utils :refer [limited-clients]]
[auto-ap.logging :refer [warn-event]]
[auto-ap.logging :as alog]
[auto-ap.routes.utils :refer [wrap-secure]]
[buddy.auth :refer [throw-unauthorized]]
[clojure.edn :as edn]
[clojure.set :as set]
[clojure.tools.logging :as log]
[datomic.api :as dc]))
@@ -30,21 +29,21 @@
clients) ))
:headers {"Content-Type" "application/edn"}})
(catch Throwable e
(log/info "here we are " (.getCause e))
(if-let [result (:result (ex-data e))]
(do (log/warn "Graphql Result error" e)
(do (alog/warn ::result-error :error e)
{:status 400
:body (pr-str result)
:headers {"Content-Type" "application/edn"}})
(if-let [message (:validation-error (ex-data (.getCause e)) )]
(do
(warn-event "GraphQL Validation error" {:message message
:data e})
(alog/warn ::graphql-validation-error
:message message
:error e)
{:status 400
:body (pr-str {:errors [{:message message}]})
:headers {"Content-Type" "application/edn"}})
(do (log/error "GraphQL error" e)
(do (alog/error ::error :error e)
{:status 500
:body (pr-str {:errors [{:message (str "Unhandled error:" (str e))}]})
:headers {"Content-Type" "application/edn"}}))))))

53
src/clj/auto_ap/routes/invoices.clj
View File

@@ -16,13 +16,12 @@
[clojure.data.csv :as csv]
[clojure.java.io :as io]
[clojure.string :as str]
[clojure.tools.logging :as log]
[config.core :refer [env]]
[datomic.api :as dc]
[digest]
[iol-ion.tx :refer [random-tempid]]
[ring.middleware.json :refer [wrap-json-response]]
[unilog.context :as lc]
[com.brunobonacci.mulog :as mu]
[auto-ap.logging :as alog])
(:import
(java.util UUID)))
@@ -251,25 +250,24 @@
(map #(assoc % :invoice/source-url-admin-only (boolean (> client-count 1))) is)))
(defn import-uploaded-invoice [user imports]
(lc/with-context {:area "upload-invoice"}
(log/info "Number of invoices to import is" (count imports))
(let [potential-invoices (->> imports
(map import->invoice)
(map #(validate-invoice % user))
admin-only-if-multiple-clients
(mapv d-invoices/code-invoice)
(mapv (fn [i] [:propose-invoice i])))]
(log/info "creating invoice" potential-invoices)
(let [tx (audit-transact potential-invoices user)]
(when-not (seq (dc/q '[:find ?i
:in $ [?i ...]
:where [?i :invoice/invoice-number]]
(:db-after tx)
(map :e (:tx-data tx))))
(throw (ex-info "No new invoices found."
{})))
tx))))
(alog/info ::importing-uploaded :count (count imports))
(let [potential-invoices (->> imports
(map import->invoice)
(map #(validate-invoice % user))
admin-only-if-multiple-clients
(mapv d-invoices/code-invoice)
(mapv (fn [i] [:propose-invoice i])))]
(alog/info ::creating-invoice :invoices potential-invoices)
(let [tx (audit-transact potential-invoices user)]
(when-not (seq (dc/q '[:find ?i
:in $ [?i ...]
:where [?i :invoice/invoice-number]]
(:db-after tx)
(map :e (:tx-data tx))))
(throw (ex-info "No new invoices found."
{})))
tx)))
(defn validate-account-rows [rows code->existing-account]
(when-let [bad-types (seq (->> rows
@@ -390,7 +388,8 @@
:body (pr-str stats)
:headers {"Content-Type" "application/edn"}})
(catch Exception e
(log/error e)
(alog/error ::couldnt-batch-upload
:error e)
{:status 500
:body (pr-str {:message (.getMessage e)
:error (.toString e)
@@ -412,8 +411,7 @@
vendor (some-> (or vendor vendor-2)
(Long/parseLong))
{:keys [filename tempfile]} files]
(lc/with-context {:parsing-file filename}
(log/info tempfile)
(mu/with-context {:parsing-file filename}
(try
(let [extension (last (str/split (.getName (io/file filename)) #"\."))
s3-location (str "invoice-files/" (str (UUID/randomUUID)) "." extension)
@@ -437,7 +435,8 @@
:body (pr-str {})
:headers {"Content-Type" "application/edn"}}
(catch Exception e
(log/warn e)
(alog/warn ::couldnt-import-upload
:error e)
{:status 400
:body (pr-str {:message (.getMessage e)
:error (.toString e)
@@ -486,7 +485,7 @@
:body (pr-str {})
:headers {"Content-Type" "application/edn"}}
(catch Exception e
(log/error e)
(alog/error ::error :error e)
{:status 500
:body (pr-str {:message (.getMessage e)
:error (.toString e)
@@ -509,7 +508,7 @@
:body (import-account-overrides client (.getPath tempfile))
:headers {"Content-Type" "application/json"}}
(catch Exception e
(log/error e)
(alog/error ::error :error e)
{:status 500
:body {:message (.getMessage e)
:data (ex-data e)}

16
src/clj/auto_ap/routes/queries.clj
View File

@@ -7,14 +7,14 @@
[clojure.edn :as edn]
[clojure.java.io :as io]
[clojure.string :as str]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[com.unbounce.dogstatsd.core :as statsd]
[com.brunobonacci.mulog :as mu]
[config.core :refer [env]]
[datomic.api :as dc]
[iol-ion.tx :refer [random-tempid]]
[ring.middleware.json :refer [wrap-json-response]]
[ring.util.request :refer [body-string]]
[unilog.context :as lc])
[ring.util.request :refer [body-string]])
(:import
(java.util UUID)))
@@ -29,12 +29,11 @@
(defn execute-query [query-params params]
(let [{:keys [query-id]} params]
(lc/with-context {:query-id query-id}
(mu/with-context {:query-id query-id}
(log/info "Executing raw query " query-id)
(alog/info ::executing-query :query-id query-id)
(let [query-string (str (slurp (:object-content (s3/get-object :bucket-name (:data-bucket env)
:key (str "queries/" (:query-id params))))))]
(log/info "Executing query " query-string)
(into (list) (apply dc/q (edn/read-string query-string)
(into [(dc/db conn)] (edn/read-string (get query-params "args" "[]")))))))))
@@ -69,7 +68,6 @@
(assert-admin identity)
(let [obj (s3/list-objects :bucket-name (:data-bucket env)
:prefix (str "queries/"))]
(log/info obj)
{:body (->> (:object-summaries obj)
(map (fn [o]
{:last-modified (.toString (:last-modified o))
@@ -77,7 +75,6 @@
(defn create-query [{:keys [query-params identity] :as request}]
(assert-admin identity)
(log/info "Note" (query-params "note"))
(put-query (str (UUID/randomUUID)) (body-string request) (query-params "note")))
(defn get-query [{:keys [identity params]} ]
@@ -86,7 +83,6 @@
obj (s3/get-object :bucket-name (:data-bucket env)
:key (str "queries/" query-id))
query-string (str (slurp (:object-content obj)))]
(log/info obj)
{:body {:query query-string
:note (:note (:user-metadata (:object-metadata obj)))
:id query-id
@@ -95,7 +91,6 @@
(defn update-query [{:keys [query-params identity params] :as request} ]
(assert-admin identity)
(log/info "Note" (query-params "note"))
(put-query (:query-id params) (body-string request) (query-params "note")))
(defn results-json-query [{:keys [query-params params]}]
@@ -108,7 +103,6 @@
obj (s3/get-object :bucket-name (:data-bucket env)
:key (str "queries/" query-id))
query-string (str (slurp (:object-content obj)))]
(log/info obj)
{:body query-string}))
(defn results-csv-query [{:keys [query-params params]}]

16
src/clj/auto_ap/routes/yodlee2.clj
View File

@@ -5,7 +5,7 @@
[auto-ap.graphql.utils :refer [assert-admin assert-can-see-client]]
[auto-ap.routes.utils :refer [wrap-secure]]
[auto-ap.yodlee.core2 :as yodlee]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[config.core :refer [env]]
[datomic.api :as dc]))
@@ -25,7 +25,7 @@
:url (:yodlee2-fastlink env)}) }))
(defn refresh-provider-accounts [{:keys [identity edn-params]}]
(assert-admin identity)
(log/info "refreshing " edn-params)
(alog/info ::refreshing :params edn-params)
(try
(yodlee/refresh-provider-account (-> (:client-id edn-params)
Long/parseLong
@@ -36,7 +36,7 @@
:headers {"Content-Type" "application/edn"}
:body "{}" }
(catch Exception e
(log/error e)
(alog/error ::error :error e)
{:status 400
:headers {"Content-Type" "application/edn"}
:body (pr-str {:message (.getMessage e)
@@ -45,7 +45,9 @@
(defn get-provider-account-detail [{:keys [identity]
{:keys [client id]} :route-params}]
(assert-admin identity)
(log/info "looking-up " client id)
(alog/info ::looking-up
:client client
:id id)
(try
{:status 200
@@ -56,7 +58,7 @@
:client/code)
id))}
(catch Exception e
(log/error e)
(alog/error ::error :error e)
{:status 400
:headers {"Content-Type" "application/edn"}
:body (pr-str {:message (.getMessage e)
@@ -75,7 +77,7 @@
(Long/parseLong id)
(dissoc data :client-id )))}
(catch Exception e
(log/error e)
(alog/error ::error :error e)
{:status 500
:headers {"Content-Type" "application/edn"}
:body (pr-str {:message (.getMessage e)
@@ -93,7 +95,7 @@
:headers {"Content-Type" "application/edn"}
:body (pr-str {}) }
(catch Exception e
(log/error e)
(alog/error ::error :error e)
{:status 400
:headers {"Content-Type" "application/edn"}
:body (pr-str {:message (.getMessage e)

4
src/clj/auto_ap/server.clj
View File

@@ -20,7 +20,7 @@
[auto-ap.jobs.sysco :as job-sysco]
[auto-ap.jobs.vendor-usages :as job-vendor-usages]
[auto-ap.jobs.yodlee2 :as job-yodlee2]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[com.unbounce.dogstatsd.core :as statsd]
[config.core :refer [env]]
[mount.core :as mount]
@@ -79,7 +79,7 @@
(statsd/gauge "requests.5xx" (double (.getResponses5xx (.getHandler jetty))))
(.statsReset (.getHandler jetty))
(catch Exception e
(log/warn e))))
(alog/warn ::cant-collect-stats :error e))))
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
(mount/defstate jetty-stats

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

@@ -11,10 +11,9 @@
[clojure.data.json :as json]
[clojure.set :as set]
[clojure.string :as str]
[clojure.tools.logging :as log]
[datomic.api :as dc]
[slingshot.slingshot :refer [try+]]
[unilog.context :as lc]))
))
(defn client-base-headers [client]
{"Square-Version" "2021-08-18"
@@ -22,7 +21,6 @@
"Content-Type" "application/json"})
(defn retry-4 [ex try-count _]
(log/warn "Retrying after failure " ex)
(if (> try-count 4) false true))
(defn lookup-dates []
@@ -42,13 +40,12 @@
:body
:locations)
(catch Exception e
(log/warn e)
[])))
(defn fetch-catalog [client i]
(if i
(try
(log/trace "looking up catalog for" (str "https://connect.squareup.com/v2/catalog/object/" i))
(->> (client/get (str "https://connect.squareup.com/v2/catalog/object/" i)
{:headers (client-base-headers client)
:query-params {"include_related_items" "true"}
@@ -56,9 +53,7 @@
:body
:object)
(catch Exception e
(log/error e)
nil))
(log/warn "Trying to look up non existant ")))
nil))))
(def fetch-catalog-fast (memoize fetch-catalog))
@@ -75,7 +70,7 @@
:else
(do
(log/warn "couldn't look up" i)
"Uncategorized"))))
(defn pc [start end]
@@ -98,7 +93,7 @@
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
(defn get-order
([client location order-id]
(log/info "Searching for" (:square-location/client-location location))
(let [result (->> (client/get (str "https://connect.squareup.com/v2/orders/" order-id)
{:headers (client-base-headers client)
:as :json})
@@ -107,7 +102,7 @@
result)))
(defn continue-search [client location start end cursor]
(log/info "Continuing search for" cursor)
(let [result (->> (client/post "https://connect.squareup.com/v2/orders/search"
{:headers (client-base-headers client)
:body (json/write-str (cond-> {"location_ids" [(:square-location/square-id location)]
@@ -116,21 +111,21 @@
start (merge (pc start end))))
:as :json})
:body)]
(log/info "found " (count (:orders result)))
(if (not-empty (:cursor result))
(concat (:orders result) (continue-search client location start end (:cursor result)))
(:orders result))))
(defn search
([client location start end]
(log/info "Searching for" (:square-location/client-location location))
(let [result (->> (client/post "https://connect.squareup.com/v2/orders/search"
{:headers (client-base-headers client)
:body (json/write-str (cond-> {"location_ids" [(:square-location/square-id location)] "limit" 10000}
start (merge (pc start end))))
:as :json})
:body)]
(log/info "found " (count (:orders result)))
(if (not-empty (:cursor result))
(concat (:orders result) (continue-search client location start end (:cursor result)))
(:orders result)))))
@@ -247,9 +242,8 @@
(try
(f)
(catch Exception e
(log/warn "error pulling http " e)
(retry f (inc i))))
(log/warn "Too many failures"))))
(retry f (inc i)))))))
(defn get-payment [client p]
(:payment (:body (retry #(client/get (str "https://connect.squareup.com/v2/payments/" p)
@@ -267,13 +261,11 @@
(async/pipeline-blocking concurrent
output-chan
(map (fn [p]
(lc/with-context {:source "Square settlements loading "}
(log/trace "looking up payment " p " for settlement " (:id settlement))
(or
(-> (get-payment client p)
:created_at
coerce/to-date)
(coerce/to-date (time/now))))))
(or
(-> (get-payment client p)
:created_at
coerce/to-date)
(coerce/to-date (time/now)))))
(async/to-chan! (->> settlement
:entries
(filter #(= "CHARGE" (:type %)))
@@ -284,9 +276,7 @@
))
true
(fn [e]
(lc/with-context {:source "Square settlements loading "}
(log/warn "Error loading sales date details" e)
e)))
e))
(->> (async/<!! (async/into [] output-chan))
(map halt-if-error)
sort
@@ -294,38 +284,32 @@
first)))
(defn get-settlement-details [client location settlements] ;; pairs of [location settlement]
(log/info "getting settlement details for " settlements)
(let [concurrent 10
output-chan (async/chan)]
(async/pipeline-blocking concurrent
output-chan
(map (fn [s]
(lc/with-context {:source "Square settlements loading "}
(log/info "Looking up settlement " s " for location " (:square-location/client-location location))
(let [settlement (:body (retry #(client/get (str "https://connect.squareup.com/v1/" (:square-location/square-id location) "/settlements/" s)
{:headers (client-base-headers client)
:as :json
:retry-handler retry-4})))
sales-date (get-settlement-sales-date client settlement)]
(log/info "sales date for " s " is " sales-date)
(assoc settlement :sales-date sales-date)))))
(let [settlement (:body (retry #(client/get (str "https://connect.squareup.com/v1/" (:square-location/square-id location) "/settlements/" s)
{:headers (client-base-headers client)
:as :json
:retry-handler retry-4})))
sales-date (get-settlement-sales-date client settlement)]
(assoc settlement :sales-date sales-date))))
(async/to-chan! settlements)
true
(fn [e]
(lc/with-context {:source "Square settlements loading "}
(log/warn "Error loading settlements details" e)
e)))
e))
(->> (async/<!! (async/into [] output-chan))
(map halt-if-error))))
(defn settlements
([client location] (settlements client location (lookup-dates)))
([client location lookup-dates]
(log/info "Searching for" (:square-location/client-location location))
(->> lookup-dates
(mapcat (fn [[start-date end-date]]
(log/info "looking up settlements for " (:square-location/client-location location) " on dates " start-date " to " end-date)
(let [settlements (->> (retry #(client/get (str "https://connect.squareup.com/v1/" (:square-location/square-id location) "/settlements")
{:headers (client-base-headers client)
:query-params {"begin_time" start-date
@@ -394,20 +378,17 @@
:when (:square-location/client-location square-location)]
(upsert client square-location (time/plus (time/now) (time/days -3)) (time/now))))
([client location start end]
(lc/with-context {:source "Square loading"}
(let [existing (->> (dc/q {:find ['?external-id]
:in ['$ '?client]
:where ['[?o :sales-order/client ?client]
'[?o :sales-order/external-id ?external-id]]}
(dc/db conn) (:db/id client))
(map first)
set)
_ (log/info (count existing) "Sales orders already exist")
to-create (filter #(not (existing (:sales-order/external-id %)))
(daily-results client location start end))]
(doseq [x (partition-all 20 to-create)]
(log/info "Loading " (count x))
@(dc/transact-async conn x))))))
(let [existing (->> (dc/q {:find ['?external-id]
:in ['$ '?client]
:where ['[?o :sales-order/client ?client]
'[?o :sales-order/external-id ?external-id]]}
(dc/db conn) (:db/id client))
(map first)
set)
to-create (filter #(not (existing (:sales-order/external-id %)))
(daily-results client location start end))]
(doseq [x (partition-all 20 to-create)]
@(dc/transact-async conn x)))))
(defn upsert-settlements
([client]
@@ -415,12 +396,9 @@
:when (:square-location/client-location square-location)]
(upsert-settlements client square-location)))
([client location]
(lc/with-context {:source "Square settlements loading"
:client (:client/code client)}
(doseq [x (partition-all 20 (daily-settlements client location))]
(log/info "Loading expected deposit" (count x))
@(dc/transact-async conn x))
(log/info "Done loading settlements"))))
(doseq [x (partition-all 20 (daily-settlements client location))]
@(dc/transact-async conn x))))
(defn upsert-refunds
([client]
@@ -428,13 +406,9 @@
:when (:square-location/client-location square-location)]
(upsert-refunds client square-location)))
([client location]
(lc/with-context {:source "Loading Square Settlements"
:client (:client/code client)
:location (:square-location/client-location client)}
(doseq [x (partition-all 20 (refunds client location))]
(log/info "Loading refund" (count x))
@(dc/transact-async conn x))
(log/info "Done loading refunds"))))
(doseq [x (partition-all 20 (refunds client location))]
@(dc/transact-async conn x))))
(def square-read [:db/id
:client/code
@@ -501,29 +475,27 @@
(defn upsert-all [ & clients]
(doseq [client (apply get-square-clients clients)
:when (seq (filter :square-location/client-location (:client/square-locations client)))]
(lc/with-context {:client (:client/code client)}
(mark-integration-status client {:integration-status/last-attempt (coerce/to-date (time/now))})
(mark-integration-status client {:integration-status/last-attempt (coerce/to-date (time/now))})
(try+
(upsert-locations client)
(upsert client)
#_(upsert-settlements client) ;; currently settlements v1 is broken
(upsert-refunds client)
(mark-integration-status client {:integration-status/state :integration-state/success
:integration-status/last-updated (coerce/to-date (time/now))})
(try+
(upsert-locations client)
(upsert client)
#_(upsert-settlements client) ;; currently settlements v1 is broken
(upsert-refunds client)
(mark-integration-status client {:integration-status/state :integration-state/success
:integration-status/last-updated (coerce/to-date (time/now))})
(catch [:status 401] data
(mark-integration-status client {:integration-status/state :integration-state/unauthorized
:integration-status/message (-> data :body str)}))
(catch [:status 503] data
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (-> data :body str)}))
(catch Object _
(log/warn &throw-context)
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (or (some-> (:wrapper &throw-context) (.getMessage ))
(some-> (:object &throw-context) str)
"Unknown error")}))))))
(catch [:status 401] data
(mark-integration-status client {:integration-status/state :integration-state/unauthorized
:integration-status/message (-> data :body str)}))
(catch [:status 503] data
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (-> data :body str)}))
(catch Object _
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (or (some-> (:wrapper &throw-context) (.getMessage ))
(some-> (:object &throw-context) str)
"Unknown error")})))))

119
src/clj/auto_ap/square/core2.clj
View File

@@ -11,11 +11,10 @@
[clojure.data.json :as json]
[clojure.set :as set]
[clojure.string :as str]
[clojure.tools.logging :as log]
[cemerick.url :as url]
[datomic.api :as dc]
[slingshot.slingshot :refer [try+]]
[unilog.context :as lc]))
))
(defn client-base-headers [client]
{"Square-Version" "2021-08-18"
@@ -23,7 +22,6 @@
"Content-Type" "application/json"})
(defn retry-4 [ex try-count _]
(log/warn "Retrying after failure " ex)
(if (> try-count 4) false true))
(defn lookup-dates []
@@ -43,13 +41,11 @@
:body
:locations)
(catch Exception e
(log/warn e)
[])))
(defn fetch-catalog [client i]
(if i
(try
(log/trace "looking up catalog for" (str "https://connect.squareup.com/v2/catalog/object/" i))
(->> (client/get (str "https://connect.squareup.com/v2/catalog/object/" i)
{:headers (client-base-headers client)
:query-params {"include_related_items" "true"}
@@ -57,9 +53,7 @@
:body
:object)
(catch Exception e
(log/error e)
nil))
(log/warn "Trying to look up non existant ")))
nil))))
(def fetch-catalog-fast (memoize fetch-catalog))
@@ -76,7 +70,7 @@
:else
(do
(log/warn "couldn't look up" i)
"Uncategorized"))))
(defn pc [start end]
@@ -99,7 +93,7 @@
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
(defn get-order
([client location order-id]
(log/info "Searching for" (:square-location/client-location location))
(let [result (->> (client/get (str "https://connect.squareup.com/v2/orders/" order-id)
{:headers (client-base-headers client)
:as :json})
@@ -108,7 +102,7 @@
result)))
(defn continue-search [client location start end cursor]
(log/info "Continuing search for" cursor)
(let [result (->> (client/post "https://connect.squareup.com/v2/orders/search"
{:headers (client-base-headers client)
:body (json/write-str (cond-> {"location_ids" [(:square-location/square-id location)]
@@ -117,21 +111,21 @@
start (merge (pc start end))))
:as :json})
:body)]
(log/info "found " (count (:orders result)))
(if (not-empty (:cursor result))
(concat (:orders result) (continue-search client location start end (:cursor result)))
(:orders result))))
(defn search
([client location start end]
(log/info "Searching for" (:square-location/client-location location))
(let [result (->> (client/post "https://connect.squareup.com/v2/orders/search"
{:headers (client-base-headers client)
:body (json/write-str (cond-> {"location_ids" [(:square-location/square-id location)] "limit" 10000}
start (merge (pc start end))))
:as :json})
:body)]
(log/info "found " (count (:orders result)))
(if (not-empty (:cursor result))
(concat (:orders result) (continue-search client location start end (:cursor result)))
(:orders result)))))
@@ -268,9 +262,9 @@
(try
(f)
(catch Exception e
(log/warn "error pulling http " e)
(retry f (inc i))))
(log/warn "Too many failures"))))
)))
(defn get-payment [client p]
(:payment (:body (retry #(client/get (str "https://connect.squareup.com/v2/payments/" p)
@@ -283,34 +277,30 @@
x))
(defn get-settlement-details [client location settlements] ;; pairs of [location settlement]
(log/info "getting settlement details for " settlements)
(let [concurrent 4
output-chan (async/chan)]
(async/pipeline-blocking concurrent
output-chan
(map (fn [s]
(lc/with-context {:source "Square settlements loading "}
(log/info "Looking up settlement " s " for location " (:square-location/client-location location))
(:body (retry #(client/get (str "https://connect.squareup.com/v1/" (:square-location/square-id location) "/settlements/" (:id s))
{:headers (client-base-headers client)
:as :json
:retry-handler retry-4}))))))
(:body (retry #(client/get (str "https://connect.squareup.com/v1/" (:square-location/square-id location) "/settlements/" (:id s))
{:headers (client-base-headers client)
:as :json
:retry-handler retry-4})))))
(async/to-chan! settlements)
true
(fn [e]
(lc/with-context {:source "Square settlements loading "}
(log/warn "Error loading settlements details" e)
e)))
e))
(->> (async/<!! (async/into [] output-chan))
(map halt-if-error))))
(defn settlements
([client location] (settlements client location (lookup-dates)))
([client location lookup-dates]
(log/info "Searching for" (:square-location/client-location location))
(->> lookup-dates
(mapcat (fn [[start-date end-date]]
(log/info "looking up settlements for " (:square-location/client-location location) " on dates " start-date " to " end-date)
(let [settlements (->> (retry #(client/get (str "https://connect.squareup.com/v1/" (:square-location/square-id location) "/settlements")
{:headers (client-base-headers client)
:query-params {"begin_time" start-date
@@ -396,10 +386,9 @@
:when (:square-location/client-location square-location)]
(upsert client square-location (time/plus (time/now) (time/days -45)) (time/now))))
([client location start end]
(lc/with-context {:source "Square loading"}
(doseq [x (partition-all 20 (daily-results client location start end))]
(log/info "Loading " (count x))
@(dc/transact conn x)))))
(doseq [x (partition-all 20 (daily-results client location start end))]
@(dc/transact conn x))))
(defn upsert-settlements
([client]
@@ -407,12 +396,9 @@
:when (:square-location/client-location square-location)]
(upsert-settlements client square-location)))
([client location]
(lc/with-context {:source "Square settlements loading"
:client (:client/code client)}
(doseq [x (partition-all 20 (daily-settlements client location))]
(log/info "Loading expected deposit" (count x))
@(dc/transact conn x))
(log/info "Done loading settlements"))))
(doseq [x (partition-all 20 (daily-settlements client location))]
@(dc/transact conn x))))
(defn upsert-refunds
([client]
@@ -420,13 +406,9 @@
:when (:square-location/client-location square-location)]
(upsert-refunds client square-location)))
([client location]
(lc/with-context {:source "Loading Square Settlements"
:client (:client/code client)
:location (:square-location/client-location client)}
(doseq [x (partition-all 20 (refunds client location))]
(log/info "Loading refund" (count x))
@(dc/transact conn x))
(log/info "Done loading refunds"))))
(doseq [x (partition-all 20 (refunds client location))]
@(dc/transact conn x))))
(def square-read [:db/id
:client/code
@@ -493,32 +475,30 @@
(defn upsert-all [ & clients]
(doseq [client (apply get-square-clients clients)
:when (seq (filter :square-location/client-location (:client/square-locations client)))]
(lc/with-context {:client (:client/code client)}
(log/info "Importing square2 " (:client/code client))
(mark-integration-status client {:integration-status/last-attempt (coerce/to-date (time/now))})
(mark-integration-status client {:integration-status/last-attempt (coerce/to-date (time/now))})
(try+
(upsert-locations client)
(upsert client)
(upsert-settlements client)
(upsert-refunds client)
(mark-integration-status client {:integration-status/state :integration-state/success
:integration-status/last-updated (coerce/to-date (time/now))})
(try+
(upsert-locations client)
(upsert client)
(upsert-settlements client)
(upsert-refunds client)
(mark-integration-status client {:integration-status/state :integration-state/success
:integration-status/last-updated (coerce/to-date (time/now))})
(catch [:status 401] data
(mark-integration-status client {:integration-status/state :integration-state/unauthorized
:integration-status/message (-> data :body str)}))
(catch [:status 401] data
(mark-integration-status client {:integration-status/state :integration-state/unauthorized
:integration-status/message (-> data :body str)}))
(catch [:status 503] data
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (-> data :body str)}))
(catch Object _
(catch [:status 503] data
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (-> data :body str)}))
(catch Object _
(log/warn &throw-context)
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (or (some-> (:wrapper &throw-context) (.getMessage ))
(some-> (:object &throw-context) str)
"Unknown error")}))))))
(mark-integration-status client {:integration-status/state :integration-state/failed
:integration-status/message (or (some-> (:wrapper &throw-context) (.getMessage ))
(some-> (:object &throw-context) str)
"Unknown error")})))))
(defn preview-changes
([client]
@@ -528,5 +508,4 @@
:when (:square-location/client-location square-location)]
(preview-changes client square-location (time/plus (time/now) (time/days -30)) (time/now)))))
([client location start end]
(lc/with-context {:source "Square loading"}
(daily-results client location start end))))
(daily-results client location start end)))

4
src/clj/auto_ap/time.clj
View File

@@ -1,7 +1,7 @@
(ns auto-ap.time
(:require [clj-time.core :as time]
[clj-time.format :as f]
[clojure.tools.logging :as log]))
[auto-ap.logging :as alog]))
(defn localize [d]
(time/to-time-zone d (time/time-zone-for-id "America/Los_Angeles")))
@@ -35,7 +35,7 @@
(try
(f/unparse (f/formatter format) v)
(catch Exception e
(log/warn e)
(alog/warn ::cant-unparse :error e)
nil)))
(defn unparse-local [v format]

40
src/clj/auto_ap/yodlee/core2.clj
View File

@@ -1,8 +1,7 @@
(ns auto-ap.yodlee.core2
(:require [clj-http.client :as client]
[auto-ap.utils :refer [by]]
[unilog.context :as lc]
[clojure.tools.logging :as log]
[auto-ap.logging :as alog]
[clojure.data.json :as json]
[clojure.core.async :as async]
[config.core :refer [env]]
@@ -11,11 +10,11 @@
#_{:clj-kondo/ignore [:unused-namespace]}
[yang.scheduler :as scheduler]
[clj-time.coerce :as coerce]
[com.brunobonacci.mulog :as mu]
[datomic.api :as dc]
[auto-ap.datomic :refer [conn]]
[auto-ap.datomic.clients :as d-clients]
[clojure.string :as str]
[auto-ap.logging :as alog]))
[clojure.string :as str]))
;; switch all of this to use tokens instead of passing around client codes, particularly because the codes
;; need to be tweaked for repeats
(defn client-code->login [client-code]
@@ -45,10 +44,10 @@
:socket-timeout 60000
:connection-timeout 60000
:retry-handler (fn [ex _ _]
(log/error "yodlee Error." ex)
(alog/error ::error :error ex)
false)}
{:retry-handler (fn [ex _ _]
(log/error "yodlee Error." ex)
(alog/error ::error :error ex)
false)
:socket-timeout 60000
:connection-timeout 60000}))
@@ -74,7 +73,7 @@
(defn login-user [client-code]
(retry-thrice
(fn []
(log/info "logging in as " client-code)
(alog/info ::logging-in :client client-code)
(-> (str (:yodlee2-base-url env) "/auth/token")
(client/post (merge {:headers (assoc base-headers
"loginName" (if (:yodlee2-test-user env)
@@ -111,14 +110,14 @@
:body
:account))
(catch Exception e
(log/error (str "Couldn't get accounts for provider account '" provider-account-id "'")
e)
(alog/error ::error
:error e)
[])))
(defn get-provider-accounts [client-code ]
(retry-thrice
(fn []
(log/info "logging in user " client-code)
(alog/info ::logging-in :client client-code)
(let [cob-session (login-user (client-code->login client-code))]
(-> (str (:yodlee2-base-url env) "/providerAccounts")
(-> (client/get (merge {:headers (merge base-headers {"Authorization" (auth-header cob-session )})
@@ -258,16 +257,15 @@
(async/pipeline-blocking concurrent
output-chan
(map (fn [provider-account]
(lc/with-context {:provider-account-id (:id provider-account)}
(mu/with-context {:provider-account-id (:id provider-account)}
(get-provider-account-detail client-code (:id provider-account)))))
(async/to-chan! provider-accounts))
(async/<!! (async/into [] output-chan))))
(defn get-accounts-for-providers [client-code provider-account-ids]
(log/info "looking up " (count provider-account-ids) " provider accounts's accounts for client " client-code ".")
(into {}
(mapv (fn [provider-account-id]
(lc/with-context {:provider-account-id provider-account-id}
(mu/with-context {:provider-account-id provider-account-id}
[provider-account-id
(get-accounts-for-provider-account client-code provider-account-id)]))
provider-account-ids)))
@@ -323,7 +321,9 @@
provider-accounts))))
(defn refresh-provider-account [client-code id]
(log/info "refreshing yodlee provider account id" id)
(alog/info ::refreshing
:client client-code
:provider-acconut-id id)
@(dc/transact conn (upsert-accounts-tx client-code
[(get-provider-account client-code id)])))
@@ -333,16 +333,16 @@
(async/pipeline-blocking concurrent
output-chan
(mapcat (fn [client]
(log/info "Upserting Yodlee Accounts for " (:client/code client))
(lc/with-context {:client-code (:client/code client)}
(alog/info ::upserting-accounts
:client (:client/code client))
(mu/with-context {:client-code (:client/code client)}
(try
(upsert-accounts-tx (:client/code client))
(catch Exception e
(log/error "Could not update client " (:client/code client)
e))))))
(alog/error ::error :error e :client (:client/code client)))))))
(async/to-chan! (d-clients/get-all)))
(let [result (async/<!! (async/into [] output-chan))]
(log/info "Current yodlee state is " result)
(alog/info ::state :state result)
@(dc/transact conn result))))
@@ -357,7 +357,7 @@
:as :json}
other-config)))
(catch Exception e
(log/error e))) )
(alog/error ::error :error e))) )
(defn reauthenticate-and-recache [client-code pa data]
(reauthenticate client-code pa data)

20
src/clj/user.clj
View File

@@ -22,27 +22,33 @@
[datomic.api :as dc]
[mount.core :as mount]
[nrepl.middleware.print]
[unilog.context :as lc]
[com.brunobonacci.mulog :as mu]
[com.brunobonacci.mulog.buffer :as rb]
[datomic.api :as d]
[puget.printer :as puget]
[clojure.data.json :as json]
[auto-ap.solr :as solr])
(:import
(org.apache.commons.io.input BOMInputStream)))
(defn println-event [item]
(printf "%s: %s - %s:%s %s\n"
(printf "%s: %s - %s:%s by %s\n"
(str (c/to-date-time (:mulog/timestamp item)))
(:mulog/namespace item) (:mulog/event-name item)
(if (:mulog/duration item)
(str " " (int (/ (:mulog/duration item) 1000000)) "ms")
"")
(pr-str (reduce
(fn [acc [k v]]
(assoc acc k v))
{}
item))))
(:user-name item)
)
(puget/cprint (reduce
(fn [acc [k v]]
(assoc acc k v))
{}
(dissoc
item
:user))
{:seq-limit 10})
(println))
(deftype DevPublisher [config buffer transform]

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

@@ -2,8 +2,8 @@
#?@
(:clj
[(:require [com.unbounce.dogstatsd.core :as statsd]
[clojure.tools.logging :as log]
[unilog.context :as lc])]))
[com.brunobonacci.mulog :as mu]
[auto-ap.logging :as alog])]))
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
(defn by
@@ -65,16 +65,16 @@
#_{:clj-kondo/ignore [:clojure-lsp/unused-public-var]}
(defn heartbeat [f id]
(fn []
#?(:clj (lc/with-context {:source id}
#?(:clj (mu/with-context {:source id}
(try
(log/info "Starting background process " id)
(alog/info ::starting-process :id id)
(f)
(log/info "Completed background process " id)
(alog/info ::ending-process :id id)
(statsd/service-check {:name (str id)
:status :ok}
nil)
(catch Exception e
(log/error e)
(alog/error ::process-failed :error e)
(statsd/service-check {:name (str id)
:status :critical}
nil))))