mirror of
https://github.com/we-promise/sure.git
synced 2026-09-09 00:24:15 +00:00
* Fix user-triggered Plaid transaction refresh Request a fresh Plaid institution update for explicit user syncs, then poll the saved cursor with bounded retries so private self-hosted instances do not depend on webhooks. Preserve the existing immediate sync and coalesce repeated refresh requests.\n\nCloses #3204 * Address Plaid refresh concurrency races * Preserve Plaid refresh handoff on retry exhaustion * Release Plaid refresh lease on enqueue errors Ensure adapter exceptions cannot leave the shared refresh cooldown occupied when no job was queued.
219 lines
6.4 KiB
Ruby
219 lines
6.4 KiB
Ruby
class Provider::Plaid
|
|
attr_reader :client, :region
|
|
|
|
SUPPORTED_PLAID_PRODUCTS = %w[transactions investments liabilities].freeze
|
|
MAX_HISTORY_DAYS = Rails.env.development? ? 90 : 730
|
|
|
|
def initialize(config, region: :us)
|
|
@client = Plaid::PlaidApi.new(
|
|
Plaid::ApiClient.new(config)
|
|
)
|
|
@region = region
|
|
end
|
|
|
|
def validate_webhook!(verification_header, raw_body)
|
|
jwks_loader = ->(options) do
|
|
key_id = options[:kid]
|
|
|
|
jwk_response = client.webhook_verification_key_get(
|
|
Plaid::WebhookVerificationKeyGetRequest.new(key_id: key_id)
|
|
)
|
|
|
|
jwks = JWT::JWK::Set.new([ jwk_response.key.to_hash ])
|
|
|
|
jwks.filter! { |key| key[:use] == "sig" }
|
|
jwks
|
|
end
|
|
|
|
payload, _header = JWT.decode(
|
|
verification_header, nil, true,
|
|
{
|
|
algorithms: [ "ES256" ],
|
|
jwks: jwks_loader,
|
|
verify_expiration: false
|
|
}
|
|
)
|
|
|
|
issued_at = Time.at(payload["iat"])
|
|
raise JWT::VerificationError, "Webhook is too old" if Time.now - issued_at > 5.minutes
|
|
|
|
expected_hash = payload["request_body_sha256"]
|
|
actual_hash = Digest::SHA256.hexdigest(raw_body)
|
|
raise JWT::VerificationError, "Invalid webhook body hash" unless ActiveSupport::SecurityUtils.secure_compare(expected_hash, actual_hash)
|
|
end
|
|
|
|
def get_link_token(user_id:, webhooks_url:, redirect_url:, accountable_type: nil, access_token: nil, account_selection_enabled: false)
|
|
request_params = {
|
|
user: { client_user_id: user_id },
|
|
client_name: "Sure Finances",
|
|
country_codes: country_codes,
|
|
language: "en",
|
|
webhook: webhooks_url,
|
|
redirect_uri: redirect_url
|
|
}
|
|
|
|
if access_token.present?
|
|
request_params[:access_token] = access_token
|
|
request_params[:update] = { account_selection_enabled: true } if account_selection_enabled
|
|
else
|
|
request_params[:transactions] = { days_requested: MAX_HISTORY_DAYS }
|
|
request_params[:products] = [ get_primary_product(accountable_type) ]
|
|
request_params[:additional_consented_products] = get_additional_consented_products(accountable_type)
|
|
end
|
|
|
|
request = Plaid::LinkTokenCreateRequest.new(request_params)
|
|
|
|
client.link_token_create(request)
|
|
end
|
|
|
|
def exchange_public_token(token)
|
|
request = Plaid::ItemPublicTokenExchangeRequest.new(
|
|
public_token: token
|
|
)
|
|
|
|
client.item_public_token_exchange(request)
|
|
end
|
|
|
|
def get_item(access_token)
|
|
request = Plaid::ItemGetRequest.new(access_token: access_token)
|
|
client.item_get(request)
|
|
end
|
|
|
|
def remove_item(access_token)
|
|
request = Plaid::ItemRemoveRequest.new(access_token: access_token)
|
|
client.item_remove(request)
|
|
end
|
|
|
|
def get_item_accounts(access_token)
|
|
request = Plaid::AccountsGetRequest.new(access_token: access_token)
|
|
client.accounts_get(request)
|
|
end
|
|
|
|
def get_transactions(access_token, next_cursor: nil)
|
|
cursor = next_cursor
|
|
added = []
|
|
modified = []
|
|
removed = []
|
|
has_more = true
|
|
|
|
while has_more
|
|
request = Plaid::TransactionsSyncRequest.new(
|
|
access_token: access_token,
|
|
cursor: cursor,
|
|
options: {
|
|
include_original_description: true
|
|
}
|
|
)
|
|
|
|
response = client.transactions_sync(request)
|
|
|
|
added += response.added
|
|
modified += response.modified
|
|
removed += response.removed
|
|
has_more = response.has_more
|
|
cursor = response.next_cursor
|
|
end
|
|
|
|
TransactionSyncResponse.new(added:, modified:, removed:, cursor:)
|
|
end
|
|
|
|
def refresh_transactions(access_token)
|
|
request = Plaid::TransactionsRefreshRequest.new(access_token: access_token)
|
|
client.transactions_refresh(request)
|
|
end
|
|
|
|
def get_item_investments(access_token, start_date: nil, end_date: Date.current)
|
|
start_date = start_date || MAX_HISTORY_DAYS.days.ago.to_date
|
|
holdings, holding_securities = get_item_holdings(access_token: access_token)
|
|
transactions, transaction_securities = get_item_investment_transactions(access_token: access_token, start_date:, end_date:)
|
|
|
|
merged_securities = ((holding_securities || []) + (transaction_securities || [])).uniq { |s| s.security_id }
|
|
|
|
InvestmentsResponse.new(holdings:, transactions:, securities: merged_securities)
|
|
end
|
|
|
|
def get_item_liabilities(access_token)
|
|
request = Plaid::LiabilitiesGetRequest.new({ access_token: access_token })
|
|
response = client.liabilities_get(request)
|
|
response.liabilities
|
|
end
|
|
|
|
def get_institution(institution_id)
|
|
request = Plaid::InstitutionsGetByIdRequest.new({
|
|
institution_id: institution_id,
|
|
country_codes: country_codes,
|
|
options: {
|
|
include_optional_metadata: true
|
|
}
|
|
})
|
|
client.institutions_get_by_id(request)
|
|
end
|
|
|
|
private
|
|
TransactionSyncResponse = Struct.new :added, :modified, :removed, :cursor, keyword_init: true
|
|
InvestmentsResponse = Struct.new :holdings, :transactions, :securities, keyword_init: true
|
|
|
|
def get_item_holdings(access_token:)
|
|
request = Plaid::InvestmentsHoldingsGetRequest.new({ access_token: access_token })
|
|
response = client.investments_holdings_get(request)
|
|
|
|
[ response.holdings, response.securities ]
|
|
end
|
|
|
|
def get_item_investment_transactions(access_token:, start_date:, end_date:)
|
|
transactions = []
|
|
securities = []
|
|
offset = 0
|
|
|
|
loop do
|
|
request = Plaid::InvestmentsTransactionsGetRequest.new(
|
|
access_token: access_token,
|
|
start_date: start_date.to_s,
|
|
end_date: end_date.to_s,
|
|
options: { offset: offset }
|
|
)
|
|
|
|
response = client.investments_transactions_get(request)
|
|
|
|
transactions += response.investment_transactions
|
|
securities += response.securities
|
|
|
|
break if transactions.length >= response.total_investment_transactions
|
|
offset = transactions.length
|
|
end
|
|
|
|
[ transactions, securities ]
|
|
end
|
|
|
|
def get_primary_product(accountable_type)
|
|
return "transactions" if eu?
|
|
|
|
case accountable_type
|
|
when "Investment"
|
|
"investments"
|
|
when "CreditCard", "Loan"
|
|
"liabilities"
|
|
else
|
|
"transactions"
|
|
end
|
|
end
|
|
|
|
def get_additional_consented_products(accountable_type)
|
|
return [] if eu?
|
|
|
|
SUPPORTED_PLAID_PRODUCTS - [ get_primary_product(accountable_type) ]
|
|
end
|
|
|
|
def eu?
|
|
region.to_sym == :eu
|
|
end
|
|
|
|
def country_codes
|
|
if eu?
|
|
[ "ES", "NL", "FR", "IE", "DE", "IT", "PL", "DK", "NO", "SE", "EE", "LT", "LV", "PT", "BE" ] # EU supported countries
|
|
else
|
|
[ "US", "CA" ] # US + CA only
|
|
end
|
|
end
|
|
end
|