From 9bf880cf6184a531fe20158c37bd7b983a79d1f9 Mon Sep 17 00:00:00 2001 From: joshuakrueger-dfx Date: Thu, 30 Jul 2026 16:34:10 +0200 Subject: [PATCH 1/2] Add aggregated partner statistics endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wallet partners integrating DFX have no way to see how their integration performs. The only partner-facing transaction endpoint returns per-transaction detail including customer IBANs, which is both more than a usage dashboard needs and more than a partner should hold. Add GET /statistic/partner and /statistic/partner/timeline, scoped to the wallet in the company JWT. There is deliberately no walletId parameter, so a partner cannot express a request for someone else's data. Both endpoints return aggregates only: volume and transaction counts by direction, active and new users, breakdowns by asset, fiat currency, blockchain and payment method, and the referral position in EUR. Aggregates alone are not anonymous. Counts below a threshold of five are suppressed; the effective count is min(transactions, distinct users). Additive groups (period totals by direction, timeline buckets) use block suppression — if any member is under the threshold the whole group is null — so a hidden value cannot be recovered via totals − visible. Period bounds snap to UTC day boundaries with a half-open interval and a one-day minimum. Referral open credit uses refCredit + partnerRefCredit − paidRefCredit on the wallet owner account; tradingUsers and referral balances are gated by the same threshold. Rate limits for these routes count by wallet id, not IP prefix. --- .env.example | 4 + ...partner-statistic-rate-limit.guard.spec.ts | 25 + .../partner-statistic.integration.spec.ts | 292 +++++++ .../partner-statistic.service.spec.ts | 792 +++++++++++++++++ .../partner-statistic.suppression.spec.ts | 327 +++++++ .../statistic/dto/partner-statistic.dto.ts | 287 +++++++ .../partner-statistic-rate-limit.guard.ts | 25 + .../statistic/partner-statistic.controller.ts | 73 ++ .../core/statistic/partner-statistic.enum.ts | 42 + .../statistic/partner-statistic.service.ts | 800 ++++++++++++++++++ .../partner-statistic.suppression.ts | 288 +++++++ .../core/statistic/statistic.module.ts | 18 +- 12 files changed, 2971 insertions(+), 2 deletions(-) create mode 100644 src/subdomains/core/statistic/__tests__/partner-statistic-rate-limit.guard.spec.ts create mode 100644 src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts create mode 100644 src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts create mode 100644 src/subdomains/core/statistic/__tests__/partner-statistic.suppression.spec.ts create mode 100644 src/subdomains/core/statistic/dto/partner-statistic.dto.ts create mode 100644 src/subdomains/core/statistic/partner-statistic-rate-limit.guard.ts create mode 100644 src/subdomains/core/statistic/partner-statistic.controller.ts create mode 100644 src/subdomains/core/statistic/partner-statistic.enum.ts create mode 100644 src/subdomains/core/statistic/partner-statistic.service.ts create mode 100644 src/subdomains/core/statistic/partner-statistic.suppression.ts 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..10510130e7 --- /dev/null +++ b/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts @@ -0,0 +1,292 @@ +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); + }); + + 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..ecce54fce7 --- /dev/null +++ b/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts @@ -0,0 +1,792 @@ +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 { + 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 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 }; + 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 }, + 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; + 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 (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')) } }, + ], + }).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], + }), + ); + + 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', + ]), + ); + }); + }); + + // --- 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], + 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], + 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(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); + }); + + 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); + }); + }); + + // --- 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..86564c31cb --- /dev/null +++ b/src/subdomains/core/statistic/dto/partner-statistic.dto.ts @@ -0,0 +1,287 @@ +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'; +} + +// --- 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: 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..6d2c57adfa --- /dev/null +++ b/src/subdomains/core/statistic/partner-statistic.service.ts @@ -0,0 +1,800 @@ +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 { TransactionSourceType } from 'src/subdomains/supporting/payment/entities/transaction.entity'; +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, + PartnerNamedBreakdownDto, + PartnerReferralDto, + 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 { + 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, + ) {} + + // --- 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, + ] = 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), + ]); + + 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 + + (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, + 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; + } + + // --- 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..359ed5fcbb 100644 --- a/src/subdomains/core/statistic/statistic.module.ts +++ b/src/subdomains/core/statistic/statistic.module.ts @@ -1,17 +1,31 @@ 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 { 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, + ], exports: [], }) export class StatisticModule {} From 7b81b9b7e377a9a7d10b3745e49d07193a51653c Mon Sep 17 00:00:00 2001 From: joshuakrueger-dfx Date: Thu, 30 Jul 2026 22:49:55 +0200 Subject: [PATCH 2/2] Add partner statistic completion funnels MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add stage A (payment-info request → payment received) and stage B (settlement received → delivered / rejected / in progress) to the partner summary. Funnel members form additive groups and use the same block suppression as period totals: if any counter is under the threshold the whole direction funnel and its rate are null, so rejected = received − delivered − inProgress cannot be reconstructed from visible fields. --- .../partner-statistic.integration.spec.ts | 8 +- .../partner-statistic.service.spec.ts | 138 +++++++++++++ .../statistic/dto/partner-statistic.dto.ts | 133 +++++++++++++ .../statistic/partner-statistic.service.ts | 181 ++++++++++++++++++ .../core/statistic/statistic.module.ts | 2 + 5 files changed, 461 insertions(+), 1 deletion(-) diff --git a/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts b/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts index 10510130e7..b8dfc10e39 100644 --- a/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts +++ b/src/subdomains/core/statistic/__tests__/partner-statistic.integration.spec.ts @@ -113,7 +113,13 @@ describeDb('PartnerStatisticService SQL path (real Postgres)', () => { `); // 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); + service = new PartnerStatisticService( + { manager: dataSource.manager } as any, + {} as any, + {} as any, + {} as any, + {} as any, + ); }); afterEach(async () => { diff --git a/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts b/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts index ecce54fce7..18c0aed1ef 100644 --- a/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts +++ b/src/subdomains/core/statistic/__tests__/partner-statistic.service.spec.ts @@ -6,6 +6,11 @@ import { BuyCryptoRepository } from 'src/subdomains/core/buy-crypto/process/repo 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, @@ -35,6 +40,12 @@ interface SettlementFixture { rejected: number; } +interface PaymentInfoRow { + type: TransactionRequestType; + status: TransactionRequestStatus; + count: number; +} + interface DirectionFixture { volume: number; transactions: number; @@ -49,6 +60,7 @@ interface WalletFixture { 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 }[]; @@ -68,6 +80,7 @@ function emptyFixture(overrides: Partial = {}): WalletFixture { 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(), @@ -148,6 +161,7 @@ describe('PartnerStatisticService', () => { 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, @@ -293,6 +307,10 @@ describe('PartnerStatisticService', () => { 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) => ({ @@ -366,6 +384,10 @@ describe('PartnerStatisticService', () => { { 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(); @@ -445,6 +467,7 @@ describe('PartnerStatisticService', () => { 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 }], }), ); @@ -467,6 +490,8 @@ describe('PartnerStatisticService', () => { 'inputAsset.blockchain', 'outputAsset.name', 'outputAsset.blockchain', + 'tr.type', + 'tr.status', ]), ); }); @@ -486,6 +511,7 @@ describe('PartnerStatisticService', () => { 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 }, @@ -504,6 +530,7 @@ describe('PartnerStatisticService', () => { 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(), @@ -527,6 +554,8 @@ describe('PartnerStatisticService', () => { 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); @@ -540,6 +569,11 @@ describe('PartnerStatisticService', () => { 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 () => { @@ -554,6 +588,110 @@ describe('PartnerStatisticService', () => { 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); }); }); diff --git a/src/subdomains/core/statistic/dto/partner-statistic.dto.ts b/src/subdomains/core/statistic/dto/partner-statistic.dto.ts index 86564c31cb..b9a9c842e3 100644 --- a/src/subdomains/core/statistic/dto/partner-statistic.dto.ts +++ b/src/subdomains/core/statistic/dto/partner-statistic.dto.ts @@ -202,6 +202,136 @@ export class PartnerReferralDto { 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 { @@ -223,6 +353,9 @@ export class PartnerStatisticDto { @ApiProperty({ type: PartnerReferralDto }) referral: PartnerReferralDto; + @ApiProperty({ type: PartnerCompletionDto }) + completion: PartnerCompletionDto; + @ApiProperty({ type: PartnerStatisticMetaDto }) meta: PartnerStatisticMetaDto; } diff --git a/src/subdomains/core/statistic/partner-statistic.service.ts b/src/subdomains/core/statistic/partner-statistic.service.ts index 6d2c57adfa..05d723ba00 100644 --- a/src/subdomains/core/statistic/partner-statistic.service.ts +++ b/src/subdomains/core/statistic/partner-statistic.service.ts @@ -6,14 +6,22 @@ import { BuyCryptoRepository } from 'src/subdomains/core/buy-crypto/process/repo 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, @@ -28,6 +36,7 @@ import { PartnerStatisticGranularity, } from './partner-statistic.enum'; import { + suppressAdditiveGroup, suppressAllTimeVolume, suppressBreakdownRows, suppressPeriodTotals, @@ -73,6 +82,7 @@ export class PartnerStatisticService { private readonly buyFiatRepo: BuyFiatRepository, private readonly userRepo: UserRepository, private readonly walletRepo: WalletRepository, + private readonly txRequestRepo: TransactionRequestRepository, ) {} // --- PUBLIC API --- // @@ -92,6 +102,7 @@ export class PartnerStatisticService { 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), @@ -104,6 +115,7 @@ export class PartnerStatisticService { () => 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 = { @@ -150,6 +162,7 @@ export class PartnerStatisticService { paymentMethods.suppressedCount + totalsSuppressed.suppressedCount + allTimeSuppressed.suppressedCount + + completionResult.suppressedCount + (activeUsers === null ? 1 : 0) + (newUsers === null ? 1 : 0) + (tradingUsersSuppressed === null ? 1 : 0) + @@ -177,6 +190,7 @@ export class PartnerStatisticService { paymentMethods: paymentMethods.rows.map(({ users: _u, ...row }) => row), }, referral, + completion: completionResult.completion, meta: { suppressionThreshold: PARTNER_STATISTIC_SUPPRESSION_THRESHOLD, suppressedBuckets, @@ -640,6 +654,173 @@ export class PartnerStatisticService { 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 --- // /** diff --git a/src/subdomains/core/statistic/statistic.module.ts b/src/subdomains/core/statistic/statistic.module.ts index 359ed5fcbb..f2022eaf1b 100644 --- a/src/subdomains/core/statistic/statistic.module.ts +++ b/src/subdomains/core/statistic/statistic.module.ts @@ -6,6 +6,7 @@ import { BuyFiatRepository } from 'src/subdomains/core/sell-crypto/process/buy-f 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'; @@ -25,6 +26,7 @@ import { StatisticService } from './statistic.service'; BuyFiatRepository, UserRepository, WalletRepository, + TransactionRequestRepository, ], exports: [], })