feat: data persistence module with transactions, migrations and audit…

.env.example

@@ -10,6 +10,12 @@ DB_PASSWORD=postgres
 DB_NAME=interchangabletrade
 DB_SYNCHRONIZE=true
 DB_LOGGING=false
+# Connection pooling
+DB_POOL_MAX=20
+DB_POOL_MIN=2
+DB_POOL_IDLE_TIMEOUT_MS=30000
+# Versioned migrations (see docs/data-persistence.md)
+DB_MIGRATIONS_RUN=false
 
 # Redis
 REDIS_HOST=localhost

docs/data-persistence.md

@@ -0,0 +1,113 @@
+# Data Persistence & Database Module
+
+How InterChangableTrade-Core persists data: connection management, transactions,
+repositories, migrations, auditing, and operational guarantees.
+
+## Components
+
+| Piece | Location | Purpose |
+| --- | --- | --- |
+| Connection & pooling | `src/config/database.config.ts` | PostgreSQL options built from validated env config (`DB_POOL_MAX`, `DB_POOL_MIN`, idle timeout). |
+| ACID transaction helper | `libs/common/src/database/transaction.helper.ts` | `TransactionHelper.run(work, isolation?)` wraps `DataSource.transaction`; everything commits or rolls back together. |
+| Base repository | `libs/common/src/database/base.repository.ts` | `BaseRepository<T>.findPaginated()` returns the shared `PaginatedResultDto` envelope with clamped page/limit. |
+| Versioned migrations | `src/database/migrations/` + `src/database/data-source.ts` | TypeORM CLI migrations; run via `npm run migration:run|revert|generate`. `DB_MIGRATIONS_RUN=true` applies pending migrations on boot. |
+| Audit trail entity | `src/database/entities/audit-trail.entity.ts` | Append-only row-level change capture (`actorId`, `action`, `before`, `after`) with CHECK constraint on `action`. |
+| Change-capture subscriber | `src/database/audit-trail.subscriber.ts` | Global TypeORM subscriber writing trail rows using the event's transaction manager; skips itself and swallows its own failures. |
+| Actor binding | `src/database/audit-context.ts`, `src/database/actor-context.interceptor.ts` | `AsyncLocalStorage` carries the request user to the subscriber without signature changes. |
+
+The activity/event log in `src/modules/audit` (who called which endpoint) is
+complementary to the row-level trail here (which row changed and how).
+
+## Entity relationship overview
+
+```mermaid
+erDiagram
+    USER ||--o{ WALLET : owns
+    USER ||--o{ AUDIT_LOG : "acted in"
+    USER ||--o{ AUDIT_TRAIL : "changed rows"
+    AUDIT_TRAIL }o--|| ENTITY : "snapshots"
+
+    USER {
+        uuid id PK
+        string email UK
+        string passwordHash
+        timestamptz createdAt
+    }
+    WALLET {
+        uuid id PK
+        uuid userId FK
+        numeric balance
+        string assetCode
+        timestamptz updatedAt
+    }
+    AUDIT_LOG {
+        uuid id PK
+        uuid userId
+        enum category
+        string action
+        jsonb metadata
+        timestamptz createdAt
+    }
+    AUDIT_TRAIL {
+        uuid id PK
+        string actorId
+        string action "CHECK insert|update|delete"
+        string entityType
+        string entityId
+        jsonb before
+        jsonb after
+        timestamptz createdAt
+    }
+```
+
+## Indexing strategy
+
+- **Composite indexes follow query patterns**: `(entityType, entityId, createdAt)`
+  serves "history of this row", `(actorId, createdAt)` serves "what did this user
+  change". Column order matches equality-first, range-last.
+- **Hot filters get indexes** (`action`, `userId`) but every index slows writes;
+  audit tables are write-heavy, so avoid speculative single-column indexes on
+  low-cardinality columns.
+- **JSONB snapshots are not indexed** by default; if diffing becomes a workload,
+  add GIN indexes on specific keys rather than whole documents.
+- Verify with `EXPLAIN (ANALYZE, BUFFERS)` before adding; monitor
+  `pg_stat_user_indexes` for unused indexes.
+
+## Backup & disaster recovery
+
+Targets: **RPO < 5 minutes**, **RTO < 15 minutes**.
+
+1. **Continuous WAL archiving** (PostgreSQL `archive_command` or a managed
+   equivalent such as Cloud SQL PITR / RDS automated backups) gives a
+   point-in-time recovery window measured in seconds-to-minutes → satisfies RPO.
+2. **Nightly base backups** (`pgBackRest` full/incremental) retained 30 days,
+   plus weekly restore drills into staging.
+3. **Recovery runbook**: promote latest base backup + replay WAL to just before
+   failure (≈5–10 min for a database of our size class); DNS/connection-string
+   cutover to the restored instance (≈2 min); app pods reconnect via pooled
+   retry logic already present in TypeORM. Total stays under the 15-minute RTO.
+4. **Audit data durability**: `audit_trails` is append-only and small per row;
+   it is included in the same backup chain. For stricter guarantees, stream
+   inserts to an append-only sink (object storage / second region) as well.
+5. Quarterly game-day: failover to warm standby replica; replication lag alarm
+   at >60s protects the RPO budget.
+
+## Sharding strategy
+
+Current scale does not require sharding; the plan below is the agreed path when
+write volume demands it:
+
+- **Tenant/region-first**: partition by `userId` hash at the application layer
+  (a thin datasource router keyed off the JWT claim). Most queries are
+  user-scoped (wallets, trades), so they hit exactly one shard.
+- **Shard key choice**: `userId`, never auto-increment ids or timestamps —
+  avoids hot shards and keeps cross-shard joins rare.
+- **Cross-shard operations**: escrow/trading flows that must touch two users'
+  rows use saga-style compensation instead of 2PC; each leg is a local ACID
+  transaction via `TransactionHelper`.
+- **Audit trails**: shard by `entityId` hash; they are written where the change
+  happened and read per-entity, so co-location preserves the access pattern.
+- **Migration path**: start with PostgreSQL declarative partitioning
+  (`PARTITION BY HASH (userId)`) on the largest tables (trades, ledger entries)
+  — it delivers most of the benefit with no application changes — before moving
+  to true multi-instance sharding.

libs/common/src/database/base.repository.ts

@@ -0,0 +1,34 @@
+import { FindOptionsOrder, FindOptionsWhere, Repository } from 'typeorm';
+import { BaseEntity } from '../entities/base.entity';
+import { PaginatedResultDto } from '../dto/paginated-result.dto';
+import { PaginationQueryDto } from '../dto/pagination-query.dto';
+
+/**
+ * Base class for feature repositories. Adds opinionated pagination on top of
+ * the stock TypeORM repository so every list endpoint shares the same
+ * envelope ({@link PaginatedResultDto}) and the same 1-based page semantics
+ * as {@link PaginationQueryDto}.
+ */
+export class BaseRepository<T extends BaseEntity> extends Repository<T> {
+  /**
+   * Returns one page of rows plus pagination metadata. Page is 1-based;
+   * limit defaults to 20 and is clamped to [1, 100] to match the shared DTO.
+   */
+  async findPaginated(
+    query: Partial<Pick<PaginationQueryDto, 'page' | 'limit'>> = {},
+    where?: FindOptionsWhere<T> | FindOptionsWhere<T>[],
+    order?: FindOptionsOrder<T>,
+  ): Promise<PaginatedResultDto<T>> {
+    const page = Math.max(1, query.page ?? 1);
+    const limit = Math.min(Math.max(1, query.limit ?? 20), 100);
+
+    const [items, total] = await this.findAndCount({
+      skip: (page - 1) * limit,
+      take: limit,
+      where,
+      order: order ?? ({ createdAt: 'DESC' } as FindOptionsOrder<T>),
+    });
+
+    return new PaginatedResultDto(items, total, page, limit);
+  }
+}

libs/common/src/database/transaction.helper.ts

@@ -0,0 +1,55 @@
+import { Injectable } from '@nestjs/common';
+import {
+  DataSource,
+  EntityManager,
+} from 'typeorm';
+
+/**
+ * Isolation levels accepted by {@link DataSource.transaction} ("READ
+ * UNCOMMITTED", "READ COMMITTED", "REPEATABLE READ", "SERIALIZABLE").
+ */
+export type TransactionIsolationLevel = Parameters<
+  DataSource['transaction']
+>[0];
+
+/**
+ * Thin ACID transaction helper. Wraps {@link DataSource.transaction} so
+ * services get a scoped {@link EntityManager}: everything executed inside the
+ * callback commits together or rolls back together.
+ *
+ * Usage:
+ * ```ts
+ * await this.transactions.run(async (tx) => {
+ *   const wallet = await tx.findOneBy(Wallet, { id });
+ *   wallet.balance -= amount;
+ *   await tx.save(wallet);
+ * }, 'SERIALIZABLE');
+ * ```
+ *
+ * A thrown error propagates after TypeORM rolls back, so callers keep their
+ * normal error handling. To compose code that must join an existing
+ * transaction, pass its manager to {@link TransactionHelper.join}.
+ */
+@Injectable()
+export class TransactionHelper {
+  constructor(private readonly dataSource: DataSource) {}
+
+  async run<T>(
+    work: (manager: EntityManager) => Promise<T>,
+    isolationLevel?: TransactionIsolationLevel,
+  ): Promise<T> {
+    return this.dataSource.transaction(isolationLevel as never, work);
+  }
+
+  /**
+   * Runs work inside an existing transaction's manager. Useful for composing
+   * repository helpers that must join a caller's transaction instead of
+   * opening a nested one.
+   */
+  static join<T>(
+    manager: EntityManager,
+    work: (tx: EntityManager) => Promise<T>,
+  ): Promise<T> {
+    return work(manager);
+  }
+}

libs/common/src/index.ts

@@ -8,3 +8,5 @@ export * from './dto/paginated-result.dto';
 export * from './interceptors/transform.interceptor';
 export * from './filters/http-exception.filter';
 export * from './decorators/api-paginated-response.decorator';
+export * from './database/base.repository';
+export * from './database/transaction.helper';

package.json

@@ -17,7 +17,10 @@
     "test": "jest",
     "test:watch": "jest --watch",
     "test:cov": "jest --coverage",
-    "test:e2e": "jest --config ./test/jest-e2e.json"
+    "test:e2e": "jest --config ./test/jest-e2e.json",
+    "migration:run": "typeorm migration:run -d src/database/data-source.ts",
+    "migration:revert": "typeorm migration:revert -d src/database/data-source.ts",
+    "migration:generate": "typeorm migration:generate -d src/database/data-source.ts"
   },
   "dependencies": {
     "@nestjs/bull": "^11.0.4",

src/app.module.ts

@@ -26,13 +26,15 @@ import { AuditModule } from './modules/audit/audit.module';
 import { EscrowModule } from './modules/escrow/escrow.module';
 import { ComplianceModule } from './modules/compliance/compliance.module';
 import { RateLimitingModule } from './modules/rate-limiting/rate-limiting.module';
+import { DatabaseModule } from './database/database.module';
 import { PortfolioModule } from './modules/portfolio/portfolio.module';
 import { WebhookModule } from './modules/webhooks/webhook.module';
 
 @Module({
   imports: [
     EventEmitterModule.forRoot(),
     ConfigModule,
+    DatabaseModule,
     TypeOrmModule.forRootAsync({
       useClass: DatabaseConfig,
     }),

src/config/configuration.ts

@@ -14,6 +14,14 @@ export default () => ({
     name: process.env.DB_NAME,
     synchronize: process.env.DB_SYNCHRONIZE === 'true',
     logging: process.env.DB_LOGGING === 'true',
+
+    // Connection pooling.
+    poolMax: parseInt(process.env.DB_POOL_MAX ?? '20', 10),
+    poolMin: parseInt(process.env.DB_POOL_MIN ?? '2', 10),
+    poolIdleTimeoutMs: parseInt(process.env.DB_POOL_IDLE_TIMEOUT_MS ?? '30000', 10),
+
+    // Run versioned migrations automatically on startup (non-production).
+    migrationsRun: process.env.DB_MIGRATIONS_RUN === 'true',
   },
 
   redis: {

src/config/database.config.ts

@@ -1,15 +1,23 @@
 import { Injectable } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { TypeOrmModuleOptions, TypeOrmOptionsFactory } from '@nestjs/typeorm';
+import { AuditTrailSubscriber } from '../database/audit-trail.subscriber';
 
 /**
  * Builds TypeORM connection options from validated configuration. Entities are
  * auto-loaded via the `autoLoadEntities` flag so feature modules only need to
  * register their entities with `TypeOrmModule.forFeature`.
+ *
+ * Pool sizing is environment-driven (`DB_POOL_MAX` / `DB_POOL_MIN`) — pg
+ * exposes these as poolSize/min. Versioned migrations are wired via
+ * `DB_MIGRATIONS_RUN`; prefer them over `synchronize` outside development.
  */
 @Injectable()
 export class DatabaseConfig implements TypeOrmOptionsFactory {
-  constructor(private readonly configService: ConfigService) {}
+  constructor(
+    private readonly configService: ConfigService,
+    private readonly auditTrailSubscriber: AuditTrailSubscriber,
+  ) {}
 
   createTypeOrmOptions(): TypeOrmModuleOptions {
     const db = this.configService.get('database');
@@ -23,6 +31,21 @@
       autoLoadEntities: true,
       synchronize: db.synchronize,
       logging: db.logging,
+
+      // Connection pooling.
+      poolSize: db.poolMax,
+      extra: {
+        min: db.poolMin,
+        max: db.poolMax,
+        // Recycle connections so they do not outlive DB/network idle timeouts.
+        idleTimeoutMillis: db.poolIdleTimeoutMs,
+      },
+
+      // Row-level change capture (see src/database/audit-trail.subscriber.ts).
+      subscribers: [this.auditTrailSubscriber],
+
+      // Versioned migrations infrastructure (see src/database/migrations/).
+      migrationsRun: db.migrationsRun,
     };
   }
 }

src/database/actor-context.interceptor.spec.ts

@@ -0,0 +1,49 @@
+import { ExecutionContext } from '@nestjs/common';
+import { firstValueFrom, Observable } from 'rxjs';
+import { ActorContextInterceptor } from './actor-context.interceptor';
+import { AuditContextService } from './audit-context';
+
+/**
+ * Verifies the interceptor binds the request user so code running downstream
+ * (including TypeORM subscribers) resolves the right audit actor.
+ */
+describe('ActorContextInterceptor', () => {
+  const buildContext = (request: unknown) =>
+    ({
+      switchToHttp: () => ({ getRequest: () => request }),
+    }) as unknown as ExecutionContext;
+
+  it('binds the authenticated user id', async () => {
+    let seen: string | undefined;
+    const interceptor = new ActorContextInterceptor();
+
+    const result$ = interceptor.intercept(buildContext({ user: { id: 'u42' } }), {
+      handle: () =>
+        new Observable<unknown>((subscriber) => {
+          seen = AuditContextService.getActor().id;
+          subscriber.next('ok');
+          subscriber.complete();
+        }),
+    } as never);
+
+    await firstValueFrom(result$ as Observable<unknown>);
+    expect(seen).toBe('u42');
+  });
+
+  it('falls back to the system actor for unauthenticated requests', async () => {
+    let seen: string | undefined;
+    const interceptor = new ActorContextInterceptor();
+
+    const result$ = interceptor.intercept(buildContext({}), {
+      handle: () =>
+        new Observable<unknown>((subscriber) => {
+          seen = AuditContextService.getActor().id;
+          subscriber.next('ok');
+          subscriber.complete();
+        }),
+    } as never);
+
+    await firstValueFrom(result$ as Observable<unknown>);
+    expect(seen).toBe('system');
+  });
+});

src/database/actor-context.interceptor.ts

@@ -0,0 +1,31 @@
+import {
+  CallHandler,
+  ExecutionContext,
+  Injectable,
+  NestInterceptor,
+} from '@nestjs/common';
+import { Observable } from 'rxjs';
+import { tap } from 'rxjs/operators';
+import { AuditContextService } from './audit-context';
+
+/**
+ * Binds the authenticated request user to {@link AuditContextService} for the
+ * duration of the handler, so the {@link AuditTrailSubscriber} can attribute
+ * row changes to an actor without any service-layer plumbing.
+ */
+@Injectable()
+export class ActorContextInterceptor implements NestInterceptor {
+  intercept(context: ExecutionContext, next: CallHandler): Observable<unknown> {
+    const request = context.switchToHttp().getRequest();
+    const user = request?.user as { id?: string } | undefined;
+
+    // Subscribe *inside* run() so every async continuation of the handler
+    // inherits the actor context.
+    return new Observable<unknown>((subscriber) =>
+      AuditContextService.run(
+        user ? { id: String(user.id) } : undefined,
+        () => next.handle().subscribe(subscriber),
+      ),
+    );
+  }
+}