diff --git a/backend/src/index.ts b/backend/src/index.ts index 75f25c6..da9534b 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -18,6 +18,8 @@ import { healthTracker } from './services/healthTracker.js'; import { observabilityService } from './services/observabilityService.js'; import { reconciliationService } from './services/reconciliationService.js'; import { reconciliationWatchdogAutoResumeService } from './services/reconciliationWatchdogAutoResumeService.js'; +import { listActiveTradeProfiles } from './services/profileRepository.js'; +import { listActiveTradingUsers } from './services/userRepository.js'; async function main() { logger.info(`Starting ${config.PRODUCT_ID} trading backend...`); @@ -46,7 +48,7 @@ async function main() { } // --- 0. Load User Configuration (Supabase) --- logger.info('Fetching active users from Supabase...'); - const users = await supabaseService.getActiveUsers(); + let users = await listActiveTradingUsers(supabaseService); // --- 1. Identify Primary Key (for Data Fetching) --- const isPlaceholder = (val: string) => !val || val === 'your_key' || val === 'your_secret'; @@ -261,7 +263,7 @@ async function main() { } // --- 0. Initialize Signal Subscribers (Users & Profiles) --- - const profiles = await supabaseService.getActiveProfiles(); + const profiles = await listActiveTradeProfiles(supabaseService); if (profiles.length > 0) { logger.info(`👥 Initializing Execution Managers for ${profiles.length} Trade Profiles...`); @@ -393,7 +395,8 @@ async function main() { // --- PROFILE HOT-RELOAD: Sync new/updated/deactivated profiles periodically --- setInterval(async () => { try { - const latestProfiles = await supabaseService.getActiveProfiles(); + users = await listActiveTradingUsers(supabaseService); + const latestProfiles = await listActiveTradeProfiles(supabaseService); const currentIds = new Set(userContexts.map(c => c.profileId)); const latestIds = new Set(latestProfiles.map((p: any) => p.id)); diff --git a/backend/src/services/userRepository.ts b/backend/src/services/userRepository.ts new file mode 100644 index 0000000..98a3ac9 --- /dev/null +++ b/backend/src/services/userRepository.ts @@ -0,0 +1,52 @@ +import logger from '../utils/logger.js'; +import type { UserConfig, supabaseService } from './SupabaseService.js'; + +type LegacySupabaseService = typeof supabaseService; + +function normalizeUser(row: Partial | null | undefined): UserConfig | null { + const userId = String(row?.user_id || '').trim(); + if (!userId) { + return null; + } + + return { + user_id: userId, + first_name: String(row?.first_name || ''), + last_name: String(row?.last_name || ''), + email: String(row?.email || ''), + ALPACA_API_KEY: String(row?.ALPACA_API_KEY || ''), + ALPACA_SECRET_KEY: String(row?.ALPACA_SECRET_KEY || ''), + REAL_ALPACA_API_KEY: String(row?.REAL_ALPACA_API_KEY || ''), + REAL_ALPACA_SECRET_KEY: String(row?.REAL_ALPACA_SECRET_KEY || ''), + role: String(row?.role || 'member'), + trade_enable: Boolean(row?.trade_enable), + drop_threshold_for_buy: Number(row?.drop_threshold_for_buy || 0), + gain_threshold_for_sell: Number(row?.gain_threshold_for_sell || 0), + market_poll_interval_in_seconds: Number(row?.market_poll_interval_in_seconds || 0), + }; +} + +export async function listActiveTradingUsers(legacyService?: LegacySupabaseService): Promise { + const client = legacyService?.getClient?.(); + if (!client) { + return []; + } + + try { + const { data, error } = await client + .from('users') + .select('*') + .eq('trade_enable', true); + + if (error || !Array.isArray(data)) { + return []; + } + + return data + .map((row) => normalizeUser(row as UserConfig)) + .filter((user): user is UserConfig => Boolean(user)); + } catch (error) { + logger.warn(`[Users] Active trading user lookup failed: ${error instanceof Error ? error.message : 'unknown error'}`); + return []; + } +}