diff --git a/app/backend/prisma/migrations/20260718000000_add_dashboard_composite_indexes/migration.sql b/app/backend/prisma/migrations/20260718000000_add_dashboard_composite_indexes/migration.sql new file mode 100644 index 00000000..8760e2a9 --- /dev/null +++ b/app/backend/prisma/migrations/20260718000000_add_dashboard_composite_indexes/migration.sql @@ -0,0 +1,7 @@ +-- Add ordered composite indexes for dashboard list and summary queries. + +-- Campaign dashboard/export filters by org and status, then orders/paginates by createdAt. +CREATE INDEX "Campaign_orgId_status_createdAt_idx" ON "Campaign"("orgId", "status", "createdAt"); + +-- Aid package dashboard filters packages within a campaign by status, then orders by createdAt. +CREATE INDEX "AidPackage_campaignId_status_createdAt_idx" ON "AidPackage"("campaignId", "status", "createdAt"); diff --git a/app/backend/prisma/schema.prisma b/app/backend/prisma/schema.prisma index 4e3567e5..1f76d0ce 100644 --- a/app/backend/prisma/schema.prisma +++ b/app/backend/prisma/schema.prisma @@ -39,7 +39,8 @@ model AidPackage { remainingAmount Float @default(0) @@index([campaignId]) - @@index([campaignId, status]) + @@index([campaignId, status]) + @@index([campaignId, status, createdAt]) } /// Tracks every balance event (lock / unlock / disburse) for a campaign. @@ -243,7 +244,7 @@ model Claim { @@index([status]) @@index([campaignId]) - @@index([campaignId, status]) + @@index([campaignId, status]) @@index([createdAt]) @@index([deletedAt]) @@index([reissuedFromId]) @@ -369,6 +370,7 @@ model Campaign { @@index([archivedAt]) @@index([ngoId]) @@index([orgId]) + @@index([orgId, status, createdAt]) @@index([deletedAt]) } diff --git a/app/backend/src/common/index.ts b/app/backend/src/common/index.ts index b6fd0fa9..8a094cf2 100644 --- a/app/backend/src/common/index.ts +++ b/app/backend/src/common/index.ts @@ -13,5 +13,8 @@ export * from './decorators/deprecated.decorator'; // Interceptors export * from './interceptors/deprecation.interceptor'; +// Streaming +export * from './streaming'; + // Budget export * from './budget/budget.service'; diff --git a/app/backend/src/common/interceptors/__tests__/http-cache.interceptor.spec.ts b/app/backend/src/common/interceptors/__tests__/http-cache.interceptor.spec.ts index 184d8e2c..e7bd1cab 100644 --- a/app/backend/src/common/interceptors/__tests__/http-cache.interceptor.spec.ts +++ b/app/backend/src/common/interceptors/__tests__/http-cache.interceptor.spec.ts @@ -8,6 +8,7 @@ import { HTTP_CACHE_METADATA, HTTP_CACHE_SKIP, } from '../../decorators/http-cache.decorator'; +import { HTTP_STREAMING_CACHE } from '../../streaming'; interface FakeResponse { headers: Record; @@ -46,6 +47,8 @@ const decorated = (key: string, value: unknown) => { return fn; }; +const nextTick = () => new Promise(resolve => setImmediate(resolve)); + const createContext = ({ method = 'GET', path = '/api/v1/campaigns', @@ -203,9 +206,7 @@ describe('HttpCacheInterceptor', () => { expect(response.getHeader('Cache-Control')).toBe( 'private, must-revalidate', ); - expect(response.getHeader('Vary')).toBe( - 'Authorization, Accept-Encoding', - ); + expect(response.getHeader('Vary')).toBe('Authorization, Accept-Encoding'); expect(response.getHeader('ETag')).toMatch(/^"[a-f0-9]{64}"$/); }); @@ -242,9 +243,7 @@ describe('HttpCacheInterceptor', () => { expect(response.getHeader('Cache-Control')).toBe( 'private, must-revalidate', ); - expect(response.getHeader('Vary')).toBe( - 'Authorization, Accept-Encoding', - ); + expect(response.getHeader('Vary')).toBe('Authorization, Accept-Encoding'); expect(response.getHeader('ETag')).toMatch(/^"[a-f0-9]{64}"$/); expect(response.getHeader('X-Http-Cache')).toBe('miss'); }); @@ -355,7 +354,10 @@ describe('HttpCacheInterceptor', () => { }), ); expect(result).toBeUndefined(); - return { status: next.response.statusCode, etag: next.response.getHeader('ETag') }; + return { + status: next.response.statusCode, + etag: next.response.getHeader('ETag'), + }; }; it('returns 304 on exact strong ETag match', async () => { @@ -485,5 +487,101 @@ describe('HttpCacheInterceptor', () => { ); } }); + + it('defers ETag hashing for @UseStreamingCache responses', async () => { + configGet.mockReturnValue(undefined); + const interceptor = buildInterceptor(); + const handler = decorated(HTTP_STREAMING_CACHE, true); + const payload = { + rows: Array.from({ length: 4_000 }, (_, id) => ({ + id, + value: `row-${id}`, + })), + }; + const { context, response } = createContext({ handler }); + + const result = await firstValueFrom( + interceptor.intercept(context, { handle: () => of(payload) }), + ); + + expect(result).toBe(payload); + expect(response.getHeader('ETag')).toBe('W/"pending"'); + expect(response.getHeader('Link')).toBe( + '; rel=etag; status=pending', + ); + expect(response.getHeader('X-Http-Cache')).toBe('pending'); + + await nextTick(); + + expect(response.getHeader('ETag')).toMatch(/^"[a-f0-9]{64}"$/); + expect(response.getHeader('Link')).toMatch( + /^<\/etag>; rel=etag; etag="[a-f0-9]{64}"$/, + ); + expect(response.getHeader('X-Http-Cache')).toBe('miss'); + }); + + it('computes identical deferred ETags for identical streaming-cache bodies', async () => { + configGet.mockReturnValue(undefined); + const interceptor = buildInterceptor(); + const handler = decorated(HTTP_STREAMING_CACHE, true); + + const firstCtx = createContext({ handler }); + await firstValueFrom( + interceptor.intercept(firstCtx.context, { + handle: () => of({ z: 1, a: 2, m: { y: 1, x: 2 } }), + }), + ); + + const secondCtx = createContext({ handler, path: '/api/v1/y' }); + await firstValueFrom( + interceptor.intercept(secondCtx.context, { + handle: () => of({ a: 2, m: { x: 2, y: 1 }, z: 1 }), + }), + ); + + await nextTick(); + + expect(firstCtx.response.getHeader('ETag')).toBeTruthy(); + expect(firstCtx.response.getHeader('ETag')).toBe( + secondCtx.response.getHeader('ETag'), + ); + }); + + it('completes a 200 KB streaming-cache JSON response under the latency budget', async () => { + configGet.mockReturnValue(undefined); + const interceptor = buildInterceptor(); + const handler = decorated(HTTP_STREAMING_CACHE, true); + const payload = { + rows: Array.from({ length: 2_500 }, (_, id) => ({ + id, + label: `recipient-${id}`, + status: 'pending', + notes: 'x'.repeat(48), + })), + }; + expect(Buffer.byteLength(JSON.stringify(payload))).toBeGreaterThan( + 200 * 1024, + ); + + const timings: number[] = []; + for (let i = 0; i < 100; i += 1) { + const { context } = createContext({ + handler, + path: `/api/v1/large-${i}`, + }); + const started = process.hrtime.bigint(); + await firstValueFrom( + interceptor.intercept(context, { handle: () => of(payload) }), + ); + const elapsedMs = Number(process.hrtime.bigint() - started) / 1_000_000; + timings.push(elapsedMs); + } + + timings.sort((a, b) => a - b); + const p99 = timings[98]; + expect(p99).toBeLessThan(8); + + await nextTick(); + }); }); }); diff --git a/app/backend/src/common/interceptors/http-cache.interceptor.ts b/app/backend/src/common/interceptors/http-cache.interceptor.ts index 83d473d2..18259028 100644 --- a/app/backend/src/common/interceptors/http-cache.interceptor.ts +++ b/app/backend/src/common/interceptors/http-cache.interceptor.ts @@ -4,6 +4,7 @@ import { ExecutionContext, CallHandler, Logger, + StreamableFile, } from '@nestjs/common'; import { Reflector } from '@nestjs/core'; import { ConfigService } from '@nestjs/config'; @@ -11,6 +12,7 @@ import { Observable } from 'rxjs'; import { catchError, map } from 'rxjs/operators'; import { throwError } from 'rxjs'; import { createHash } from 'node:crypto'; +import { Readable } from 'node:stream'; import type { Request, Response } from 'express'; import { canonicalStringify } from '../utils/json-canonicalize.util'; import { @@ -18,6 +20,7 @@ import { HTTP_CACHE_SKIP, HttpCacheOptions, } from '../decorators/http-cache.decorator'; +import { HTTP_STREAMING_CACHE } from '../streaming'; /** * Maximum size (bytes) for which we compute an ETag. Larger payloads @@ -44,6 +47,8 @@ const SKIP_PATH_PREFIXES: readonly string[] = [ * Cache-Control / ETag surface as the original GET. */ const SAFE_METHODS = new Set(['GET', 'HEAD']); +const PENDING_ETAG = 'W/"pending"'; +const ETAG_LINK_TARGET = '; rel=etag'; /** * HTTP methods whose responses must never be persisted by any cache @@ -105,9 +110,10 @@ export class HttpCacheInterceptor implements NestInterceptor { private readonly reflector: Reflector, configService: ConfigService, ) { - this.enabled = (configService.get('HTTP_CACHE_ENABLED') ?? 'true') - .toLowerCase() - !== 'false'; + this.enabled = + ( + configService.get('HTTP_CACHE_ENABLED') ?? 'true' + ).toLowerCase() !== 'false'; const ttlRaw = configService.get('HTTP_CACHE_DEFAULT_TTL'); const parsedTtl = ttlRaw !== undefined ? Number.parseInt(ttlRaw, 10) : NaN; @@ -127,8 +133,7 @@ export class HttpCacheInterceptor implements NestInterceptor { : MAX_ETAG_PAYLOAD_BYTES; this.debugHeaders = - (configService.get('NODE_ENV') ?? 'development') !== - 'production'; + (configService.get('NODE_ENV') ?? 'development') !== 'production'; } intercept(context: ExecutionContext, next: CallHandler): Observable { @@ -175,6 +180,11 @@ export class HttpCacheInterceptor implements NestInterceptor { context.getHandler(), context.getClass(), ]) ?? {}; + const useStreamingCache = + this.reflector.getAllAndOverride(HTTP_STREAMING_CACHE, [ + context.getHandler(), + context.getClass(), + ]) === true; response.setHeader('Cache-Control', this.buildCacheControl(options)); response.setHeader('Vary', 'Authorization, Accept-Encoding'); @@ -188,7 +198,11 @@ export class HttpCacheInterceptor implements NestInterceptor { this.markNonCacheable(response); return throwError(() => err); }), - map((data) => this.applyGetHeaders(request, response, data)), + map(data => + useStreamingCache + ? this.applyStreamingGetHeaders(request, response, data) + : this.applyGetHeaders(request, response, data), + ), ); } @@ -222,10 +236,7 @@ export class HttpCacheInterceptor implements NestInterceptor { // Don't bother hashing huge payloads; ETag is a "free" win only on // small responses that we expect to be re-served often. const serialized = canonicalStringify(data); - if ( - serialized.length === 0 || - serialized.length > this.maxPayloadBytes - ) { + if (serialized.length === 0 || serialized.length > this.maxPayloadBytes) { if (serialized.length > this.maxPayloadBytes) { this.logger.warn( `Skipping ETag hash for ${request.method} ${ @@ -264,6 +275,100 @@ export class HttpCacheInterceptor implements NestInterceptor { return data; } + private applyStreamingGetHeaders( + request: Request, + response: Response, + data: unknown, + ): unknown { + if (!this.isCacheableBody(data, response)) { + this.setDebugHeader(response, 'bypass'); + return data; + } + + response.setHeader('ETag', PENDING_ETAG); + response.setHeader('Link', `${ETAG_LINK_TARGET}; status=pending`); + response.setHeader('Trailer', 'ETag, Link'); + this.setDebugHeader(response, 'pending'); + + this.deferEtagHash(request, response, data); + return this.streamJsonResponse(response, data); + } + + private deferEtagHash( + request: Request, + response: Response, + data: unknown, + ): void { + setImmediate(() => { + try { + const serialized = canonicalStringify(data); + if (serialized.length === 0) { + this.setDebugHeader(response, 'bypass'); + return; + } + + const etagHash = createHash('sha256').update(serialized).digest('hex'); + const etag = `"${etagHash}"`; + const link = `${ETAG_LINK_TARGET}; etag=${etag}`; + + if (!response.headersSent) { + response.setHeader('ETag', etag); + response.setHeader('Link', link); + } + this.addTrailers(response, { ETag: etag, Link: link }); + + if (this.ifNoneMatchMatches(request, etag)) { + this.setDebugHeader(response, 'hit'); + } else { + this.setDebugHeader(response, 'miss'); + } + } catch (err) { + this.logger.warn( + `Deferred ETag hash failed for ${request.method} ${ + request.originalUrl ?? request.url + }: ${err instanceof Error ? err.message : String(err)}`, + ); + this.setDebugHeader(response, 'bypass'); + } + }); + } + + private streamJsonResponse(response: Response, data: unknown): unknown { + if (!supportsStreamingResponse(response)) { + return data; + } + + let payload: string; + try { + payload = JSON.stringify(data) ?? 'null'; + } catch { + return data; + } + + const chunkSize = 16 * 1024; + function* chunks(): Generator { + for (let offset = 0; offset < payload.length; offset += chunkSize) { + yield payload.slice(offset, offset + chunkSize); + } + } + + return new StreamableFile(Readable.from(chunks()), { + type: 'application/json; charset=utf-8', + }); + } + + private addTrailers( + response: Response, + trailers: Record, + ): void { + const candidate = response as Response & { + addTrailers?: (headers: Record) => void; + }; + if (typeof candidate.addTrailers === 'function') { + candidate.addTrailers(trailers); + } + } + private markNonCacheable(response: Response): void { response.setHeader('Cache-Control', 'no-store'); response.setHeader('Pragma', 'no-cache'); @@ -310,12 +415,19 @@ export class HttpCacheInterceptor implements NestInterceptor { // Node Readables expose `.pipe`; Web streams expose `.pipeTo`. const candidate = data as { pipe?: unknown; pipeTo?: unknown }; - if (typeof candidate.pipe === 'function' || typeof candidate.pipeTo === 'function') { + if ( + typeof candidate.pipe === 'function' || + typeof candidate.pipeTo === 'function' + ) { return false; } // Web Readables expose an async iterator (ReadableStream / Readable.fromWeb). - if (typeof (data as { [Symbol.asyncIterator]?: unknown })[Symbol.asyncIterator] === 'function') { + if ( + typeof (data as { [Symbol.asyncIterator]?: unknown })[ + Symbol.asyncIterator + ] === 'function' + ) { return false; } @@ -348,10 +460,7 @@ export class HttpCacheInterceptor implements NestInterceptor { return true; } - private ifNoneMatchMatches( - request: Request, - etag: string, - ): boolean { + private ifNoneMatchMatches(request: Request, etag: string): boolean { const raw = request.headers['if-none-match']; if (raw === undefined) return false; const values = normalizeIfNoneMatch(raw); @@ -397,3 +506,11 @@ function normalizeIfNoneMatch(raw: string | string[]): string[] { function stripWeakPrefix(value: string): string { return value.startsWith('W/') ? value.slice(2) : value; } + +function supportsStreamingResponse(response: Response): boolean { + const candidate = response as Response & { + write?: unknown; + }; + + return typeof candidate.write === 'function'; +} diff --git a/app/backend/src/common/streaming/index.ts b/app/backend/src/common/streaming/index.ts new file mode 100644 index 00000000..a265bd8d --- /dev/null +++ b/app/backend/src/common/streaming/index.ts @@ -0,0 +1 @@ +export * from './streaming-cache.decorator'; diff --git a/app/backend/src/common/streaming/streaming-cache.decorator.ts b/app/backend/src/common/streaming/streaming-cache.decorator.ts new file mode 100644 index 00000000..16f5cbfe --- /dev/null +++ b/app/backend/src/common/streaming/streaming-cache.decorator.ts @@ -0,0 +1,12 @@ +import { SetMetadata } from '@nestjs/common'; + +export const HTTP_STREAMING_CACHE = 'http_cache:streaming'; + +/** + * Opt in to deferred ETag computation for large JSON GET / HEAD responses. + * + * The cache interceptor emits `ETag: W/"pending"` immediately and schedules + * canonical hashing after the response path has yielded. + */ +export const UseStreamingCache = (): MethodDecorator & ClassDecorator => + SetMetadata(HTTP_STREAMING_CACHE, true); diff --git a/docs/db/indexes.md b/docs/db/indexes.md new file mode 100644 index 00000000..17542296 --- /dev/null +++ b/docs/db/indexes.md @@ -0,0 +1,50 @@ +# Database Indexes + +This document records non-obvious composite indexes and the query shapes they +are intended to support. + +## Dashboard Composites + +The dashboard and export paths filter by tenant/campaign and status before +ordering or constraining by creation time. Single-column indexes on `orgId`, +`campaignId`, `status`, or `createdAt` do not give PostgreSQL the same ordered +access path as a composite index with equality filters first and the ordered +column last. + +| Model | Index | Query shape | +| ------------ | ------------------------------------------ | ----------------------------------------------------------------------------------------------- | +| `Claim` | `@@index([campaignId, status])` | Claims within a campaign filtered by status. | +| `Campaign` | `@@index([orgId, status, createdAt])` | Organization dashboard/export queries filtered by `orgId` and `status`, ordered by `createdAt`. | +| `AidPackage` | `@@index([campaignId, status, createdAt])` | Package dashboard queries filtered by `campaignId` and `status`, ordered by `createdAt`. | + +## CI EXPLAIN Check + +For PostgreSQL-backed CI environments, verify the dashboard plan with +`EXPLAIN ANALYZE` after migrations are applied. The expected plan should use an +index scan or bitmap index scan over the matching composite index, not a +sequential scan. + +```sql +EXPLAIN ANALYZE +SELECT id, name, status, "createdAt" +FROM "Campaign" +WHERE "orgId" = 'org_test' + AND status = 'active' +ORDER BY "createdAt" DESC +LIMIT 50; + +EXPLAIN ANALYZE +SELECT id, status, "createdAt" +FROM "AidPackage" +WHERE "campaignId" = 'campaign_test' + AND status = 'active' +ORDER BY "createdAt" DESC +LIMIT 50; + +EXPLAIN ANALYZE +SELECT id, status +FROM "Claim" +WHERE "campaignId" = 'campaign_test' + AND status = 'approved' +LIMIT 50; +```