diff --git a/app/controllers/accounts_controller.rb b/app/controllers/accounts_controller.rb index ce2447c56..20d31811c 100644 --- a/app/controllers/accounts_controller.rb +++ b/app/controllers/accounts_controller.rb @@ -27,6 +27,7 @@ class AccountsController < ApplicationController @indexa_capital_items = visible_provider_items(family.indexa_capital_items.ordered.includes(:syncs, :indexa_capital_accounts)) @sophtron_items = visible_provider_items(family.sophtron_items.ordered.includes(:syncs, :sophtron_accounts)) @binance_items = visible_provider_items(family.binance_items.ordered.includes(:binance_accounts, :accounts, :syncs)) + @questrade_items = visible_provider_items(family.questrade_items.ordered.includes(:syncs, questrade_accounts: :account_provider)) # Build sync stats maps for all providers build_sync_stats_maps @@ -437,5 +438,18 @@ class AccountsController < ApplicationController .count @binance_unlinked_count_map[item.id] = count end + + # Questrade sync stats and account counts + @questrade_sync_stats_map = {} + @questrade_account_counts_map = {} + @questrade_items.each do |item| + latest_sync = item.syncs.ordered.first + @questrade_sync_stats_map[item.id] = latest_sync&.sync_stats || {} + accounts = item.questrade_accounts.to_a + linked = accounts.count { |a| a.account_provider.present? } + @questrade_account_counts_map[item.id] = { + linked: linked, unlinked: accounts.size - linked, total: accounts.size + } + end end end diff --git a/app/controllers/questrade_items_controller.rb b/app/controllers/questrade_items_controller.rb new file mode 100644 index 000000000..5289cec5d --- /dev/null +++ b/app/controllers/questrade_items_controller.rb @@ -0,0 +1,407 @@ +# frozen_string_literal: true + +class QuestradeItemsController < ApplicationController + ALLOWED_ACCOUNTABLE_TYPES = %w[Depository CreditCard Investment Loan OtherAsset OtherLiability Crypto Property Vehicle].freeze + + before_action :set_questrade_item, only: [ :show, :edit, :update, :destroy, :sync, :setup_accounts, :complete_account_setup ] + before_action :require_admin!, only: [ :create, :update, :destroy, :sync, :preload_accounts, :select_accounts, :link_accounts, :select_existing_account, :link_existing_account, :setup_accounts, :complete_account_setup ] + + def index + @questrade_items = Current.family.questrade_items.ordered + end + + def show + end + + def new + @questrade_item = Current.family.questrade_items.build + end + + def edit + end + + def create + @questrade_item = Current.family.questrade_items.build(questrade_item_params) + @questrade_item.name = I18n.t("questrade_items.default_name") if @questrade_item.name.blank? + + if @questrade_item.save + # Kick off an initial sync so accounts are discovered and appear under the + # Accounts tab for setup. + @questrade_item.sync_later unless @questrade_item.syncing? + + if turbo_frame_request? + flash.now[:notice] = t(".success", default: "Successfully configured Questrade.") + @questrade_items = Current.family.questrade_items.ordered + render turbo_stream: [ + turbo_stream.replace( + "questrade-providers-panel", + partial: "settings/providers/questrade_panel", + locals: { questrade_items: @questrade_items } + ), + *flash_notification_stream_items + ] + else + redirect_to settings_providers_path, notice: t(".success"), status: :see_other + end + else + @error_message = @questrade_item.errors.full_messages.join(", ") + + if turbo_frame_request? + render turbo_stream: turbo_stream.replace( + "questrade-providers-panel", + partial: "settings/providers/questrade_panel", + locals: { error_message: @error_message } + ), status: :unprocessable_entity + else + redirect_to settings_providers_path, alert: @error_message + end + end + end + + def update + update_attrs = update_params + # A fresh, non-blank token re-arms a connection that was marked requires_update. + update_attrs = update_attrs.merge(status: :good) if update_attrs[:refresh_token].present? + + if @questrade_item.update(update_attrs) + if turbo_frame_request? + flash.now[:notice] = t(".success", default: "Successfully updated Questrade configuration.") + @questrade_items = Current.family.questrade_items.ordered + render turbo_stream: [ + turbo_stream.replace( + "questrade-providers-panel", + partial: "settings/providers/questrade_panel", + locals: { questrade_items: @questrade_items } + ), + *flash_notification_stream_items + ] + else + redirect_to settings_providers_path, notice: t(".success"), status: :see_other + end + else + @error_message = @questrade_item.errors.full_messages.join(", ") + + if turbo_frame_request? + render turbo_stream: turbo_stream.replace( + "questrade-providers-panel", + partial: "settings/providers/questrade_panel", + locals: { error_message: @error_message } + ), status: :unprocessable_entity + else + redirect_to settings_providers_path, alert: @error_message + end + end + end + + def destroy + @questrade_item.destroy_later + redirect_to settings_providers_path, notice: t(".success", default: "Disconnected Questrade."), status: :see_other + end + + def sync + unless @questrade_item.syncing? + @questrade_item.sync_later + end + + respond_to do |format| + format.html { redirect_back_or_to accounts_path } + format.json { head :ok } + end + end + + # Collection actions for account linking flow + + def preload_accounts + # Trigger a sync to fetch accounts from the provider + questrade_item = Current.family.questrade_items.first + unless questrade_item&.credentials_configured? + redirect_to settings_providers_path, alert: t(".no_credentials_configured") + return + end + + questrade_item.sync_later unless questrade_item.syncing? + redirect_to select_accounts_questrade_items_path(accountable_type: params[:accountable_type], return_to: params[:return_to]) + end + + def select_accounts + questrade_item = Current.family.questrade_items.first + unless questrade_item&.credentials_configured? + if turbo_frame_request? + render partial: "questrade_items/setup_required", layout: false + else + redirect_to settings_providers_path, alert: t(".no_credentials_configured") + end + return + end + + # The account-linking UI lives in the setup_accounts view (mirrors IBKR). + redirect_to setup_accounts_questrade_item_path(questrade_item, return_to: safe_return_to_path) + end + + def link_accounts + questrade_item = Current.family.questrade_items.first + unless questrade_item&.credentials_configured? + redirect_to settings_providers_path, alert: t(".no_api_key") + return + end + + selected_ids = params[:selected_account_ids] || [] + if selected_ids.empty? + redirect_to select_accounts_questrade_items_path, alert: t(".no_accounts_selected") + return + end + + accountable_type = params[:accountable_type] || "Depository" + created_count = 0 + already_linked_count = 0 + invalid_count = 0 + + questrade_item.questrade_accounts.where(id: selected_ids).find_each do |questrade_account| + # Skip if already linked + if questrade_account.account_provider.present? + already_linked_count += 1 + next + end + + # Skip if invalid name + if questrade_account.name.blank? + invalid_count += 1 + next + end + + # Create Sure account and link + link_questrade_account(questrade_account, accountable_type) + created_count += 1 + rescue => e + DebugLogEntry.capture( + category: "provider_sync_error", + level: "error", + message: "Failed to link Questrade account", + source: "QuestradeItemsController", + provider_key: "questrade", + family: Current.family, + metadata: { questrade_account_id: questrade_account.id, error_class: e.class.name, error_message: e.message } + ) + end + + if created_count > 0 + questrade_item.sync_later unless questrade_item.syncing? + redirect_to accounts_path, notice: t(".success", count: created_count) + else + redirect_to select_accounts_questrade_items_path, alert: t(".link_failed") + end + end + + def select_existing_account + @account = find_writable_account!(params[:account_id]) + @questrade_item = Current.family.questrade_items.first + + unless @questrade_item&.credentials_configured? + if turbo_frame_request? + render partial: "questrade_items/setup_required", layout: false + else + redirect_to settings_providers_path, alert: t(".no_credentials_configured") + end + return + end + + @questrade_accounts = @questrade_item.questrade_accounts + .left_joins(:account_provider) + .where(account_providers: { id: nil }) + .order(:name) + end + + def link_existing_account + account = find_writable_account!(params[:account_id]) + questrade_item = Current.family.questrade_items.first + + unless questrade_item&.credentials_configured? + redirect_to settings_providers_path, alert: t(".no_api_key") + return + end + + questrade_account = questrade_item.questrade_accounts.find(params[:questrade_account_id]) + + if questrade_account.account_provider.present? + redirect_to account_path(account), alert: t(".provider_account_already_linked") + return + end + + questrade_account.ensure_account_provider!(account) + questrade_item.sync_later unless questrade_item.syncing? + + redirect_to account_path(account), notice: t(".success", account_name: account.name) + end + + def setup_accounts + @unlinked_accounts = @questrade_item.unlinked_questrade_accounts.order(:name) + # When empty, the view renders an "all linked" message inside the modal + # frame; redirecting to accounts_path here would dead-end the Turbo modal. + end + + def complete_account_setup + account_configs = params[:accounts] || {} + + if account_configs.empty? + redirect_to setup_accounts_questrade_item_path(@questrade_item), alert: t(".no_accounts") + return + end + + created_count = 0 + skipped_count = 0 + + account_configs.each do |questrade_account_id, config| + next if config[:account_type] == "skip" + + questrade_account = @questrade_item.questrade_accounts.find_by(id: questrade_account_id) + next unless questrade_account + next if questrade_account.account_provider.present? + + accountable_type = infer_accountable_type(config[:account_type], config[:subtype]) + + # Atomic: roll back the manual account if linking the provider fails. + ActiveRecord::Base.transaction do + account = create_account_from_questrade(questrade_account, accountable_type, config) + questrade_account.ensure_account_provider!(account) + questrade_account.update!(sync_start_date: config[:sync_start_date]) if config[:sync_start_date].present? + end + created_count += 1 + rescue => e + DebugLogEntry.capture( + category: "provider_sync_error", + level: "error", + message: "Failed to create account during Questrade setup", + source: "QuestradeItemsController", + provider_key: "questrade", + family: Current.family, + metadata: { questrade_account_id: questrade_account_id, error_class: e.class.name, error_message: e.message } + ) + skipped_count += 1 + end + + return_to = safe_return_to_path + + if created_count > 0 + @questrade_item.sync_later unless @questrade_item.syncing? + redirect_to return_to || accounts_path, notice: t(".success", count: created_count) + elsif skipped_count > 0 && created_count == 0 + redirect_to return_to || accounts_path, notice: t(".all_skipped") + else + redirect_to setup_accounts_questrade_item_path(@questrade_item, return_to: return_to), alert: t(".creation_failed", error: "Unknown error") + end + end + + private + + def set_questrade_item + @questrade_item = Current.family.questrade_items.find(params[:id]) + end + + # Mirror AccountsController's access gate: only accounts the user can reach + # and write to may be inspected or linked to a provider. + def find_writable_account!(account_id) + account = Current.user.accessible_accounts.find(account_id) + raise ActiveRecord::RecordNotFound unless account.permission_for(Current.user).in?([ :owner, :full_control ]) + account + end + + def questrade_item_params + params.require(:questrade_item).permit( + :name, + :sync_start_date, + :refresh_token + ) + end + + # Params for update: drop a blank refresh_token so an empty submission + # never wipes the stored (still-valid) token. + def update_params + permitted = questrade_item_params + permitted = permitted.except(:refresh_token) if permitted[:refresh_token].blank? + permitted + end + + def link_questrade_account(questrade_account, accountable_type) + accountable_class = validated_accountable_class(accountable_type) + + # Atomic: a failure in ensure_account_provider! must roll back the account + # so we never leave an orphan manual account behind. + ActiveRecord::Base.transaction do + account = Current.family.accounts.create!( + name: questrade_account.name, + balance: questrade_account.current_balance || 0, + currency: questrade_account.currency || "USD", + accountable: accountable_class.new + ) + + questrade_account.ensure_account_provider!(account) + account + end + end + + def create_account_from_questrade(questrade_account, accountable_type, config) + accountable_class = validated_accountable_class(accountable_type) + accountable_attrs = {} + + # Set subtype if the accountable supports it + if config[:subtype].present? && accountable_class.respond_to?(:subtypes) + accountable_attrs[:subtype] = config[:subtype] + end + + Current.family.accounts.create!( + name: questrade_account.name, + balance: config[:balance].present? ? config[:balance].to_d : (questrade_account.current_balance || 0), + currency: questrade_account.currency || "USD", + accountable: accountable_class.new(accountable_attrs) + ) + end + + def infer_accountable_type(account_type, subtype = nil) + case account_type&.downcase + when "depository" + "Depository" + when "credit_card" + "CreditCard" + when "investment" + "Investment" + when "loan" + "Loan" + when "other_asset" + "OtherAsset" + when "other_liability" + "OtherLiability" + when "crypto" + "Crypto" + when "property" + "Property" + when "vehicle" + "Vehicle" + else + "Depository" + end + end + + def validated_accountable_class(accountable_type) + unless ALLOWED_ACCOUNTABLE_TYPES.include?(accountable_type) + raise ArgumentError, "Invalid accountable type: #{accountable_type}" + end + + accountable_type.constantize + end + + def safe_return_to_path + return nil if params[:return_to].blank? + + return_to = params[:return_to].to_s + + begin + uri = URI.parse(return_to) + return nil if uri.scheme.present? + return nil if uri.host.present? + return nil unless return_to.start_with?("/") + return_to + rescue URI::InvalidURIError + nil + end + end +end diff --git a/app/controllers/settings/providers_controller.rb b/app/controllers/settings/providers_controller.rb index 8af7dd0ce..5fdb82657 100644 --- a/app/controllers/settings/providers_controller.rb +++ b/app/controllers/settings/providers_controller.rb @@ -196,7 +196,8 @@ class Settings::ProvidersController < ApplicationController { key: "snaptrade", title: "SnapTrade", turbo_id: "snaptrade", partial: "snaptrade_panel", auto_open: "manage" }, { key: "ibkr", title: "Interactive Brokers", turbo_id: "ibkr", partial: "ibkr_panel" }, { key: "indexa_capital", title: "Indexa Capital", turbo_id: "indexa_capital", partial: "indexa_capital_panel" }, - { key: "sophtron", title: "Sophtron", turbo_id: "sophtron", partial: "sophtron_panel" } + { key: "sophtron", title: "Sophtron", turbo_id: "sophtron", partial: "sophtron_panel" }, + { key: "questrade", title: "Questrade", turbo_id: "questrade", partial: "questrade_panel" } ].freeze FAMILY_PANEL_KEYS = FAMILY_PANELS.map { |p| p[:key] }.freeze @@ -215,6 +216,7 @@ class Settings::ProvidersController < ApplicationController "binance" => "BinanceItem", "kraken" => "KrakenItem", "snaptrade" => "SnaptradeItem", + "questrade" => "QuestradeItem", "ibkr" => "IbkrItem", "indexa_capital" => "IndexaCapitalItem", "sophtron" => "SophtronItem" @@ -252,6 +254,8 @@ class Settings::ProvidersController < ApplicationController @indexa_capital_items = Current.family.indexa_capital_items.ordered when "sophtron" @sophtron_items = Current.family.sophtron_items.ordered + when "questrade" + @questrade_items = Current.family.questrade_items.active.ordered end end @@ -280,6 +284,7 @@ class Settings::ProvidersController < ApplicationController @indexa_capital_items = Current.family.indexa_capital_items.ordered.select(:id) @binance_items = Current.family.binance_items.active.ordered @kraken_items = Current.family.kraken_items.active.ordered + @questrade_items = Current.family.questrade_items.active.ordered.select(:id) @provider_sync_health = compute_provider_sync_health(family_panel_items) @@ -309,6 +314,7 @@ class Settings::ProvidersController < ApplicationController "binance" => @binance_items, "kraken" => @kraken_items, "snaptrade" => @snaptrade_items, + "questrade" => @questrade_items, "ibkr" => @ibkr_items, "indexa_capital" => @indexa_capital_items, "sophtron" => @sophtron_items diff --git a/app/helpers/settings_helper.rb b/app/helpers/settings_helper.rb index b88499aae..155f8ec92 100644 --- a/app/helpers/settings_helper.rb +++ b/app/helpers/settings_helper.rb @@ -107,6 +107,9 @@ module SettingsHelper when "sophtron" return { status: :off } unless @sophtron_items&.any? sync_based_summary(key) + when "questrade" + return { status: :off } unless @questrade_items&.any? + sync_based_summary(key) else { status: :off } end diff --git a/app/jobs/questrade_activities_fetch_job.rb b/app/jobs/questrade_activities_fetch_job.rb new file mode 100644 index 000000000..26337a424 --- /dev/null +++ b/app/jobs/questrade_activities_fetch_job.rb @@ -0,0 +1,131 @@ +# frozen_string_literal: true + +class QuestradeActivitiesFetchJob < ApplicationJob + include QuestradeAccount::DataHelpers + + queue_as :default + + MAX_RETRIES = 6 + RETRY_INTERVAL = 10.seconds + + sidekiq_options lock: :until_executed, + lock_args_method: ->(args) { args.first }, + on_conflict: :log + + def perform(questrade_account, start_date: nil, retry_count: 0) + @questrade_account = questrade_account + @start_date = start_date || 3.years.ago.to_date + @retry_count = retry_count + + return clear_pending_flag unless valid_for_fetch? + + fetch_and_process_activities + rescue => e + Rails.logger.error("QuestradeActivitiesFetchJob error: #{e.class} - #{e.message}") + clear_pending_flag + raise + end + + private + + def valid_for_fetch? + return false unless @questrade_account + return false unless @questrade_account.questrade_item + return false unless @questrade_account.current_account + true + end + + def fetch_and_process_activities + activities = fetch_activities + + if activities.blank? && @retry_count < MAX_RETRIES + schedule_retry + return + end + + if activities.any? + merged = merge_activities(existing_activities, activities) + @questrade_account.upsert_activities_snapshot!(merged) + QuestradeAccount::ActivitiesProcessor.new(@questrade_account).process + end + + # Always record the fetch as completed (even for legitimately empty + # accounts) so the importer's fresh-account check stops re-queueing this. + @questrade_account.update!(last_activities_sync: Time.current) + clear_pending_flag + broadcast_updates + end + + def fetch_activities + provider = @questrade_account.questrade_item.questrade_provider + return [] unless provider + + response = provider.get_activities( + account_id: @questrade_account.questrade_account_id, + start_date: @start_date, + end_date: Date.current + ) + Array(response.is_a?(Hash) ? response[:activities] : response) + rescue Provider::Questrade::AuthenticationError + # Re-raise auth errors - they need immediate attention + raise + rescue => e + # Transient errors trigger retry via blank response + Rails.logger.error("QuestradeActivitiesFetchJob - API error: #{e.message}") + [] + end + + def existing_activities + @questrade_account.raw_activities_payload || [] + end + + def merge_activities(existing, new_activities) + by_id = {} + existing.each { |a| by_id[activity_key(a)] = a } + new_activities.each do |a| + activity_hash = sdk_object_to_hash(a) + by_id[activity_key(activity_hash)] = activity_hash + end + by_id.values + end + + def activity_key(activity) + activity = activity.with_indifferent_access if activity.is_a?(Hash) + # Questrade activities have no id; key on the immutable fields (same basis + # as QuestradeItem::Importer#activity_key) to dedup across syncs. + [ activity[:transactionDate], activity[:action], activity[:symbolId], + activity[:netAmount], activity[:description], activity[:currency], activity[:type] ].join("-") + end + + def schedule_retry + Rails.logger.info( + "QuestradeActivitiesFetchJob - No activities found, scheduling retry " \ + "#{@retry_count + 1}/#{MAX_RETRIES} in #{RETRY_INTERVAL.to_i}s" + ) + + self.class.set(wait: RETRY_INTERVAL).perform_later( + @questrade_account, + start_date: @start_date, + retry_count: @retry_count + 1 + ) + end + + def clear_pending_flag + @questrade_account.update!(activities_fetch_pending: false) + rescue => e + # Best-effort: never let clearing the flag mask the original error in + # perform's rescue (which then re-raises). + Rails.logger.warn("QuestradeActivitiesFetchJob - failed to clear pending flag: #{e.message}") + end + + def broadcast_updates + @questrade_account.current_account&.broadcast_sync_complete + @questrade_account.questrade_item&.broadcast_replace_to( + @questrade_account.questrade_item.family, + target: "questrade_item_#{@questrade_account.questrade_item.id}", + partial: "questrade_items/questrade_item" + ) + rescue => e + Rails.logger.warn("QuestradeActivitiesFetchJob - Broadcast failed: #{e.message}") + end +end diff --git a/app/models/data_enrichment.rb b/app/models/data_enrichment.rb index 59de42251..781d9a6dd 100644 --- a/app/models/data_enrichment.rb +++ b/app/models/data_enrichment.rb @@ -16,6 +16,7 @@ class DataEnrichment < ApplicationRecord brex: "brex", indexa_capital: "indexa_capital", sophtron: "sophtron", - ibkr: "ibkr" + ibkr: "ibkr", + questrade: "questrade" } end diff --git a/app/models/family.rb b/app/models/family.rb index 908fa5702..0645c6670 100644 --- a/app/models/family.rb +++ b/app/models/family.rb @@ -4,6 +4,7 @@ class Family < ApplicationRecord include CoinbaseConnectable, BinanceConnectable, KrakenConnectable, CoinstatsConnectable, SnaptradeConnectable, MercuryConnectable, BrexConnectable, SophtronConnectable include IndexaCapitalConnectable, IbkrConnectable include UpConnectable + include QuestradeConnectable DATE_FORMATS = [ [ "MM-DD-YYYY", "%m-%d-%Y" ], diff --git a/app/models/family/financial_data_reset.rb b/app/models/family/financial_data_reset.rb index 75141a5e4..dbd971627 100644 --- a/app/models/family/financial_data_reset.rb +++ b/app/models/family/financial_data_reset.rb @@ -48,6 +48,7 @@ class Family::FinancialDataReset ibkr_items indexa_capital_items kraken_items + questrade_items lunchflow_items mercury_items plaid_items diff --git a/app/models/family/questrade_connectable.rb b/app/models/family/questrade_connectable.rb new file mode 100644 index 000000000..1a337b690 --- /dev/null +++ b/app/models/family/questrade_connectable.rb @@ -0,0 +1,28 @@ +module Family::QuestradeConnectable + extend ActiveSupport::Concern + + included do + has_many :questrade_items, dependent: :destroy + end + + def can_connect_questrade? + # Families can configure their own Questrade credentials + true + end + + def create_questrade_item!(refresh_token:, api_server: nil, item_name: nil) + questrade_item = questrade_items.create!( + name: item_name || "Questrade Connection", + refresh_token: refresh_token, + api_server: api_server + ) + + questrade_item.sync_later + + questrade_item + end + + def has_questrade_credentials? + questrade_items.where.not(refresh_token: nil).exists? + end +end diff --git a/app/models/provider/metadata.rb b/app/models/provider/metadata.rb index 21d9436bf..384547b77 100644 --- a/app/models/provider/metadata.rb +++ b/app/models/provider/metadata.rb @@ -17,7 +17,8 @@ class Provider indexa_capital: { region: "ES", kinds: %w[Investment], maturity: :alpha, logo_text: "IC", logo_bg: "bg-red-600" }, sophtron: { region: "US", kinds: %w[Bank Investment], maturity: :alpha, logo_text: "SO", logo_bg: "bg-teal-600" }, plaid: { region: "US", kinds: %w[Bank], maturity: :stable, logo_text: "PL", logo_bg: "bg-indigo-600", tier: "Paid" }, - plaid_eu: { region: "EU", kinds: %w[Bank], maturity: :stable, logo_text: "PL", logo_bg: "bg-indigo-600", tier: "Paid", name: "Plaid EU" } + plaid_eu: { region: "EU", kinds: %w[Bank], maturity: :stable, logo_text: "PL", logo_bg: "bg-indigo-600", tier: "Paid", name: "Plaid EU" }, + questrade: { region: "CA", kinds: %w[Investment], maturity: :beta, logo_text: "QT", logo_bg: "bg-teal-600" } }.freeze def self.for(provider_key) diff --git a/app/models/provider/questrade.rb b/app/models/provider/questrade.rb new file mode 100644 index 000000000..8bd605962 --- /dev/null +++ b/app/models/provider/questrade.rb @@ -0,0 +1,268 @@ +# frozen_string_literal: true + +# Questrade API client. +# +# Auth model (the important part): Questrade uses single-use, rotating refresh +# tokens that EXPIRE 7 DAYS after generation. Exchanging a refresh token returns: +# - access_token (Bearer, ~30 min TTL) +# - api_server (the base URL you must use for all data calls) +# - refresh_token (a BRAND NEW one — the old one is now dead) +# +# Because the refresh token is single-use, the new one MUST be persisted +# immediately and the exchange MUST be serialized (no two syncs refreshing at +# once). This SDK performs the exchange and hands the new credentials back to +# the caller via the `on_token_refresh` callback so the item model can persist +# them inside its own row lock / transaction. +class Provider::Questrade + include HTTParty + + headers "User-Agent" => "Sure Finance Questrade Client" + default_options.merge!(verify: true, ssl_verify_mode: OpenSSL::SSL::VERIFY_PEER, timeout: 120) + + LOGIN_URL = "https://login.questrade.com/oauth2/token" + # Questrade STRICTLY caps activity ranges at 31 days (err 1003). Used as an + # inclusive day count, so a single window spans at most 30 days. + MAX_ACTIVITY_DAYS = 30 + ACCESS_TOKEN_SKEW = 60 # refresh slightly early to avoid mid-call expiry + + class Error < StandardError + attr_reader :error_type + + def initialize(message, error_type = :unknown) + super(message) + @error_type = error_type + end + end + + class ConfigurationError < Error; end + class AuthenticationError < Error; end + class RetryableResponseError < Error; end + + attr_reader :refresh_token, :api_server + + # @param refresh_token [String] current (single-use) refresh token + # @param api_server [String, nil] cached base URL from the last exchange + # @param on_token_refresh [#call] called with the new credentials hash + # { refresh_token:, api_server:, access_token:, expires_at: } so the + # caller can persist them. REQUIRED for durable operation. + def initialize(refresh_token:, api_server: nil, on_token_refresh: nil, synchronize_exchange: nil) + @refresh_token = refresh_token + @api_server = api_server + @on_token_refresh = on_token_refresh + @synchronize_exchange = synchronize_exchange + @access_token = nil + @access_expires_at = nil + validate_configuration! + end + + # GET /v1/accounts -> { accounts: [...], userId: ... } + def list_accounts + get_json("v1/accounts") + end + + # GET /v1/accounts/:id/positions (Sure calls these "holdings") + def get_holdings(account_id:) + get_json("v1/accounts/#{account_id}/positions") + end + + # GET /v1/accounts/:id/balances + def get_balances(account_id:) + get_json("v1/accounts/#{account_id}/balances") + end + + # GET /v1/symbols?ids=1,2,3 -> currency/description per symbol. Questrade's + # positions endpoint omits currency, so we use this to tag USD vs CAD holdings. + def get_symbols(ids:) + ids = Array(ids).compact.uniq.join(",") + return { symbols: [] } if ids.blank? + + get_json("v1/symbols", query: { ids: ids }) + end + + # GET /v1/accounts/:id/activities?startTime=&endTime= + # Chunked into <=30-day windows because Questrade rejects ranges > 31 days. + def get_activities(account_id:, start_date:, end_date: Date.current) + activities = [] + window_start = start_date.to_date + end_date = end_date.to_date + + while window_start <= end_date + # (MAX - 1) because the range is inclusive of both endpoints; this keeps the + # span strictly under Questrade's 31-day ceiling even after UTC conversion. + window_end = [ window_start + (MAX_ACTIVITY_DAYS - 1), end_date ].min + page = get_json( + "v1/accounts/#{account_id}/activities", + query: { startTime: iso(window_start.beginning_of_day), + endTime: iso(window_end.end_of_day) } + ) + activities.concat(Array(page[:activities])) + window_start = window_end + 1 + end + + { activities: activities } + end + + private + + RETRYABLE_ERRORS = [ + SocketError, Net::OpenTimeout, Net::ReadTimeout, + Errno::ECONNRESET, Errno::ECONNREFUSED, Errno::ETIMEDOUT, EOFError + ].freeze + + MAX_RETRIES = 3 + INITIAL_RETRY_DELAY = 2 # seconds + + def validate_configuration! + raise ConfigurationError.new("Refresh token is required", :missing_credentials) if @refresh_token.blank? + end + + def get_json(path, query: {}) + ensure_authenticated! + with_retries(path) do + response = self.class.get("#{api_base}#{path}", headers: auth_headers, query: query) + # Access token can expire mid-sync; refresh once and retry on 401. + if response.code == 401 + authenticate!(force: true) + response = self.class.get("#{api_base}#{path}", headers: auth_headers, query: query) + end + handle_response(response) + end + end + + # Exchange the refresh token unless we already hold a valid access token. + def ensure_authenticated! + authenticate! if @access_token.nil? || @access_expires_at.nil? || Time.current >= @access_expires_at + end + + def authenticate!(force: false) + return if @access_token && !force && Time.current < @access_expires_at + + # Spending a single-use refresh token must be serialized across workers + # and must use the freshest persisted token. The caller supplies a lock + # that yields the current token; without one, exchange directly. + if @synchronize_exchange + @synchronize_exchange.call do |fresh_token| + @refresh_token = fresh_token if fresh_token.present? + exchange_token! + end + else + exchange_token! + end + end + + def exchange_token! + response = with_retries("oauth_token") do + # POST with a form body keeps the single-use refresh token out of the + # URL (and therefore out of access logs / error-tracking breadcrumbs). + self.class.post(LOGIN_URL, body: { grant_type: "refresh_token", refresh_token: @refresh_token }) + end + + unless response.code == 200 + # 400/401 here usually means the refresh token expired (>7 days) or was + # already used. The connection must be re-authorized by the user. + raise AuthenticationError.new( + "Questrade token exchange failed (#{response.code}). Re-authorization required.", + :reauth_required + ) + end + + body = JSON.parse(response.body, symbolize_names: true) + @access_token = body[:access_token] + @api_server = body[:api_server] + @refresh_token = body[:refresh_token] # rotate in-memory immediately + @access_expires_at = Time.current + (body[:expires_in].to_i - ACCESS_TOKEN_SKEW).seconds + + # Hand the new credentials to the caller to persist (single-use token!). + @on_token_refresh&.call( + refresh_token: @refresh_token, + api_server: @api_server, + access_token: @access_token, + expires_at: @access_expires_at + ) + end + + def api_base + raise ConfigurationError.new("No api_server; authenticate first", :missing_api_server) if @api_server.blank? + @api_server.end_with?("/") ? @api_server : "#{@api_server}/" + end + + def auth_headers + { + "Authorization" => "Bearer #{@access_token}", + "Accept" => "application/json" + } + end + + def iso(time) + time.utc.iso8601 + end + + def with_retries(operation_name, max_retries: MAX_RETRIES) + retries = 0 + + begin + yield + rescue *RETRYABLE_ERRORS, RetryableResponseError => e + retries += 1 + + if retries <= max_retries + delay = calculate_retry_delay(retries) + Rails.logger.warn( + "Questrade API: #{operation_name} failed (attempt #{retries}/#{max_retries}): " \ + "#{e.class}: #{e.message}. Retrying in #{delay}s..." + ) + sleep(delay) + retry + else + Rails.logger.error( + "Questrade API: #{operation_name} failed after #{max_retries} retries: " \ + "#{e.class}: #{e.message}" + ) + raise Error.new("Network error after #{max_retries} retries: #{e.message}", :network_error) + end + end + end + + def calculate_retry_delay(retry_count) + base_delay = INITIAL_RETRY_DELAY * (2 ** (retry_count - 1)) + jitter = base_delay * rand * 0.25 + [ base_delay + jitter, 30 ].min + end + + # Record provider failure detail to the super-admin /settings/debug log + # (the sanctioned channel) instead of echoing the raw response body into + # application logs or exception messages, where the importer re-logs it. + def capture_response_error(reason, response) + DebugLogEntry.capture( + category: "provider_sync", + level: "error", + message: "Questrade API #{reason} (#{response.code})", + source: self.class.name, + provider_key: "questrade", + metadata: { status: response.code, body: response.body.to_s.first(1000) } + ) + end + + def handle_response(response) + case response.code + when 200, 201 + JSON.parse(response.body, symbolize_names: true) + when 400 + capture_response_error("bad_request", response) + raise Error.new("Questrade bad request (#{response.code})", :bad_request) + when 401 + raise AuthenticationError.new("Invalid or expired Questrade credentials", :unauthorized) + when 403 + raise AuthenticationError.new("Access forbidden - check your permissions", :access_forbidden) + when 404 + raise Error.new("Resource not found", :not_found) + when 429 + raise RetryableResponseError.new("Questrade rate limit exceeded. Please try again later.", :rate_limited) + when 500..599 + raise RetryableResponseError.new("Questrade server error (#{response.code}). Please try again later.", :server_error) + else + capture_response_error("unexpected_response", response) + raise Error.new("Questrade unexpected response (#{response.code})", :unknown) + end + end +end diff --git a/app/models/provider/questrade_adapter.rb b/app/models/provider/questrade_adapter.rb new file mode 100644 index 000000000..e4f5fa708 --- /dev/null +++ b/app/models/provider/questrade_adapter.rb @@ -0,0 +1,103 @@ +class Provider::QuestradeAdapter < Provider::Base + include Provider::Syncable + include Provider::InstitutionMetadata + + # Register this adapter with the factory + Provider::Factory.register("QuestradeAccount", self) + + # Define which account types this provider supports + def self.supported_account_types + # Questrade is a stock/ETF/options brokerage, so it only links to + # Investment accounts (not Crypto). + %w[Investment] + end + + # Returns connection configurations for this provider + def self.connection_configs(family:) + return [] unless family.can_connect_questrade? + + [ { + key: "questrade", + name: "Questrade", + description: "Connect to your brokerage via Questrade", + can_connect: true, + new_account_path: ->(accountable_type, return_to) { + Rails.application.routes.url_helpers.select_accounts_questrade_items_path( + accountable_type: accountable_type, + return_to: return_to + ) + }, + existing_account_path: ->(account_id) { + Rails.application.routes.url_helpers.select_existing_account_questrade_items_path( + account_id: account_id + ) + } + } ] + end + + def provider_name + "questrade" + end + + # Build a Questrade provider instance with family-specific credentials + # @param family [Family] The family to get credentials for (required) + # @return [Provider::Questrade, nil] Returns nil if credentials are not configured + def self.build_provider(family: nil) + return nil unless family.present? + + # Get family-specific credentials + questrade_item = family.questrade_items.where.not(refresh_token: nil).first + return nil unless questrade_item&.credentials_configured? + + questrade_item.questrade_provider + end + + def sync_path + Rails.application.routes.url_helpers.sync_questrade_item_path(item) + end + + def item + provider_account.questrade_item + end + + def can_delete_holdings? + false + end + + def institution_domain + metadata = provider_account.institution_metadata + return nil unless metadata.present? + + domain = metadata["domain"] + url = metadata["url"] + + # Derive domain from URL if missing + if domain.blank? && url.present? + begin + domain = URI.parse(url).host&.gsub(/^www\./, "") + rescue URI::InvalidURIError + Rails.logger.warn("Invalid institution URL for Questrade account #{provider_account.id}: #{url}") + end + end + + domain + end + + def institution_name + metadata = provider_account.institution_metadata + return nil unless metadata.present? + + metadata["name"] || item&.institution_name + end + + def institution_url + metadata = provider_account.institution_metadata + return nil unless metadata.present? + + metadata["url"] || item&.institution_url + end + + def institution_color + item&.institution_color + end +end diff --git a/app/models/provider_connection_status.rb b/app/models/provider_connection_status.rb index 298ea538f..3b5e54ccc 100644 --- a/app/models/provider_connection_status.rb +++ b/app/models/provider_connection_status.rb @@ -17,7 +17,8 @@ class ProviderConnectionStatus { key: "mercury", type: "MercuryItem", association: :mercury_items, accounts: :mercury_accounts }, { key: "brex", type: "BrexItem", association: :brex_items, accounts: :brex_accounts }, { key: "sophtron", type: "SophtronItem", association: :sophtron_items, accounts: :sophtron_accounts }, - { key: "indexa_capital", type: "IndexaCapitalItem", association: :indexa_capital_items, accounts: :indexa_capital_accounts } + { key: "indexa_capital", type: "IndexaCapitalItem", association: :indexa_capital_items, accounts: :indexa_capital_accounts }, + { key: "questrade", type: "QuestradeItem", association: :questrade_items, accounts: :questrade_accounts } ].freeze class << self diff --git a/app/models/provider_merchant.rb b/app/models/provider_merchant.rb index 719c6800f..54515e5bb 100644 --- a/app/models/provider_merchant.rb +++ b/app/models/provider_merchant.rb @@ -1,5 +1,5 @@ class ProviderMerchant < Merchant - enum :source, { plaid: "plaid", simplefin: "simplefin", lunchflow: "lunchflow", akahu: "akahu", up: "up", synth: "synth", ai: "ai", enable_banking: "enable_banking", coinstats: "coinstats", mercury: "mercury", brex: "brex", indexa_capital: "indexa_capital", sophtron: "sophtron" } + enum :source, { plaid: "plaid", simplefin: "simplefin", lunchflow: "lunchflow", akahu: "akahu", up: "up", synth: "synth", ai: "ai", enable_banking: "enable_banking", coinstats: "coinstats", mercury: "mercury", brex: "brex", indexa_capital: "indexa_capital", sophtron: "sophtron", questrade: "questrade" } validates :name, uniqueness: { scope: [ :source ] } validates :source, presence: true diff --git a/app/models/questrade_account.rb b/app/models/questrade_account.rb new file mode 100644 index 000000000..c6523a4eb --- /dev/null +++ b/app/models/questrade_account.rb @@ -0,0 +1,153 @@ +# frozen_string_literal: true + +class QuestradeAccount < ApplicationRecord + include CurrencyNormalizable, Encryptable + include QuestradeAccount::DataHelpers + + if encryption_ready? + encrypts :raw_payload + encrypts :raw_holdings_payload + encrypts :raw_activities_payload + encrypts :raw_balances_payload + end + + belongs_to :questrade_item + + # Association through account_providers + has_one :account_provider, as: :provider, dependent: :destroy + has_one :account, through: :account_provider, source: :account + has_one :linked_account, through: :account_provider, source: :account + + validates :name, :currency, presence: true + + # Scopes + scope :with_linked, -> { joins(:account_provider) } + scope :without_linked, -> { left_joins(:account_provider).where(account_providers: { id: nil }) } + scope :ordered, -> { order(created_at: :desc) } + + + # Helper to get account using account_providers system + def current_account + account + end + + # Idempotently create or update AccountProvider link + # CRITICAL: After creation, reload association to avoid stale nil + def ensure_account_provider!(linked_account) + return nil unless linked_account + + provider = account_provider || build_account_provider + provider.account = linked_account + provider.save! + + # Reload to clear cached nil value + reload_account_provider + account_provider + end + + def upsert_from_questrade!(account_data) + # Convert SDK object to hash if needed + data = sdk_object_to_hash(account_data).with_indifferent_access + + # Questrade's list_accounts returns: number, type (TFSA/RRSP/Margin/Cash...), + # status, clientAccountType, isPrimary, isBilling. There's no friendly name, + # so we build one from the type + number. + number = (data[:number] || data[:id]).to_s + type = data[:type].presence || "Account" + + update!( + questrade_account_id: number, + name: "#{type} (#{number})", + # Default; the real home currency is inferred from per-currency balances + # in upsert_balances! (Questrade exposes no account-level currency field). + currency: "CAD", + account_status: data[:status], + account_type: type, + provider: "Questrade", + institution_metadata: { name: "Questrade", domain: "questrade.com" }, + raw_payload: data + ) + end + + # Store holdings snapshot - return early if empty to avoid setting timestamps incorrectly + def upsert_holdings_snapshot!(holdings_data) + return if holdings_data.blank? + + update!( + raw_holdings_payload: holdings_data, + last_holdings_sync: Time.current + ) + end + + # Store activities snapshot - return early if empty to avoid setting timestamps incorrectly + def upsert_activities_snapshot!(activities_data) + return if activities_data.blank? + + update!( + raw_activities_payload: activities_data, + last_activities_sync: Time.current + ) + end + + # Store per-currency balances. Primary (account currency) cash goes in + # cash_balance; the full set is kept so the processor can surface + # non-primary-currency cash as holdings (issue #1809). + def upsert_balances!(per_currency_balances) + data = Array(per_currency_balances).map { |b| sdk_object_to_hash(b).with_indifferent_access } + + # Questrade has no account-level currency field, so infer the home currency + # from the per-currency balances (the currency holding the cash wins). This + # keeps genuinely USD-denominated accounts from being mislabelled CAD. + self.currency = primary_balance_currency(data) if data.any? + + primary = primary_cash_entry(data) + cash_value = primary ? primary[:cash] : 0 + update!(currency: currency, cash_balance: cash_value, raw_balances_payload: data) + end + + # Cash held in currencies other than the account's primary currency, surfaced + # as synthetic cash holdings (issue #1809). Primary cash lives in cash_balance. + def non_primary_cash_entries + entries = Array(raw_balances_payload).map do |e| + e.respond_to?(:with_indifferent_access) ? e.with_indifferent_access : {} + end + entries.filter_map do |e| + code = e[:currency] + next if code.blank? || code == currency + amount = e[:cash] + next if amount.blank? || amount.to_d.abs < BigDecimal("0.01") + { currency: code, amount: amount } + end + end + + private + + # Infer the account home currency from per-currency balances: the currency + # holding the most cash wins (ties broken by total equity), default CAD. + def primary_balance_currency(entries) + ranked = entries.max_by do |b| + [ b[:cash].to_f.abs, (b[:totalEquity] || b[:marketValue]).to_f.abs ] + end + ranked&.dig(:currency).presence || "CAD" + end + + # Primary cash entry: account currency first, then USD, then first entry. + def primary_cash_entry(entries) + entries = entries.map { |e| e.respond_to?(:with_indifferent_access) ? e.with_indifferent_access : {} } + # Only the account-currency (CAD) entry is primary; other currencies + # surface as separate cash holdings. + entries.find { |b| b[:currency] == currency } + end + + def extract_institution_metadata(data) + { + name: data[:institution_name] || data.dig(:institution, :name), + logo: data[:institution_logo] || data.dig(:institution, :logo), + domain: data[:institution_domain] || data.dig(:institution, :domain) + }.compact + end + + def log_invalid_currency(currency_value) + Rails.logger.warn("Invalid currency code '#{currency_value}' for Questrade account #{id}, defaulting to USD") + end +end diff --git a/app/models/questrade_account/activities_processor.rb b/app/models/questrade_account/activities_processor.rb new file mode 100644 index 000000000..edddc3b34 --- /dev/null +++ b/app/models/questrade_account/activities_processor.rb @@ -0,0 +1,212 @@ +# frozen_string_literal: true + +class QuestradeAccount::ActivitiesProcessor + include QuestradeAccount::DataHelpers + + # Questrade groups activities by `type` (category) and `action` (sub-action, + # e.g. Buy/Sell/CON/FCH). We route on `type`, then use `action` for direction. + # + # Sign convention: Questrade `netAmount` is +money-in / -money-out, but Sure + # stores transactions as -inflow / +outflow, so cash signed_amount = -netAmount. + TRADE_TYPE = "Trades" + + # Questrade `type` -> Sure investment activity label (cash transactions) + CASH_TYPE_TO_LABEL = { + "Deposits" => "Contribution", + "Withdrawals" => "Withdrawal", + "Dividends" => "Dividend", + "Interest" => "Interest", + "Fees and rebates" => "Fee" + }.freeze + + # Still unmapped: FX conversions (cash currency exchanges) and corporate + # actions. Skipped with a log rather than imported wrong. + UNSUPPORTED_TYPES = [ "FX conversion", "Corporate actions" ].freeze + + # Norbert's Gambit / in-kind transfers: shares journaled between symbols or + # currencies with no cash impact. Recorded as zero-cost "Transfer" trades. + JOURNAL_TYPES = [ "Other", "Transfers" ].freeze + + def initialize(questrade_account) + @questrade_account = questrade_account + end + + def process + return { trades: 0, transactions: 0 } unless account.present? + + @trades_count = 0 + @transactions_count = 0 + + activities.each do |raw| + process_activity(raw.with_indifferent_access) + rescue => e + Rails.logger.error "QuestradeAccount::ActivitiesProcessor - Failed to process activity: #{e.message}" + Rails.logger.error e.backtrace.first(5).join("\n") if e.backtrace + end + + { trades: @trades_count, transactions: @transactions_count } + end + + private + + def account + @questrade_account.current_account + end + + def import_adapter + @import_adapter ||= Account::ProviderImportAdapter.new(account) + end + + # raw_activities_payload may be the array itself or the { activities: [...] } + # hash returned by Provider::Questrade#get_activities. + def activities + payload = @questrade_account.raw_activities_payload + payload = payload.with_indifferent_access[:activities] if payload.is_a?(Hash) + Array(payload) + end + + def process_activity(data) + type = data[:type].to_s.strip + return if type.blank? + + if type == TRADE_TYPE + process_trade(data) + elsif CASH_TYPE_TO_LABEL.key?(type) + process_cash_activity(data, CASH_TYPE_TO_LABEL[type]) + elsif JOURNAL_TYPES.include?(type) + if journal?(data) + process_journal(data) + else + Rails.logger.info "QuestradeAccount::ActivitiesProcessor - Skipping non-journal '#{type}'" + end + elsif UNSUPPORTED_TYPES.include?(type) + Rails.logger.info "QuestradeAccount::ActivitiesProcessor - Skipping unsupported type '#{type}'" + else + Rails.logger.warn "QuestradeAccount::ActivitiesProcessor - Unmapped activity type '#{type}'" + end + end + + # Questrade activities carry no unique id, so synthesize a stable one from + # the immutable fields to keep re-syncs idempotent. + def external_id(data, prefix) + digest = Digest::SHA256.hexdigest( + [ data[:transactionDate], data[:action], data[:symbolId], data[:quantity], data[:netAmount], data[:description] ].join("|") + ) + "questrade_#{prefix}_#{digest.first(24)}" + end + + def process_trade(data) + ticker = data[:symbol].to_s.strip + return if ticker.blank? + + security = resolve_security(ticker, { name: data[:description], currency: data[:currency] }) + return unless security + + quantity = parse_decimal(data[:quantity]) + price = parse_decimal(data[:price]) + return if quantity.nil? || quantity.zero? + + sell = data[:action].to_s.casecmp("Sell").zero? + signed_quantity = sell ? -quantity.abs : quantity.abs + # Buy => positive cost, Sell => negative (matches Sure's trade convention). + amount = price ? signed_quantity * price : parse_decimal(data[:netAmount])&.abs + return if amount.nil? + + date = parse_date(data[:tradeDate]) || parse_date(data[:transactionDate]) || Date.current + currency = extract_currency(data, fallback: account.currency) + + result = import_adapter.import_trade( + external_id: external_id(data, "trade"), + security: security, + quantity: signed_quantity, + price: price, + amount: amount, + currency: currency, + date: date, + name: data[:description].presence || "#{sell ? 'Sell' : 'Buy'} #{ticker}", + source: "questrade", + activity_label: sell ? "Sell" : "Buy" + ) + @trades_count += 1 if result + + import_commission(data, ticker, date, currency) + end + + def import_commission(data, ticker, date, currency) + commission = parse_decimal(data[:commission]) + return if commission.nil? || commission.zero? + + result = import_adapter.import_transaction( + external_id: external_id(data, "fee"), + amount: commission.abs, # money out + currency: currency, + date: date, + name: "Commission for #{ticker}", + source: "questrade", + investment_activity_label: "Fee" + ) + @transactions_count += 1 if result + end + + def journal?(data) + data[:symbol].to_s.strip.present? && !(parse_decimal(data[:quantity]) || 0).zero? + end + + # A journal (Norbert's Gambit / transfer) moves shares with no cash impact. + # Recorded as a zero-cost Transfer trade; the holding's cost basis still + # comes from the positions snapshot. + def process_journal(data) + ticker = data[:symbol].to_s.strip + security = resolve_security(ticker, { name: data[:description], currency: data[:currency] }) + return unless security + + quantity = parse_decimal(data[:quantity]) + return if quantity.nil? || quantity.zero? + + date = parse_date(data[:tradeDate]) || parse_date(data[:transactionDate]) || Date.current + currency = extract_currency(data, fallback: account.currency) + + result = import_adapter.import_trade( + external_id: external_id(data, "journal"), + security: security, + quantity: quantity, + price: 0, + amount: 0, + currency: currency, + date: date, + name: data[:description].presence || "Journal #{ticker}", + source: "questrade", + activity_label: "Transfer" + ) + @trades_count += 1 if result + end + + def process_cash_activity(data, label) + net = parse_decimal(data[:netAmount]) + return if net.nil? + + signed_amount = -net # Questrade +in / -out -> Sure -in / +out + date = parse_date(data[:settlementDate]) || + parse_date(data[:transactionDate]) || + parse_date(data[:tradeDate]) || + Date.current + currency = extract_currency(data, fallback: account.currency) + + symbol = data[:symbol].to_s.strip + security = symbol.present? ? resolve_security(symbol, { name: data[:description], currency: data[:currency] }) : nil + + name = data[:description].presence || (symbol.present? ? "#{label} - #{symbol}" : label) + + result = import_adapter.import_transaction( + external_id: external_id(data, "cash"), + amount: signed_amount, + currency: currency, + date: date, + name: name, + source: "questrade", + investment_activity_label: label, + extra: { security_id: security&.id }.compact + ) + @transactions_count += 1 if result + end +end diff --git a/app/models/questrade_account/data_helpers.rb b/app/models/questrade_account/data_helpers.rb new file mode 100644 index 000000000..c477c0d6b --- /dev/null +++ b/app/models/questrade_account/data_helpers.rb @@ -0,0 +1,156 @@ +# frozen_string_literal: true + +module QuestradeAccount::DataHelpers + extend ActiveSupport::Concern + + private + + # Convert SDK objects to hashes via JSON round-trip + # Many SDKs return objects that don't have proper #to_h methods + def sdk_object_to_hash(obj) + return obj if obj.is_a?(Hash) + + if obj.respond_to?(:to_json) + JSON.parse(obj.to_json) + elsif obj.respond_to?(:to_h) + obj.to_h + else + obj + end + rescue JSON::ParserError, TypeError + obj.respond_to?(:to_h) ? obj.to_h : {} + end + + def parse_decimal(value) + return nil if value.nil? + + case value + when BigDecimal + value + when String + BigDecimal(value) + when Numeric + BigDecimal(value.to_s) + else + nil + end + rescue ArgumentError => e + Rails.logger.error("QuestradeAccount::DataHelpers - Failed to parse decimal value: #{value.inspect} - #{e.message}") + nil + end + + def parse_date(date_value) + return nil if date_value.nil? + + case date_value + when Date + date_value + when String + # Use Time.zone.parse for external timestamps (Rails timezone guidelines) + Time.zone.parse(date_value)&.to_date + when Time, DateTime, ActiveSupport::TimeWithZone + date_value.to_date + else + nil + end + rescue ArgumentError, TypeError => e + Rails.logger.error("QuestradeAccount::DataHelpers - Failed to parse date: #{date_value.inspect} - #{e.message}") + nil + end + + # Find or create security with race condition handling + def resolve_security(symbol, symbol_data = {}) + ticker = symbol.to_s.upcase.strip + return nil if ticker.blank? + + security = Security.find_by(ticker: ticker) + + # If security exists but has a bad name (looks like a hash), update it + if security && security.name&.start_with?("{") + new_name = extract_security_name(symbol_data, ticker) + Rails.logger.info "QuestradeAccount::DataHelpers - Fixing security name: #{security.name.first(50)}... -> #{new_name}" + security.update!(name: new_name) + end + + return security if security + + # Create new security + security_name = extract_security_name(symbol_data, ticker) + + Rails.logger.info "QuestradeAccount::DataHelpers - Creating security: ticker=#{ticker}, name=#{security_name}" + + Security.create!( + ticker: ticker, + name: security_name, + exchange_mic: extract_exchange(symbol_data), + country_code: extract_country_code(symbol_data) + ) + rescue ActiveRecord::RecordInvalid, ActiveRecord::RecordNotUnique => e + # Handle race condition - another process may have created it + Rails.logger.error "QuestradeAccount::DataHelpers - Failed to create security #{ticker}: #{e.message}" + Security.find_by(ticker: ticker) + end + + def extract_security_name(symbol_data, fallback_ticker) + symbol_data = symbol_data.with_indifferent_access if symbol_data.respond_to?(:with_indifferent_access) + + # Try various paths where the name might be + name = symbol_data[:name] || symbol_data[:description] + + # If description is missing or looks like a type description, use ticker + if name.blank? || name.is_a?(Hash) || name =~ /^(COMMON STOCK|CRYPTOCURRENCY|ETF|MUTUAL FUND)$/i + name = fallback_ticker + end + + # Titleize for readability if it's all caps + name = name.titleize if name == name.upcase && name.length > 4 + + name + end + + def extract_exchange(symbol_data) + symbol_data = symbol_data.with_indifferent_access if symbol_data.respond_to?(:with_indifferent_access) + + exchange = symbol_data[:exchange] + return nil unless exchange.is_a?(Hash) + + exchange.with_indifferent_access[:mic_code] || exchange.with_indifferent_access[:id] + end + + def extract_country_code(symbol_data) + symbol_data = symbol_data.with_indifferent_access if symbol_data.respond_to?(:with_indifferent_access) + + # Try to extract country from currency or exchange + currency = symbol_data[:currency] + currency = currency.dig(:code) if currency.is_a?(Hash) + + case currency + when "USD" + "US" + when "CAD" + "CA" + when "GBP", "GBX" + "GB" + when "EUR" + nil # Could be many countries + else + nil + end + end + + # Handle currency as string or object (API inconsistency) + def extract_currency(data, fallback: nil) + data = data.with_indifferent_access if data.respond_to?(:with_indifferent_access) + + currency_data = data[:currency] + return fallback if currency_data.blank? + + if currency_data.is_a?(Hash) + currency_data.with_indifferent_access[:code] || fallback + elsif currency_data.is_a?(String) + currency_data.upcase + else + fallback + end + end +end diff --git a/app/models/questrade_account/holdings_processor.rb b/app/models/questrade_account/holdings_processor.rb new file mode 100644 index 000000000..af9a64dd5 --- /dev/null +++ b/app/models/questrade_account/holdings_processor.rb @@ -0,0 +1,108 @@ +# frozen_string_literal: true + +class QuestradeAccount::HoldingsProcessor + include QuestradeAccount::DataHelpers + + def initialize(questrade_account) + @questrade_account = questrade_account + end + + def process + return unless account.present? + + positions.each do |raw| + process_holding(raw.with_indifferent_access) + rescue => e + Rails.logger.error "QuestradeAccount::HoldingsProcessor - Failed to process holding: #{e.class} - #{e.message}" + Rails.logger.error e.backtrace.first(5).join("\n") if e.backtrace + end + + # Surface non-primary-currency cash as synthetic holdings. + process_cash_holdings + end + + private + + # Surface cash held in currencies other than the account's primary currency + # as synthetic cash holdings (issue #1809). Primary-currency cash stays in + # account.cash_balance. + def process_cash_holdings + @questrade_account.non_primary_cash_entries.each do |entry| + amount = parse_decimal(entry[:amount]) + next if amount.nil? + + security = Security.cash_for(account, currency: entry[:currency]) + date = Date.current + import_adapter.import_holding( + security: security, + quantity: amount, + amount: amount, + currency: entry[:currency], + date: date, + price: 1, + # Date-scope the id so each sync writes a daily snapshot instead of + # rewriting the previous cash holding row. + external_id: "questrade_cash_#{entry[:currency].to_s.downcase}_#{date}", + account_provider_id: @questrade_account.account_provider&.id, + source: "questrade", + delete_future_holdings: false + ) + rescue => e + Rails.logger.error "QuestradeAccount::HoldingsProcessor - Failed to import #{entry[:currency]} cash holding: #{e.message}" + end + end + + def account + @questrade_account.current_account + end + + def import_adapter + @import_adapter ||= Account::ProviderImportAdapter.new(account) + end + + # raw_holdings_payload may be the array itself or the { positions: [...] } + # hash returned by Provider::Questrade#get_holdings. + def positions + payload = @questrade_account.raw_holdings_payload + payload = payload.with_indifferent_access[:positions] if payload.is_a?(Hash) + Array(payload) + end + + def process_holding(data) + ticker = data[:symbol].to_s.strip + return if ticker.blank? + + security = resolve_security(ticker, { name: ticker, currency: data[:currency] }) + return unless security + + quantity = parse_decimal(data[:openQuantity]) # may be fractional + price = parse_decimal(data[:currentPrice]) + return if quantity.nil? || quantity.zero? || price.nil? + + # Prefer Questrade's authoritative market value; fall back to qty * price. + amount = parse_decimal(data[:currentMarketValue]) || (quantity * price) + cost_basis = parse_decimal(data[:averageEntryPrice]) # per-share cost + + # Questrade positions don't carry a currency, so we fall back to the + # account currency. This is fine: Sure values each holding via the + # security's own price/currency (Security#current_price), so this arg is + # just a consistent bookkeeping/dedup key, not the source of valuation. + currency = extract_currency(data, fallback: account.currency) + date = Date.current + external_id = [ "questrade", @questrade_account.questrade_account_id, data[:symbolId], date ].join("_") + + import_adapter.import_holding( + security: security, + quantity: quantity, + amount: amount, + currency: currency, + date: date, + price: price, + cost_basis: cost_basis, + external_id: external_id, + source: "questrade", + account_provider_id: @questrade_account.account_provider&.id, + delete_future_holdings: false + ) + end +end diff --git a/app/models/questrade_account/processor.rb b/app/models/questrade_account/processor.rb new file mode 100644 index 000000000..e1526f1a6 --- /dev/null +++ b/app/models/questrade_account/processor.rb @@ -0,0 +1,57 @@ +# frozen_string_literal: true + +class QuestradeAccount::Processor + include QuestradeAccount::DataHelpers + + attr_reader :questrade_account + + def initialize(questrade_account) + @questrade_account = questrade_account + end + + def process + account = questrade_account.current_account + return unless account + + Rails.logger.info "QuestradeAccount::Processor - Processing account #{questrade_account.id} -> Sure account #{account.id}" + + # Anchor the account at its reported total (cash + holdings) and store the + # primary-currency cash. Non-primary cash is surfaced as holdings below. + update_account_balance(account) + + if questrade_account.raw_holdings_payload.present? || questrade_account.non_primary_cash_entries.any? + QuestradeAccount::HoldingsProcessor.new(questrade_account).process + end + + if questrade_account.raw_activities_payload.present? + QuestradeAccount::ActivitiesProcessor.new(questrade_account).process + end + + account.broadcast_sync_complete + Rails.logger.info "QuestradeAccount::Processor - Broadcast sync complete for account #{account.id}" + + { + holdings_processed: questrade_account.raw_holdings_payload.present?, + activities_processed: questrade_account.raw_activities_payload.present? + } + end + + private + + def update_account_balance(account) + total = questrade_account.current_balance + return if total.blank? + + cash = questrade_account.cash_balance || 0 + account.assign_attributes( + balance: total, + cash_balance: cash, + currency: questrade_account.currency || account.currency + ) + account.save! + + # Current-balance anchor = the reported total (cash + holdings). The value + # is composed from the holdings + per-currency cash, not a made-up figure. + account.set_current_balance(total) + end +end diff --git a/app/models/questrade_item.rb b/app/models/questrade_item.rb new file mode 100644 index 000000000..12a419189 --- /dev/null +++ b/app/models/questrade_item.rb @@ -0,0 +1,202 @@ +# frozen_string_literal: true + +class QuestradeItem < ApplicationRecord + include Syncable, Provided, Unlinking, Encryptable + + enum :status, { good: "good", requires_update: "requires_update" }, default: :good + + # Encrypt sensitive credentials if ActiveRecord encryption is configured + # (encryption_ready? is provided by Encryptable, shared with the other providers). + if encryption_ready? + encrypts :refresh_token, deterministic: true + encrypts :raw_payload + encrypts :raw_institution_payload + end + + validates :name, presence: true + validates :refresh_token, presence: true, unless: :scheduled_for_deletion? + + belongs_to :family + has_one_attached :logo, dependent: :purge_later + + has_many :questrade_accounts, dependent: :destroy + has_many :accounts, through: :questrade_accounts + + scope :active, -> { where(scheduled_for_deletion: false) } + scope :syncable, -> { active.where.not(refresh_token: nil) } + scope :ordered, -> { order(created_at: :desc) } + scope :needs_update, -> { where(status: :requires_update) } + + def syncer + QuestradeItem::Syncer.new(self) + end + + def destroy_later + update!(scheduled_for_deletion: true) + DestroyJob.perform_later(self) + end + + # Override syncing? to include background activities fetch + def syncing? + super || questrade_accounts.where(activities_fetch_pending: true).exists? + end + + # Import data from provider API + def import_latest_questrade_data(sync: nil) + provider = questrade_provider + unless provider + DebugLogEntry.capture( + category: "provider_sync_error", + level: "error", + message: "Cannot import: Questrade provider is not configured", + source: self.class.name, + provider_key: "questrade", + family: family, + metadata: { questrade_item_id: id } + ) + raise StandardError, I18n.t("questrade_items.errors.provider_not_configured") + end + + QuestradeItem::Importer.new(self, questrade_provider: provider, sync: sync).import + rescue => e + DebugLogEntry.capture( + category: "provider_sync_error", + level: "error", + message: "Failed to import data", + source: self.class.name, + provider_key: "questrade", + family: family, + metadata: { questrade_item_id: id, error_class: e.class.name, error_message: e.message } + ) + raise + end + + # Process linked accounts after data import + def process_accounts + return [] if questrade_accounts.empty? + + results = [] + linked_questrade_accounts.includes(account_provider: :account).each do |questrade_account| + begin + result = QuestradeAccount::Processor.new(questrade_account).process + results << { questrade_account_id: questrade_account.id, success: true, result: result } + rescue => e + DebugLogEntry.capture( + category: "provider_sync_error", + level: "error", + message: "Failed to process account", + source: self.class.name, + provider_key: "questrade", + family: family, + metadata: { questrade_item_id: id, questrade_account_id: questrade_account.id, error_class: e.class.name, error_message: e.message } + ) + results << { questrade_account_id: questrade_account.id, success: false, error: e.message } + end + end + + results + end + + # Schedule sync jobs for all linked accounts + def schedule_account_syncs(parent_sync: nil, window_start_date: nil, window_end_date: nil) + return [] if accounts.empty? + + results = [] + accounts.visible.each do |account| + begin + account.sync_later( + parent_sync: parent_sync, + window_start_date: window_start_date, + window_end_date: window_end_date + ) + results << { account_id: account.id, success: true } + rescue => e + DebugLogEntry.capture( + category: "provider_sync_error", + level: "error", + message: "Failed to schedule sync for account", + source: self.class.name, + provider_key: "questrade", + family: family, + metadata: { questrade_item_id: id, account_id: account.id, error_class: e.class.name, error_message: e.message } + ) + results << { account_id: account.id, success: false, error: e.message } + end + end + + results + end + + def upsert_questrade_snapshot!(accounts_snapshot) + assign_attributes( + raw_payload: accounts_snapshot + ) + + save! + end + + def has_completed_initial_setup? + accounts.any? + end + + # Linked accounts (have AccountProvider association) + def linked_questrade_accounts + questrade_accounts.joins(:account_provider) + end + + # Unlinked accounts (no AccountProvider association) + def unlinked_questrade_accounts + questrade_accounts.left_joins(:account_provider).where(account_providers: { id: nil }) + end + + def sync_status_summary + total_accounts = total_accounts_count + linked_count = linked_accounts_count + unlinked_count = unlinked_accounts_count + + if total_accounts == 0 + I18n.t("questrade_items.sync_status.no_accounts") + elsif unlinked_count == 0 + I18n.t("questrade_items.sync_status.synced", count: linked_count) + else + I18n.t("questrade_items.sync_status.synced_with_setup", linked: linked_count, unlinked: unlinked_count) + end + end + + def linked_accounts_count + questrade_accounts.joins(:account_provider).count + end + + def unlinked_accounts_count + questrade_accounts.left_joins(:account_provider).where(account_providers: { id: nil }).count + end + + def total_accounts_count + questrade_accounts.count + end + + def institution_display_name + institution_name.presence || institution_domain.presence || name + end + + def connected_institutions + questrade_accounts.includes(:account) + .where.not(institution_metadata: nil) + .map { |acc| acc.institution_metadata } + .uniq { |inst| inst["name"] || inst["institution_name"] } + end + + def institution_summary + institutions = connected_institutions + case institutions.count + when 0 + I18n.t("questrade_items.institution_summary.none") + else + I18n.t("questrade_items.institution_summary.count", count: institutions.count) + end + end + + def credentials_configured? + refresh_token.present? + end +end diff --git a/app/models/questrade_item/importer.rb b/app/models/questrade_item/importer.rb new file mode 100644 index 000000000..99647874f --- /dev/null +++ b/app/models/questrade_item/importer.rb @@ -0,0 +1,291 @@ +# frozen_string_literal: true + +class QuestradeItem::Importer + include SyncStats::Collector + include QuestradeAccount::DataHelpers + + # Chunk size for fetching activities + ACTIVITY_CHUNK_DAYS = 365 + MAX_ACTIVITY_CHUNKS = 3 # Up to 3 years of history + + # Minimum existing activities required before using incremental sync + MINIMUM_HISTORY_FOR_INCREMENTAL = 10 + + attr_reader :questrade_item, :questrade_provider, :sync + + def initialize(questrade_item, questrade_provider:, sync: nil) + @questrade_item = questrade_item + @questrade_provider = questrade_provider + @sync = sync + end + + class CredentialsError < StandardError; end + + def import + Rails.logger.info "QuestradeItem::Importer - Starting import for item #{questrade_item.id}" + + credentials = questrade_item.questrade_credentials + unless credentials + raise CredentialsError, "No Questrade credentials configured for item #{questrade_item.id}" + end + + # Step 1: Fetch and store all accounts + import_accounts(credentials) + + # Step 2: For LINKED accounts only, fetch data + # Unlinked accounts just need basic info (name, balance) for the setup modal + linked_accounts = QuestradeAccount + .where(questrade_item_id: questrade_item.id) + .joins(:account_provider) + + Rails.logger.info "QuestradeItem::Importer - Found #{linked_accounts.count} linked accounts to process" + + linked_accounts.each do |questrade_account| + Rails.logger.info "QuestradeItem::Importer - Processing linked account #{questrade_account.id}" + import_account_data(questrade_account, credentials) + end + + # Update raw payload on the item + questrade_item.upsert_questrade_snapshot!(stats) + rescue Provider::Questrade::AuthenticationError => e + questrade_item.update!(status: :requires_update) + raise + end + + private + + def stats + @stats ||= {} + end + + def persist_stats! + return unless sync&.respond_to?(:sync_stats) + merged = (sync.sync_stats || {}).merge(stats) + sync.update_columns(sync_stats: merged) + end + + def import_accounts(credentials) + Rails.logger.info "QuestradeItem::Importer - Fetching accounts" + + response = questrade_provider.list_accounts + accounts_data = Array(response.is_a?(Hash) ? response[:accounts] : response) + + stats["api_requests"] = stats.fetch("api_requests", 0) + 1 + stats["total_accounts"] = accounts_data.size + + # Track upstream account IDs to detect removed accounts + upstream_account_ids = [] + + accounts_data.each do |account_data| + begin + account_data = account_data.with_indifferent_access if account_data.is_a?(Hash) + import_account(account_data, credentials) + number = (account_data[:number] || account_data[:id]).to_s + upstream_account_ids << number if number.present? + rescue => e + Rails.logger.error "QuestradeItem::Importer - Failed to import account: #{e.message}" + stats["accounts_skipped"] = stats.fetch("accounts_skipped", 0) + 1 + register_error(e, account_data: account_data) + end + end + + persist_stats! + + # Clean up accounts that no longer exist upstream + prune_removed_accounts(upstream_account_ids) + end + + def import_account(account_data, credentials) + questrade_account_id = (account_data[:number] || account_data[:id]).to_s + return if questrade_account_id.blank? + + questrade_account = questrade_item.questrade_accounts.find_or_initialize_by( + questrade_account_id: questrade_account_id + ) + questrade_account.upsert_from_questrade!(account_data) + + stats["accounts_imported"] = stats.fetch("accounts_imported", 0) + 1 + end + + + def import_account_data(questrade_account, credentials) + # Per-currency balances (cash) -> total anchor + cash holdings + store_balances(questrade_account) + + # Import holdings + import_holdings(questrade_account, credentials) + + # Import activities + import_activities(questrade_account, credentials) + end + + # Fetch per-currency balances. Stores primary-currency cash in cash_balance + # (the rest become cash holdings) and the combined total equity used as the + # account's current-balance anchor. + def store_balances(questrade_account) + response = questrade_provider.get_balances(account_id: questrade_account.questrade_account_id) + stats["api_requests"] = stats.fetch("api_requests", 0) + 1 + + per = Array(response.is_a?(Hash) ? response[:perCurrencyBalances] : nil) + questrade_account.upsert_balances!(per) if per.any? + + combined = Array(response.is_a?(Hash) ? response[:combinedBalances] : nil).map { |b| b.with_indifferent_access } + entry = combined.find { |b| b[:currency] == questrade_account.currency } || combined.first + total = entry && (entry[:totalEquity] || entry[:marketValue]) + questrade_account.update!(current_balance: total) if total.present? + rescue => e + Rails.logger.warn "QuestradeItem::Importer - Failed to fetch balances for account #{questrade_account.id}: #{e.message}" + end + + def import_holdings(questrade_account, credentials) + Rails.logger.info "QuestradeItem::Importer - Fetching holdings for account #{questrade_account.id}" + + begin + response = questrade_provider.get_holdings(account_id: questrade_account.questrade_account_id) + holdings_data = Array(response.is_a?(Hash) ? response[:positions] : response) + + stats["api_requests"] = stats.fetch("api_requests", 0) + 1 + + if holdings_data.any? + # Convert SDK objects to hashes for storage + holdings_hashes = holdings_data.map { |h| sdk_object_to_hash(h) } + holdings_hashes = enrich_positions_with_currency(holdings_hashes) + questrade_account.upsert_holdings_snapshot!(holdings_hashes) + stats["holdings_found"] = stats.fetch("holdings_found", 0) + holdings_data.size + end + rescue => e + Rails.logger.warn "QuestradeItem::Importer - Failed to fetch holdings: #{e.message}" + register_error(e, context: "holdings", account_id: questrade_account.id) + end + end + + # Questrade positions omit currency. Tag each position with its symbol's + # currency (via /v1/symbols) so USD holdings aren't mislabeled as the + # account's CAD currency. + def enrich_positions_with_currency(positions) + ids = positions.filter_map { |p| p.with_indifferent_access[:symbolId] }.uniq + return positions if ids.empty? + + currency_by_id = {} + begin + resp = questrade_provider.get_symbols(ids: ids) + stats["api_requests"] = stats.fetch("api_requests", 0) + 1 + Array(resp.is_a?(Hash) ? resp[:symbols] : nil).each do |sym| + sym = sym.with_indifferent_access + currency_by_id[sym[:symbolId]] = sym[:currency] + end + rescue => e + Rails.logger.warn "QuestradeItem::Importer - symbol currency lookup failed: #{e.message}" + return positions + end + + positions.map do |p| + p = p.with_indifferent_access + cur = currency_by_id[p[:symbolId]] + p[:currency] = cur if cur.present? && p[:currency].blank? + p + end + end + + def import_activities(questrade_account, credentials) + Rails.logger.info "QuestradeItem::Importer - Fetching activities for account #{questrade_account.id}" + + begin + # Determine date range + start_date = calculate_start_date(questrade_account) + end_date = Date.current + + response = questrade_provider.get_activities( + account_id: questrade_account.questrade_account_id, + start_date: start_date, + end_date: end_date + ) + activities_data = Array(response.is_a?(Hash) ? response[:activities] : response) + + stats["api_requests"] = stats.fetch("api_requests", 0) + 1 + + if activities_data.any? + # Convert SDK objects to hashes and merge with existing + activities_hashes = activities_data.map { |a| sdk_object_to_hash(a) } + merged = merge_activities(questrade_account.raw_activities_payload || [], activities_hashes) + questrade_account.upsert_activities_snapshot!(merged) + stats["activities_found"] = stats.fetch("activities_found", 0) + activities_data.size + elsif fresh_linked_account?(questrade_account) + # Fresh account with no activities - schedule background fetch + schedule_background_activities_fetch(questrade_account, start_date) + end + rescue => e + Rails.logger.warn "QuestradeItem::Importer - Failed to fetch activities: #{e.message}" + register_error(e, context: "activities", account_id: questrade_account.id) + end + end + + def calculate_start_date(questrade_account) + # Use user-specified start date if available + user_start = questrade_account.sync_start_date + return user_start if user_start.present? + + # For accounts with existing history, use incremental sync + existing_count = (questrade_account.raw_activities_payload || []).size + if existing_count >= MINIMUM_HISTORY_FOR_INCREMENTAL && questrade_account.last_activities_sync.present? + # Incremental: go back 30 days from last sync to catch updates + (questrade_account.last_activities_sync - 30.days).to_date + else + # Full sync: go back up to 3 years + (ACTIVITY_CHUNK_DAYS * MAX_ACTIVITY_CHUNKS).days.ago.to_date + end + end + + def fresh_linked_account?(questrade_account) + # Account was just linked and has no activity history yet + questrade_account.last_activities_sync.nil? && + (questrade_account.raw_activities_payload || []).empty? + end + + def schedule_background_activities_fetch(questrade_account, start_date) + return if questrade_account.activities_fetch_pending? + + Rails.logger.info "QuestradeItem::Importer - Scheduling background activities fetch for account #{questrade_account.id}" + + questrade_account.update!(activities_fetch_pending: true) + QuestradeActivitiesFetchJob.perform_later(questrade_account, start_date: start_date) + end + + def merge_activities(existing, new_activities) + # Merge by ID, preferring newer data + by_id = {} + existing.each { |a| by_id[activity_key(a)] = a } + new_activities.each { |a| by_id[activity_key(a)] = a } + by_id.values + end + + def activity_key(activity) + activity = activity.with_indifferent_access if activity.is_a?(Hash) + # Questrade activities have no id; key on the immutable fields (same basis + # as the processor's synthesized external_id) to dedup across syncs. + [ activity[:transactionDate], activity[:action], activity[:symbolId], + activity[:netAmount], activity[:description], activity[:currency], activity[:type] ].join("-") + end + + def prune_removed_accounts(upstream_account_ids) + return if upstream_account_ids.empty? + + # Find accounts that exist locally but not upstream + removed = questrade_item.questrade_accounts + .where.not(questrade_account_id: upstream_account_ids) + + if removed.any? + Rails.logger.info "QuestradeItem::Importer - Pruning #{removed.count} removed accounts" + removed.destroy_all + end + end + + def register_error(error, **context) + stats["errors"] ||= [] + stats["errors"] << { + message: error.message, + context: context.to_s, + timestamp: Time.current.iso8601 + } + end +end diff --git a/app/models/questrade_item/provided.rb b/app/models/questrade_item/provided.rb new file mode 100644 index 000000000..cc7ab86bc --- /dev/null +++ b/app/models/questrade_item/provided.rb @@ -0,0 +1,38 @@ +# frozen_string_literal: true + +module QuestradeItem::Provided + extend ActiveSupport::Concern + + def questrade_provider + return nil unless credentials_configured? + + Provider::Questrade.new( + refresh_token: refresh_token, + api_server: api_server, + # Questrade refresh tokens are single-use. Persist the rotated token + + # api_server. This runs inside synchronize_exchange's row lock. + on_token_refresh: ->(creds) { + update!(refresh_token: creds[:refresh_token], api_server: creds[:api_server]) + }, + # Serialize the single-use token exchange across workers: take the row + # lock, reload to get the freshest persisted token, and hand it to the + # SDK to spend. Two concurrent syncs/jobs can no longer burn the same + # token (the brief lock is held across the short token-exchange call). + synchronize_exchange: ->(&blk) { + with_lock do + reload + blk.call(refresh_token) + end + } + ) + end + + # Returns credentials hash for API calls that need them passed explicitly + def questrade_credentials + return nil unless credentials_configured? + + { + refresh_token: refresh_token + } + end +end diff --git a/app/models/questrade_item/syncer.rb b/app/models/questrade_item/syncer.rb new file mode 100644 index 000000000..ef9e8c7c6 --- /dev/null +++ b/app/models/questrade_item/syncer.rb @@ -0,0 +1,97 @@ +# frozen_string_literal: true + +class QuestradeItem::Syncer + include SyncStats::Collector + + attr_reader :questrade_item + + def initialize(questrade_item) + @questrade_item = questrade_item + end + + def perform_sync(sync) + Rails.logger.info "QuestradeItem::Syncer - Starting sync for item #{questrade_item.id}" + + # Phase 1: Import data from provider API + update_sync_status(sync, :importing) + questrade_item.import_latest_questrade_data(sync: sync) + + # Phase 2: Collect setup statistics + finalize_setup_counts(sync) + + # Phase 3: Process data for linked accounts + sync_errors = nil + linked_questrade_accounts = questrade_item.linked_questrade_accounts.includes(account_provider: :account) + if linked_questrade_accounts.any? + update_sync_status(sync, :processing) + mark_import_started(sync) + process_results = questrade_item.process_accounts + + # Phase 4: Schedule balance calculations + update_sync_status(sync, :calculating) + schedule_results = questrade_item.schedule_account_syncs( + parent_sync: sync, + window_start_date: sync.window_start_date, + window_end_date: sync.window_end_date + ) + + # Surface per-account processing/scheduling failures in sync health. + sync_errors = [ *process_results, *schedule_results ].filter_map do |result| + next if result[:success] + { message: result[:error], category: "sync_error" } + end.presence + + # Phase 5: Collect statistics + account_ids = linked_questrade_accounts.filter_map { |pa| pa.current_account&.id } + collect_transaction_stats(sync, account_ids: account_ids, source: "questrade") + collect_trades_stats(sync, account_ids: account_ids, source: "questrade") + collect_holdings_stats(sync, holdings_count: count_holdings, label: "processed") + end + + # Mark sync health + collect_health_stats(sync, errors: sync_errors) + rescue Provider::Questrade::AuthenticationError => e + questrade_item.update!(status: :requires_update) + collect_health_stats(sync, errors: [ { message: e.message, category: "auth_error" } ]) + raise + rescue => e + collect_health_stats(sync, errors: [ { message: e.message, category: "sync_error" } ]) + raise + end + + # Public: called by Sync after finalization + def perform_post_sync + # Override for post-sync cleanup if needed + end + + private + + def count_holdings + questrade_item.linked_questrade_accounts.sum { |pa| Array(pa.raw_holdings_payload).size } + end + + def update_sync_status(sync, key, **i18n_options) + sync.update!(status_text: I18n.t("questrade_items.sync.status.#{key}", **i18n_options)) if sync.respond_to?(:status_text) + end + + def mark_import_started(sync) + # Mark that we're now processing imported data + update_sync_status(sync, :importing_data) + end + + def finalize_setup_counts(sync) + update_sync_status(sync, :checking_setup) + + unlinked_count = questrade_item.unlinked_accounts_count + + if unlinked_count > 0 + questrade_item.update!(pending_account_setup: true) + update_sync_status(sync, :needs_setup, count: unlinked_count) + else + questrade_item.update!(pending_account_setup: false) + end + + # Collect setup stats + collect_setup_stats(sync, provider_accounts: questrade_item.questrade_accounts) + end +end diff --git a/app/models/questrade_item/unlinking.rb b/app/models/questrade_item/unlinking.rb new file mode 100644 index 000000000..86bef73fe --- /dev/null +++ b/app/models/questrade_item/unlinking.rb @@ -0,0 +1,56 @@ +# frozen_string_literal: true + +module QuestradeItem::Unlinking + # Concern that encapsulates unlinking logic for a Questrade item. + extend ActiveSupport::Concern + + # Idempotently remove all connections between this Questrade item and local accounts. + # - Detaches any AccountProvider links for each QuestradeAccount + # - Detaches Holdings that point at the AccountProvider links + # Returns a per-account result payload for observability + def unlink_all!(dry_run: false) + results = [] + + questrade_accounts.find_each do |provider_account| + links = AccountProvider.where(provider_type: "QuestradeAccount", provider_id: provider_account.id).to_a + link_ids = links.map(&:id) + result = { + provider_account_id: provider_account.id, + name: provider_account.name, + provider_link_ids: link_ids + } + results << result + + next if dry_run + + begin + ActiveRecord::Base.transaction do + # Detach holdings for any provider links found + if link_ids.any? + # Bulk-detach: update_all is intentional (no per-row callbacks needed, + # and it avoids loading every Holding into memory). + Holding.where(account_provider_id: link_ids).update_all(account_provider_id: nil) + end + + # Destroy all provider links + links.each do |ap| + ap.destroy! + end + end + rescue StandardError => e + DebugLogEntry.capture( + category: "provider_sync", + level: "warn", + message: "QuestradeItem: failed to fully unlink provider account #\#{provider_account.id}", + source: self.class.name, + provider_key: "questrade", + family: family, + metadata: { provider_account_id: provider_account.id, link_ids: link_ids, error_class: e.class.name, error_message: e.message } + ) + result[:error] = e.message + end + end + + results + end +end diff --git a/app/views/accounts/index.html.erb b/app/views/accounts/index.html.erb index 0145d1830..fb3fe4395 100644 --- a/app/views/accounts/index.html.erb +++ b/app/views/accounts/index.html.erb @@ -17,7 +17,7 @@ ) %> <% end %> -<% if @manual_accounts.empty? && @plaid_items.empty? && @simplefin_items.empty? && @lunchflow_items.empty? && @akahu_items.empty? && @up_items.empty? && @enable_banking_items.empty? && @coinstats_items.empty? && @coinbase_items.empty? && @mercury_items.empty? && @brex_items.empty? && @ibkr_items.empty? && @snaptrade_items.empty? && @indexa_capital_items.empty? && @sophtron_items.empty? && @binance_items.empty? %> +<% if @manual_accounts.empty? && @plaid_items.empty? && @simplefin_items.empty? && @lunchflow_items.empty? && @akahu_items.empty? && @up_items.empty? && @enable_banking_items.empty? && @coinstats_items.empty? && @coinbase_items.empty? && @mercury_items.empty? && @brex_items.empty? && @ibkr_items.empty? && @snaptrade_items.empty? && @indexa_capital_items.empty? && @sophtron_items.empty? && @binance_items.empty? && @questrade_items.empty? %> <%= render "empty" %> <% else %>
<%= provider[:name] %>
diff --git a/app/views/questrade_items/_questrade_item.html.erb b/app/views/questrade_items/_questrade_item.html.erb new file mode 100644 index 000000000..07f2d50ae --- /dev/null +++ b/app/views/questrade_items/_questrade_item.html.erb @@ -0,0 +1,109 @@ +<%# locals: (questrade_item:) %> + +<%= tag.div id: dom_id(questrade_item) do %> + <%= render DS::Disclosure.new(variant: :card, open: true) do |disclosure| %> + <% disclosure.with_summary_content do %> +<%= t(".deletion_in_progress", default: "Removing…") %>
+ <% end %> +<%= t(".kind", default: "Brokerage") %>
+ <% if questrade_item.syncing? %> ++ <% if questrade_item.last_synced_at %> + <% if questrade_item.sync_status_summary %> + <%= t(".status_with_summary", default: "Synced %{timestamp} ago · %{summary}", timestamp: time_ago_in_words(questrade_item.last_synced_at), summary: questrade_item.sync_status_summary) %> + <% else %> + <%= t(".status", default: "Synced %{timestamp} ago", timestamp: time_ago_in_words(questrade_item.last_synced_at)) %> + <% end %> + <% else %> + <%= t(".status_never", default: "Not synced yet") %> + <% end %> +
+ <% end %> +<%= t(".setup_needed", default: "Accounts need setup") %>
+<%= t(".setup_description", default: "%{linked} of %{total} accounts linked", linked: linked_count, total: total_count) %>
+ <%= render DS::Link.new(text: t(".setup_action", default: "Set up accounts"), icon: "settings", variant: "primary", href: setup_accounts_questrade_item_path(questrade_item), frame: :modal) %> +<%= t(".no_accounts_title", default: "No accounts found") %>
+<%= t(".no_accounts_description", default: "We could not find any Questrade accounts to import.") %>
+ <%= render DS::Link.new(text: t(".setup_action", default: "Set up accounts"), icon: "settings", variant: "primary", href: setup_accounts_questrade_item_path(questrade_item), frame: :modal) %> +<%= t(".linking_label", default: "Linking") %>
+<%= @account.name %>
+<%= t(".none_available", default: "No unlinked Questrade accounts available.") %>
+ <%= render DS::Link.new(text: t(".buttons.cancel", default: "Cancel"), variant: "secondary", href: account_path(@account), frame: "_top") %> +<%= questrade_account.name %>
+<%= [ questrade_account.account_type, questrade_account.currency ].compact.join(" · ") %>
+<%= t(".all_linked", default: "All Questrade accounts are already linked.") %>
+ <%= render DS::Link.new(text: t(".buttons.done", default: "Done"), variant: "primary", href: accounts_path, frame: "_top") %> ++ <%= [ questrade_account.account_type, questrade_account.currency ].compact.join(" · ") %> +
+ + + +<%= t("questrade_items.panel.status_configured_html", accounts_path: accounts_path).html_safe %>
+<%= t("questrade_items.panel.token_refresh_hint") %>
+ + <%= styled_form_with model: items.first, + url: questrade_item_path(items.first), + scope: :questrade_item, + method: :patch, + data: { turbo: true }, + class: "space-y-3" do |form| %> + <%= form.text_field :refresh_token, + label: t("questrade_items.panel.fields.refresh_token.label"), + placeholder: t("questrade_items.panel.fields.refresh_token.placeholder_update"), + type: :password, + value: nil %> + +