bytelyst/learning_ai_invt_trdg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(backend): resolve legacy Supabase client inside user, profile, and snapshot repositories

Browse Source 
Made-with: Cursor
This commit is contained in:
Saravana Achu Mac 2026-04-04 18:47:23 -07:00
parent b632a0d946
commit f0dd2055bf
8 changed files with 83 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
backend/src/index.ts
View File

@ -13,7 +13,6 @@ import { ManualTrader } from './services/ManualTrader.js';
import { TradeMonitor } from './services/tradeMonitor.js'; import { TradeMonitor } from './services/tradeMonitor.js';
import { OrderStatusSyncEvent, OrderStatusSyncService } from './services/OrderStatusSyncService.js'; import { OrderStatusSyncEvent, OrderStatusSyncService } from './services/OrderStatusSyncService.js';
import { supabaseService } from './services/SupabaseService.js';
import { healthTracker } from './services/healthTracker.js'; import { healthTracker } from './services/healthTracker.js';
import { observabilityService } from './services/observabilityService.js'; import { observabilityService } from './services/observabilityService.js';
import { reconciliationService } from './services/reconciliationService.js'; import { reconciliationService } from './services/reconciliationService.js';
@ -49,7 +48,7 @@ async function main() {
} }
// --- 0. Load user configuration (Cosmos-first; legacy Supabase optional) --- // --- 0. Load user configuration (Cosmos-first; legacy Supabase optional) ---
logger.info('Fetching active trading users...'); logger.info('Fetching active trading users...');
let users = await listActiveTradingUsers(supabaseService); let users = await listActiveTradingUsers();
// --- 1. Identify Primary Key (for Data Fetching) --- // --- 1. Identify Primary Key (for Data Fetching) ---
const isPlaceholder = (val: string) => !val || val === 'your_key' || val === 'your_secret'; const isPlaceholder = (val: string) => !val || val === 'your_key' || val === 'your_secret';
@ -264,7 +263,7 @@ async function main() {
} }
// --- 0. Initialize Signal Subscribers (Users & Profiles) --- // --- 0. Initialize Signal Subscribers (Users & Profiles) ---
const profiles = await listActiveTradeProfiles(supabaseService); const profiles = await listActiveTradeProfiles();
if (profiles.length > 0) { if (profiles.length > 0) {
logger.info(`👥 Initializing Execution Managers for ${profiles.length} Trade Profiles...`); logger.info(`👥 Initializing Execution Managers for ${profiles.length} Trade Profiles...`);
@ -396,8 +395,8 @@ async function main() {
// --- PROFILE HOT-RELOAD: Sync new/updated/deactivated profiles periodically --- // --- PROFILE HOT-RELOAD: Sync new/updated/deactivated profiles periodically ---
setInterval(async () => { setInterval(async () => {
try { try {
users = await listActiveTradingUsers(supabaseService); users = await listActiveTradingUsers();
const latestProfiles = await listActiveTradeProfiles(supabaseService); const latestProfiles = await listActiveTradeProfiles();
const currentIds = new Set(userContexts.map(c => c.profileId)); const currentIds = new Set(userContexts.map(c => c.profileId));
const latestIds = new Set(latestProfiles.map((p: any) => p.id)); const latestIds = new Set(latestProfiles.map((p: any) => p.id));

2
backend/src/services/CapitalLedger.ts
View File

@ -36,7 +36,7 @@ export class CapitalLedger {
private async ensureLedger(profileId: string, allocatedCapital?: number): Promise<CapitalLedgerRecord | null> { private async ensureLedger(profileId: string, allocatedCapital?: number): Promise<CapitalLedgerRecord | null> {
try { try {
const profileCapital = await getTradeProfileCapital(profileId, supabaseService); const profileCapital = await getTradeProfileCapital(profileId);
const allocation = toNumeric(allocatedCapital ?? profileCapital?.allocatedCapital ?? config.TOTAL_CAPITAL); const allocation = toNumeric(allocatedCapital ?? profileCapital?.allocatedCapital ?? config.TOTAL_CAPITAL);
const existing = await getCapitalLedger(profileId, supabaseService); const existing = await getCapitalLedger(profileId, supabaseService);
const nextRecord: CapitalLedgerRecord = { const nextRecord: CapitalLedgerRecord = {

3
backend/src/services/ManualTrader.ts
View File

@ -1,7 +1,6 @@
import type { TradeExecutor } from './TradeExecutor.js'; import type { TradeExecutor } from './TradeExecutor.js';
import { SignalDirection } from '../strategies/rules/types.js'; import { SignalDirection } from '../strategies/rules/types.js';
import logger from '../utils/logger.js'; import logger from '../utils/logger.js';
import { supabaseService } from './SupabaseService.js';
import { config } from '../config/index.js'; import { config } from '../config/index.js';
import { getTradeProfileCapital } from './profileRepository.js'; import { getTradeProfileCapital } from './profileRepository.js';
@ -34,7 +33,7 @@ export class ManualTrader {
return Number(config.TOTAL_CAPITAL || 0); return Number(config.TOTAL_CAPITAL || 0);
} }
const profileCapital = await getTradeProfileCapital(profileId, supabaseService); const profileCapital = await getTradeProfileCapital(profileId);
if (profileCapital?.allocatedCapital && profileCapital.allocatedCapital > 0) { if (profileCapital?.allocatedCapital && profileCapital.allocatedCapital > 0) {
return profileCapital.allocatedCapital; return profileCapital.allocatedCapital;
} }

46
backend/src/services/apiServer.ts
View File

@ -1113,7 +1113,7 @@ export class ApiServer {
private async resolveSnapshotOwnerId(): Promise<string | null> { private async resolveSnapshotOwnerId(): Promise<string | null> {
if (this.snapshotOwnerId) return this.snapshotOwnerId; if (this.snapshotOwnerId) return this.snapshotOwnerId;
const owner = await resolveSnapshotOwnerIdFromRepository(supabaseService); const owner = await resolveSnapshotOwnerIdFromRepository();
this.snapshotOwnerId = owner; this.snapshotOwnerId = owner;
return owner; return owner;
} }
@ -1124,7 +1124,7 @@ export class ApiServer {
if (!ownerId) { if (!ownerId) {
logger.warn('[API] Snapshot owner not resolved; skipping snapshot restore.'); logger.warn('[API] Snapshot owner not resolved; skipping snapshot restore.');
} else { } else {
const snapshot = await loadLatestBotStateSnapshotFromRepository(ownerId, supabaseService); const snapshot = await loadLatestBotStateSnapshotFromRepository(ownerId);
if (snapshot && snapshot.state) { if (snapshot && snapshot.state) {
const restoredState = snapshot.state as Partial<BotState>; const restoredState = snapshot.state as Partial<BotState>;
this.state = { this.state = {
@ -1189,7 +1189,7 @@ export class ApiServer {
try { try {
const ownerId = await this.resolveSnapshotOwnerId(); const ownerId = await this.resolveSnapshotOwnerId();
if (!ownerId) return; if (!ownerId) return;
await saveBotStateSnapshotFromRepository(ownerId, this.getPersistableState(), supabaseService); await saveBotStateSnapshotFromRepository(ownerId, this.getPersistableState());
this.lastSnapshotWriteAt = Date.now(); this.lastSnapshotWriteAt = Date.now();
logger.info(`[API] Persisted snapshot for ${ownerId}. Interval: ${Math.round(elapsed / 1000)}s`); logger.info(`[API] Persisted snapshot for ${ownerId}. Interval: ${Math.round(elapsed / 1000)}s`);
} catch (error: any) { } catch (error: any) {
@ -1311,8 +1311,8 @@ export class ApiServer {
); );
const profileRows = isAdmin const profileRows = isAdmin
? await listAllTradeProfiles(supabaseService) ? await listAllTradeProfiles()
: await listTradeProfilesForUser(authUserId, supabaseService); : await listTradeProfilesForUser(authUserId);
if (requestedProfileId && !profileRows.some((row) => row.id === requestedProfileId)) { if (requestedProfileId && !profileRows.some((row) => row.id === requestedProfileId)) {
res.status(403).json({ success: false, error: 'Forbidden: profile does not belong to scoped user context' }); res.status(403).json({ success: false, error: 'Forbidden: profile does not belong to scoped user context' });
@ -1672,7 +1672,7 @@ export class ApiServer {
first_name: displayNameParts[0] || '', first_name: displayNameParts[0] || '',
last_name: displayNameParts.slice(1).join(' '), last_name: displayNameParts.slice(1).join(' '),
trade_enable: true, trade_enable: true,
}, supabaseService); });
res.json({ profile }); res.json({ profile });
}); });
@ -1693,7 +1693,7 @@ export class ApiServer {
first_name: displayNameParts[0] || '', first_name: displayNameParts[0] || '',
last_name: displayNameParts.slice(1).join(' '), last_name: displayNameParts.slice(1).join(' '),
trade_enable: true, trade_enable: true,
}, supabaseService); });
res.json({ profile }); res.json({ profile });
} catch (error: any) { } catch (error: any) {
res.status(400).json({ error: `Failed to update profile: ${error.message}` }); res.status(400).json({ error: `Failed to update profile: ${error.message}` });
@ -1715,15 +1715,15 @@ export class ApiServer {
let profiles; let profiles;
if (ensureDefault && !wantsAll) { if (ensureDefault && !wantsAll) {
profiles = await ensureDefaultTradeProfileForUser(authUserId, supabaseService); profiles = await ensureDefaultTradeProfileForUser(authUserId);
} else if (wantsAll) { } else if (wantsAll) {
if (!isAdmin) { if (!isAdmin) {
res.status(403).json({ error: 'Forbidden: Admin role required' }); res.status(403).json({ error: 'Forbidden: Admin role required' });
return; return;
} }
profiles = await listAllTradeProfiles(supabaseService); profiles = await listAllTradeProfiles();
} else { } else {
profiles = await listTradeProfilesForUser(authUserId, supabaseService); profiles = await listTradeProfilesForUser(authUserId);
} }
res.json({ profiles }); res.json({ profiles });
} catch (error: any) { } catch (error: any) {
@ -1743,7 +1743,7 @@ export class ApiServer {
const isAdmin = await isTradingAdmin(authUserId, authReq.authRole); const isAdmin = await isTradingAdmin(authUserId, authReq.authRole);
const requestedUserId = String(req.body?.user_id || '').trim(); const requestedUserId = String(req.body?.user_id || '').trim();
const targetUserId = isAdmin && requestedUserId ? requestedUserId : authUserId; const targetUserId = isAdmin && requestedUserId ? requestedUserId : authUserId;
const profile = await saveTradeProfileForUser(req.body || {}, targetUserId, supabaseService); const profile = await saveTradeProfileForUser(req.body || {}, targetUserId);
res.status(201).json({ profile }); res.status(201).json({ profile });
} catch (error: any) { } catch (error: any) {
res.status(400).json({ error: `Failed to save profile: ${error.message}` }); res.status(400).json({ error: `Failed to save profile: ${error.message}` });
@ -1762,8 +1762,8 @@ export class ApiServer {
const profileId = String(req.params.id || '').trim(); const profileId = String(req.params.id || '').trim();
const isAdmin = await isTradingAdmin(authUserId, authReq.authRole); const isAdmin = await isTradingAdmin(authUserId, authReq.authRole);
const existingProfiles = isAdmin const existingProfiles = isAdmin
? await listAllTradeProfiles(supabaseService) ? await listAllTradeProfiles()
: await listTradeProfilesForUser(authUserId, supabaseService); : await listTradeProfilesForUser(authUserId);
const existing = existingProfiles.find((profile) => profile.id === profileId); const existing = existingProfiles.find((profile) => profile.id === profileId);
if (!existing) { if (!existing) {
res.status(404).json({ error: 'Profile not found' }); res.status(404).json({ error: 'Profile not found' });
@ -1774,7 +1774,7 @@ export class ApiServer {
...existing, ...existing,
...(req.body || {}), ...(req.body || {}),
id: profileId, id: profileId,
}, existing.user_id, supabaseService); }, existing.user_id);
res.json({ profile }); res.json({ profile });
} catch (error: any) { } catch (error: any) {
res.status(400).json({ error: `Failed to update profile: ${error.message}` }); res.status(400).json({ error: `Failed to update profile: ${error.message}` });
@ -1793,8 +1793,8 @@ export class ApiServer {
const profileId = String(req.params.id || '').trim(); const profileId = String(req.params.id || '').trim();
const isAdmin = await isTradingAdmin(authUserId, authReq.authRole); const isAdmin = await isTradingAdmin(authUserId, authReq.authRole);
const existingProfiles = isAdmin const existingProfiles = isAdmin
? await listAllTradeProfiles(supabaseService) ? await listAllTradeProfiles()
: await listTradeProfilesForUser(authUserId, supabaseService); : await listTradeProfilesForUser(authUserId);
const existing = existingProfiles const existing = existingProfiles
.find((profile) => profile.id === profileId); .find((profile) => profile.id === profileId);
if (!existing) { if (!existing) {
@ -1805,7 +1805,7 @@ export class ApiServer {
const profile = await saveTradeProfileForUser({ const profile = await saveTradeProfileForUser({
...existing, ...existing,
is_active: Boolean(req.body?.is_active), is_active: Boolean(req.body?.is_active),
}, existing.user_id, supabaseService); }, existing.user_id);
res.json({ profile }); res.json({ profile });
} catch (error: any) { } catch (error: any) {
res.status(400).json({ error: `Failed to update profile state: ${error.message}` }); res.status(400).json({ error: `Failed to update profile state: ${error.message}` });
@ -1824,15 +1824,15 @@ export class ApiServer {
const profileId = String(req.params.id || '').trim(); const profileId = String(req.params.id || '').trim();
const isAdmin = await isTradingAdmin(authUserId, authReq.authRole); const isAdmin = await isTradingAdmin(authUserId, authReq.authRole);
const existingProfiles = isAdmin const existingProfiles = isAdmin
? await listAllTradeProfiles(supabaseService) ? await listAllTradeProfiles()
: await listTradeProfilesForUser(authUserId, supabaseService); : await listTradeProfilesForUser(authUserId);
const existing = existingProfiles.find((profile) => profile.id === profileId); const existing = existingProfiles.find((profile) => profile.id === profileId);
if (!existing) { if (!existing) {
res.status(404).json({ error: 'Profile not found' }); res.status(404).json({ error: 'Profile not found' });
return; return;
} }
await deleteTradeProfileForUser(profileId, existing.user_id, supabaseService); await deleteTradeProfileForUser(profileId, existing.user_id);
res.json({ success: true }); res.json({ success: true });
} catch (error: any) { } catch (error: any) {
res.status(400).json({ error: `Failed to delete profile: ${error.message}` }); res.status(400).json({ error: `Failed to delete profile: ${error.message}` });
@ -1879,8 +1879,8 @@ export class ApiServer {
limit: orderLimit limit: orderLimit
}), }),
wantsAll wantsAll
? listAllTradeProfiles(supabaseService) ? listAllTradeProfiles()
: listTradeProfilesForUser(authUserId, supabaseService) : listTradeProfilesForUser(authUserId)
]); ]);
res.json({ res.json({
@ -2045,7 +2045,7 @@ export class ApiServer {
const profileId = String(body.profileId || '').trim(); const profileId = String(body.profileId || '').trim();
let profileSettings: any = undefined; let profileSettings: any = undefined;
if (profileId) { if (profileId) {
profileSettings = await getTradeProfileForUser(profileId, authUserId, supabaseService); profileSettings = await getTradeProfileForUser(profileId, authUserId);
if (!profileSettings) { if (!profileSettings) {
res.status(404).json({ success: false, error: 'Backtest profile not found for current user' }); res.status(404).json({ success: false, error: 'Backtest profile not found for current user' });
return; return;

86
backend/src/services/profileRepository.ts
View File

@ -2,9 +2,7 @@ import { getContainer } from '@bytelyst/cosmos';
import { randomUUID } from 'node:crypto'; import { randomUUID } from 'node:crypto';
import { config } from '../config/index.js'; import { config } from '../config/index.js';
import logger from '../utils/logger.js'; import logger from '../utils/logger.js';
import type { supabaseService } from './SupabaseService.js'; import { supabaseService } from './SupabaseService.js';
type LegacySupabaseService = typeof supabaseService;
export interface TradingUserProfile { export interface TradingUserProfile {
user_id: string; user_id: string;
@ -63,6 +61,10 @@ function isCosmosConfigured(): boolean {
return Boolean(config.COSMOS_ENDPOINT && config.COSMOS_KEY); return Boolean(config.COSMOS_ENDPOINT && config.COSMOS_KEY);
} }
function getLegacyClient() {
return supabaseService.getClient();
}
function normalizeProfile(row: Partial<TradeProfileRecord> | null | undefined): TradeProfileRecord | null { function normalizeProfile(row: Partial<TradeProfileRecord> | null | undefined): TradeProfileRecord | null {
const id = String(row?.id || '').trim(); const id = String(row?.id || '').trim();
const userId = String(row?.user_id || '').trim(); const userId = String(row?.user_id || '').trim();
@ -223,8 +225,8 @@ async function listAllProfilesFromCosmos(): Promise<TradeProfileRecord[]> {
.filter((profile): profile is TradeProfileRecord => Boolean(profile)); .filter((profile): profile is TradeProfileRecord => Boolean(profile));
} }
async function listProfilesFromSupabase(userId: string, legacyService?: LegacySupabaseService): Promise<TradeProfileRecord[]> { async function listProfilesFromSupabase(userId: string): Promise<TradeProfileRecord[]> {
const client = legacyService?.getClient?.(); const client = getLegacyClient();
if (!client) { if (!client) {
return []; return [];
} }
@ -265,8 +267,8 @@ async function seedProfilesToCosmos(profiles: TradeProfileRecord[]): Promise<Tra
return profiles; return profiles;
} }
async function listAllProfilesFromSupabase(legacyService?: LegacySupabaseService): Promise<TradeProfileRecord[]> { async function listAllProfilesFromSupabase(): Promise<TradeProfileRecord[]> {
const client = legacyService?.getClient?.(); const client = getLegacyClient();
if (!client) { if (!client) {
return []; return [];
} }
@ -317,8 +319,8 @@ async function getProfileFromCosmos(profileId: string): Promise<TradeProfileReco
}); });
} }
async function getProfileFromSupabase(profileId: string, legacyService?: LegacySupabaseService): Promise<TradeProfileRecord | null> { async function getProfileFromSupabase(profileId: string): Promise<TradeProfileRecord | null> {
const client = legacyService?.getClient?.(); const client = getLegacyClient();
if (!client || !profileId) { if (!client || !profileId) {
return null; return null;
} }
@ -341,8 +343,8 @@ async function getProfileFromSupabase(profileId: string, legacyService?: LegacyS
} }
} }
async function mirrorProfileToSupabase(profile: TradeProfileRecord, legacyService?: LegacySupabaseService): Promise<void> { async function mirrorProfileToSupabase(profile: TradeProfileRecord): Promise<void> {
const client = legacyService?.getClient?.(); const client = getLegacyClient();
if (!client) return; if (!client) return;
try { try {
@ -362,8 +364,8 @@ async function mirrorProfileToSupabase(profile: TradeProfileRecord, legacyServic
} }
} }
async function deleteProfileFromSupabase(profileId: string, userId: string, legacyService?: LegacySupabaseService): Promise<void> { async function deleteProfileFromSupabase(profileId: string, userId: string): Promise<void> {
const client = legacyService?.getClient?.(); const client = getLegacyClient();
if (!client) return; if (!client) return;
try { try {
@ -381,9 +383,9 @@ async function deleteProfileFromSupabase(profileId: string, userId: string, lega
} }
} }
export async function listTradeProfilesForUser(userId: string, legacyService?: LegacySupabaseService): Promise<TradeProfileRecord[]> { export async function listTradeProfilesForUser(userId: string): Promise<TradeProfileRecord[]> {
if (!isCosmosConfigured()) { if (!isCosmosConfigured()) {
return listProfilesFromSupabase(userId, legacyService); return listProfilesFromSupabase(userId);
} }
try { try {
@ -391,7 +393,7 @@ export async function listTradeProfilesForUser(userId: string, legacyService?: L
if (cosmosProfiles.length > 0) { if (cosmosProfiles.length > 0) {
return cosmosProfiles; return cosmosProfiles;
} }
const seededProfiles = await seedProfilesToCosmos(await listProfilesFromSupabase(userId, legacyService)); const seededProfiles = await seedProfilesToCosmos(await listProfilesFromSupabase(userId));
if (seededProfiles.length > 0) { if (seededProfiles.length > 0) {
logger.info(`[Profiles] Seeded ${seededProfiles.length} user profiles from legacy store into Cosmos for user ${userId}.`); logger.info(`[Profiles] Seeded ${seededProfiles.length} user profiles from legacy store into Cosmos for user ${userId}.`);
} }
@ -402,9 +404,9 @@ export async function listTradeProfilesForUser(userId: string, legacyService?: L
} }
} }
export async function listAllTradeProfiles(legacyService?: LegacySupabaseService): Promise<TradeProfileRecord[]> { export async function listAllTradeProfiles(): Promise<TradeProfileRecord[]> {
if (!isCosmosConfigured()) { if (!isCosmosConfigured()) {
return listAllProfilesFromSupabase(legacyService); return listAllProfilesFromSupabase();
} }
try { try {
@ -412,7 +414,7 @@ export async function listAllTradeProfiles(legacyService?: LegacySupabaseService
if (cosmosProfiles.length > 0) { if (cosmosProfiles.length > 0) {
return cosmosProfiles; return cosmosProfiles;
} }
const seededProfiles = await seedProfilesToCosmos(await listAllProfilesFromSupabase(legacyService)); const seededProfiles = await seedProfilesToCosmos(await listAllProfilesFromSupabase());
if (seededProfiles.length > 0) { if (seededProfiles.length > 0) {
logger.info(`[Profiles] Seeded ${seededProfiles.length} profiles from legacy store into Cosmos.`); logger.info(`[Profiles] Seeded ${seededProfiles.length} profiles from legacy store into Cosmos.`);
} }
@ -423,19 +425,19 @@ export async function listAllTradeProfiles(legacyService?: LegacySupabaseService
} }
} }
export async function listActiveTradeProfiles(legacyService?: LegacySupabaseService): Promise<TradeProfileRecord[]> { export async function listActiveTradeProfiles(): Promise<TradeProfileRecord[]> {
const profiles = await listAllTradeProfiles(legacyService); const profiles = await listAllTradeProfiles();
return profiles.filter((profile) => Boolean(profile.is_active)); return profiles.filter((profile) => Boolean(profile.is_active));
} }
export async function getTradeProfileById(profileId: string, legacyService?: LegacySupabaseService): Promise<TradeProfileRecord | null> { export async function getTradeProfileById(profileId: string): Promise<TradeProfileRecord | null> {
const normalizedId = String(profileId || '').trim(); const normalizedId = String(profileId || '').trim();
if (!normalizedId) { if (!normalizedId) {
return null; return null;
} }
if (!isCosmosConfigured()) { if (!isCosmosConfigured()) {
return getProfileFromSupabase(normalizedId, legacyService); return getProfileFromSupabase(normalizedId);
} }
try { try {
@ -443,7 +445,7 @@ export async function getTradeProfileById(profileId: string, legacyService?: Leg
if (cosmosProfile) { if (cosmosProfile) {
return cosmosProfile; return cosmosProfile;
} }
const legacyProfile = await getProfileFromSupabase(normalizedId, legacyService); const legacyProfile = await getProfileFromSupabase(normalizedId);
if (!legacyProfile) { if (!legacyProfile) {
return null; return null;
} }
@ -456,8 +458,8 @@ export async function getTradeProfileById(profileId: string, legacyService?: Leg
} }
} }
export async function getTradeProfileCapital(profileId: string, legacyService?: LegacySupabaseService): Promise<TradeProfileCapitalSummary | null> { export async function getTradeProfileCapital(profileId: string): Promise<TradeProfileCapitalSummary | null> {
const profile = await getTradeProfileById(profileId, legacyService); const profile = await getTradeProfileById(profileId);
if (!profile) { if (!profile) {
return null; return null;
} }
@ -469,28 +471,27 @@ export async function getTradeProfileCapital(profileId: string, legacyService?:
}; };
} }
export async function getTradeProfileForUser(profileId: string, userId: string, legacyService?: LegacySupabaseService): Promise<TradeProfileRecord | null> { export async function getTradeProfileForUser(profileId: string, userId: string): Promise<TradeProfileRecord | null> {
const profile = await getTradeProfileById(profileId, legacyService); const profile = await getTradeProfileById(profileId);
if (!profile || String(profile.user_id || '').trim() !== String(userId || '').trim()) { if (!profile || String(profile.user_id || '').trim() !== String(userId || '').trim()) {
return null; return null;
} }
return profile; return profile;
} }
export async function ensureDefaultTradeProfileForUser(userId: string, legacyService?: LegacySupabaseService): Promise<TradeProfileRecord[]> { export async function ensureDefaultTradeProfileForUser(userId: string): Promise<TradeProfileRecord[]> {
const profiles = await listTradeProfilesForUser(userId, legacyService); const profiles = await listTradeProfilesForUser(userId);
if (profiles.length > 0) { if (profiles.length > 0) {
return profiles; return profiles;
} }
const created = await saveTradeProfileForUser(buildDefaultTradeProfile(userId), userId, legacyService); const created = await saveTradeProfileForUser(buildDefaultTradeProfile(userId), userId);
return [created]; return [created];
} }
export async function saveTradeProfileForUser( export async function saveTradeProfileForUser(
input: Partial<TradeProfileRecord>, input: Partial<TradeProfileRecord>,
userId: string, userId: string
legacyService?: LegacySupabaseService
): Promise<TradeProfileRecord> { ): Promise<TradeProfileRecord> {
const now = new Date().toISOString(); const now = new Date().toISOString();
const normalized = normalizeProfile({ const normalized = normalizeProfile({
@ -520,14 +521,13 @@ export async function saveTradeProfileForUser(
} }
} }
await mirrorProfileToSupabase(normalized, legacyService); await mirrorProfileToSupabase(normalized);
return normalized; return normalized;
} }
export async function deleteTradeProfileForUser( export async function deleteTradeProfileForUser(
profileId: string, profileId: string,
userId: string, userId: string
legacyService?: LegacySupabaseService
): Promise<void> { ): Promise<void> {
if (!profileId || !userId) { if (!profileId || !userId) {
return; return;
@ -542,13 +542,12 @@ export async function deleteTradeProfileForUser(
} }
} }
await deleteProfileFromSupabase(profileId, userId, legacyService); await deleteProfileFromSupabase(profileId, userId);
} }
export async function getCurrentUserProfile( export async function getCurrentUserProfile(
userId: string, userId: string,
fallback: Partial<TradingUserProfile> = {}, fallback: Partial<TradingUserProfile> = {}
legacyService?: LegacySupabaseService
): Promise<TradingUserProfile> { ): Promise<TradingUserProfile> {
if (isCosmosConfigured()) { if (isCosmosConfigured()) {
try { try {
@ -561,7 +560,7 @@ export async function getCurrentUserProfile(
} }
} }
const client = legacyService?.getClient?.(); const client = getLegacyClient();
if (client) { if (client) {
try { try {
const { data, error } = await client const { data, error } = await client
@ -614,10 +613,9 @@ export async function getCurrentUserProfile(
export async function saveCurrentUserProfile( export async function saveCurrentUserProfile(
userId: string, userId: string,
input: Partial<TradingUserProfile>, input: Partial<TradingUserProfile>,
fallback: Partial<TradingUserProfile> = {}, fallback: Partial<TradingUserProfile> = {}
legacyService?: LegacySupabaseService
): Promise<TradingUserProfile> { ): Promise<TradingUserProfile> {
const existing = await getCurrentUserProfile(userId, fallback, legacyService); const existing = await getCurrentUserProfile(userId, fallback);
const merged: TradingUserProfile = { const merged: TradingUserProfile = {
...existing, ...existing,
...input, ...input,
@ -638,7 +636,7 @@ export async function saveCurrentUserProfile(
logger.warn(`[Profiles] Cosmos user profile save failed: ${error instanceof Error ? error.message : 'unknown error'}`); logger.warn(`[Profiles] Cosmos user profile save failed: ${error instanceof Error ? error.message : 'unknown error'}`);
} }
const client = legacyService?.getClient?.(); const client = getLegacyClient();
if (client) { if (client) {
try { try {
const { error } = await client const { error } = await client

3
backend/src/services/reconciliationParityHeartbeatService.ts
View File

@ -3,7 +3,6 @@ import { config } from '../config/index.js';
import logger from '../utils/logger.js'; import logger from '../utils/logger.js';
import { healthTracker } from './healthTracker.js'; import { healthTracker } from './healthTracker.js';
import { observabilityService } from './observabilityService.js'; import { observabilityService } from './observabilityService.js';
import { supabaseService } from './SupabaseService.js';
import { getTradeProfileCapital } from './profileRepository.js'; import { getTradeProfileCapital } from './profileRepository.js';
import type { TradeExecutor } from './TradeExecutor.js'; import type { TradeExecutor } from './TradeExecutor.js';
import { buildAlpacaSubTag } from '../utils/alpacaSubTag.js'; import { buildAlpacaSubTag } from '../utils/alpacaSubTag.js';
@ -277,7 +276,7 @@ export class ReconciliationParityHeartbeatService {
const throttleMs = Math.max(0, toNumber(config.RECON_INTEGRITY_WATCHDOG_THROTTLE_MS || 600_000)); const throttleMs = Math.max(0, toNumber(config.RECON_INTEGRITY_WATCHDOG_THROTTLE_MS || 600_000));
const requireSubTagAttribution = Boolean(config.RECON_POSITION_PARITY_REQUIRE_SUBTAG_ATTRIBUTION); const requireSubTagAttribution = Boolean(config.RECON_POSITION_PARITY_REQUIRE_SUBTAG_ATTRIBUTION);
const allowLegacyEntryAttribution = Boolean(config.RECON_POSITION_PARITY_ALLOW_LEGACY_ENTRY_ATTRIBUTION); const allowLegacyEntryAttribution = Boolean(config.RECON_POSITION_PARITY_ALLOW_LEGACY_ENTRY_ATTRIBUTION);
const profileCapital = await getTradeProfileCapital(profileId, supabaseService); const profileCapital = await getTradeProfileCapital(profileId);
const allocatedCapitalUsd = Math.max(0, toNumber(profileCapital?.allocatedCapital)); const allocatedCapitalUsd = Math.max(0, toNumber(profileCapital?.allocatedCapital));
let mismatchTrades = 0; let mismatchTrades = 0;

18
backend/src/services/snapshotRepository.ts
View File

@ -1,11 +1,9 @@
import { getContainer } from '@bytelyst/cosmos'; import { getContainer } from '@bytelyst/cosmos';
import { config } from '../config/index.js'; import { config } from '../config/index.js';
import logger from '../utils/logger.js'; import logger from '../utils/logger.js';
import type { supabaseService } from './SupabaseService.js'; import { supabaseService } from './SupabaseService.js';
import { listActiveTradingUsers } from './userRepository.js'; import { listActiveTradingUsers } from './userRepository.js';
type LegacySupabaseService = typeof supabaseService;
const CONTAINER_NAME = 'bot_state_snapshots'; const CONTAINER_NAME = 'bot_state_snapshots';
interface BotStateSnapshotDocument { interface BotStateSnapshotDocument {
@ -28,13 +26,13 @@ function buildSnapshotId(ownerId: string): string {
return `${config.PRODUCT_ID}:${ownerId}`; return `${config.PRODUCT_ID}:${ownerId}`;
} }
export async function resolveSnapshotOwnerId(legacyService?: LegacySupabaseService): Promise<string | null> { export async function resolveSnapshotOwnerId(): Promise<string | null> {
const configured = String(config.SNAPSHOT_USER_ID || '').trim().toLowerCase(); const configured = String(config.SNAPSHOT_USER_ID || '').trim().toLowerCase();
if (isUuid(configured)) { if (isUuid(configured)) {
return configured; return configured;
} }
const users = await listActiveTradingUsers(legacyService); const users = await listActiveTradingUsers();
const firstUserId = String(users[0]?.user_id || '').trim().toLowerCase(); const firstUserId = String(users[0]?.user_id || '').trim().toLowerCase();
if (isUuid(firstUserId)) { if (isUuid(firstUserId)) {
return firstUserId; return firstUserId;
@ -44,12 +42,11 @@ export async function resolveSnapshotOwnerId(legacyService?: LegacySupabaseServi
} }
export async function loadLatestBotStateSnapshot( export async function loadLatestBotStateSnapshot(
ownerId: string, ownerId: string
legacyService?: LegacySupabaseService
): Promise<{ state: unknown } | null> { ): Promise<{ state: unknown } | null> {
if (!ownerId) return null; if (!ownerId) return null;
const loadLegacySnapshot = async () => legacyService?.loadLatestBotStateSnapshot(ownerId) ?? null; const loadLegacySnapshot = async () => supabaseService.loadLatestBotStateSnapshot(ownerId);
if (!isCosmosConfigured()) { if (!isCosmosConfigured()) {
return loadLegacySnapshot(); return loadLegacySnapshot();
@ -90,8 +87,7 @@ export async function loadLatestBotStateSnapshot(
export async function saveBotStateSnapshot( export async function saveBotStateSnapshot(
ownerId: string, ownerId: string,
state: unknown, state: unknown
legacyService?: LegacySupabaseService
): Promise<void> { ): Promise<void> {
if (!ownerId) return; if (!ownerId) return;
@ -112,5 +108,5 @@ export async function saveBotStateSnapshot(
} }
} }
await legacyService?.saveBotStateSnapshot(ownerId, state); await supabaseService.saveBotStateSnapshot(ownerId, state);
} }

8
backend/src/services/userRepository.ts
View File

@ -1,9 +1,9 @@
import { getContainer } from '@bytelyst/cosmos'; import { getContainer } from '@bytelyst/cosmos';
import { config } from '../config/index.js'; import { config } from '../config/index.js';
import logger from '../utils/logger.js'; import logger from '../utils/logger.js';
import type { UserConfig, supabaseService } from './SupabaseService.js'; import type { UserConfig } from './SupabaseService.js';
import { supabaseService } from './SupabaseService.js';
type LegacySupabaseService = typeof supabaseService;
const USER_PROFILE_CONTAINER = 'trading_users'; const USER_PROFILE_CONTAINER = 'trading_users';
interface TradingUserDocument extends UserConfig { interface TradingUserDocument extends UserConfig {
@ -39,7 +39,7 @@ function normalizeUser(row: Partial<UserConfig> | null | undefined): UserConfig
}; };
} }
export async function listActiveTradingUsers(legacyService?: LegacySupabaseService): Promise<UserConfig[]> { export async function listActiveTradingUsers(): Promise<UserConfig[]> {
if (isCosmosConfigured()) { if (isCosmosConfigured()) {
try { try {
const container = getContainer(USER_PROFILE_CONTAINER); const container = getContainer(USER_PROFILE_CONTAINER);
@ -63,7 +63,7 @@ export async function listActiveTradingUsers(legacyService?: LegacySupabaseServi
} }
} }
const client = legacyService?.getClient?.(); const client = supabaseService.getClient();
if (!client) { if (!client) {
return []; return [];
} }