fix: translate permanent invalid-token errors on online entrypoints (WealthLedger/backend#59) (#110)

src/integrations/tbank/tbank-access.service.ts

@@ -121,19 +121,11 @@ export class TbankAccessService {
     return work ? this.tokens.decrypt(work.connection.encryptedToken) : null;
   }
 
-  async onlineToken(userId: string) {
-    const connection = await this.prisma.tbankConnection.findUnique({
-      where: { userId },
-      select: { encryptedToken: true, invalidToken: true },
-    });
-    if (!connection) throw new ConflictException({ code: TbankErrorCode.NotConnected });
-    if (connection.invalidToken) {
-      throw new BadRequestException({ code: TbankErrorCode.InvalidToken });
-    }
-    return this.tokens.decrypt(connection.encryptedToken);
-  }
-
-  async current(userId: string): Promise<{ revision: string; token: string }> {
+  // Online reads: a permanent upstream auth refusal becomes 400/INVALID_TBANK_TOKEN here, once, for every caller.
+  async online<T>(
+    userId: string,
+    run: (connection: { revision: string; token: string }) => Promise<T>,
+  ): Promise<T> {
     const connection = await this.prisma.tbankConnection.findUnique({
       where: { userId },
       select: { encryptedToken: true, invalidToken: true, revision: true },
@@ -142,7 +134,8 @@ export class TbankAccessService {
     if (connection.invalidToken) {
       throw new BadRequestException({ code: TbankErrorCode.InvalidToken });
     }
-    return { revision: connection.revision, token: this.tokens.decrypt(connection.encryptedToken) };
+    const token = this.tokens.decrypt(connection.encryptedToken);
+    return this.tokens.translateInvalidToken(() => run({ revision: connection.revision, token }));
   }
 
   async publishMembership(

src/modules/investments/instruments/application/catalog-discovery.service.ts

@@ -122,11 +122,11 @@ export class CatalogDiscoveryService {
     observe = false,
   ) {
     if (!observe) {
-      const token = await this.access.onlineToken(userId);
-      const instrument =
+      const instrument = await this.access.online(userId, ({ token }) =>
         instrumentType === 'share'
-          ? await this.tbank.getShare(token, uid)
-          : await this.tbank.getBond(token, uid);
+          ? this.tbank.getShare(token, uid)
+          : this.tbank.getBond(token, uid),
+      );
       if (!instrument) throw new NotFoundException();
       return {
         data: { ...publicInstrument({ ...instrument, uid }), uid, slug: instrument.ticker?.toUpperCase() ?? uid },
@@ -201,11 +201,11 @@ export class CatalogDiscoveryService {
   ) {
     if (!observe) {
       const query = ticker.trim().toUpperCase();
-      const token = await this.access.onlineToken(userId);
-      const found =
+      const found = await this.access.online(userId, ({ token }) =>
         instrumentType === 'share'
-          ? await this.tbank.findShares(token, query)
-          : await this.tbank.findBonds(token, query);
+          ? this.tbank.findShares(token, query)
+          : this.tbank.findBonds(token, query),
+      );
       const matches = selectCatalogInstruments(
         found.filter((item) => item.ticker?.trim().toUpperCase() === query),
         query,
@@ -268,30 +268,32 @@ export class CatalogDiscoveryService {
   }
 
   private async onlineSearch(userId: string, input: CatalogQuery) {
-    const token = await this.access.onlineToken(userId);
-    const found =
-      input.instrumentType === 'share'
-        ? await this.tbank.findShares(token, input.query)
-        : await this.tbank.findBonds(token, input.query);
-    const instruments = selectCatalogInstruments(
-      input.exactTicker
-        ? found.filter((item) => item.ticker?.trim().toUpperCase() === input.query)
-        : found,
-      input.query,
-      input.instrumentType,
-    ).slice(FIRST_RESULT_OFFSET, input.exactTicker ? tickerAmbiguityLimit : catalogSearchLimit);
-    const [details, prices] = await Promise.all([
-      Promise.all(
-        instruments.map((item) =>
-          input.instrumentType === 'share'
-            ? this.tbank.getShare(token, item.uid)
-            : this.tbank.getBond(token, item.uid),
+    const { details, instruments, prices } = await this.access.online(userId, async ({ token }) => {
+      const found =
+        input.instrumentType === 'share'
+          ? await this.tbank.findShares(token, input.query)
+          : await this.tbank.findBonds(token, input.query);
+      const instruments = selectCatalogInstruments(
+        input.exactTicker
+          ? found.filter((item) => item.ticker?.trim().toUpperCase() === input.query)
+          : found,
+        input.query,
+        input.instrumentType,
+      ).slice(FIRST_RESULT_OFFSET, input.exactTicker ? tickerAmbiguityLimit : catalogSearchLimit);
+      const [details, prices] = await Promise.all([
+        Promise.all(
+          instruments.map((item) =>
+            input.instrumentType === 'share'
+              ? this.tbank.getShare(token, item.uid)
+              : this.tbank.getBond(token, item.uid),
+          ),
         ),
-      ),
-      instruments.length
-        ? this.tbank.getLastPrices(token, instruments.map((item) => item.uid))
-        : [],
-    ]);
+        instruments.length
+          ? this.tbank.getLastPrices(token, instruments.map((item) => item.uid))
+          : [],
+      ]);
+      return { details, instruments, prices };
+    });
     const fetchedAt = new Date();
     const data = instruments.flatMap((item, index) => {
       const detail = details.at(index);

src/modules/investments/instruments/application/catalog-enrichment.service.ts

@@ -231,7 +231,6 @@ export class CatalogEnrichmentService {
     return reservation;
   }
 
-  // eslint-disable-next-line sonarjs/cognitive-complexity -- Section composition owns the source dependency policy.
   async section(
     userId: string,
     instrumentType: 'bond' | 'share',
@@ -240,18 +239,20 @@ export class CatalogEnrichmentService {
     observe = false,
   ) {
     if (section === 'company' && !observe) {
-      const token = await this.access.onlineToken(userId);
-      const instrument =
-        instrumentType === 'share'
-          ? await this.tbank.getShare(token, uid)
-          : await this.tbank.getBond(token, uid);
-      if (!instrument) throw new NotFoundException();
-      const [description, fundamentals] = instrument.assetUid
-        ? await Promise.all([
-            this.tbank.getAssetDescription(token, instrument.assetUid),
-            this.tbank.getFundamentals(token, instrument.assetUid),
-          ])
-        : [null, null];
+      const { description, fundamentals, instrument } = await this.access.online(userId, async ({ token }) => {
+        const instrument =
+          instrumentType === 'share'
+            ? await this.tbank.getShare(token, uid)
+            : await this.tbank.getBond(token, uid);
+        if (!instrument) throw new NotFoundException();
+        const [description, fundamentals] = instrument.assetUid
+          ? await Promise.all([
+              this.tbank.getAssetDescription(token, instrument.assetUid),
+              this.tbank.getFundamentals(token, instrument.assetUid),
+            ])
+          : [null, null];
+        return { description, fundamentals, instrument };
+      });
       const meta = investmentResourceMeta({
         resourceKey: `${instrumentType}/company/${uid}`,
         dataRevision: `online:${Date.now()}`,

src/modules/investments/instruments/stocks/application/stock-screener.service.spec.ts

@@ -22,14 +22,16 @@ function share(index: number, overrides: Partial<TbankShare> = {}): TbankShare {
 }
 
 describe('StockScreenerService', () => {
-  const access = { current: jest.fn(), local: jest.fn() };
+  const access = { online: jest.fn(), local: jest.fn() };
   const client = { getAllShares: jest.fn() };
   const logger = { event: jest.fn() };
   let service: StockScreenerService;
 
   beforeEach(() => {
     jest.clearAllMocks();
-    access.current.mockResolvedValue({ revision, token });
+    access.online.mockImplementation((_userId: string, run: (connection: { revision: string; token: string }) => Promise<unknown>) =>
+      run({ revision, token }),
+    );
     access.local.mockResolvedValue({ revision });
     client.getAllShares.mockResolvedValue(
       Array.from({ length: pageSize + 1 }, (_, index) => share(index)),

src/modules/investments/instruments/stocks/application/stock-screener.service.ts

@@ -48,8 +48,8 @@ type StockScreenerPage = {
 };
 type Snapshot = { asOf: string; expiresAt: number; items: StockScreenerItem[]; revision: string; userId: string };
 type TbankAccess = {
-  current(userId: string): Promise<{ revision: string; token: string }>;
   local(userId: string): Promise<{ revision: string }>;
+  online<T>(userId: string, run: (connection: { revision: string; token: string }) => Promise<T>): Promise<T>;
 };
 
 @Injectable()
@@ -63,8 +63,10 @@ export class StockScreenerService {
   ) {}
 
   async search(userId: string, conditions: StockScreenerConditions): Promise<StockScreenerPage> {
-    const connection = await this.access.current(userId);
-    const instruments = await this.client.getAllShares(connection.token);
+    const { instruments, revision } = await this.access.online(userId, async (connection) => ({
+      instruments: await this.client.getAllShares(connection.token),
+      revision: connection.revision,
+    }));
     const asOf = new Date().toISOString();
     const selected = selectStockScreenerInstruments({ asOf: new Date(asOf), instruments }, conditions)
       .instruments.map((instrument) => this.item(instrument));
@@ -74,7 +76,7 @@ export class StockScreenerService {
       asOf,
       expiresAt: Date.now() + SNAPSHOT_TTL_MS,
       items: selected,
-      revision: connection.revision,
+      revision,
       userId,
     });
     this.logger.event('info', 'stock.screener.search.completed', { asOf, count: selected.length, operation: 'stock-screener-search' });

test/bonds.e2e-spec.ts

@@ -1,5 +1,6 @@
 import type { Server } from 'node:http';
 
+import { status } from '@grpc/grpc-js';
 import {
   ConflictException,
   INestApplication,
@@ -16,6 +17,13 @@ import { BondCatalogService } from '../src/modules/investments/instruments/bonds
 import { BondIncomeService } from '../src/modules/investments/instruments/bonds/application/bond-income.service';
 import { BondMarketService } from '../src/modules/investments/instruments/bonds/application/bond-market.service';
 import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler';
+import {
+  bootOnlineApp,
+  expectSafeBody,
+  failUpstream,
+  onlineUserId,
+  refusals,
+} from './online-upstream';
 
 const uid = 'e6123145-9665-43e0-8413-cd61b8aa9b13';
 describe('Bond HTTP contract', () => {
@@ -201,3 +209,39 @@ describe('Bond HTTP contract', () => {
     expect(response.body).not.toHaveProperty('stack');
   });
 });
+
+// Real services and TbankClient error mapping; only the gRPC transport is stubbed.
+// Bonds expose no company or screener route; their online entrypoints are search, detail and ticker.
+describe('Bond online reads: upstream failure translation', () => {
+  const rpc = jest.fn();
+  let ctx: Awaited<ReturnType<typeof bootOnlineApp>>;
+  const get = (path: string) => request(ctx.app.getHttpServer()).get(path);
+  // [entrypoint, request, RPC that fails second]: a failure after the first call must translate too.
+  const entrypoints: [string, () => request.Test, string][] = [
+    ['search', () => get('/api/v1/bonds?query=SBER'), 'getLastPrices'],
+    ['detail', () => get(`/api/v1/bonds/${uid}`), 'bondBy'],
+    ['ticker', () => get('/api/v1/bonds/by-ticker/SBER'), 'bondBy'],
+  ];
+  const cases = entrypoints.flatMap(([name, send, later]) =>
+    refusals.flatMap((code) => [undefined, later].map((at) => [name, code, at, send] as const)),
+  );
+
+  beforeAll(async () => {
+    ctx = await bootOnlineApp(rpc);
+  });
+  afterAll(() => ctx.app.close());
+
+  it.each(cases)('%s: upstream auth refusal %s (failing RPC %s) becomes 400/INVALID_TBANK_TOKEN', async (_name, code, at, send) => {
+    failUpstream(rpc, 'bond', code, at);
+    const response = await send().set('Authorization', ctx.bearer).expect(400);
+    expectSafeBody(response.body, 400, 'INVALID_TBANK_TOKEN');
+    const saved = await ctx.prisma.tbankConnection.findUniqueOrThrow({ where: { userId: onlineUserId } });
+    expect(saved).toMatchObject({ invalidToken: false, revision: ctx.connection.revision });
+  });
+
+  it.each(entrypoints)('%s: temporary unavailability stays 503/TBANK_UNAVAILABLE', async (_name, send) => {
+    failUpstream(rpc, 'bond', status.UNAVAILABLE);
+    const response = await send().set('Authorization', ctx.bearer).expect(503);
+    expectSafeBody(response.body, 503, 'TBANK_UNAVAILABLE');
+  });
+});

test/online-upstream.ts

@@ -0,0 +1,87 @@
+import type { Server } from 'node:http';
+
+import { status } from '@grpc/grpc-js';
+import { INestApplication } from '@nestjs/common';
+import { JwtService } from '@nestjs/jwt';
+import { Test } from '@nestjs/testing';
+
+import { AppModule } from '../src/app.module';
+import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
+import { TbankClient } from '../src/integrations/tbank/tbank.client';
+import { configureApp } from '../src/main';
+import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler';
+import { PrismaService } from '../src/prisma/prisma.service';
+
+export const onlineUserId = 'online-reader';
+export const onlineToken = 'synthetic-online-token';
+export const rawUpstreamText = 'raw upstream refusal text';
+export const refusals = [status.UNAUTHENTICATED, status.PERMISSION_DENIED];
+
+// Real TbankClient with only the transport stubbed: the gRPC status goes through its real error mapping.
+export async function bootOnlineApp(rpc: jest.Mock) {
+  const upstream = Object.assign(Object.create(TbankClient.prototype) as TbankClient, {
+    logger: { event: jest.fn() },
+    queue: { run: (operation: () => Promise<unknown>) => operation(), size: 0 },
+    request: rpc,
+  });
+  const module = await Test.createTestingModule({ imports: [AppModule] })
+    .overrideProvider(TbankClient)
+    .useValue(upstream)
+    .overrideProvider(TbankHourlyScheduler)
+    .useValue({ tick: jest.fn() })
+    .compile();
+  const app: INestApplication<Server> = module.createNestApplication();
+  configureApp(app);
+  await app.init();
+  await app.listen(0, '127.0.0.1');
+  const prisma = app.get(PrismaService);
+  await prisma.user.create({
+    data: { id: onlineUserId, email: 'online@example.invalid', passwordHash: 'synthetic' },
+  });
+  const connection = await prisma.tbankConnection.create({
+    data: {
+      userId: onlineUserId,
+      encryptedToken: app.get(TbankTokenService).encrypt(onlineToken),
+      connectedAt: new Date(),
+    },
+  });
+  const bearer = `Bearer ${new JwtService().sign(
+    { sub: onlineUserId, type: 'access' },
+    { secret: process.env.JWT_SECRET },
+  )}`;
+  return { app, bearer, connection, prisma };
+}
+
+// Healthy answers for every RPC an online entrypoint uses; `at` picks the single RPC that fails.
+export function failUpstream(rpc: jest.Mock, kind: 'bond' | 'share', code: number, at?: string) {
+  const instrument = {
+    apiTradeAvailableFlag: true,
+    assetUid: 'asset-uid',
+    instrumentKind: kind === 'bond' ? 'INSTRUMENT_TYPE_BOND' : 'INSTRUMENT_TYPE_SHARE',
+    ticker: 'SBER',
+    uid: 'e6123145-9665-43e0-8413-cd61b8aa9b13',
+  };
+  const healthy = new Map<string, unknown>([
+    ['bondBy', { instrument }],
+    ['findInstrument', { instruments: [instrument] }],
+    ['getAssetBy', { asset: { description: 'description' } }],
+    ['getAssetFundamentals', { fundamentals: [] }],
+    ['getLastPrices', { lastPrices: [] }],
+    ['shareBy', { instrument }],
+  ]);
+  const failure = Object.assign(new Error('grpc failure'), {
+    code,
+    details: `${rawUpstreamText} Bearer ${onlineToken}`,
+  });
+  rpc.mockReset();
+  rpc.mockImplementation((_token: string, _client: unknown, method: string) =>
+    !at || method === at ? Promise.reject(failure) : Promise.resolve(healthy.get(method) ?? {}),
+  );
+}
+
+// Safe public error: stable code only, no upstream text, no credentials.
+export function expectSafeBody(body: unknown, status: number, code: string) {
+  expect(body).toEqual({ code, status });
+  expect(JSON.stringify(body)).not.toContain(rawUpstreamText);
+  expect(JSON.stringify(body)).not.toContain(onlineToken);
+}

test/stocks.e2e-spec.ts

@@ -1,5 +1,6 @@
 import type { Server } from 'node:http';
 
+import { status } from '@grpc/grpc-js';
 import {
   ConflictException,
   INestApplication,
@@ -17,6 +18,13 @@ import { StockIncomeService } from '../src/modules/investments/instruments/stock
 import { StockMarketService } from '../src/modules/investments/instruments/stocks/application/stock-market.service';
 import { StockScreenerService } from '../src/modules/investments/instruments/stocks/application/stock-screener.service';
 import { TbankHourlyScheduler } from '../src/modules/investments/tbank-sync/tbank-hourly.scheduler';
+import {
+  bootOnlineApp,
+  expectSafeBody,
+  failUpstream,
+  onlineUserId,
+  refusals,
+} from './online-upstream';
 
 const uid = 'e6123145-9665-43e0-8413-cd61b8aa9b13';
 describe('Stock HTTP contract', () => {
@@ -201,3 +209,40 @@ describe('Stock HTTP contract', () => {
     expect(response.body).not.toHaveProperty('stack');
   });
 });
+
+// Real services and TbankClient error mapping; only the gRPC transport is stubbed.
+describe('Stock online reads: upstream failure translation', () => {
+  const rpc = jest.fn();
+  let ctx: Awaited<ReturnType<typeof bootOnlineApp>>;
+  const get = (path: string) => request(ctx.app.getHttpServer()).get(path);
+  // [entrypoint, request, RPC that fails second]: a failure after the first call must translate too.
+  const entrypoints: [string, () => request.Test, string][] = [
+    ['search', () => get('/api/v1/stocks?query=SBER'), 'getLastPrices'],
+    ['detail', () => get(`/api/v1/stocks/${uid}`), 'shareBy'],
+    ['ticker', () => get('/api/v1/stocks/by-ticker/SBER'), 'shareBy'],
+    ['company', () => get(`/api/v1/stocks/${uid}/company`), 'getAssetBy'],
+    ['screener', () => request(ctx.app.getHttpServer()).post('/api/v1/stocks/screener/search').send({}), 'shares'],
+  ];
+  const cases = entrypoints.flatMap(([name, send, later]) =>
+    refusals.flatMap((code) => [undefined, later].map((at) => [name, code, at, send] as const)),
+  );
+
+  beforeAll(async () => {
+    ctx = await bootOnlineApp(rpc);
+  });
+  afterAll(() => ctx.app.close());
+
+  it.each(cases)('%s: upstream auth refusal %s (failing RPC %s) becomes 400/INVALID_TBANK_TOKEN', async (_name, code, at, send) => {
+    failUpstream(rpc, 'share', code, at);
+    const response = await send().set('Authorization', ctx.bearer).expect(400);
+    expectSafeBody(response.body, 400, 'INVALID_TBANK_TOKEN');
+    const saved = await ctx.prisma.tbankConnection.findUniqueOrThrow({ where: { userId: onlineUserId } });
+    expect(saved).toMatchObject({ invalidToken: false, revision: ctx.connection.revision });
+  });
+
+  it.each(entrypoints)('%s: temporary unavailability stays 503/TBANK_UNAVAILABLE', async (_name, send) => {
+    failUpstream(rpc, 'share', status.UNAVAILABLE);
+    const response = await send().set('Authorization', ctx.bearer).expect(503);
+    expectSafeBody(response.body, 503, 'TBANK_UNAVAILABLE');
+  });
+});