diff --git a/.env.example b/.env.example index 5e7045785c..cf8e5287bb 100644 --- a/.env.example +++ b/.env.example @@ -344,3 +344,7 @@ REALUNIT_W2W_GAS_LOW_BALANCE_THRESHOLD=0.05 REQUEST_KNOWN_IPS= CRON_JOB_DELAY= + +# Optional: connection string to a throwaway Postgres for the specs that exercise real SQL. +# Unset, the "(real Postgres)" describe blocks skip. Existing convention across the repo. +MIGRATION_TEST_PG= diff --git a/src/subdomains/core/statistic/__tests__/partner-statistic-rate-limit.guard.spec.ts b/src/subdomains/core/statistic/__tests__/partner-statistic-rate-limit.guard.spec.ts new file mode 100644 index 0000000000..6db8a7e5a9 --- /dev/null +++ b/src/subdomains/core/statistic/__tests__/partner-statistic-rate-limit.guard.spec.ts @@ -0,0 +1,25 @@ +import { PartnerStatisticRateLimitGuard } from '../partner-statistic-rate-limit.guard'; + +describe('PartnerStatisticRateLimitGuard', () => { + const guard = new PartnerStatisticRateLimitGuard({} as any, {} as any, {} as any); + + const getTracker = (req: Record): string => guard['getTracker'](req); + + it('keys by jwt.user (wallet id) when present', () => { + expect(getTracker({ user: { user: 42 }, realIp: '1.2.3.4' })).toBe('partner-stat:wallet:42'); + expect(getTracker({ user: { user: 99 }, realIp: '1.2.3.4' })).toBe('partner-stat:wallet:99'); + expect(getTracker({ user: { user: 42 }, realIp: '1.2.3.4' })).not.toBe( + getTracker({ user: { user: 99 }, realIp: '1.2.3.4' }), + ); + }); + + it('does not share a bucket across wallets on the same IP', () => { + const a = getTracker({ user: { user: 1 }, realIp: '185.12.34.56' }); + const b = getTracker({ user: { user: 2 }, realIp: '185.12.34.56' }); + expect(a).not.toEqual(b); + }); + + it('falls back to IP when jwt.user is missing', () => { + expect(getTracker({ realIp: '9.9.9.9' })).toBe('partner-stat:ip:9.9.9.9'); + }); +}); diff --git a/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts b/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts new file mode 100644 index 0000000000..b8dfc10e39 --- /dev/null +++ b/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts @@ -0,0 +1,298 @@ +import { DataSource } from 'typeorm'; +import { ConfigService } from 'src/config/config'; +import { PartnerStatisticService } from '../partner-statistic.service'; +import { suppressPeriodTotals, suppressTimelineBuckets } from '../partner-statistic.suppression'; + +/** + * Real-Postgres integration for the partner-statistic SQL shapes. + * Skipped unless MIGRATION_TEST_PG is set (CI / local disposable DB). + * + * Covers: half-open period filter, UNION active-user count, breakdown GROUP BY + * rows fed through mergeNamedRows, and a timeline DATE_TRUNC bucket — paths the + * unit mocks previously never exercised with real rows. + * + * Uses string-based QueryBuilder (no full entity graph) against a minimal schema + * that mirrors the production join columns. + */ +const PG_URL = process.env.MIGRATION_TEST_PG; +const describeDb = PG_URL ? describe : describe.skip; +const SCHEMA = 'partner_statistic_spec'; + +process.env.TZ = 'UTC'; + +describeDb('PartnerStatisticService SQL path (real Postgres)', () => { + let dataSource: DataSource; + let service: PartnerStatisticService; + + beforeAll(async () => { + new ConfigService(); + dataSource = new DataSource({ type: 'postgres', url: PG_URL }); + await dataSource.initialize(); + }); + + beforeEach(async () => { + await dataSource.query(`DROP SCHEMA IF EXISTS "${SCHEMA}" CASCADE`); + await dataSource.query(`CREATE SCHEMA "${SCHEMA}"`); + await dataSource.query(`SET search_path TO "${SCHEMA}"`); + + await dataSource.query(` + CREATE TABLE "user" ( + "id" SERIAL PRIMARY KEY, + "walletId" int NOT NULL, + "buyVolume" numeric DEFAULT 0, + "sellVolume" numeric DEFAULT 0, + "created" TIMESTAMP NOT NULL DEFAULT NOW() + ); + CREATE TABLE "buy" ( + "id" SERIAL PRIMARY KEY, + "userId" int NOT NULL + ); + CREATE TABLE "sell" ( + "id" SERIAL PRIMARY KEY, + "userId" int NOT NULL + ); + CREATE TABLE "asset" ( + "id" SERIAL PRIMARY KEY, + "name" varchar(256), + "blockchain" varchar(256) + ); + CREATE TABLE "transaction" ( + "id" SERIAL PRIMARY KEY, + "sourceType" varchar(256) + ); + CREATE TABLE "buy_crypto" ( + "id" SERIAL PRIMARY KEY, + "buyId" int, + "outputAssetId" int, + "transactionId" int, + "amountInChf" numeric DEFAULT 0, + "inputAsset" varchar(256), + "amlCheck" varchar(64), + "isComplete" boolean DEFAULT false, + "created" TIMESTAMP NOT NULL + ); + CREATE TABLE "buy_fiat" ( + "id" SERIAL PRIMARY KEY, + "sellId" int, + "outputAssetId" int, + "transactionId" int, + "amountInChf" numeric DEFAULT 0, + "inputAsset" varchar(256), + "amlCheck" varchar(64), + "isComplete" boolean DEFAULT false, + "created" TIMESTAMP NOT NULL + ); + `); + + await dataSource.query(` + INSERT INTO "user" ("id", "walletId", "buyVolume", "sellVolume", "created") VALUES + (1, 1, 100, 0, '2024-06-01 10:00:00'), + (2, 1, 100, 0, '2024-06-01 11:00:00'), + (3, 1, 100, 0, '2024-06-01 12:00:00'), + (4, 1, 100, 0, '2024-06-01 13:00:00'), + (5, 1, 100, 50, '2024-06-01 14:00:00'), + (6, 2, 999, 0, '2024-06-01 10:00:00'); + INSERT INTO "buy" ("id", "userId") VALUES (1,1),(2,2),(3,3),(4,4),(5,5),(6,6); + INSERT INTO "sell" ("id", "userId") VALUES (1,5); + INSERT INTO "asset" ("id", "name", "blockchain") VALUES (1, 'BTC', 'Bitcoin'), (2, 'ETH', 'Ethereum'); + INSERT INTO "transaction" ("id", "sourceType") VALUES + (1, 'BankTx'), (2, 'BankTx'), (3, 'BankTx'), (4, 'BankTx'), (5, 'BankTx'), (6, 'BankTx'); + INSERT INTO "buy_crypto" + ("buyId", "outputAssetId", "transactionId", "amountInChf", "amlCheck", "isComplete", "created", "inputAsset") + VALUES + (1, 1, 1, 100, 'Pass', true, '2024-06-10 10:00:00', 'CHF'), + (2, 1, 2, 100, 'Pass', true, '2024-06-10 11:00:00', 'CHF'), + (3, 1, 3, 100, 'Pass', true, '2024-06-10 12:00:00', 'CHF'), + (4, 1, 4, 100, 'Pass', true, '2024-06-11 10:00:00', 'CHF'), + (5, 1, 5, 100, 'Pass', true, '2024-06-11 11:00:00', 'CHF'), + (6, 2, 6, 9999, 'Pass', true, '2024-06-10 10:00:00', 'CHF'); + INSERT INTO "buy_fiat" + ("sellId", "outputAssetId", "transactionId", "amountInChf", "amlCheck", "isComplete", "created", "inputAsset") + VALUES + (1, 1, 1, 50, 'Pass', true, '2024-06-11 12:00:00', 'BTC'); + `); + + // Lightweight service shell for mergeNamedRows only (repos unused for SQL below). + service = new PartnerStatisticService( + { manager: dataSource.manager } as any, + {} as any, + {} as any, + {} as any, + {} as any, + ); + }); + + afterEach(async () => { + await dataSource.query(`SET search_path TO public`); + await dataSource.query(`DROP SCHEMA IF EXISTS "${SCHEMA}" CASCADE`); + }); + + afterAll(async () => { + if (dataSource?.isInitialized) await dataSource.destroy(); + }); + + it('aggregates volume, UNION active users, breakdown rows, and timeline buckets on real rows', async () => { + const from = new Date('2024-06-01T00:00:00.000Z'); + const to = new Date('2024-07-01T00:00:00.000Z'); + const walletId = 1; + + // Direction aggregate (buy) — half-open, scoped, aml Pass + const buyAgg = await dataSource + .createQueryBuilder() + .select('COALESCE(SUM(tx."amountInChf"), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT u.id)', 'users') + .from('buy_crypto', 'tx') + .innerJoin('buy', 'route', 'route.id = tx."buyId"') + .innerJoin('user', 'u', 'u.id = route."userId"') + .where('u."walletId" = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }) + .andWhere('tx."amlCheck" = :check', { check: 'Pass' }) + .getRawOne<{ volume: string; transactions: string; users: string }>(); + + expect(+buyAgg.volume).toBe(500); + expect(+buyAgg.transactions).toBe(5); + expect(+buyAgg.users).toBe(5); + + // Foreign wallet row must not leak + expect(+buyAgg.volume).not.toBe(500 + 9999); + + // UNION distinct active users across buy + sell + const buyIds = dataSource + .createQueryBuilder() + .select('u.id', 'id') + .from('buy_crypto', 'tx') + .innerJoin('buy', 'route', 'route.id = tx."buyId"') + .innerJoin('user', 'u', 'u.id = route."userId"') + .where('u."walletId" = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }) + .andWhere('tx."amlCheck" = :check', { check: 'Pass' }); + const sellIds = dataSource + .createQueryBuilder() + .select('u.id', 'id') + .from('buy_fiat', 'tx') + .innerJoin('sell', 'route', 'route.id = tx."sellId"') + .innerJoin('user', 'u', 'u.id = route."userId"') + .where('u."walletId" = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }) + .andWhere('tx."amlCheck" = :check', { check: 'Pass' }); + + const unionSql = `(${buyIds.getQuery()}) UNION (${sellIds.getQuery()})`; + const active = await dataSource + .createQueryBuilder() + .select('COUNT(*)', 'count') + .from(`(${unionSql})`, 'active_users') + .setParameters({ ...buyIds.getParameters(), ...sellIds.getParameters() }) + .getRawOne<{ count: string }>(); + expect(+active.count).toBe(5); + + // Breakdown by asset name + blockchain (GROUP BY qualified columns) + const assetRows = await dataSource + .createQueryBuilder() + .select('a.name', 'name') + .addSelect('a.blockchain', 'blockchain') + .addSelect('COALESCE(SUM(tx."amountInChf"), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT u.id)', 'users') + .from('buy_crypto', 'tx') + .innerJoin('buy', 'route', 'route.id = tx."buyId"') + .innerJoin('user', 'u', 'u.id = route."userId"') + .leftJoin('asset', 'a', 'a.id = tx."outputAssetId"') + .where('u."walletId" = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }) + .andWhere('tx."amlCheck" = :check', { check: 'Pass' }) + .groupBy('a.name') + .addGroupBy('a.blockchain') + .getRawMany<{ name: string; blockchain: string; volume: string; transactions: string; users: string }>(); + + const merged = service.mergeNamedRows( + assetRows.map((r) => ({ + name: r.name, + blockchain: r.blockchain, + volume: r.volume, + transactions: r.transactions, + users: r.users, + })), + ); + expect(merged.find((r) => r.name === 'BTC')?.volume).toBe(500); + expect(merged.find((r) => r.name === 'BTC')?.transactions).toBe(5); + // mergeNamedRows must actually run — a no-op would leave volume as string or wrong + expect(typeof merged[0].volume).toBe('number'); + + // Timeline DATE_TRUNC day buckets (UTC-tagged, same expression shape as the service) + const timelineTrunc = `DATE_TRUNC('day', tx.created) AT TIME ZONE 'UTC'`; + const timelineRows = await dataSource + .createQueryBuilder() + .select(timelineTrunc, 'bucket') + .addSelect('COALESCE(SUM(tx."amountInChf"), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT u.id)', 'users') + .from('buy_crypto', 'tx') + .innerJoin('buy', 'route', 'route.id = tx."buyId"') + .innerJoin('user', 'u', 'u.id = route."userId"') + .where('u."walletId" = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { + from: new Date('2024-06-10T00:00:00.000Z'), + to: new Date('2024-06-12T00:00:00.000Z'), + }) + .andWhere('tx."amlCheck" = :check', { check: 'Pass' }) + .groupBy(timelineTrunc) + .orderBy(timelineTrunc, 'ASC') + .getRawMany<{ bucket: Date; volume: string; transactions: string; users: string }>(); + + expect(timelineRows.length).toBe(2); + expect(+timelineRows[0].transactions).toBe(3); + expect(+timelineRows[1].transactions).toBe(2); + + // Suppression integration: day-1 (3 txs) under k → suppressed + const buckets = timelineRows.map((r) => ({ + date: new Date(r.bucket), + volume: { buy: +r.volume, sell: 0, swap: 0 }, + transactions: { buy: +r.transactions, sell: 0, swap: 0 }, + users: { buy: +r.users, sell: 0, swap: 0 }, + suppressed: false, + partial: false, + })); + const { buckets: suppressed } = suppressTimelineBuckets(buckets); + expect(suppressed[0].suppressed).toBe(true); + expect(suppressed[1].suppressed).toBe(true); // complementary or under k + + // Period totals block rule on real aggregates + const sellAgg = await dataSource + .createQueryBuilder() + .select('COALESCE(SUM(tx."amountInChf"), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT u.id)', 'users') + .from('buy_fiat', 'tx') + .innerJoin('sell', 'route', 'route.id = tx."sellId"') + .innerJoin('user', 'u', 'u.id = route."userId"') + .where('u."walletId" = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }) + .andWhere('tx."amlCheck" = :check', { check: 'Pass' }) + .getRawOne<{ volume: string; transactions: string; users: string }>(); + + const totals = suppressPeriodTotals( + { + buy: +buyAgg.volume, + sell: +sellAgg.volume, + swap: 0, + total: +buyAgg.volume + +sellAgg.volume, + }, + { + buy: +buyAgg.transactions, + sell: +sellAgg.transactions, + swap: 0, + total: +buyAgg.transactions + +sellAgg.transactions, + }, + { + buy: +buyAgg.users, + sell: +sellAgg.users, + swap: 0, + total: +active.count, + }, + ); + // sell has 1 tx → entire totals group null + expect(totals.volume.total).toBeNull(); + expect(totals.volume.buy).toBeNull(); + }); +}); diff --git a/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts b/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts new file mode 100644 index 0000000000..18c0aed1ef --- /dev/null +++ b/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts @@ -0,0 +1,930 @@ +import { BadRequestException } from '@nestjs/common'; +import { Test, TestingModule } from '@nestjs/testing'; +import { ConfigService } from 'src/config/config'; +import { Util } from 'src/shared/utils/util'; +import { BuyCryptoRepository } from 'src/subdomains/core/buy-crypto/process/repositories/buy-crypto.repository'; +import { BuyFiatRepository } from 'src/subdomains/core/sell-crypto/process/buy-fiat.repository'; +import { UserRepository } from 'src/subdomains/generic/user/models/user/user.repository'; +import { WalletRepository } from 'src/subdomains/generic/user/models/wallet/wallet.repository'; +import { + TransactionRequestStatus, + TransactionRequestType, +} from 'src/subdomains/supporting/payment/entities/transaction-request.entity'; +import { TransactionRequestRepository } from 'src/subdomains/supporting/payment/repositories/transaction-request.repository'; +import { + PARTNER_STATISTIC_DEFAULT_PERIOD_DAYS, + PARTNER_STATISTIC_MAX_PERIOD_DAYS, + PartnerStatisticDirection, +} from '../partner-statistic.enum'; +import { PartnerStatisticService } from '../partner-statistic.service'; + +// Timeline and period resolution are UTC-only (no process.TZ pin). Specs must stay green under +// arbitrary host timezones — see "timezone independence" below. + +type Direction = PartnerStatisticDirection; + +/** + * Single calendar anchor for all period fixtures. Every from/to is derived from this + * (or from `new Date()` via fake timers) so tests do not go stale as wall-clock years pass. + */ +const TEST_NOW = new Date('2024-06-15T12:00:00.000Z'); +const PERIOD_FROM = Util.daysBefore(30, TEST_NOW); +const PERIOD_TO = TEST_NOW; +/** Wednesday UTC before TEST_NOW (Saturday) — mid-week so week-edge buckets are partial. */ +const MID_WEEK_FROM = Util.daysBefore(3, TEST_NOW); +const MID_WEEK_TO = Util.daysAfter(7, MID_WEEK_FROM); + +interface SettlementFixture { + received: number; + delivered: number; + rejected: number; +} + +interface PaymentInfoRow { + type: TransactionRequestType; + status: TransactionRequestStatus; + count: number; +} + +interface DirectionFixture { + volume: number; + transactions: number; + users: number; +} + +interface WalletFixture { + buy: DirectionFixture; + sell: DirectionFixture; + swap: DirectionFixture; + activeUserIds: number[]; + newUsers: number; + allTime: { buy: number; sell: number; registeredUsers: number; tradingUsers: number }; + referral: { volume: number; partnerRefCredit: number; refCredit: number; paidRefCredit: number }; + paymentInfo: PaymentInfoRow[]; + settlement: Record; + /** Named breakdown rows returned by getRawMany for asset/fiat/blockchain/payment queries. */ + namedRows: { name: string; blockchain?: string; volume: number; transactions: number; users: number }[]; + timelineRows: { bucket: Date; volume: number; transactions: number; users: number }[]; +} + +function emptySettlement(): SettlementFixture { + return { received: 0, delivered: 0, rejected: 0 }; +} + +function emptyFixture(overrides: Partial = {}): WalletFixture { + return { + buy: { volume: 0, transactions: 0, users: 0 }, + sell: { volume: 0, transactions: 0, users: 0 }, + swap: { volume: 0, transactions: 0, users: 0 }, + activeUserIds: [], + newUsers: 0, + allTime: { buy: 0, sell: 0, registeredUsers: 0, tradingUsers: 0 }, + referral: { volume: 0, partnerRefCredit: 0, refCredit: 0, paidRefCredit: 0 }, + paymentInfo: [], + settlement: { + [PartnerStatisticDirection.BUY]: emptySettlement(), + [PartnerStatisticDirection.SELL]: emptySettlement(), + [PartnerStatisticDirection.SWAP]: emptySettlement(), + }, + namedRows: [], + timelineRows: [], + ...overrides, + }; +} + +interface QbState { + walletId?: number; + direction?: Direction; + selects: string[]; + selectAliases: string[]; + groupBys: string[]; + isTimeline: boolean; + isAllTime: boolean; + isReferral: boolean; + isSettlement: boolean; + isPaymentInfo: boolean; + amlPassOnly: boolean; +} + +interface GroupByCapture { + groupBys: string[]; + selectAliases: string[]; +} + +describe('PartnerStatisticService', () => { + let service: PartnerStatisticService; + let fixtures: Map; + let lastWalletIds: number[]; + let whereClauses: string[]; + let amlFilterClauses: string[]; + let settlementAmlPassOnly: boolean[]; + let groupByCapture: GroupByCapture; + let managerCreateQueryBuilderCalls: number; + let activeUserCountFromManager: number | undefined; + let getRawManyCalls: number; + + /** + * Records walletId from query params when present and returns it for fixture routing. + * `params &&` already excludes null/undefined — no separate null check (CodeQL inconvertible types). + */ + function trackWalletId(params?: unknown): number | undefined { + if (params && typeof params === 'object' && 'walletId' in params) { + const id = (params as { walletId: unknown }).walletId; + if (typeof id === 'number') { + lastWalletIds.push(id); + return id; + } + } + return undefined; + } + + function fixtureFor(walletId?: number): WalletFixture { + if (walletId == null) { + const merged = emptyFixture(); + for (const f of fixtures.values()) { + merged.buy.volume += f.buy.volume; + merged.buy.transactions += f.buy.transactions; + merged.buy.users += f.buy.users; + merged.sell.volume += f.sell.volume; + merged.sell.transactions += f.sell.transactions; + merged.sell.users += f.sell.users; + merged.swap.volume += f.swap.volume; + merged.swap.transactions += f.swap.transactions; + merged.swap.users += f.swap.users; + merged.activeUserIds.push(...f.activeUserIds); + merged.newUsers += f.newUsers; + merged.allTime.buy += f.allTime.buy; + merged.allTime.sell += f.allTime.sell; + merged.allTime.registeredUsers += f.allTime.registeredUsers; + merged.allTime.tradingUsers += f.allTime.tradingUsers; + merged.referral.volume += f.referral.volume; + merged.referral.partnerRefCredit += f.referral.partnerRefCredit; + merged.referral.refCredit += f.referral.refCredit; + merged.referral.paidRefCredit += f.referral.paidRefCredit; + merged.paymentInfo.push(...f.paymentInfo); + for (const d of [ + PartnerStatisticDirection.BUY, + PartnerStatisticDirection.SELL, + PartnerStatisticDirection.SWAP, + ]) { + merged.settlement[d].received += f.settlement[d].received; + merged.settlement[d].delivered += f.settlement[d].delivered; + merged.settlement[d].rejected += f.settlement[d].rejected; + } + merged.namedRows.push(...f.namedRows); + merged.timelineRows.push(...f.timelineRows); + } + return merged; + } + return fixtures.get(walletId) ?? emptyFixture(); + } + + function createQb(kind: 'buyCrypto' | 'buyFiat' | 'user' | 'wallet' | 'txRequest') { + const state: QbState = { + selects: [], + selectAliases: [], + groupBys: [], + isTimeline: false, + isAllTime: false, + isReferral: false, + isSettlement: false, + isPaymentInfo: kind === 'txRequest', + amlPassOnly: false, + }; + + const qb: Record unknown)> = {}; + const self = () => qb; + + const recordSelect = (clause: unknown, alias?: unknown) => { + const expr = String(clause); + const label = alias != null ? String(alias) : expr; + state.selects.push(expr); + if (alias != null) { + state.selectAliases.push(String(alias)); + groupByCapture.selectAliases.push(String(alias)); + } + if (expr.includes('DATE_TRUNC') || label === 'bucket') state.isTimeline = true; + if (label === 'registeredUsers' || expr.includes('registeredUsers')) state.isAllTime = true; + if (label === 'partnerRefCredit' || expr.includes('partnerRefCredit')) state.isReferral = true; + if (label === 'received' || label === 'delivered' || label === 'rejected') state.isSettlement = true; + if (kind === 'txRequest' || label === 'type' || label === 'status') state.isPaymentInfo = true; + }; + + const recordGroupBy = (clause: unknown) => { + const g = String(clause); + state.groupBys.push(g); + groupByCapture.groupBys.push(g); + }; + + qb.select = jest.fn((clause: unknown, alias?: unknown) => { + recordSelect(clause, alias); + return self(); + }); + qb.addSelect = jest.fn((clause: unknown, alias?: unknown) => { + recordSelect(clause, alias); + return self(); + }); + qb.innerJoin = jest.fn((path: string) => { + if (path === 'tx.buy') state.direction = PartnerStatisticDirection.BUY; + if (path === 'tx.cryptoRoute') state.direction = PartnerStatisticDirection.SWAP; + if (path === 'tx.sell') state.direction = PartnerStatisticDirection.SELL; + return self(); + }); + qb.leftJoin = jest.fn(() => self()); + qb.groupBy = jest.fn((clause: unknown) => { + recordGroupBy(clause); + return self(); + }); + qb.addGroupBy = jest.fn((clause: unknown) => { + recordGroupBy(clause); + return self(); + }); + qb.orderBy = jest.fn(() => self()); + qb.setParameter = jest.fn(() => self()); + qb.where = jest.fn((clause: unknown, params?: unknown) => { + whereClauses.push(String(clause)); + const walletId = trackWalletId(params); + if (walletId !== undefined) state.walletId = walletId; + return self(); + }); + qb.andWhere = jest.fn((clause: unknown, params?: unknown) => { + const clauseStr = String(clause); + whereClauses.push(clauseStr); + const walletId = trackWalletId(params); + if (walletId !== undefined) state.walletId = walletId; + if (clauseStr.includes('amlCheck')) { + amlFilterClauses.push(clauseStr); + if (params && typeof params === 'object' && 'check' in (params as object)) { + state.amlPassOnly = true; + } + } + return self(); + }); + qb.getQuery = jest.fn(() => `SELECT user.id AS id FROM mock_${kind}_${state.direction ?? 'x'}`); + qb.getParameters = jest.fn(() => ({ + walletId: state.walletId, + check: 'Pass', + from: new Date(TEST_NOW), + to: new Date(TEST_NOW), + })); + + qb.getRawOne = jest.fn(async () => { + const f = fixtureFor(state.walletId); + + if (kind === 'wallet' || state.isReferral) { + return { + volume: f.referral.volume, + partnerRefCredit: f.referral.partnerRefCredit, + refCredit: f.referral.refCredit, + paidRefCredit: f.referral.paidRefCredit, + }; + } + + if (kind === 'user' || state.isAllTime) { + return { + buy: f.allTime.buy, + sell: f.allTime.sell, + registeredUsers: f.allTime.registeredUsers, + tradingUsers: f.allTime.tradingUsers, + }; + } + + if (state.isTimeline) return null; + + if (state.isSettlement) { + settlementAmlPassOnly.push(state.amlPassOnly); + const dir = + state.direction ?? (kind === 'buyFiat' ? PartnerStatisticDirection.SELL : PartnerStatisticDirection.BUY); + const s = f.settlement[dir]; + return { received: s.received, delivered: s.delivered, rejected: s.rejected }; + } + + const dir = + state.direction ?? (kind === 'buyFiat' ? PartnerStatisticDirection.SELL : PartnerStatisticDirection.BUY); + const agg = f[dir]; + return { volume: agg.volume, transactions: agg.transactions, users: agg.users }; + }); + + qb.getRawMany = jest.fn(async () => { + getRawManyCalls += 1; + if (kind === 'txRequest' || state.isPaymentInfo) { + const f = fixtureFor(state.walletId); + return f.paymentInfo.map((r) => ({ type: r.type, status: r.status, count: r.count })); + } + if (state.isTimeline) { + const f = fixtureFor(state.walletId); + return f.timelineRows.map((r) => ({ + bucket: r.bucket, + volume: r.volume, + transactions: r.transactions, + users: r.users, + })); + } + // Breakdown path — exercise mergeNamedRows + const f = fixtureFor(state.walletId); + return f.namedRows.map((r) => ({ + name: r.name, + blockchain: r.blockchain ?? null, + volume: r.volume, + transactions: r.transactions, + users: r.users, + })); + }); + + qb.getCount = jest.fn(async () => fixtureFor(state.walletId).newUsers); + + return qb; + } + + function createManagerQb() { + const state: { walletId?: number } = {}; + const qb: Record = {}; + const self = () => qb; + + qb.select = jest.fn(() => self()); + qb.from = jest.fn(() => self()); + qb.setParameters = jest.fn((params: Record) => { + const walletId = trackWalletId(params); + if (walletId !== undefined) state.walletId = walletId; + return self(); + }); + qb.getRawOne = jest.fn(async () => { + managerCreateQueryBuilderCalls += 1; + const f = fixtureFor(state.walletId); + const count = activeUserCountFromManager ?? f.activeUserIds.length; + return { count }; + }); + + return qb; + } + + beforeEach(async () => { + lastWalletIds = []; + whereClauses = []; + amlFilterClauses = []; + settlementAmlPassOnly = []; + fixtures = new Map(); + groupByCapture = { groupBys: [], selectAliases: [] }; + managerCreateQueryBuilderCalls = 0; + activeUserCountFromManager = undefined; + getRawManyCalls = 0; + new ConfigService(); + + const buyCryptoRepo = { + createQueryBuilder: jest.fn(() => createQb('buyCrypto')), + manager: { + createQueryBuilder: jest.fn(() => createManagerQb()), + }, + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + PartnerStatisticService, + { provide: BuyCryptoRepository, useValue: buyCryptoRepo }, + { provide: BuyFiatRepository, useValue: { createQueryBuilder: jest.fn(() => createQb('buyFiat')) } }, + { provide: UserRepository, useValue: { createQueryBuilder: jest.fn(() => createQb('user')) } }, + { provide: WalletRepository, useValue: { createQueryBuilder: jest.fn(() => createQb('wallet')) } }, + { + provide: TransactionRequestRepository, + useValue: { createQueryBuilder: jest.fn(() => createQb('txRequest')) }, + }, + ], + }).compile(); + + service = module.get(PartnerStatisticService); + }); + + // --- PERIOD VALIDATION --- // + + describe('resolvePeriod', () => { + it('defaults to the last 30 days snapped to UTC day boundaries when from/to are omitted', () => { + jest.useFakeTimers().setSystemTime(TEST_NOW); + + const period = service.resolvePeriod(); + // to = start of day after TEST_NOW's UTC day + expect(period.to.toISOString()).toBe('2024-06-16T00:00:00.000Z'); + expect(period.from.toISOString()).toBe( + new Date(Date.UTC(2024, 5, 15 - PARTNER_STATISTIC_DEFAULT_PERIOD_DAYS)).toISOString(), + ); + + jest.useRealTimers(); + }); + + it('snaps from to UTC day start and to to exclusive end of day', () => { + const period = service.resolvePeriod('2024-06-01T15:30:00.000Z', '2024-06-03T08:00:00.000Z'); + expect(period.from.toISOString()).toBe('2024-06-01T00:00:00.000Z'); + expect(period.to.toISOString()).toBe('2024-06-04T00:00:00.000Z'); + }); + + it('rejects from ≥ to after snap', () => { + expect(() => service.resolvePeriod('2024-06-05', '2024-06-04')).toThrow(BadRequestException); + expect(() => service.resolvePeriod('2024-06-05', '2024-06-04')).toThrow(/From must be before to/); + }); + + it('rejects periods longer than the max span', () => { + const from = Util.daysBefore(PARTNER_STATISTIC_MAX_PERIOD_DAYS + 50, TEST_NOW); + const to = TEST_NOW; + expect(() => service.resolvePeriod(from, to)).toThrow(BadRequestException); + expect(() => service.resolvePeriod(from, to)).toThrow(String(PARTNER_STATISTIC_MAX_PERIOD_DAYS)); + }); + + it('accepts a period of exactly the max span', () => { + const from = Util.daysBefore(PARTNER_STATISTIC_MAX_PERIOD_DAYS - 1, TEST_NOW); + const to = TEST_NOW; + expect(() => service.resolvePeriod(from, to)).not.toThrow(); + }); + + it('accepts a single full day', () => { + const period = service.resolvePeriod('2024-06-01T00:00:00.000Z', '2024-06-01T23:59:59.000Z'); + expect(period.from.toISOString()).toBe('2024-06-01T00:00:00.000Z'); + expect(period.to.toISOString()).toBe('2024-06-02T00:00:00.000Z'); + }); + }); + + describe('parseGranularity', () => { + it('rejects invalid granularity with a clear message', () => { + expect(() => service.parseGranularity('year')).toThrow(BadRequestException); + expect(() => service.parseGranularity('year')).toThrow(/day, week, month/); + }); + + it('accepts day|week|month', () => { + expect(service.parseGranularity('day')).toBe('day'); + expect(service.parseGranularity('week')).toBe('week'); + expect(service.parseGranularity('month')).toBe('month'); + }); + }); + + // --- B1: GROUP BY must not use SELECT aliases --- // + + describe('groupBy uses qualified columns, never SELECT aliases (B1)', () => { + it('records only qualified columns or DATE_TRUNC expressions for every groupBy across all breakdowns', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 1000, transactions: 20, users: 10 }, + sell: { volume: 200, transactions: 10, users: 8 }, + swap: { volume: 50, transactions: 10, users: 6 }, + allTime: { buy: 5000, sell: 1000, registeredUsers: 100, tradingUsers: 40 }, + newUsers: 8, + activeUserIds: [1, 2, 3, 4, 5, 6], + paymentInfo: [{ type: TransactionRequestType.BUY, status: TransactionRequestStatus.CREATED, count: 10 }], + }), + ); + + await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + await service.getTimeline(1, PERIOD_FROM, Util.daysBefore(16, PERIOD_TO), 'day'); + + expect(groupByCapture.groupBys.length).toBeGreaterThan(0); + + const aliases = new Set(groupByCapture.selectAliases); + for (const g of groupByCapture.groupBys) { + const isQualified = g.includes('.'); + const isDateTrunc = g.startsWith('DATE_TRUNC('); + expect(isQualified || isDateTrunc).toBe(true); + expect(aliases.has(g)).toBe(false); + } + + expect(groupByCapture.groupBys).toEqual( + expect.arrayContaining([ + 'tx.inputAsset', + 'inputAsset.blockchain', + 'outputAsset.name', + 'outputAsset.blockchain', + 'tr.type', + 'tr.status', + ]), + ); + }); + }); + + // --- SCOPE ISOLATION --- // + + describe('wallet scope isolation', () => { + beforeEach(() => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 1000, transactions: 10, users: 8 }, + sell: { volume: 200, transactions: 5, users: 5 }, + swap: { volume: 50, transactions: 5, users: 5 }, + allTime: { buy: 5000, sell: 1000, registeredUsers: 100, tradingUsers: 40 }, + referral: { volume: 50, partnerRefCredit: 10, refCredit: 0, paidRefCredit: 4 }, + newUsers: 8, + activeUserIds: [11, 12, 13, 14, 15, 16], + paymentInfo: [{ type: TransactionRequestType.BUY, status: TransactionRequestStatus.COMPLETED, count: 20 }], + settlement: { + // inProgress = received − delivered − rejected must also be 0 or ≥ k + [PartnerStatisticDirection.BUY]: { received: 20, delivered: 15, rejected: 0 }, + [PartnerStatisticDirection.SELL]: emptySettlement(), + [PartnerStatisticDirection.SWAP]: emptySettlement(), + }, + }), + ); + fixtures.set( + 2, + emptyFixture({ + buy: { volume: 99999, transactions: 999, users: 100 }, + sell: { volume: 88888, transactions: 888, users: 90 }, + swap: { volume: 77777, transactions: 777, users: 80 }, + allTime: { buy: 77777, sell: 66666, registeredUsers: 9999, tradingUsers: 8888 }, + referral: { volume: 12345, partnerRefCredit: 999, refCredit: 0, paidRefCredit: 111 }, + newUsers: 500, + activeUserIds: [21, 22, 23, 24, 25, 26, 27, 28, 29, 30], + paymentInfo: [{ type: TransactionRequestType.BUY, status: TransactionRequestStatus.COMPLETED, count: 9000 }], + settlement: { + [PartnerStatisticDirection.BUY]: { received: 9000, delivered: 8000, rejected: 100 }, + [PartnerStatisticDirection.SELL]: emptySettlement(), + [PartnerStatisticDirection.SWAP]: emptySettlement(), + }, + }), + ); + }); + + it('returns only wallet A aggregates and scopes SQL to user.walletId on every user-related query', async () => { + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + + expect(lastWalletIds.length).toBeGreaterThan(0); + expect(lastWalletIds.every((id) => id === 1)).toBe(true); + + // Every walletId-binding clause must use the wallet column — removing scope from one query fails this. + const scopeClauses = whereClauses.filter((c) => c.includes('walletId')); + expect(scopeClauses.length).toBeGreaterThan(0); + expect(scopeClauses.every((c) => c.includes('user.walletId') || c.includes('wallet.id'))).toBe(true); + expect(whereClauses.every((c) => !/user\.id\s*=\s*:walletId/.test(c))).toBe(true); + + expect(amlFilterClauses.length).toBeGreaterThan(0); + expect(amlFilterClauses.every((c) => /amlCheck\s*=\s*:check/.test(c))).toBe(true); + expect(settlementAmlPassOnly.length).toBeGreaterThan(0); + expect(settlementAmlPassOnly.every((v) => v === false)).toBe(true); + + expect(result.totals.volume.buy).toBe(1000); + expect(result.totals.volume.sell).toBe(200); + expect(result.totals.volume.swap).toBe(50); + expect(result.allTime.registeredUsers).toBe(100); + expect(result.referral.volume).toBe(50); + expect(result.referral.creditEarned).toBe(10); + expect(result.totals.newUsers).toBe(8); + + expect(result.totals.volume.buy).not.toBe(99999); + expect(result.allTime.registeredUsers).not.toBe(9999); + expect(result.referral.volume).not.toBe(12345); + expect(result.totals.newUsers).not.toBe(500); + + expect(result.completion.paymentInfoRequests.buy.paymentReceived).toBe(20); + expect(result.completion.paymentInfoRequests.buy.paymentReceived).not.toBe(9000); + expect(result.completion.settlement.buy.received).toBe(20); + expect(result.completion.settlement.buy.received).not.toBe(9000); + }); + + it('returns only wallet B aggregates when called with wallet B (not wallet A)', async () => { + lastWalletIds = []; + whereClauses = []; + const result = await service.getStatistics(2, PERIOD_FROM, PERIOD_TO); + + expect(lastWalletIds.length).toBeGreaterThan(0); + expect(lastWalletIds.every((id) => id === 2)).toBe(true); + const scopeClauses = whereClauses.filter((c) => c.includes('walletId')); + expect(scopeClauses.every((c) => c.includes('user.walletId') || c.includes('wallet.id'))).toBe(true); + + expect(result.totals.volume.buy).toBe(99999); + expect(result.totals.volume.buy).not.toBe(1000); + expect(result.completion.paymentInfoRequests.buy.paymentReceived).toBe(9000); + expect(result.completion.settlement.buy.received).toBe(9000); + }); + }); + + // --- COMPLETION --- // + + describe('completion funnels', () => { + it('computes stage A payment-info funnel counts and receivedRate as paymentReceived/requested', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 1000, transactions: 20, users: 10 }, + allTime: { buy: 1000, sell: 0, registeredUsers: 20, tradingUsers: 10 }, + activeUserIds: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], + paymentInfo: [ + { type: TransactionRequestType.BUY, status: TransactionRequestStatus.CREATED, count: 80 }, + { type: TransactionRequestType.BUY, status: TransactionRequestStatus.WAITING_FOR_PAYMENT, count: 5 }, + { type: TransactionRequestType.BUY, status: TransactionRequestStatus.COMPLETED, count: 15 }, + { type: TransactionRequestType.SELL, status: TransactionRequestStatus.CREATED, count: 10 }, + { type: TransactionRequestType.SELL, status: TransactionRequestStatus.COMPLETED, count: 5 }, + { type: TransactionRequestType.SWAP, status: TransactionRequestStatus.COMPLETED, count: 5 }, + ], + settlement: { + [PartnerStatisticDirection.BUY]: { received: 20, delivered: 15, rejected: 0 }, + [PartnerStatisticDirection.SELL]: { received: 5, delivered: 5, rejected: 0 }, + [PartnerStatisticDirection.SWAP]: { received: 5, delivered: 5, rejected: 0 }, + }, + }), + ); + + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + const buyA = result.completion.paymentInfoRequests.buy; + + expect(buyA.requested).toBe(100); + expect(buyA.noPaymentReceived).toBe(80); + expect(buyA.waitingForPayment).toBe(5); + expect(buyA.paymentReceived).toBe(15); + expect(buyA.receivedRate).toBe(0.15); + expect(buyA.receivedRate).not.toBe(Util.round(100 / 15, 4)); + + expect(result.completion.paymentInfoRequests.sell.requested).toBe(15); + expect(result.completion.paymentInfoRequests.sell.paymentReceived).toBe(5); + expect(result.completion.paymentInfoRequests.swap.paymentReceived).toBe(5); + }); + + it('computes stage B settlement counts and deliveredRate as delivered/received', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 1000, transactions: 20, users: 10 }, + allTime: { buy: 1000, sell: 0, registeredUsers: 20, tradingUsers: 10 }, + activeUserIds: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], + settlement: { + [PartnerStatisticDirection.BUY]: { received: 40, delivered: 30, rejected: 5 }, + [PartnerStatisticDirection.SELL]: { received: 0, delivered: 0, rejected: 0 }, + [PartnerStatisticDirection.SWAP]: { received: 20, delivered: 10, rejected: 5 }, + }, + }), + ); + + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + const buyB = result.completion.settlement.buy; + + expect(buyB.received).toBe(40); + expect(buyB.delivered).toBe(30); + expect(buyB.rejected).toBe(5); + expect(buyB.inProgress).toBe(5); + expect(buyB.deliveredRate).toBe(0.75); + + expect(result.completion.settlement.sell.received).toBe(0); + expect(result.completion.settlement.sell.deliveredRate).toBeNull(); + expect(result.completion.settlement.swap.inProgress).toBe(5); + }); + + it('block-suppresses the entire funnel when any counter is under k', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 1000, transactions: 20, users: 10 }, + allTime: { buy: 1000, sell: 0, registeredUsers: 20, tradingUsers: 10 }, + activeUserIds: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], + paymentInfo: [{ type: TransactionRequestType.BUY, status: TransactionRequestStatus.CREATED, count: 4 }], + settlement: { + [PartnerStatisticDirection.BUY]: { received: 10, delivered: 3, rejected: 0 }, + [PartnerStatisticDirection.SELL]: emptySettlement(), + [PartnerStatisticDirection.SWAP]: emptySettlement(), + }, + }), + ); + + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + + // Stage A: requested=4 under k → entire group null (zeros stay 0) + expect(result.completion.paymentInfoRequests.buy.requested).toBeNull(); + expect(result.completion.paymentInfoRequests.buy.noPaymentReceived).toBeNull(); + expect(result.completion.paymentInfoRequests.buy.paymentReceived).toBe(0); + expect(result.completion.paymentInfoRequests.buy.receivedRate).toBeNull(); + + // Stage B: delivered=3 under k → block entire settlement group + expect(result.completion.settlement.buy.received).toBeNull(); + expect(result.completion.settlement.buy.delivered).toBeNull(); + expect(result.completion.settlement.buy.deliveredRate).toBeNull(); + expect(result.completion.settlement.buy.rejected).toBe(0); + }); + }); + + // --- B3: totals / allTime suppression via service --- // + + describe('totals and allTime suppression (B3)', () => { + it('nulls totals when overall transaction count is below k (boundary at k)', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 99, transactions: 4, users: 4 }, + allTime: { buy: 99, sell: 0, registeredUsers: 10, tradingUsers: 4 }, + newUsers: 0, + activeUserIds: [1, 2, 3, 4], + }), + ); + + const under = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + expect(under.totals.volume.total).toBeNull(); + expect(under.totals.volume.buy).toBeNull(); + expect(under.totals.transactions.total).toBeNull(); + expect(under.totals.averageTransactionVolume).toBeNull(); + expect(under.allTime.volume.total).toBeNull(); + expect(under.allTime.volume.buy).toBeNull(); + expect(under.allTime.registeredUsers).toBe(10); + expect(under.allTime.tradingUsers).toBeNull(); + expect(under.referral.volume).toBeNull(); + expect(under.meta.suppressedBuckets).toBeGreaterThanOrEqual(2); + + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 100, transactions: 5, users: 5 }, + allTime: { buy: 100, sell: 0, registeredUsers: 10, tradingUsers: 5 }, + newUsers: 0, + activeUserIds: [1, 2, 3, 4, 5], + }), + ); + + const atK = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + expect(atK.totals.volume.total).toBe(100); + expect(atK.totals.transactions.total).toBe(5); + expect(atK.allTime.volume.total).toBe(100); + expect(atK.allTime.tradingUsers).toBe(5); + }); + + it('nulls totals when person count is under k even with high transaction counts', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 1000, transactions: 20, users: 2 }, + allTime: { buy: 1000, sell: 0, registeredUsers: 10, tradingUsers: 5 }, + newUsers: 0, + activeUserIds: [1, 2], + }), + ); + activeUserCountFromManager = 2; + + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + expect(result.totals.volume.total).toBeNull(); + expect(result.totals.activeUsers).toBeNull(); + }); + }); + + // --- M2: active users via manager UNION count --- // + + describe('countActiveUsers uses DB COUNT over UNION (M2)', () => { + it('returns the manager COUNT and does not load user id rows via getRawMany', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 1000, transactions: 20, users: 10 }, + allTime: { buy: 1000, sell: 0, registeredUsers: 50, tradingUsers: 20 }, + activeUserIds: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], + newUsers: 0, + }), + ); + activeUserCountFromManager = 7; + + const beforeMany = getRawManyCalls; + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + + expect(managerCreateQueryBuilderCalls).toBeGreaterThanOrEqual(1); + expect(result.totals.activeUsers).toBe(7); + expect(result.totals.activeUsers).not.toBe(10); + expect(getRawManyCalls).toBeGreaterThanOrEqual(beforeMany); + }); + }); + + // --- CURRENCY / REFERRAL --- // + + describe('currency separation and referral creditOpen', () => { + it('keeps referral in EUR and computes creditOpen as ref + partner − paid', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 100, transactions: 10, users: 5 }, + // partner earned 12.25, personal ref 5, paid 2.25 → open = 15 + referral: { volume: 42.5, partnerRefCredit: 12.25, refCredit: 5, paidRefCredit: 2.25 }, + allTime: { buy: 100, sell: 0, registeredUsers: 10, tradingUsers: 5 }, + newUsers: 5, + activeUserIds: [1, 2, 3, 4, 5], + }), + ); + + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + + expect(result.currency).toBe('CHF'); + expect(result.referral.currency).toBe('EUR'); + expect(result.referral.volume).toBe(42.5); + expect(result.referral.creditEarned).toBe(12.25); + expect(result.referral.creditPaid).toBe(2.25); + expect(result.referral.creditOpen).toBe(15); + // Old formula partner − paid would yield 10 — must not regress + expect(result.referral.creditOpen).not.toBe(10); + expect(result.totals.volume.buy).toBe(100); + }); + }); + + // --- mergeNamedRows / breakdown pipeline --- // + + describe('mergeNamedRows and breakdown pipeline', () => { + it('merges same-name rows across directions and surfaces them in the response', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 500, transactions: 20, users: 10 }, + sell: { volume: 300, transactions: 10, users: 8 }, + allTime: { buy: 500, sell: 300, registeredUsers: 20, tradingUsers: 10 }, + activeUserIds: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10], + namedRows: [ + { name: 'BTC', blockchain: 'Bitcoin', volume: 200, transactions: 10, users: 6 }, + { name: 'BTC', blockchain: 'Bitcoin', volume: 100, transactions: 5, users: 4 }, + { name: 'ETH', blockchain: 'Ethereum', volume: 50, transactions: 5, users: 5 }, + { name: 'CHF', volume: 400, transactions: 15, users: 10 }, + ], + }), + ); + + const result = await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + + // mergeNamedRows is used for fiat/blockchain/payment; asset rows are not merged by name across dirs + expect(result.breakdown.fiatCurrencies.length + result.breakdown.blockchains.length).toBeGreaterThan(0); + // Direct unit check of mergeNamedRows (mutation: neutralize → this fails) + const merged = service.mergeNamedRows([ + { name: 'BTC', volume: 200, transactions: 10, users: 6 }, + { name: 'BTC', volume: 100, transactions: 5, users: 4 }, + { name: 'ETH', volume: 50, transactions: 5, users: 5 }, + ]); + expect(merged.find((r) => r.name === 'BTC')?.volume).toBe(300); + expect(merged.find((r) => r.name === 'BTC')?.transactions).toBe(15); + expect(merged).toHaveLength(2); + }); + }); + + // --- K1: partial edge buckets --- // + + describe('timeline partial edge buckets (K1)', () => { + it('marks week-edge buckets as partial when from/to cut mid-week', async () => { + const result = await service.getTimeline(1, MID_WEEK_FROM, MID_WEEK_TO, 'week'); + + expect(result.buckets.length).toBeGreaterThan(0); + expect(result.buckets[0].partial).toBe(true); + expect(result.buckets[result.buckets.length - 1].partial).toBe(true); + }); + + it('marks full day range buckets as non-partial when aligned to midnight span', async () => { + const result = await service.getTimeline(1, '2024-06-10T00:00:00.000Z', '2024-06-12T23:59:59.000Z', 'day'); + + expect(result.buckets.length).toBe(3); + expect(result.buckets.every((b) => b.partial === false)).toBe(true); + }); + }); + + // --- TIMEZONE INDEPENDENCE --- // + + describe('timezone independence of timeline buckets', () => { + it('emits exactly three UTC day buckets for a known three-day period (regression for local TZ drift)', async () => { + const result = await service.getTimeline(1, '2024-06-10T00:00:00.000Z', '2024-06-12T23:59:59.000Z', 'day'); + + expect(result.buckets).toHaveLength(3); + expect(result.buckets.map((b) => b.date.toISOString())).toEqual([ + '2024-06-10T00:00:00.000Z', + '2024-06-11T00:00:00.000Z', + '2024-06-12T00:00:00.000Z', + ]); + expect(result.period.from.toISOString()).toBe('2024-06-10T00:00:00.000Z'); + expect(result.period.to.toISOString()).toBe('2024-06-13T00:00:00.000Z'); + }); + + it('binds DATE_TRUNC to UTC in the timeline SQL expression', async () => { + await service.getTimeline(1, '2024-06-10T00:00:00.000Z', '2024-06-12T23:59:59.000Z', 'day'); + + const truncs = groupByCapture.groupBys.filter((g) => g.includes('DATE_TRUNC')); + expect(truncs.length).toBeGreaterThan(0); + for (const g of truncs) { + expect(g).toContain("AT TIME ZONE 'UTC'"); + expect(g).toMatch(/DATE_TRUNC\('day',\s*tx\.created\)/); + } + }); + }); + + // --- TIMELINE VALIDATION PATH --- // + + describe('getTimeline validation', () => { + it('throws 400 for invalid granularity', async () => { + await expect(service.getTimeline(1, PERIOD_FROM, PERIOD_TO, 'hour')).rejects.toThrow(BadRequestException); + }); + + it('throws 400 for period over max days', async () => { + const from = Util.daysBefore(PARTNER_STATISTIC_MAX_PERIOD_DAYS + 100, TEST_NOW); + await expect(service.getTimeline(1, from, TEST_NOW, 'day')).rejects.toThrow(BadRequestException); + }); + }); + + // --- HALF-OPEN INTERVAL --- // + + describe('half-open period filter', () => { + it('uses >= from AND < to (not BETWEEN inclusive)', async () => { + fixtures.set( + 1, + emptyFixture({ + buy: { volume: 100, transactions: 10, users: 5 }, + allTime: { buy: 100, sell: 0, registeredUsers: 10, tradingUsers: 5 }, + activeUserIds: [1, 2, 3, 4, 5], + }), + ); + + await service.getStatistics(1, PERIOD_FROM, PERIOD_TO); + + const halfOpen = whereClauses.filter((c) => c.includes('created')); + expect(halfOpen.some((c) => c.includes('>=') && c.includes('<'))).toBe(true); + expect(halfOpen.every((c) => !/BETWEEN/i.test(c))).toBe(true); + }); + }); +}); diff --git a/src/subdomains/core/statistic/__tests__/partner-statistic.suppression.spec.ts b/src/subdomains/core/statistic/__tests__/partner-statistic.suppression.spec.ts new file mode 100644 index 0000000000..cfd81eaede --- /dev/null +++ b/src/subdomains/core/statistic/__tests__/partner-statistic.suppression.spec.ts @@ -0,0 +1,327 @@ +import { PARTNER_STATISTIC_SUPPRESSION_THRESHOLD } from '../partner-statistic.enum'; +import { + suppressAdditiveGroup, + suppressAllTimeVolume, + suppressBreakdownRows, + suppressCount, + suppressPeriodTotals, + suppressRate, + suppressScalar, + suppressTimelineBuckets, +} from '../partner-statistic.suppression'; + +/** Single calendar anchor for payload dates — suppression logic is count-based, not calendar-based. */ +const TEST_BUCKET_DATE = new Date(); + +describe('Partner statistic suppression', () => { + describe('suppressScalar (M1)', () => { + it('keeps 0 as 0 and nulls only 1..k-1 (boundary is strict < k)', () => { + expect(suppressScalar(0)).toBe(0); + expect(suppressScalar(1)).toBeNull(); + expect(suppressScalar(4)).toBeNull(); + expect(suppressScalar(PARTNER_STATISTIC_SUPPRESSION_THRESHOLD - 1)).toBeNull(); + expect(suppressScalar(PARTNER_STATISTIC_SUPPRESSION_THRESHOLD)).toBe(5); + expect(suppressScalar(5)).toBe(5); + expect(suppressScalar(10)).toBe(10); + expect(suppressScalar(PARTNER_STATISTIC_SUPPRESSION_THRESHOLD - 1)).not.toBe( + PARTNER_STATISTIC_SUPPRESSION_THRESHOLD - 1, + ); + expect(suppressScalar(PARTNER_STATISTIC_SUPPRESSION_THRESHOLD)).not.toBeNull(); + }); + + it('uses min(transactions, users) as the effective count (person gate)', () => { + // 10 txs but only 3 users → under k + expect(suppressScalar(10, undefined, 3)).toBeNull(); + // 10 txs and 5 users → visible + expect(suppressScalar(10, undefined, 5)).toBe(10); + // 3 txs and 10 users → under k on tx side + expect(suppressScalar(3, undefined, 10)).toBeNull(); + }); + }); + + describe('suppressRate (completion)', () => { + it('returns numerator/denominator (not swapped) and nulls zero or suppressed sides', () => { + expect(suppressRate(0, 0)).toBeNull(); + expect(suppressRate(5, 0)).toBeNull(); + expect(suppressRate(null, 10)).toBeNull(); + expect(suppressRate(5, null)).toBeNull(); + expect(suppressRate(15, 100)).toBe(0.15); + expect(suppressRate(15, 100)).not.toBe(suppressRate(100, 15)); + expect(suppressRate(1, 3)).toBe(0.3333); + }); + }); + + describe('suppressCount', () => { + it('flags suppression only for 1..k-1', () => { + expect(suppressCount(0)).toEqual({ value: 0, suppressed: false }); + expect(suppressCount(3)).toEqual({ value: null, suppressed: true }); + expect(suppressCount(5)).toEqual({ value: 5, suppressed: false }); + }); + }); + + describe('suppressBreakdownRows', () => { + it('drops rows with fewer than k transactions (boundary at k)', () => { + const { rows, suppressedCount } = suppressBreakdownRows([ + { name: 'A', volume: 100, transactions: 4 }, + { name: 'B', volume: 50, transactions: 3 }, + { name: 'C', volume: 200, transactions: 5 }, + { name: 'D', volume: 300, transactions: 10 }, + ]); + + expect(rows.map((r) => r.name).sort()).toEqual(['C', 'D']); + expect(suppressedCount).toBe(2); + expect(rows.find((r) => r.name === 'C')?.transactions).toBe(5); + expect(rows.some((r) => r.transactions === PARTNER_STATISTIC_SUPPRESSION_THRESHOLD)).toBe(true); + }); + + it('drops rows when person count is under k even if transactions are above k', () => { + const { rows, suppressedCount } = suppressBreakdownRows([ + { name: 'A', volume: 100, transactions: 20, users: 2 }, + { name: 'B', volume: 200, transactions: 20, users: 10 }, + { name: 'C', volume: 300, transactions: 30, users: 15 }, + ]); + // A under k on person gate; complementary drops B (smallest remaining); C stays + expect(rows.map((r) => r.name)).toEqual(['C']); + expect(suppressedCount).toBe(2); + expect(rows.find((r) => r.name === 'A')).toBeUndefined(); + }); + + it('applies complementary suppression only when exactly one row is under threshold', () => { + const { rows, suppressedCount } = suppressBreakdownRows([ + { name: 'A', volume: 10, transactions: 3 }, + { name: 'B', volume: 20, transactions: 6 }, + { name: 'C', volume: 100, transactions: 50 }, + ]); + + expect(rows.map((r) => r.name)).toEqual(['C']); + expect(suppressedCount).toBe(2); + expect(rows.find((r) => r.name === 'B')).toBeUndefined(); + }); + + it('does not apply complementary suppression when zero or two+ rows are under threshold', () => { + const none = suppressBreakdownRows([ + { name: 'A', volume: 10, transactions: 5 }, + { name: 'B', volume: 20, transactions: 6 }, + ]); + expect(none.rows).toHaveLength(2); + expect(none.suppressedCount).toBe(0); + + const two = suppressBreakdownRows([ + { name: 'A', volume: 10, transactions: 2 }, + { name: 'B', volume: 20, transactions: 3 }, + { name: 'C', volume: 100, transactions: 50 }, + ]); + expect(two.rows.map((r) => r.name)).toEqual(['C']); + expect(two.suppressedCount).toBe(2); + expect(two.rows).toHaveLength(1); + }); + }); + + describe('suppressTimelineBuckets (B2)', () => { + const bucket = ( + buyTx: number, + sellTx = 0, + swapTx = 0, + buyUsers?: number, + sellUsers?: number, + swapUsers?: number, + ) => ({ + date: new Date(TEST_BUCKET_DATE), + volume: { buy: buyTx * 10, sell: sellTx * 10, swap: swapTx * 10 }, + transactions: { buy: buyTx, sell: sellTx, swap: swapTx }, + users: { + buy: buyUsers ?? buyTx, + sell: sellUsers ?? sellTx, + swap: swapUsers ?? swapTx, + }, + suppressed: false, + partial: false, + }); + + it('nullifies a bucket with 4 transactions and keeps one with 5 (plus complementary)', () => { + const { buckets, suppressedCount } = suppressTimelineBuckets([bucket(4), bucket(5), bucket(20)]); + + expect(buckets[0].suppressed).toBe(true); + expect(buckets[0].volume).toBeNull(); + expect(buckets[0].transactions).toBeNull(); + expect(buckets[1].suppressed).toBe(true); + expect(buckets[2].suppressed).toBe(false); + expect(buckets[2].volume).toEqual({ buy: 200, sell: 0, swap: 0 }); + expect(suppressedCount).toBe(2); + }); + + it('suppresses mixed bucket when any single direction is under k (block rule)', () => { + // buy=5 (ok), sell=1 (under) → whole bucket suppressed even though total=6 ≥ k + const mixed = bucket(5, 1, 0); + const large = bucket(20, 20, 20); + const { buckets, suppressedCount } = suppressTimelineBuckets([mixed, large]); + + expect(buckets[0].suppressed).toBe(true); + expect(buckets[0].volume).toBeNull(); + expect(buckets[0].transactions).toBeNull(); + // complementary may also suppress large if it is the only other filled + expect(suppressedCount).toBeGreaterThanOrEqual(1); + // total − visible must not recover sell=1 + if (!buckets[1].suppressed) { + // if complementary did not fire (shouldn't with only one under), buy stays + expect(buckets[1].transactions?.sell).not.toBe(1); + } + }); + + it('keeps a bucket at exactly k when no complementary case applies', () => { + const { buckets, suppressedCount } = suppressTimelineBuckets([bucket(4), bucket(3), bucket(5)]); + + expect(buckets[0].suppressed).toBe(true); + expect(buckets[1].suppressed).toBe(true); + expect(buckets[2].suppressed).toBe(false); + expect(buckets[2].transactions).toEqual({ buy: 5, sell: 0, swap: 0 }); + expect(suppressedCount).toBe(2); + }); + + it('leaves empty (0-tx) buckets as visible zeros, not suppressed', () => { + const empty = bucket(0); + const filled = bucket(10); + const { buckets, suppressedCount } = suppressTimelineBuckets([empty, filled]); + + expect(buckets[0].suppressed).toBe(false); + expect(buckets[0].volume).toEqual({ buy: 0, sell: 0, swap: 0 }); + expect(buckets[0].transactions).toEqual({ buy: 0, sell: 0, swap: 0 }); + expect(buckets[1].suppressed).toBe(false); + expect(suppressedCount).toBe(0); + }); + + it('still applies complementary when empty days would otherwise defeat it (B2 security)', () => { + const empties = [bucket(0), bucket(0), bucket(0), bucket(0), bucket(0)]; + const under = bucket(3); + const small = bucket(6); + const large = bucket(50); + const large2 = bucket(40); + + const { buckets, suppressedCount } = suppressTimelineBuckets([...empties, under, small, large, large2]); + + for (let i = 0; i < empties.length; i++) { + expect(buckets[i].suppressed).toBe(false); + expect(buckets[i].transactions).toEqual({ buy: 0, sell: 0, swap: 0 }); + expect(buckets[i].volume).toEqual({ buy: 0, sell: 0, swap: 0 }); + } + + expect(buckets[empties.length].suppressed).toBe(true); + expect(buckets[empties.length].volume).toBeNull(); + expect(buckets[empties.length + 1].suppressed).toBe(true); + expect(buckets[empties.length + 2].suppressed).toBe(false); + expect(buckets[empties.length + 3].suppressed).toBe(false); + expect(suppressedCount).toBe(2); + }); + }); + + describe('suppressPeriodTotals (B3 / block rule)', () => { + it('nulls all totals fields when overall transactions.total is in 1..k-1', () => { + const { volume, transactions, averageTransactionVolume, suppressedCount } = suppressPeriodTotals( + { buy: 100, sell: 0, swap: 0, total: 100 }, + { buy: 3, sell: 0, swap: 0, total: 3 }, + { buy: 3, sell: 0, swap: 0, total: 3 }, + ); + + expect(volume).toEqual({ buy: null, sell: null, swap: null, total: null }); + expect(transactions).toEqual({ buy: null, sell: null, swap: null, total: null }); + expect(averageTransactionVolume).toBeNull(); + expect(suppressedCount).toBe(1); + }); + + it('block-suppresses the entire group when any direction is under k (no partial nulling)', () => { + // Previously leaked: sell=null but total/buy/swap visible → sell = total − buy − swap. + const mixed = suppressPeriodTotals( + { buy: 1000, sell: 10, swap: 200, total: 1210 }, + { buy: 20, sell: 2, swap: 10, total: 32 }, + { buy: 10, sell: 2, swap: 8, total: 15 }, + ); + expect(mixed.volume).toEqual({ buy: null, sell: null, swap: null, total: null }); + expect(mixed.transactions).toEqual({ buy: null, sell: null, swap: null, total: null }); + expect(mixed.averageTransactionVolume).toBeNull(); + expect(mixed.suppressedCount).toBe(1); + // Reconstruction must fail + expect(mixed.volume.total).toBeNull(); + }); + + it('keeps totals at exactly k for every direction and overall', () => { + const atK = suppressPeriodTotals( + { buy: 500, sell: 0, swap: 0, total: 500 }, + { buy: 5, sell: 0, swap: 0, total: 5 }, + { buy: 5, sell: 0, swap: 0, total: 5 }, + ); + expect(atK.volume.total).toBe(500); + expect(atK.transactions.total).toBe(5); + expect(atK.volume.buy).toBe(500); + expect(atK.suppressedCount).toBe(0); + }); + + it('suppresses when transaction count is high but person count is under k', () => { + const personGate = suppressPeriodTotals( + { buy: 1000, sell: 0, swap: 0, total: 1000 }, + { buy: 20, sell: 0, swap: 0, total: 20 }, + { buy: 2, sell: 0, swap: 0, total: 2 }, + ); + expect(personGate.volume.total).toBeNull(); + expect(personGate.suppressedCount).toBe(1); + }); + + it('leaves all-zero totals as zeros, not null', () => { + const { volume, transactions, averageTransactionVolume, suppressedCount } = suppressPeriodTotals( + { buy: 0, sell: 0, swap: 0, total: 0 }, + { buy: 0, sell: 0, swap: 0, total: 0 }, + { buy: 0, sell: 0, swap: 0, total: 0 }, + ); + expect(volume.total).toBe(0); + expect(transactions.total).toBe(0); + expect(averageTransactionVolume).toBeNull(); + expect(suppressedCount).toBe(0); + }); + }); + + describe('suppressAdditiveGroup (funnel block rule)', () => { + it('nulls every non-zero member and the rate when any member is under k', () => { + const { values, rate, suppressedCount } = suppressAdditiveGroup( + { received: 20, delivered: 12, rejected: 3, inProgress: 5 }, + { numeratorKey: 'delivered', denominatorKey: 'received' }, + ); + // rejected=3 under k → block + expect(values.received).toBeNull(); + expect(values.delivered).toBeNull(); + expect(values.rejected).toBeNull(); + expect(values.inProgress).toBeNull(); + expect(rate).toBeNull(); + expect(suppressedCount).toBeGreaterThanOrEqual(1); + // reconstruction of rejected via 20-12-5 must not be possible from visible non-nulls + expect(Object.values(values).every((v) => v === null || v === 0)).toBe(true); + }); + + it('keeps the group when every member is 0 or ≥ k', () => { + const { values, rate, suppressedCount } = suppressAdditiveGroup( + { received: 20, delivered: 15, rejected: 0, inProgress: 5 }, + { numeratorKey: 'delivered', denominatorKey: 'received' }, + ); + expect(values.received).toBe(20); + expect(values.delivered).toBe(15); + expect(values.rejected).toBe(0); + expect(values.inProgress).toBe(5); + expect(rate).toBe(0.75); + expect(suppressedCount).toBe(0); + }); + }); + + describe('suppressAllTimeVolume (B3)', () => { + it('nulls all-time volume when tradingUsers is in 1..k-1 and keeps it at k', () => { + const under = suppressAllTimeVolume({ buy: 100, sell: 50, total: 150 }, 3); + expect(under.volume).toEqual({ buy: null, sell: null, total: null }); + expect(under.suppressedCount).toBe(1); + + const atK = suppressAllTimeVolume({ buy: 100, sell: 50, total: 150 }, 5); + expect(atK.volume).toEqual({ buy: 100, sell: 50, total: 150 }); + expect(atK.suppressedCount).toBe(0); + + const zero = suppressAllTimeVolume({ buy: 0, sell: 0, total: 0 }, 0); + expect(zero.volume).toEqual({ buy: 0, sell: 0, total: 0 }); + expect(zero.suppressedCount).toBe(0); + }); + }); +}); diff --git a/src/subdomains/core/statistic/dto/partner-statistic.dto.ts b/src/subdomains/core/statistic/dto/partner-statistic.dto.ts new file mode 100644 index 0000000000..b9a9c842e3 --- /dev/null +++ b/src/subdomains/core/statistic/dto/partner-statistic.dto.ts @@ -0,0 +1,420 @@ +import { ApiProperty } from '@nestjs/swagger'; +import { PartnerStatisticDirection, PartnerStatisticGranularity } from '../partner-statistic.enum'; + +// --- PERIOD / META --- // + +export class PartnerStatisticPeriodDto { + @ApiProperty({ description: 'Inclusive period start, snapped to UTC midnight' }) + from: Date; + + @ApiProperty({ + description: 'Exclusive period end, snapped to UTC midnight of the day after the last included day', + }) + to: Date; +} + +export class PartnerStatisticMetaDto { + @ApiProperty({ description: 'Minimum effective count (min of transactions and distinct users) for disclosure' }) + suppressionThreshold: number; + + @ApiProperty({ description: 'Number of buckets/rows suppressed under the threshold' }) + suppressedBuckets: number; + + @ApiProperty({ nullable: true, required: false }) + generatedAt?: Date; +} + +// --- VOLUME / COUNTS --- // + +export class PartnerVolumeByTypeDto { + @ApiProperty({ type: Number, nullable: true, description: 'CHF; null when the totals group is suppressed' }) + buy: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'CHF; null when the totals group is suppressed' }) + sell: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'CHF; null when the totals group is suppressed' }) + swap: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'CHF; null when the totals group is suppressed' }) + total: number | null; +} + +export class PartnerVolumeBuySellDto { + @ApiProperty({ type: Number, nullable: true, description: 'CHF; null when tradingUsers < k' }) + buy: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'CHF; null when tradingUsers < k' }) + sell: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'CHF; null when tradingUsers < k' }) + total: number | null; +} + +export class PartnerTransactionsByTypeDto { + @ApiProperty({ type: Number, nullable: true, description: 'Null when the totals group is suppressed' }) + buy: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'Null when the totals group is suppressed' }) + sell: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'Null when the totals group is suppressed' }) + swap: number | null; + + @ApiProperty({ type: Number, nullable: true, description: 'Null when the totals group is suppressed' }) + total: number | null; +} + +export class PartnerTotalsDto { + @ApiProperty({ + type: PartnerVolumeByTypeDto, + description: 'Volume in CHF; entire group null when any direction is under the suppression threshold', + }) + volume: PartnerVolumeByTypeDto; + + @ApiProperty({ type: PartnerTransactionsByTypeDto }) + transactions: PartnerTransactionsByTypeDto; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'Average volume per transaction in CHF; null if no transactions or totals suppressed', + }) + averageTransactionVolume: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'Distinct users with ≥1 counted transaction in the period; null if 1..k-1', + }) + activeUsers: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'Users of this wallet created in the period; null if 1..k-1', + }) + newUsers: number | null; +} + +export class PartnerAllTimeDto { + @ApiProperty({ + type: PartnerVolumeBuySellDto, + description: 'Lifetime volume in CHF; null fields when tradingUsers < k', + }) + volume: PartnerVolumeBuySellDto; + + @ApiProperty({ description: 'Always visible (installation count, no transaction linkage)' }) + registeredUsers: number; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'Users of this wallet with buyVolume > 0 or sellVolume > 0; null if 1..k-1', + }) + tradingUsers: number | null; +} + +// --- BREAKDOWN --- // + +export class PartnerAssetBreakdownDto { + @ApiProperty() + name: string; + + @ApiProperty({ nullable: true }) + blockchain: string | null; + + @ApiProperty({ enum: PartnerStatisticDirection }) + direction: PartnerStatisticDirection; + + @ApiProperty({ description: 'Volume in CHF' }) + volume: number; + + @ApiProperty() + transactions: number; +} + +export class PartnerNamedBreakdownDto { + @ApiProperty() + name: string; + + @ApiProperty({ description: 'Volume in CHF' }) + volume: number; + + @ApiProperty() + transactions: number; +} + +export class PartnerBreakdownDto { + @ApiProperty({ type: PartnerAssetBreakdownDto, isArray: true }) + assets: PartnerAssetBreakdownDto[]; + + @ApiProperty({ type: PartnerNamedBreakdownDto, isArray: true }) + fiatCurrencies: PartnerNamedBreakdownDto[]; + + @ApiProperty({ type: PartnerNamedBreakdownDto, isArray: true }) + blockchains: PartnerNamedBreakdownDto[]; + + @ApiProperty({ type: PartnerNamedBreakdownDto, isArray: true }) + paymentMethods: PartnerNamedBreakdownDto[]; +} + +// --- REFERRAL --- // + +export class PartnerReferralDto { + @ApiProperty({ + type: Number, + nullable: true, + description: + 'Partner referral volume in EUR on the wallet owner’s account (all wallets of that owner). ' + + 'Null when tradingUsers < k (moves with individual customer trades).', + }) + volume: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: + 'Partner referral credit earned (owner.partnerRefCredit only) in EUR, account-wide. ' + + 'Null when tradingUsers < k.', + }) + creditEarned: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: + 'Referral credit already paid out (owner.paidRefCredit) in EUR, account-wide across both ' + + 'personal and partner pots. Null when tradingUsers < k.', + }) + creditPaid: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: + 'Open referral credit in EUR: owner.refCredit + owner.partnerRefCredit − owner.paidRefCredit ' + + '(account-wide). Null when tradingUsers < k.', + }) + creditOpen: number | null; + + @ApiProperty({ enum: ['EUR'], description: 'Native currency of the referral system' }) + currency: 'EUR'; +} + +// --- COMPLETION (funnels) --- // + +/** + * Stage A: payment-info quote requests → payment received. + * Each UI amount change creates a new transaction_request — this is NOT a user conversion rate. + */ +export class PartnerPaymentInfoDirectionDto { + @ApiProperty({ + type: Number, + nullable: true, + description: + 'Count of payment-info quote requests (transaction_request rows) in the period. ' + + 'One row is created on every payment-info fetch (e.g. amount change in the UI), not once per purchase intent. ' + + 'Null when any funnel member is under the suppression threshold (block suppression).', + }) + requested: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'Requests that reached status Completed (a payment was linked).', + }) + paymentReceived: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'Requests still in status WaitingForPayment.', + }) + waitingForPayment: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'Requests left in status Created (quote shown, no payment followed).', + }) + noPaymentReceived: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: + 'paymentReceived / requested (0..1, 4 dp). Share of payment-info requests that got a payment — ' + + 'not a per-user conversion rate. Null if denominator is 0 or the funnel group was suppressed.', + }) + receivedRate: number | null; +} + +export class PartnerPaymentInfoRequestsDto { + @ApiProperty({ type: PartnerPaymentInfoDirectionDto }) + buy: PartnerPaymentInfoDirectionDto; + + @ApiProperty({ type: PartnerPaymentInfoDirectionDto }) + sell: PartnerPaymentInfoDirectionDto; + + @ApiProperty({ type: PartnerPaymentInfoDirectionDto }) + swap: PartnerPaymentInfoDirectionDto; +} + +/** + * Stage B: payment received (buy_crypto / buy_fiat created) → delivered or rejected. + * Note: `delivered` requires amlCheck=Pass AND isComplete=true, which is a subset of + * `totals.transactions` (those only require amlCheck=Pass, including still-in-flight payouts). + */ +export class PartnerSettlementDirectionDto { + @ApiProperty({ + type: Number, + nullable: true, + description: + 'All buy_crypto/buy_fiat rows in the period for this direction (no amlCheck filter). ' + + 'Null when any funnel member is under the suppression threshold (block suppression).', + }) + received: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: + 'amlCheck=Pass AND isComplete=true. Stricter than totals.transactions (which counts all Pass, ' + + 'including rows not yet fully paid out).', + }) + delivered: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'amlCheck=Fail (regardless of isComplete).', + }) + rejected: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'received − delivered − rejected (pending AML, in-flight payout, etc.).', + }) + inProgress: number | null; + + @ApiProperty({ + type: Number, + nullable: true, + description: 'delivered / received (0..1, 4 dp). Null if denominator is 0 or the funnel group was suppressed.', + }) + deliveredRate: number | null; +} + +export class PartnerSettlementDto { + @ApiProperty({ type: PartnerSettlementDirectionDto }) + buy: PartnerSettlementDirectionDto; + + @ApiProperty({ type: PartnerSettlementDirectionDto }) + sell: PartnerSettlementDirectionDto; + + @ApiProperty({ type: PartnerSettlementDirectionDto }) + swap: PartnerSettlementDirectionDto; +} + +export class PartnerCompletionDto { + @ApiProperty({ + type: PartnerPaymentInfoRequestsDto, + description: 'Stage A: payment-info requests → payment received (transaction_request funnel).', + }) + paymentInfoRequests: PartnerPaymentInfoRequestsDto; + + @ApiProperty({ + type: PartnerSettlementDto, + description: 'Stage B: payment entered the system → delivered / rejected / in progress.', + }) + settlement: PartnerSettlementDto; +} + +// --- SUMMARY RESPONSE --- // + +export class PartnerStatisticDto { + @ApiProperty({ type: PartnerStatisticPeriodDto }) + period: PartnerStatisticPeriodDto; + + @ApiProperty({ enum: ['CHF'] }) + currency: 'CHF'; + + @ApiProperty({ type: PartnerTotalsDto }) + totals: PartnerTotalsDto; + + @ApiProperty({ type: PartnerAllTimeDto }) + allTime: PartnerAllTimeDto; + + @ApiProperty({ type: PartnerBreakdownDto }) + breakdown: PartnerBreakdownDto; + + @ApiProperty({ type: PartnerReferralDto }) + referral: PartnerReferralDto; + + @ApiProperty({ type: PartnerCompletionDto }) + completion: PartnerCompletionDto; + + @ApiProperty({ type: PartnerStatisticMetaDto }) + meta: PartnerStatisticMetaDto; +} + +// --- TIMELINE --- // + +/** Volume or transaction counts split by trade direction. */ +export class PartnerTimelineByDirectionDto { + @ApiProperty() + buy: number; + + @ApiProperty() + sell: number; + + @ApiProperty() + swap: number; +} + +export class PartnerTimelineBucketDto { + @ApiProperty() + date: Date; + + @ApiProperty({ + type: PartnerTimelineByDirectionDto, + nullable: true, + description: 'Volume in CHF; null when suppressed', + }) + volume: PartnerTimelineByDirectionDto | null; + + @ApiProperty({ + type: PartnerTimelineByDirectionDto, + nullable: true, + description: 'Transaction counts; null when suppressed', + }) + transactions: PartnerTimelineByDirectionDto | null; + + @ApiProperty({ description: 'True when values are withheld under the suppression threshold' }) + suppressed: boolean; + + @ApiProperty({ + description: + 'True when the bucket’s natural range extends outside the requested period (edge week/month truncated by from/to)', + }) + partial: boolean; +} + +export class PartnerTimelineDto { + @ApiProperty({ type: PartnerStatisticPeriodDto }) + period: PartnerStatisticPeriodDto; + + @ApiProperty({ enum: ['CHF'] }) + currency: 'CHF'; + + @ApiProperty({ enum: PartnerStatisticGranularity }) + granularity: PartnerStatisticGranularity; + + @ApiProperty({ type: PartnerTimelineBucketDto, isArray: true }) + buckets: PartnerTimelineBucketDto[]; + + @ApiProperty({ type: PartnerStatisticMetaDto }) + meta: PartnerStatisticMetaDto; +} diff --git a/src/subdomains/core/statistic/partner-statistic-rate-limit.guard.ts b/src/subdomains/core/statistic/partner-statistic-rate-limit.guard.ts new file mode 100644 index 0000000000..6919d2ace9 --- /dev/null +++ b/src/subdomains/core/statistic/partner-statistic-rate-limit.guard.ts @@ -0,0 +1,25 @@ +import { ExecutionContext, Injectable } from '@nestjs/common'; +import { ThrottlerGuard } from '@nestjs/throttler'; +import { Config } from 'src/config/config'; + +/** + * Wallet-scoped rate limit for partner statistic routes. + * + * The shared RateLimitGuard keys by IP prefix and bypasses known/Azure IPs, so it does not + * bound repeated scrapes by a single partner JWT. These routes authenticate first (AuthGuard), + * then count by `jwt.user` (wallet id). There is no other user/wallet tracker in the repo. + */ +@Injectable() +export class PartnerStatisticRateLimitGuard extends ThrottlerGuard { + protected getTracker(req: Record): string { + const walletId = req.user?.user; + if (walletId != null) return `partner-stat:wallet:${walletId}`; + // AuthGuard should have run first; fall back so a mis-ordered guard still keys something. + return `partner-stat:ip:${req.realIp ?? 'unknown'}`; + } + + async handleRequest(context: ExecutionContext, limit: number, ttl: number): Promise { + if (!Config.request.limitCheck) return true; + return super.handleRequest(context, limit, ttl); + } +} diff --git a/src/subdomains/core/statistic/partner-statistic.controller.ts b/src/subdomains/core/statistic/partner-statistic.controller.ts new file mode 100644 index 0000000000..8f854f0427 --- /dev/null +++ b/src/subdomains/core/statistic/partner-statistic.controller.ts @@ -0,0 +1,73 @@ +import { Controller, Get, Query, UseGuards } from '@nestjs/common'; +import { AuthGuard } from '@nestjs/passport'; +import { ApiBearerAuth, ApiOkResponse, ApiQuery, ApiTags } from '@nestjs/swagger'; +import { Throttle } from '@nestjs/throttler/dist/throttler.decorator'; +import { GetJwt } from 'src/shared/auth/get-jwt.decorator'; +import { JwtPayload } from 'src/shared/auth/jwt-payload.interface'; +import { RoleGuard } from 'src/shared/auth/role.guard'; +import { UserRole } from 'src/shared/auth/user-role.enum'; +import { PartnerStatisticDto, PartnerTimelineDto } from './dto/partner-statistic.dto'; +import { PartnerStatisticRateLimitGuard } from './partner-statistic-rate-limit.guard'; +import { PartnerStatisticGranularity } from './partner-statistic.enum'; +import { PartnerStatisticService } from './partner-statistic.service'; + +@ApiTags('Statistic') +@Controller('statistic') +export class PartnerStatisticController { + constructor(private readonly partnerStatisticService: PartnerStatisticService) {} + + @Get('partner') + @ApiBearerAuth() + @UseGuards(AuthGuard(), RoleGuard(UserRole.CLIENT_COMPANY), PartnerStatisticRateLimitGuard) + // 120 req/h per wallet: dashboard auto-refresh (~1/min) for summary + headroom + @Throttle(120, 3600) + @ApiOkResponse({ type: PartnerStatisticDto }) + @ApiQuery({ + name: 'from', + required: false, + description: 'Period start (ISO date). Snapped to UTC day start. Default: 30 days before `to`.', + }) + @ApiQuery({ + name: 'to', + required: false, + description: 'Period end (ISO date). Snapped to exclusive UTC day end. Default: now.', + }) + async getPartnerStatistics( + @GetJwt() jwt: JwtPayload, + @Query('from') from?: string, + @Query('to') to?: string, + ): Promise { + return this.partnerStatisticService.getStatistics(jwt.user, from, to); + } + + @Get('partner/timeline') + @ApiBearerAuth() + @UseGuards(AuthGuard(), RoleGuard(UserRole.CLIENT_COMPANY), PartnerStatisticRateLimitGuard) + // 120 req/h per wallet: same budget as summary so a dual-widget dashboard can refresh without 429s + @Throttle(120, 3600) + @ApiOkResponse({ type: PartnerTimelineDto }) + @ApiQuery({ + name: 'from', + required: false, + description: 'Period start (ISO date). Snapped to UTC day start. Default: 30 days before `to`.', + }) + @ApiQuery({ + name: 'to', + required: false, + description: 'Period end (ISO date). Snapped to exclusive UTC day end. Default: now.', + }) + @ApiQuery({ + name: 'granularity', + required: false, + enum: PartnerStatisticGranularity, + description: 'Bucket size. Default: day.', + }) + async getPartnerTimeline( + @GetJwt() jwt: JwtPayload, + @Query('from') from?: string, + @Query('to') to?: string, + @Query('granularity') granularity?: PartnerStatisticGranularity, + ): Promise { + return this.partnerStatisticService.getTimeline(jwt.user, from, to, granularity); + } +} diff --git a/src/subdomains/core/statistic/partner-statistic.enum.ts b/src/subdomains/core/statistic/partner-statistic.enum.ts new file mode 100644 index 0000000000..16c26116c5 --- /dev/null +++ b/src/subdomains/core/statistic/partner-statistic.enum.ts @@ -0,0 +1,42 @@ +import { TransactionSourceType } from 'src/subdomains/supporting/payment/entities/transaction.entity'; + +/** k-anonymity threshold: disclosure units need min(transactions, distinct users) ≥ k. */ +export const PARTNER_STATISTIC_SUPPRESSION_THRESHOLD = 5; + +/** Default lookback when `from`/`to` are omitted (calendar days before `to`). */ +export const PARTNER_STATISTIC_DEFAULT_PERIOD_DAYS = 30; + +/** Maximum allowed period span in calendar days (half-open [from, to)). */ +export const PARTNER_STATISTIC_MAX_PERIOD_DAYS = 366; + +/** Max concurrent SQL queries per partner-statistic request (pool size is 10). */ +export const PARTNER_STATISTIC_QUERY_CONCURRENCY = 4; + +export enum PartnerStatisticGranularity { + DAY = 'day', + WEEK = 'week', + MONTH = 'month', +} + +/** Partner-facing trade direction (API contract; lowercase). */ +export enum PartnerStatisticDirection { + BUY = 'buy', + SELL = 'sell', + SWAP = 'swap', +} + +export enum PartnerPaymentMethodName { + BANK = 'Bank', + CARD = 'Card', + ON_CHAIN = 'OnChain', + REFERRAL = 'Referral', +} + +/** Maps transaction.sourceType to a partner-facing payment method label. */ +export const PartnerPaymentMethodMap: { [key in TransactionSourceType]: PartnerPaymentMethodName } = { + [TransactionSourceType.BANK_TX]: PartnerPaymentMethodName.BANK, + [TransactionSourceType.CHECKOUT_TX]: PartnerPaymentMethodName.CARD, + [TransactionSourceType.CRYPTO_INPUT]: PartnerPaymentMethodName.ON_CHAIN, + [TransactionSourceType.REF]: PartnerPaymentMethodName.REFERRAL, + [TransactionSourceType.MANUAL_REF]: PartnerPaymentMethodName.REFERRAL, +}; diff --git a/src/subdomains/core/statistic/partner-statistic.service.ts b/src/subdomains/core/statistic/partner-statistic.service.ts new file mode 100644 index 0000000000..05d723ba00 --- /dev/null +++ b/src/subdomains/core/statistic/partner-statistic.service.ts @@ -0,0 +1,981 @@ +import { BadRequestException, Injectable } from '@nestjs/common'; +import { Config } from 'src/config/config'; +import { Util } from 'src/shared/utils/util'; +import { CheckStatus } from 'src/subdomains/core/aml/enums/check-status.enum'; +import { BuyCryptoRepository } from 'src/subdomains/core/buy-crypto/process/repositories/buy-crypto.repository'; +import { BuyFiatRepository } from 'src/subdomains/core/sell-crypto/process/buy-fiat.repository'; +import { UserRepository } from 'src/subdomains/generic/user/models/user/user.repository'; +import { WalletRepository } from 'src/subdomains/generic/user/models/wallet/wallet.repository'; +import { + TransactionRequestStatus, + TransactionRequestType, +} from 'src/subdomains/supporting/payment/entities/transaction-request.entity'; +import { TransactionSourceType } from 'src/subdomains/supporting/payment/entities/transaction.entity'; +import { TransactionRequestRepository } from 'src/subdomains/supporting/payment/repositories/transaction-request.repository'; +import { SelectQueryBuilder } from 'typeorm'; +import { BuyCrypto } from '../buy-crypto/process/entities/buy-crypto.entity'; +import { BuyFiat } from '../sell-crypto/process/buy-fiat.entity'; +import { + PartnerAssetBreakdownDto, + PartnerCompletionDto, + PartnerNamedBreakdownDto, + PartnerPaymentInfoDirectionDto, + PartnerReferralDto, + PartnerSettlementDirectionDto, + PartnerStatisticDto, + PartnerTimelineBucketDto, + PartnerTimelineDto, +} from './dto/partner-statistic.dto'; +import { + PARTNER_STATISTIC_DEFAULT_PERIOD_DAYS, + PARTNER_STATISTIC_MAX_PERIOD_DAYS, + PARTNER_STATISTIC_QUERY_CONCURRENCY, + PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, + PartnerPaymentMethodMap, + PartnerStatisticDirection, + PartnerStatisticGranularity, +} from './partner-statistic.enum'; +import { + suppressAdditiveGroup, + suppressAllTimeVolume, + suppressBreakdownRows, + suppressPeriodTotals, + suppressScalar, + suppressTimelineBuckets, +} from './partner-statistic.suppression'; + +type Direction = PartnerStatisticDirection; + +interface BaseTxQueryOptions { + /** When true (default), only amlCheck=Pass rows. Settlement stage B sets false. */ + amlPassOnly?: boolean; +} + +interface AggregateRow { + volume: string | number | null; + transactions: string | number | null; + users: string | number | null; +} + +interface NamedAggregateRow extends AggregateRow { + name: string | null; + blockchain?: string | null; +} + +interface TimelineRawRow { + bucket: Date | string; + volume: string | number | null; + transactions: string | number | null; + users: string | number | null; +} + +interface DirectionAgg { + volume: number; + transactions: number; + users: number; +} + +@Injectable() +export class PartnerStatisticService { + constructor( + private readonly buyCryptoRepo: BuyCryptoRepository, + private readonly buyFiatRepo: BuyFiatRepository, + private readonly userRepo: UserRepository, + private readonly walletRepo: WalletRepository, + private readonly txRequestRepo: TransactionRequestRepository, + ) {} + + // --- PUBLIC API --- // + + async getStatistics(walletId: number, from?: string | Date, to?: string | Date): Promise { + const period = this.resolvePeriod(from, to); + + const [ + buyAgg, + sellAgg, + swapAgg, + activeUsersRaw, + newUsersRaw, + allTimeRaw, + referralRaw, + assetRows, + fiatRows, + blockchainRows, + paymentMethodRows, + completionResult, + ] = await this.runLimited([ + () => this.aggregateByDirection(walletId, period.from, period.to, PartnerStatisticDirection.BUY), + () => this.aggregateByDirection(walletId, period.from, period.to, PartnerStatisticDirection.SELL), + () => this.aggregateByDirection(walletId, period.from, period.to, PartnerStatisticDirection.SWAP), + () => this.countActiveUsers(walletId, period.from, period.to), + () => this.countNewUsers(walletId, period.from, period.to), + () => this.getAllTime(walletId), + () => this.getReferralRaw(walletId), + () => this.aggregateAssets(walletId, period.from, period.to), + () => this.aggregateFiatCurrencies(walletId, period.from, period.to), + () => this.aggregateBlockchains(walletId, period.from, period.to), + () => this.aggregatePaymentMethods(walletId, period.from, period.to), + () => this.getCompletion(walletId, period.from, period.to), + ]); + + const rawVolume = { + buy: buyAgg.volume, + sell: sellAgg.volume, + swap: swapAgg.volume, + total: Util.round(buyAgg.volume + sellAgg.volume + swapAgg.volume, Config.defaultVolumeDecimal), + }; + const rawTransactions = { + buy: buyAgg.transactions, + sell: sellAgg.transactions, + swap: swapAgg.transactions, + total: buyAgg.transactions + sellAgg.transactions + swapAgg.transactions, + }; + const rawUsers = { + buy: buyAgg.users, + sell: sellAgg.users, + swap: swapAgg.users, + total: activeUsersRaw, + }; + + const totalsSuppressed = suppressPeriodTotals(rawVolume, rawTransactions, rawUsers); + const averageTransactionVolume = + totalsSuppressed.averageTransactionVolume != null + ? Util.round(totalsSuppressed.averageTransactionVolume, Config.defaultVolumeDecimal) + : null; + + const tradingUsersSuppressed = suppressScalar(allTimeRaw.tradingUsers); + const allTimeSuppressed = suppressAllTimeVolume(allTimeRaw.volume, allTimeRaw.tradingUsers); + const referral = this.applyReferralSuppression(referralRaw, allTimeRaw.tradingUsers); + + const assets = suppressBreakdownRows(assetRows); + const fiatCurrencies = suppressBreakdownRows(fiatRows); + const blockchains = suppressBreakdownRows(blockchainRows); + const paymentMethods = suppressBreakdownRows(paymentMethodRows); + + const activeUsers = suppressScalar(activeUsersRaw); + const newUsers = suppressScalar(newUsersRaw); + + const suppressedBuckets = + assets.suppressedCount + + fiatCurrencies.suppressedCount + + blockchains.suppressedCount + + paymentMethods.suppressedCount + + totalsSuppressed.suppressedCount + + allTimeSuppressed.suppressedCount + + completionResult.suppressedCount + + (activeUsers === null ? 1 : 0) + + (newUsers === null ? 1 : 0) + + (tradingUsersSuppressed === null ? 1 : 0) + + (referral.volume === null && referralRaw.volume !== 0 ? 1 : 0); + + return { + period, + currency: 'CHF', + totals: { + volume: totalsSuppressed.volume, + transactions: totalsSuppressed.transactions, + averageTransactionVolume, + activeUsers, + newUsers, + }, + allTime: { + volume: allTimeSuppressed.volume, + registeredUsers: allTimeRaw.registeredUsers, + tradingUsers: tradingUsersSuppressed, + }, + breakdown: { + assets: assets.rows.map(({ users: _u, ...row }) => row), + fiatCurrencies: fiatCurrencies.rows.map(({ users: _u, ...row }) => row), + blockchains: blockchains.rows.map(({ users: _u, ...row }) => row), + paymentMethods: paymentMethods.rows.map(({ users: _u, ...row }) => row), + }, + referral, + completion: completionResult.completion, + meta: { + suppressionThreshold: PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, + suppressedBuckets, + generatedAt: new Date(), + }, + }; + } + + async getTimeline( + walletId: number, + from?: string | Date, + to?: string | Date, + granularity: string = PartnerStatisticGranularity.DAY, + ): Promise { + const resolvedGranularity = this.parseGranularity(granularity); + const period = this.resolvePeriod(from, to); + + const [buyRows, sellRows, swapRows] = await this.runLimited([ + () => + this.timelineByDirection(walletId, period.from, period.to, PartnerStatisticDirection.BUY, resolvedGranularity), + () => + this.timelineByDirection(walletId, period.from, period.to, PartnerStatisticDirection.SELL, resolvedGranularity), + () => + this.timelineByDirection(walletId, period.from, period.to, PartnerStatisticDirection.SWAP, resolvedGranularity), + ]); + + const filled = this.fillTimelineGaps(period.from, period.to, resolvedGranularity, buyRows, sellRows, swapRows); + const { buckets, suppressedCount } = suppressTimelineBuckets(filled); + + // Drop internal users field from the public payload. + const publicBuckets: PartnerTimelineBucketDto[] = buckets.map(({ users: _u, ...rest }) => rest); + + return { + period, + currency: 'CHF', + granularity: resolvedGranularity, + buckets: publicBuckets, + meta: { + suppressionThreshold: PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, + suppressedBuckets: suppressedCount, + }, + }; + } + + // --- PERIOD / VALIDATION --- // + + /** + * Snaps `from`/`to` to UTC day boundaries (half-open [from, to)), enforces min 1 day and max span. + * Accepts ISO strings or Date; parsing lives here so the controller stays free of date logic. + */ + resolvePeriod(from?: string | Date, to?: string | Date): { from: Date; to: Date } { + const resolvedTo = this.parseDate(to) ?? new Date(); + const resolvedFrom = this.parseDate(from) ?? Util.daysBefore(PARTNER_STATISTIC_DEFAULT_PERIOD_DAYS, resolvedTo); + + if (isNaN(resolvedFrom.getTime()) || isNaN(resolvedTo.getTime())) { + throw new BadRequestException('Invalid date'); + } + + // Snap: from → start of its UTC day; to → start of the UTC day after the last included day (exclusive). + const fromDay = this.startOfUtcDay(resolvedFrom); + const toDayStart = this.startOfUtcDay(resolvedTo); + const toExclusive = this.addUtcDays(toDayStart, 1); + + if (fromDay.getTime() >= toExclusive.getTime()) { + throw new BadRequestException('From must be before to'); + } + + const spanDays = (toExclusive.getTime() - fromDay.getTime()) / (24 * 3600 * 1000); + if (spanDays < 1) { + throw new BadRequestException('Period must cover at least one full day'); + } + if (spanDays > PARTNER_STATISTIC_MAX_PERIOD_DAYS) { + throw new BadRequestException( + `Period must not exceed ${PARTNER_STATISTIC_MAX_PERIOD_DAYS} days (got ${Math.ceil(spanDays)})`, + ); + } + + return { from: fromDay, to: toExclusive }; + } + + parseGranularity(value: string): PartnerStatisticGranularity { + const allowed = Object.values(PartnerStatisticGranularity) as string[]; + if (!allowed.includes(value)) { + throw new BadRequestException(`Invalid granularity '${value}'. Allowed: ${allowed.join(', ')}`); + } + return value as PartnerStatisticGranularity; + } + + parseDate(value?: string | Date): Date | undefined { + if (value == null || value === '') return undefined; + if (value instanceof Date) { + if (isNaN(value.getTime())) throw new BadRequestException('Invalid date'); + return value; + } + const date = new Date(value); + if (isNaN(date.getTime())) throw new BadRequestException(`Invalid date: ${value}`); + return date; + } + + // --- AGGREGATES (SQL GROUP BY — never load user rows) --- // + + private async aggregateByDirection( + walletId: number, + from: Date, + to: Date, + direction: Direction, + ): Promise { + const qb = this.baseTxQuery(direction, walletId, from, to); + qb.select('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users'); + + const raw = await qb.getRawOne(); + return { + volume: this.toVolume(raw?.volume), + transactions: this.toCount(raw?.transactions), + users: this.toCount(raw?.users), + }; + } + + /** + * Distinct active users across buy/sell/swap — COUNT happens in the DB via UNION, not in Node. + */ + private async countActiveUsers(walletId: number, from: Date, to: Date): Promise { + const buyQb = this.baseTxQuery(PartnerStatisticDirection.BUY, walletId, from, to).select('user.id', 'id'); + const sellQb = this.baseTxQuery(PartnerStatisticDirection.SELL, walletId, from, to).select('user.id', 'id'); + const swapQb = this.baseTxQuery(PartnerStatisticDirection.SWAP, walletId, from, to).select('user.id', 'id'); + + const unionSql = `(${buyQb.getQuery()}) UNION (${sellQb.getQuery()}) UNION (${swapQb.getQuery()})`; + + const raw = await this.buyCryptoRepo.manager + .createQueryBuilder() + .from(`(${unionSql})`, 'active_users') + .select('COUNT(*)', 'count') + .setParameters({ + ...buyQb.getParameters(), + ...sellQb.getParameters(), + ...swapQb.getParameters(), + }) + .getRawOne<{ count: string | number }>(); + + return this.toCount(raw?.count); + } + + private async countNewUsers(walletId: number, from: Date, to: Date): Promise { + return this.userRepo + .createQueryBuilder('user') + .where('user.walletId = :walletId', { walletId }) + .andWhere('user.created >= :from AND user.created < :to', { from, to }) + .getCount(); + } + + private async getAllTime( + walletId: number, + ): Promise<{ volume: { buy: number; sell: number; total: number }; registeredUsers: number; tradingUsers: number }> { + const raw = await this.userRepo + .createQueryBuilder('user') + .select('COALESCE(SUM(user.buyVolume), 0)', 'buy') + .addSelect('COALESCE(SUM(user.sellVolume), 0)', 'sell') + .addSelect('COUNT(*)', 'registeredUsers') + .addSelect( + 'COALESCE(SUM(CASE WHEN user.buyVolume > 0 OR user.sellVolume > 0 THEN 1 ELSE 0 END), 0)', + 'tradingUsers', + ) + .where('user.walletId = :walletId', { walletId }) + .getRawOne<{ buy: string; sell: string; registeredUsers: string; tradingUsers: string }>(); + + const buy = this.toVolume(raw?.buy); + const sell = this.toVolume(raw?.sell); + + return { + volume: { + buy, + sell, + total: Util.round(buy + sell, Config.defaultVolumeDecimal), + }, + registeredUsers: this.toCount(raw?.registeredUsers), + tradingUsers: this.toCount(raw?.tradingUsers), + }; + } + + /** + * Reads the wallet **owner** account (all wallets of that owner), not only the querying wallet. + * Open credit formula matches user.service / ref-reward.service: + * refCredit + partnerRefCredit − paidRefCredit. + */ + private async getReferralRaw(walletId: number): Promise<{ + volume: number; + creditEarned: number; + creditPaid: number; + creditOpen: number; + }> { + const raw = await this.walletRepo + .createQueryBuilder('wallet') + .leftJoin('wallet.owner', 'owner') + .select('COALESCE(owner.partnerRefVolume, 0)', 'volume') + .addSelect('COALESCE(owner.partnerRefCredit, 0)', 'partnerRefCredit') + .addSelect('COALESCE(owner.refCredit, 0)', 'refCredit') + .addSelect('COALESCE(owner.paidRefCredit, 0)', 'paidRefCredit') + .where('wallet.id = :walletId', { walletId }) + .getRawOne<{ volume: string; partnerRefCredit: string; refCredit: string; paidRefCredit: string }>(); + + const volume = this.toVolume(raw?.volume); + const partnerRefCredit = this.toVolume(raw?.partnerRefCredit); + const refCredit = this.toVolume(raw?.refCredit); + const paidRefCredit = this.toVolume(raw?.paidRefCredit); + + return { + volume, + creditEarned: partnerRefCredit, + creditPaid: paidRefCredit, + creditOpen: Util.round(refCredit + partnerRefCredit - paidRefCredit, Config.defaultVolumeDecimal), + }; + } + + private applyReferralSuppression( + raw: { volume: number; creditEarned: number; creditPaid: number; creditOpen: number }, + tradingUsers: number, + ): PartnerReferralDto { + // Referral balances move with individual customer trades. Gate them on tradingUsers so two + // successive polls cannot recover a single trade’s credit delta when the cohort is under k. + // These are the partner’s own account figures (owner-scoped), but the leakage path is the same. + if (suppressScalar(tradingUsers) === null && tradingUsers > 0) { + return { + volume: null, + creditEarned: null, + creditPaid: null, + creditOpen: null, + currency: 'EUR', + }; + } + return { ...raw, currency: 'EUR' }; + } + + private async aggregateAssets( + walletId: number, + from: Date, + to: Date, + ): Promise<(PartnerAssetBreakdownDto & { users: number })[]> { + const [buy, sell, swap] = await this.runLimited([ + () => this.assetQuery(PartnerStatisticDirection.BUY, walletId, from, to), + () => this.assetQuery(PartnerStatisticDirection.SELL, walletId, from, to), + () => this.assetQuery(PartnerStatisticDirection.SWAP, walletId, from, to), + ]); + + return [...buy, ...sell, ...swap].sort((a, b) => b.volume - a.volume); + } + + private async assetQuery( + direction: Direction, + walletId: number, + from: Date, + to: Date, + ): Promise<(PartnerAssetBreakdownDto & { users: number })[]> { + const qb = this.baseTxQuery(direction, walletId, from, to); + + if (direction === PartnerStatisticDirection.SELL) { + // GROUP BY must use qualified columns — never SELECT aliases (Postgres resolves aliases as input columns). + qb.leftJoin('tx.cryptoInput', 'cryptoInput') + .leftJoin('cryptoInput.asset', 'inputAsset') + .select('tx.inputAsset', 'name') + .addSelect('inputAsset.blockchain', 'blockchain') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy('tx.inputAsset') + .addGroupBy('inputAsset.blockchain'); + } else { + qb.leftJoin('tx.outputAsset', 'outputAsset') + .select('outputAsset.name', 'name') + .addSelect('outputAsset.blockchain', 'blockchain') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy('outputAsset.name') + .addGroupBy('outputAsset.blockchain'); + } + + const rows = await qb.getRawMany(); + return rows + .filter((r) => r.name) + .map((r) => ({ + name: r.name as string, + blockchain: r.blockchain ?? null, + direction, + volume: this.toVolume(r.volume), + transactions: this.toCount(r.transactions), + users: this.toCount(r.users), + })); + } + + private async aggregateFiatCurrencies( + walletId: number, + from: Date, + to: Date, + ): Promise<(PartnerNamedBreakdownDto & { users: number })[]> { + // Buy: inputAsset is the fiat ticker. Sell: outputAsset is Fiat. Swap has no fiat leg. + const buyQb = this.baseTxQuery(PartnerStatisticDirection.BUY, walletId, from, to) + .select('tx.inputAsset', 'name') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy('tx.inputAsset'); + + const sellQb = this.baseTxQuery(PartnerStatisticDirection.SELL, walletId, from, to) + .leftJoin('tx.outputAsset', 'fiat') + .select('fiat.name', 'name') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy('fiat.name'); + + const [buyRows, sellRows] = await this.runLimited([ + () => buyQb.getRawMany(), + () => sellQb.getRawMany(), + ]); + + return this.mergeNamedRows([...buyRows, ...sellRows]); + } + + private async aggregateBlockchains( + walletId: number, + from: Date, + to: Date, + ): Promise<(PartnerNamedBreakdownDto & { users: number })[]> { + const queries = ([PartnerStatisticDirection.BUY, PartnerStatisticDirection.SWAP] as Direction[]).map( + (direction) => () => + this.baseTxQuery(direction, walletId, from, to) + .leftJoin('tx.outputAsset', 'outputAsset') + .select('outputAsset.blockchain', 'name') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy('outputAsset.blockchain') + .getRawMany(), + ); + + const sellQ = () => + this.baseTxQuery(PartnerStatisticDirection.SELL, walletId, from, to) + .leftJoin('tx.cryptoInput', 'cryptoInput') + .leftJoin('cryptoInput.asset', 'inputAsset') + .select('inputAsset.blockchain', 'name') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy('inputAsset.blockchain') + .getRawMany(); + + const rows = (await this.runLimited([...queries, sellQ])).flat(); + return this.mergeNamedRows(rows); + } + + private async aggregatePaymentMethods( + walletId: number, + from: Date, + to: Date, + ): Promise<(PartnerNamedBreakdownDto & { users: number })[]> { + const rows = ( + await this.runLimited( + ( + [PartnerStatisticDirection.BUY, PartnerStatisticDirection.SELL, PartnerStatisticDirection.SWAP] as Direction[] + ).map( + (direction) => () => + this.baseTxQuery(direction, walletId, from, to) + .innerJoin('tx.transaction', 'transaction') + .select('transaction.sourceType', 'name') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy('transaction.sourceType') + .getRawMany(), + ), + ) + ).flat(); + + const mapped = rows.map((r) => ({ + ...r, + name: PartnerPaymentMethodMap[r.name as TransactionSourceType] ?? r.name, + })); + + return this.mergeNamedRows(mapped); + } + + private async timelineByDirection( + walletId: number, + from: Date, + to: Date, + direction: Direction, + granularity: PartnerStatisticGranularity, + ): Promise> { + // Truncate on the stored wall clock (UTC values in TIMESTAMP without TZ), then tag the + // result as UTC so the driver returns an unambiguous absolute instant. Session TimeZone + // must not shift buckets — DATE_TRUNC on timestamp without tz is field-only; AT TIME ZONE + // 'UTC' only re-labels the truncated value as timestamptz. + const trunc = `DATE_TRUNC('${granularity}', tx.created) AT TIME ZONE 'UTC'`; + const qb = this.baseTxQuery(direction, walletId, from, to) + .select(trunc, 'bucket') + .addSelect('COALESCE(SUM(tx.amountInChf), 0)', 'volume') + .addSelect('COUNT(*)', 'transactions') + .addSelect('COUNT(DISTINCT user.id)', 'users') + .groupBy(trunc) + .orderBy(trunc, 'ASC'); + + const rows = await qb.getRawMany(); + const map = new Map(); + for (const row of rows) { + // UTC keys on both sides (SQL buckets + fill loop). Period bounds are UTC-normalized in + // resolvePeriod; this module deliberately does not follow support-issue.service local + // date-parts — that path has no UTC period snap. + const key = this.bucketKey(this.startOfBucket(new Date(row.bucket), granularity)); + const volume = this.toVolume(row.volume); + const transactions = this.toCount(row.transactions); + const users = this.toCount(row.users); + const existing = map.get(key); + if (existing) { + existing.volume = Util.round(existing.volume + volume, Config.defaultVolumeDecimal); + existing.transactions += transactions; + // users can double-count on key collision; take max as a lower-bound person estimate + existing.users = Math.max(existing.users, users); + } else { + map.set(key, { volume, transactions, users }); + } + } + return map; + } + + private fillTimelineGaps( + from: Date, + to: Date, + granularity: PartnerStatisticGranularity, + buy: Map, + sell: Map, + swap: Map, + ): (PartnerTimelineBucketDto & { users: { buy: number; sell: number; swap: number } })[] { + const buckets: (PartnerTimelineBucketDto & { users: { buy: number; sell: number; swap: number } })[] = []; + let cursor = this.startOfBucket(from, granularity); + // `to` is exclusive (half-open period). + const end = to.getTime(); + + while (cursor.getTime() < end) { + const key = this.bucketKey(cursor); + const b = buy.get(key) ?? { volume: 0, transactions: 0, users: 0 }; + const s = sell.get(key) ?? { volume: 0, transactions: 0, users: 0 }; + const w = swap.get(key) ?? { volume: 0, transactions: 0, users: 0 }; + const next = this.addBucket(cursor, granularity); + // Edge buckets whose natural range extends outside [from, to) are partial. + const partial = cursor.getTime() < from.getTime() || next.getTime() > to.getTime(); + + buckets.push({ + date: new Date(cursor), + volume: { buy: b.volume, sell: s.volume, swap: w.volume }, + transactions: { buy: b.transactions, sell: s.transactions, swap: w.transactions }, + users: { buy: b.users, sell: s.users, swap: w.users }, + suppressed: false, + partial, + }); + + cursor = next; + } + + return buckets; + } + + // --- COMPLETION (stage A + B) --- // + + private async getCompletion( + walletId: number, + from: Date, + to: Date, + ): Promise<{ completion: PartnerCompletionDto; suppressedCount: number }> { + const [paymentInfoRaw, buySettlement, sellSettlement, swapSettlement] = await this.runLimited([ + () => this.aggregatePaymentInfoRequests(walletId, from, to), + () => this.aggregateSettlement(walletId, from, to, PartnerStatisticDirection.BUY), + () => this.aggregateSettlement(walletId, from, to, PartnerStatisticDirection.SELL), + () => this.aggregateSettlement(walletId, from, to, PartnerStatisticDirection.SWAP), + ]); + + let suppressedCount = 0; + + const paymentInfoRequests = { + buy: this.suppressPaymentInfoFunnel(paymentInfoRaw.buy, (n) => { + suppressedCount += n; + }), + sell: this.suppressPaymentInfoFunnel(paymentInfoRaw.sell, (n) => { + suppressedCount += n; + }), + swap: this.suppressPaymentInfoFunnel(paymentInfoRaw.swap, (n) => { + suppressedCount += n; + }), + }; + + const settlement = { + buy: this.suppressSettlementFunnel(buySettlement, (n) => { + suppressedCount += n; + }), + sell: this.suppressSettlementFunnel(sellSettlement, (n) => { + suppressedCount += n; + }), + swap: this.suppressSettlementFunnel(swapSettlement, (n) => { + suppressedCount += n; + }), + }; + + return { completion: { paymentInfoRequests, settlement }, suppressedCount }; + } + + private async aggregatePaymentInfoRequests( + walletId: number, + from: Date, + to: Date, + ): Promise< + Record< + Direction, + { requested: number; paymentReceived: number; waitingForPayment: number; noPaymentReceived: number } + > + > { + const empty = () => ({ requested: 0, paymentReceived: 0, waitingForPayment: 0, noPaymentReceived: 0 }); + const byDir: Record> = { + [PartnerStatisticDirection.BUY]: empty(), + [PartnerStatisticDirection.SELL]: empty(), + [PartnerStatisticDirection.SWAP]: empty(), + }; + + const rows = await this.txRequestRepo + .createQueryBuilder('tr') + .innerJoin('tr.user', 'user') + .select('tr.type', 'type') + .addSelect('tr.status', 'status') + .addSelect('COUNT(*)', 'count') + .where('user.walletId = :walletId', { walletId }) + .andWhere('tr.created >= :from AND tr.created < :to', { from, to }) + .groupBy('tr.type') + .addGroupBy('tr.status') + .getRawMany<{ type: string; status: string; count: string | number }>(); + + const typeToDir: Record = { + [TransactionRequestType.BUY]: PartnerStatisticDirection.BUY, + [TransactionRequestType.SELL]: PartnerStatisticDirection.SELL, + [TransactionRequestType.SWAP]: PartnerStatisticDirection.SWAP, + }; + + for (const row of rows) { + const dir = typeToDir[row.type]; + if (!dir) continue; + const count = this.toCount(row.count); + byDir[dir].requested += count; + if (row.status === TransactionRequestStatus.COMPLETED) byDir[dir].paymentReceived += count; + else if (row.status === TransactionRequestStatus.WAITING_FOR_PAYMENT) byDir[dir].waitingForPayment += count; + else if (row.status === TransactionRequestStatus.CREATED) byDir[dir].noPaymentReceived += count; + } + + return byDir; + } + + private async aggregateSettlement( + walletId: number, + from: Date, + to: Date, + direction: Direction, + ): Promise<{ received: number; delivered: number; rejected: number; inProgress: number }> { + const pass = CheckStatus.PASS; + const fail = CheckStatus.FAIL; + + const qb = this.baseTxQuery(direction, walletId, from, to, { amlPassOnly: false }); + qb.select('COUNT(*)', 'received') + .addSelect( + `COALESCE(SUM(CASE WHEN tx.amlCheck = :passCheck AND tx.isComplete = true THEN 1 ELSE 0 END), 0)`, + 'delivered', + ) + .addSelect(`COALESCE(SUM(CASE WHEN tx.amlCheck = :failCheck THEN 1 ELSE 0 END), 0)`, 'rejected') + .setParameter('passCheck', pass) + .setParameter('failCheck', fail); + + const raw = await qb.getRawOne<{ received: string; delivered: string; rejected: string }>(); + const received = this.toCount(raw?.received); + const delivered = this.toCount(raw?.delivered); + const rejected = this.toCount(raw?.rejected); + const inProgress = Math.max(received - delivered - rejected, 0); + + return { received, delivered, rejected, inProgress }; + } + + private suppressPaymentInfoFunnel( + raw: { requested: number; paymentReceived: number; waitingForPayment: number; noPaymentReceived: number }, + onSuppressed: (n: number) => void, + ): PartnerPaymentInfoDirectionDto { + const { values, rate, suppressedCount } = suppressAdditiveGroup( + { + requested: raw.requested, + paymentReceived: raw.paymentReceived, + waitingForPayment: raw.waitingForPayment, + noPaymentReceived: raw.noPaymentReceived, + }, + { numeratorKey: 'paymentReceived', denominatorKey: 'requested' }, + ); + onSuppressed(suppressedCount); + + return { + requested: values.requested, + paymentReceived: values.paymentReceived, + waitingForPayment: values.waitingForPayment, + noPaymentReceived: values.noPaymentReceived, + receivedRate: rate, + }; + } + + private suppressSettlementFunnel( + raw: { received: number; delivered: number; rejected: number; inProgress: number }, + onSuppressed: (n: number) => void, + ): PartnerSettlementDirectionDto { + const { values, rate, suppressedCount } = suppressAdditiveGroup( + { + received: raw.received, + delivered: raw.delivered, + rejected: raw.rejected, + inProgress: raw.inProgress, + }, + { numeratorKey: 'delivered', denominatorKey: 'received' }, + ); + onSuppressed(suppressedCount); + + return { + received: values.received, + delivered: values.delivered, + rejected: values.rejected, + inProgress: values.inProgress, + deliveredRate: rate, + }; + } + + // --- QUERY BUILDERS --- // + + /** + * Base query for partner transactions of one direction. + * Scope is always `user.walletId = :walletId` — never accept walletId from the client. + * Default filters amlCheck=Pass (volume/totals/breakdown). Settlement stage B passes amlPassOnly: false. + * Period is half-open: created >= from AND created < to. + */ + private baseTxQuery( + direction: Direction, + walletId: number, + from: Date, + to: Date, + options: BaseTxQueryOptions = {}, + ): SelectQueryBuilder { + // Fail-closed: without an explicit opt-out, only amlCheck=Pass rows are counted. That is the safe + // direction for volume/totals — omitting the filter would inflate figures with rejected traffic. + // Settlement stage B is the sole caller that sets amlPassOnly: false (needs Fail / in-progress too). + const amlPassOnly = options.amlPassOnly ?? true; + + let qb: SelectQueryBuilder; + + if (direction === PartnerStatisticDirection.SELL) { + qb = this.buyFiatRepo + .createQueryBuilder('tx') + .innerJoin('tx.sell', 'route') + .innerJoin('route.user', 'user') + .where('user.walletId = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }); + } else if (direction === PartnerStatisticDirection.BUY) { + qb = this.buyCryptoRepo + .createQueryBuilder('tx') + .innerJoin('tx.buy', 'route') + .innerJoin('route.user', 'user') + .where('user.walletId = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }); + } else { + qb = this.buyCryptoRepo + .createQueryBuilder('tx') + .innerJoin('tx.cryptoRoute', 'route') + .innerJoin('route.user', 'user') + .where('user.walletId = :walletId', { walletId }) + .andWhere('tx.created >= :from AND tx.created < :to', { from, to }); + } + + if (amlPassOnly) { + qb.andWhere('tx.amlCheck = :check', { check: CheckStatus.PASS }); + } + + return qb; + } + + // --- HELPERS --- // + + /** + * Runs async tasks with a hard cap on concurrency so a single partner request cannot saturate + * the TypeORM pool (default size 10). Max concurrency: PARTNER_STATISTIC_QUERY_CONCURRENCY (4). + */ + private async runLimited Promise)[]>( + tasks: [...T], + concurrency = PARTNER_STATISTIC_QUERY_CONCURRENCY, + ): Promise<{ [K in keyof T]: T[K] extends () => Promise ? R : never }> { + const results: unknown[] = new Array(tasks.length); + let next = 0; + + const workers = Array.from({ length: Math.min(concurrency, tasks.length) }, async () => { + while (next < tasks.length) { + const i = next++; + results[i] = await tasks[i](); + } + }); + + await Promise.all(workers); + return results as { [K in keyof T]: T[K] extends () => Promise ? R : never }; + } + + /** Exposed for tests that exercise mergeNamedRows / breakdown mapping without full SQL. */ + mergeNamedRows(rows: NamedAggregateRow[]): (PartnerNamedBreakdownDto & { users: number })[] { + const map = new Map(); + + for (const row of rows) { + if (!row.name) continue; + const existing = map.get(row.name); + const volume = this.toVolume(row.volume); + const transactions = this.toCount(row.transactions); + const users = this.toCount(row.users); + if (existing) { + existing.volume = Util.round(existing.volume + volume, Config.defaultVolumeDecimal); + existing.transactions += transactions; + existing.users = Math.max(existing.users, users); + } else { + map.set(row.name, { name: row.name, volume, transactions, users }); + } + } + + return [...map.values()].sort((a, b) => b.volume - a.volume); + } + + /** + * SQL aggregates always use COALESCE(SUM(...), 0) or COUNT(*), so an empty match set yields 0 + * (a row is still returned). Null/undefined only appears when getRawOne finds no row at all + * (e.g. missing wallet on a left join) — that is absence of data, which is correctly 0 volume. + */ + private toVolume(value: string | number | null | undefined): number { + return Util.round(+(value ?? 0), Config.defaultVolumeDecimal); + } + + /** + * Same as toVolume: COUNT(*) is 0 over an empty set; null/undefined means no row, i.e. zero count. + */ + private toCount(value: string | number | null | undefined): number { + return Math.trunc(+(value ?? 0)); + } + + /** + * UTC date-part key — matches resolvePeriod (UTC day snap) and SQL + * `DATE_TRUNC(...) AT TIME ZONE 'UTC'`. Not process-local: unlike support-issue.service, + * this timeline is period-normalized to UTC, so keys must be UTC too. + */ + private bucketKey(date: Date): string { + const pad = (n: number) => String(n).padStart(2, '0'); + return `${date.getUTCFullYear()}-${pad(date.getUTCMonth() + 1)}-${pad(date.getUTCDate())}T${pad(date.getUTCHours())}:${pad(date.getUTCMinutes())}:${pad(date.getUTCSeconds())}`; + } + + private startOfBucket(date: Date, granularity: PartnerStatisticGranularity): Date { + const d = new Date(date); + d.setUTCHours(0, 0, 0, 0); + + if (granularity === PartnerStatisticGranularity.MONTH) { + d.setUTCDate(1); + } else if (granularity === PartnerStatisticGranularity.WEEK) { + // Align to Monday UTC (Postgres DATE_TRUNC('week') is ISO week starting Monday). + const day = d.getUTCDay(); // 0=Sun … 6=Sat + const diff = day === 0 ? 6 : day - 1; + d.setUTCDate(d.getUTCDate() - diff); + } + + return d; + } + + private addBucket(date: Date, granularity: PartnerStatisticGranularity): Date { + const d = new Date(date); + if (granularity === PartnerStatisticGranularity.DAY) d.setUTCDate(d.getUTCDate() + 1); + else if (granularity === PartnerStatisticGranularity.WEEK) d.setUTCDate(d.getUTCDate() + 7); + else d.setUTCMonth(d.getUTCMonth() + 1); + return d; + } + + private startOfUtcDay(date: Date): Date { + return new Date(Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate())); + } + + private addUtcDays(date: Date, days: number): Date { + const d = new Date(date); + d.setUTCDate(d.getUTCDate() + days); + return d; + } +} diff --git a/src/subdomains/core/statistic/partner-statistic.suppression.ts b/src/subdomains/core/statistic/partner-statistic.suppression.ts new file mode 100644 index 0000000000..d836f25994 --- /dev/null +++ b/src/subdomains/core/statistic/partner-statistic.suppression.ts @@ -0,0 +1,288 @@ +import { PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, PartnerStatisticDirection } from './partner-statistic.enum'; + +/** + * k-anonymity helpers for partner statistics. + * + * A disclosure unit is withheld when its effective count is in 1..k-1. + * The effective count is min(transactions, distinctUsers) when both are known; + * otherwise the transaction count alone. + * + * Additive groups (period totals by direction, payment-info funnel, settlement + * funnel, timeline bucket directions) use block suppression: if any member is + * under the threshold, the entire group (members + derived sums/rates) is + * suppressed. Partial nulling would let totals − visible recover the hidden + * member exactly. + * + * Zero is never suppressed: 0 means “none”, null means “withheld”. + */ + +export function effectiveCount(transactions: number, users?: number): number { + if (users == null) return transactions; + return Math.min(transactions, users); +} + +export function isUnderThreshold( + transactions: number, + users?: number, + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, +): boolean { + const n = effectiveCount(transactions, users); + return n > 0 && n < threshold; +} + +export function suppressScalar( + value: number, + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, + users?: number, +): number | null { + if (value === 0) return 0; + return isUnderThreshold(value, users, threshold) ? null : value; +} + +/** Share numerator/denominator (0..1). Null if either side is suppressed or denominator is 0. */ +export function suppressRate(numerator: number | null, denominator: number | null, decimals = 4): number | null { + if (numerator == null || denominator == null || denominator === 0) return null; + const factor = 10 ** decimals; + return Math.round((numerator / denominator) * factor) / factor; +} + +/** Applies suppressScalar and reports whether the value was withheld (not zero). */ +export function suppressCount( + value: number, + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, + users?: number, +): { value: number | null; suppressed: boolean } { + const next = suppressScalar(value, threshold, users); + return { value: next, suppressed: next === null }; +} + +export interface SuppressibleRow { + transactions: number; + /** Distinct users contributing to this row; when set, min(tx, users) is the gate. */ + users?: number; +} + +/** + * Removes rows with effective count in 1..k-1. If exactly one row is removed, also removes the + * smallest remaining filled row so never fewer than two are suppressed when complementary applies. + */ +export function suppressBreakdownRows( + rows: T[], + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, +): { rows: T[]; suppressedCount: number } { + const visible = rows.filter((r) => !isUnderThreshold(r.transactions, r.users, threshold)); + let suppressedCount = rows.length - visible.length; + + if (suppressedCount === 1 && visible.length > 0) { + const candidates = visible.filter((r) => effectiveCount(r.transactions, r.users) > 0); + if (candidates.length > 0) { + const smallest = candidates.reduce((a, b) => + effectiveCount(a.transactions, a.users) <= effectiveCount(b.transactions, b.users) ? a : b, + ); + visible.splice(visible.indexOf(smallest), 1); + suppressedCount += 1; + } + } + + return { rows: visible, suppressedCount }; +} + +export interface TimelineSuppressible { + transactions: { buy: number; sell: number; swap: number } | null; + volume: { buy: number; sell: number; swap: number } | null; + /** Distinct users per direction (internal; not required on the public DTO). */ + users?: { buy: number; sell: number; swap: number } | null; + suppressed: boolean; +} + +const DIRECTIONS = [ + PartnerStatisticDirection.BUY, + PartnerStatisticDirection.SELL, + PartnerStatisticDirection.SWAP, +] as const; + +/** + * Nullifies volume/transactions when any direction is under the threshold (block rule), + * or when the combined effective total is under k. + * Empty (0-tx) buckets stay visible as zeros — they are not suppressed. + * Complementary: if exactly one real suppression, also suppress the smallest filled visible bucket. + */ +export function suppressTimelineBuckets( + buckets: T[], + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, +): { buckets: T[]; suppressedCount: number } { + const result = buckets.map((b) => { + if (bucketNeedsSuppression(b, threshold)) { + return { ...b, volume: null, transactions: null, users: null, suppressed: true } as T; + } + return { ...b, suppressed: false }; + }); + + let suppressedCount = result.filter((b) => b.suppressed).length; + + if (suppressedCount === 1) { + const visible = result + .map((b, index) => ({ b, index, total: txTotal(b) })) + .filter((x) => !x.b.suppressed && x.total > 0); + + if (visible.length > 0) { + visible.sort((a, b) => a.total - b.total); + const target = visible[0].index; + result[target] = { + ...result[target], + volume: null, + transactions: null, + users: null, + suppressed: true, + } as T; + suppressedCount += 1; + } + } + + return { buckets: result, suppressedCount }; +} + +function bucketNeedsSuppression(b: TimelineSuppressible, threshold: number): boolean { + if (!b.transactions) return false; + const txs = b.transactions; + const users = b.users; + + // Per-direction gate (block rule for the bucket as an additive group). + for (const dir of DIRECTIONS) { + if (isUnderThreshold(txs[dir], users?.[dir], threshold)) return true; + } + + // Overall bucket total: if the whole day is under k when summed, hide it too. + const totalTx = txs.buy + txs.sell + txs.swap; + if (totalTx === 0) return false; + // Distinct users across directions can overlap; without a true UNION we use max as a lower bound + // on the person count only when every direction that has txs also has a user count. + if (users) { + const totalUsers = Math.max(users.buy, users.sell, users.swap); + return isUnderThreshold(totalTx, totalUsers, threshold); + } + return isUnderThreshold(totalTx, undefined, threshold); +} + +export interface DirectionTotals { + buy: number; + sell: number; + swap: number; + total: number; +} + +export interface DirectionUsers { + buy: number; + sell: number; + swap: number; + /** Distinct users across all directions (UNION), used for the total gate. */ + total: number; +} + +export interface SuppressedDirectionTotals { + buy: number | null; + sell: number | null; + swap: number | null; + total: number | null; +} + +/** + * Block-suppresses period totals: if any direction (or the overall total) is under k, + * every field is null — including derived average. No partial nulling of single directions. + */ +export function suppressPeriodTotals( + volume: DirectionTotals, + transactions: DirectionTotals, + users?: DirectionUsers, + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, +): { + volume: SuppressedDirectionTotals; + transactions: SuppressedDirectionTotals; + averageTransactionVolume: number | null; + suppressedCount: number; +} { + const under = (dir: keyof DirectionTotals): boolean => isUnderThreshold(transactions[dir], users?.[dir], threshold); + + if (under('total') || under('buy') || under('sell') || under('swap')) { + return { + volume: { buy: null, sell: null, swap: null, total: null }, + transactions: { buy: null, sell: null, swap: null, total: null }, + averageTransactionVolume: null, + suppressedCount: 1, + }; + } + + const averageTransactionVolume = + volume.total != null && transactions.total > 0 ? volume.total / transactions.total : null; + + return { + volume: { ...volume }, + transactions: { ...transactions }, + averageTransactionVolume, + suppressedCount: 0, + }; +} + +export interface AllTimeVolume { + buy: number; + sell: number; + total: number; +} + +/** + * When tradingUsers is in 1..k-1, all-time volumes are withheld. registeredUsers stays visible + * (installation count, not a per-trade disclosure). + */ +export function suppressAllTimeVolume( + volume: AllTimeVolume, + tradingUsers: number, + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, +): { volume: { buy: number | null; sell: number | null; total: number | null }; suppressedCount: number } { + if (isUnderThreshold(tradingUsers, undefined, threshold)) { + return { + volume: { buy: null, sell: null, total: null }, + suppressedCount: 1, + }; + } + return { volume, suppressedCount: 0 }; +} + +/** + * Block-suppress an additive funnel: if any count is under k, null every member and the rate. + * Returns how many fields were suppressed (0 or memberCount). + */ +export function suppressAdditiveGroup( + counts: Record, + rate?: { numeratorKey: string; denominatorKey: string }, + threshold = PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, +): { values: Record; rate: number | null; suppressedCount: number } { + const keys = Object.keys(counts); + const anyUnder = keys.some((k) => isUnderThreshold(counts[k], undefined, threshold)); + + if (anyUnder) { + const values: Record = {}; + for (const k of keys) values[k] = counts[k] === 0 ? 0 : null; + // Zero stays 0 even under block suppression of the group — but if the group is suppressed + // because another member is under k, zeros remain visible and non-zeros become null. + // Wait: block rule says entire group suppressed. Zeros are still “none” and safe. + // Non-zero under-threshold and non-zero over-threshold both null when any member is under. + for (const k of keys) { + if (counts[k] !== 0) values[k] = null; + } + return { values, rate: null, suppressedCount: keys.filter((k) => counts[k] !== 0).length || 1 }; + } + + const values: Record = {}; + for (const k of keys) values[k] = counts[k]; + + let rateValue: number | null = null; + if (rate) { + rateValue = suppressRate(values[rate.numeratorKey], values[rate.denominatorKey]); + } + return { values, rate: rateValue, suppressedCount: 0 }; +} + +function txTotal(b: TimelineSuppressible): number { + if (b.suppressed || !b.transactions) return 0; + return b.transactions.buy + b.transactions.sell + b.transactions.swap; +} diff --git a/src/subdomains/core/statistic/statistic.module.ts b/src/subdomains/core/statistic/statistic.module.ts index ecbe419f2e..f2022eaf1b 100644 --- a/src/subdomains/core/statistic/statistic.module.ts +++ b/src/subdomains/core/statistic/statistic.module.ts @@ -1,17 +1,33 @@ import { Module } from '@nestjs/common'; import { BitcoinModule } from 'src/integration/blockchain/bitcoin/bitcoin.module'; import { SharedModule } from 'src/shared/shared.module'; +import { BuyCryptoRepository } from 'src/subdomains/core/buy-crypto/process/repositories/buy-crypto.repository'; +import { BuyFiatRepository } from 'src/subdomains/core/sell-crypto/process/buy-fiat.repository'; +import { UserRepository } from 'src/subdomains/generic/user/models/user/user.repository'; +import { WalletRepository } from 'src/subdomains/generic/user/models/wallet/wallet.repository'; import { UserModule } from 'src/subdomains/generic/user/user.module'; +import { TransactionRequestRepository } from 'src/subdomains/supporting/payment/repositories/transaction-request.repository'; import { BuyCryptoModule } from '../buy-crypto/buy-crypto.module'; import { ReferralModule } from '../referral/referral.module'; import { SellCryptoModule } from '../sell-crypto/sell-crypto.module'; +import { PartnerStatisticController } from './partner-statistic.controller'; +import { PartnerStatisticService } from './partner-statistic.service'; import { StatisticController } from './statistic.controller'; import { StatisticService } from './statistic.service'; @Module({ imports: [SharedModule, BuyCryptoModule, SellCryptoModule, ReferralModule, UserModule, BitcoinModule], - controllers: [StatisticController], - providers: [StatisticService], + controllers: [StatisticController, PartnerStatisticController], + providers: [ + StatisticService, + PartnerStatisticService, + // Repositories not re-exported by every parent module — same pattern as FiatOutputModule. + BuyCryptoRepository, + BuyFiatRepository, + UserRepository, + WalletRepository, + TransactionRequestRepository, + ], exports: [], }) export class StatisticModule {}