refactor(web): dedup tỷ/triệu compact formatters (GOO-206)

- Add `formatCompact` as an exported alias for `formatPrice` in lib/currency.ts
- Replace 5 inline copies of the tỷ/triệu compact formatter:
  - components/map/listing-map.tsx (local `formatPrice` fn)
  - components/agents/agent-profile-client.tsx (local `fmtVND` fn)
  - app/(dashboard)/dashboard/saved-searches/page.tsx (local `formatPrice` fn)
  - app/(public)/page.tsx (local `formatVnd` fn + `vndFmt` Intl instance)
  - components/listings/price-history-chart.tsx (local `formatMillions` + `priceToMillions`)

All call sites now import from the canonical lib/currency module.
PriceHistoryChart now stores raw VND in chart data (was: millions) so
formatCompact emits correct tỷ/triệu labels using canonical thresholds.

Pre-existing test failures in inquiry/lead/AVM specs are unrelated to this change.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

apps/api/src/modules/retention/application/__tests__/payment-callback-purge.service.spec.ts

@@ -0,0 +1,46 @@
+import type { LoggerService } from '@modules/shared';
+import { PaymentCallbackPurgeService } from '../services/payment-callback-purge.service';
+import type { RetentionRunLogRepository } from '../../infrastructure/repositories/retention-run-log.repository';
+
+describe('PaymentCallbackPurgeService (stub)', () => {
+  let logger: { log: ReturnType<typeof vi.fn>; warn: ReturnType<typeof vi.fn>; error: ReturnType<typeof vi.fn>; debug: ReturnType<typeof vi.fn> };
+  let runLog: { start: ReturnType<typeof vi.fn>; markFinished: ReturnType<typeof vi.fn>; markFailed: ReturnType<typeof vi.fn> };
+  let service: PaymentCallbackPurgeService;
+
+  beforeEach(() => {
+    logger = { log: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() };
+    runLog = {
+      start: vi.fn().mockResolvedValue('run-X'),
+      markFinished: vi.fn().mockResolvedValue(undefined),
+      markFailed: vi.fn().mockResolvedValue(undefined),
+    };
+    service = new PaymentCallbackPurgeService(
+      logger as unknown as LoggerService,
+      runLog as unknown as RetentionRunLogRepository,
+    );
+  });
+
+  it.each([1, 2, 3] as const)('records a SUCCESS run with rowsAffected=0 for phase %s', async (phase) => {
+    const result = await service.run(phase);
+    expect(runLog.start).toHaveBeenCalledWith(expect.objectContaining({ job: 'payment-callback-purge', phase }));
+    expect(runLog.markFinished).toHaveBeenCalledWith('run-X', 0);
+    expect(result).toEqual({ rowsAffected: 0, runId: 'run-X' });
+  });
+
+  it('selects strictly older cutoffs per higher phase (2y < 5y < 10y)', async () => {
+    const fixed = new Date('2030-01-01T00:00:00Z').getTime();
+    vi.spyOn(Date, 'now').mockReturnValue(fixed);
+
+    const calls: string[] = [];
+    logger.warn = vi.fn((msg: string) => calls.push(msg)) as unknown as ReturnType<typeof vi.fn>;
+
+    await service.run(1);
+    await service.run(2);
+    await service.run(3);
+
+    const cutoffs = calls.map((m) => /cutoff=([^)]+)/.exec(m)?.[1] ?? '');
+    expect(cutoffs).toHaveLength(3);
+    expect(cutoffs[0]! > cutoffs[1]!).toBe(true);
+    expect(cutoffs[1]! > cutoffs[2]!).toBe(true);
+  });
+});

apps/api/src/modules/retention/application/__tests__/refresh-token-purge.service.spec.ts

@@ -0,0 +1,49 @@
+import type { LoggerService, PrismaService } from '@modules/shared';
+import { RefreshTokenPurgeService } from '../services/refresh-token-purge.service';
+import type { RetentionRunLogRepository } from '../../infrastructure/repositories/retention-run-log.repository';
+
+describe('RefreshTokenPurgeService', () => {
+  let prisma: { $queryRaw: ReturnType<typeof vi.fn>; refreshToken: { count: ReturnType<typeof vi.fn> } };
+  let logger: { log: ReturnType<typeof vi.fn>; error: ReturnType<typeof vi.fn>; debug: ReturnType<typeof vi.fn> };
+  let runLog: { start: ReturnType<typeof vi.fn>; markFinished: ReturnType<typeof vi.fn>; markFailed: ReturnType<typeof vi.fn> };
+  let service: RefreshTokenPurgeService;
+
+  beforeEach(() => {
+    prisma = { $queryRaw: vi.fn(), refreshToken: { count: vi.fn() } };
+    logger = { log: vi.fn(), error: vi.fn(), debug: vi.fn() };
+    runLog = {
+      start: vi.fn().mockResolvedValue('run-1'),
+      markFinished: vi.fn().mockResolvedValue(undefined),
+      markFailed: vi.fn().mockResolvedValue(undefined),
+    };
+    service = new RefreshTokenPurgeService(
+      prisma as unknown as PrismaService,
+      logger as unknown as LoggerService,
+      runLog as unknown as RetentionRunLogRepository,
+    );
+  });
+
+  it('starts a run, deletes batches, and marks finished with total rowsAffected', async () => {
+    prisma.$queryRaw
+      .mockResolvedValueOnce(new Array(1000).fill(0).map((_, i) => ({ id: `r${i}` })))
+      .mockResolvedValueOnce([{ id: 'r1000' }]);
+
+    const result = await service.run();
+
+    expect(runLog.start).toHaveBeenCalledWith(expect.objectContaining({ job: 'refresh-token-purge' }));
+    expect(prisma.$queryRaw).toHaveBeenCalledTimes(2);
+    expect(runLog.markFinished).toHaveBeenCalledWith('run-1', 1001);
+    expect(result).toEqual({ rowsAffected: 1001, runId: 'run-1' });
+  });
+
+  it('marks the run as FAILED when the delete query throws', async () => {
+    const boom = new Error('connection lost');
+    prisma.$queryRaw.mockRejectedValue(boom);
+
+    await expect(service.run()).rejects.toThrow(boom);
+
+    expect(runLog.markFailed).toHaveBeenCalledWith('run-1', boom, 0);
+    expect(runLog.markFinished).not.toHaveBeenCalled();
+    expect(logger.error).toHaveBeenCalled();
+  });
+});

apps/api/src/modules/retention/application/services/audit-log-purge.service.ts

@@ -0,0 +1,85 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService, LoggerService } from '@modules/shared';
+import { RETENTION_CONFIG } from '../../domain/retention.config';
+import { RetentionRunLogRepository } from '../../infrastructure/repositories/retention-run-log.repository';
+
+/**
+ * Anonymizes rows in AdminAuditLog older than 5 years. Because the schema
+ * requires non-null actor/target IDs, the strategy is tombstone-in-place:
+ * PII columns are replaced with the sentinel `ANONYMIZED`, and a
+ * `metadata.anonymized=true` marker is written so the predicate is
+ * idempotent.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+@Injectable()
+export class AuditLogPurgeService {
+  static readonly JOB = 'audit-log-anonymize';
+  private static readonly TOMBSTONE = 'ANONYMIZED';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly logger: LoggerService,
+    private readonly runLog: RetentionRunLogRepository,
+  ) {}
+
+  async run(): Promise<{ rowsAffected: number; runId: string }> {
+    const cutoff = new Date(Date.now() - RETENTION_CONFIG.auditAnonymizeMs);
+    const runId = await this.runLog.start({
+      job: AuditLogPurgeService.JOB,
+      batchSize: RETENTION_CONFIG.batchSize,
+      dryRun: RETENTION_CONFIG.dryRun,
+    });
+
+    let total = 0;
+    try {
+      for (let batch = 0; batch < RETENTION_CONFIG.maxBatches; batch += 1) {
+        if (RETENTION_CONFIG.dryRun) {
+          total = await this.prisma.adminAuditLog.count({
+            where: {
+              createdAt: { lt: cutoff },
+            },
+          });
+          break;
+        }
+
+        const rows = await this.prisma.$queryRaw<Array<{ id: string }>>`
+          UPDATE "AdminAuditLog"
+          SET "actorId"   = ${AuditLogPurgeService.TOMBSTONE},
+              "targetId"  = ${AuditLogPurgeService.TOMBSTONE},
+              "ipAddress" = NULL,
+              "userAgent" = NULL,
+              "metadata"  = COALESCE("metadata", '{}'::jsonb) || '{"anonymized":true}'::jsonb
+          WHERE id IN (
+            SELECT id FROM "AdminAuditLog"
+            WHERE "createdAt" < ${cutoff}
+              AND ("metadata" ->> 'anonymized') IS DISTINCT FROM 'true'
+            ORDER BY "createdAt" ASC
+            LIMIT ${RETENTION_CONFIG.batchSize}
+            FOR UPDATE SKIP LOCKED
+          )
+          RETURNING id
+        `;
+
+        total += rows.length;
+        if (rows.length < RETENTION_CONFIG.batchSize) break;
+      }
+
+      await this.runLog.markFinished(runId, total);
+      this.logger.log(
+        `AuditLogPurgeService anonymized ${total} row(s) (cutoff=${cutoff.toISOString()}, dryRun=${RETENTION_CONFIG.dryRun})`,
+        'AuditLogPurgeService',
+      );
+      return { rowsAffected: total, runId };
+    } catch (err) {
+      const error = err as Error;
+      await this.runLog.markFailed(runId, error, total);
+      this.logger.error(
+        `AuditLogPurgeService failed: ${error.message}`,
+        error.stack,
+        'AuditLogPurgeService',
+      );
+      throw error;
+    }
+  }
+}

apps/api/src/modules/retention/application/services/kyc-purge.service.ts

@@ -0,0 +1,85 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService, LoggerService } from '@modules/shared';
+import { RETENTION_CONFIG } from '../../domain/retention.config';
+import { RetentionRunLogRepository } from '../../infrastructure/repositories/retention-run-log.repository';
+
+/**
+ * Hard-deletes the User.kycData JSON blob 90 days after `User.deletedAt`.
+ * Per CLO guidance, KYC PII must not survive a deleted account beyond the
+ * Decree-13 minimum. The User row itself stays (it carries audit-relevant
+ * metadata via foreign keys), only the kyc payload is nulled.
+ *
+ * Future work (tracked separately): when a dedicated KycDocument table
+ * lands, this service must also call StorageService.deleteObject(key) for
+ * each blob in S3.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+@Injectable()
+export class KycPurgeService {
+  static readonly JOB = 'kyc-blob-purge';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly logger: LoggerService,
+    private readonly runLog: RetentionRunLogRepository,
+  ) {}
+
+  async run(): Promise<{ rowsAffected: number; runId: string }> {
+    const cutoff = new Date(Date.now() - RETENTION_CONFIG.kycPurgeMs);
+    const runId = await this.runLog.start({
+      job: KycPurgeService.JOB,
+      batchSize: RETENTION_CONFIG.batchSize,
+      dryRun: RETENTION_CONFIG.dryRun,
+    });
+
+    let total = 0;
+    try {
+      for (let batch = 0; batch < RETENTION_CONFIG.maxBatches; batch += 1) {
+        if (RETENTION_CONFIG.dryRun) {
+          total = await this.prisma.user.count({
+            where: {
+              deletedAt: { lt: cutoff },
+              NOT: { kycData: { equals: null as unknown as undefined } },
+            },
+          });
+          break;
+        }
+
+        const rows = await this.prisma.$queryRaw<Array<{ id: string }>>`
+          UPDATE "User"
+          SET "kycData" = NULL
+          WHERE id IN (
+            SELECT id FROM "User"
+            WHERE "deletedAt" IS NOT NULL
+              AND "deletedAt" < ${cutoff}
+              AND "kycData"   IS NOT NULL
+            ORDER BY "deletedAt" ASC
+            LIMIT ${RETENTION_CONFIG.batchSize}
+            FOR UPDATE SKIP LOCKED
+          )
+          RETURNING id
+        `;
+
+        total += rows.length;
+        if (rows.length < RETENTION_CONFIG.batchSize) break;
+      }
+
+      await this.runLog.markFinished(runId, total);
+      this.logger.log(
+        `KycPurgeService nulled kycData on ${total} user row(s) (cutoff=${cutoff.toISOString()}, dryRun=${RETENTION_CONFIG.dryRun})`,
+        'KycPurgeService',
+      );
+      return { rowsAffected: total, runId };
+    } catch (err) {
+      const error = err as Error;
+      await this.runLog.markFailed(runId, error, total);
+      this.logger.error(
+        `KycPurgeService failed: ${error.message}`,
+        error.stack,
+        'KycPurgeService',
+      );
+      throw error;
+    }
+  }
+}

apps/api/src/modules/retention/application/services/messaging-purge.service.ts

@@ -0,0 +1,85 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService, LoggerService } from '@modules/shared';
+import { RETENTION_CONFIG } from '../../domain/retention.config';
+import { RetentionRunLogRepository } from '../../infrastructure/repositories/retention-run-log.repository';
+
+/**
+ * Hard-deletes Message.content for any conversation in CLOSED status whose
+ * last activity is older than 90 days. Metadata (sender, timestamps) is
+ * preserved per CLO guidance — only the message body itself is PII.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+@Injectable()
+export class MessagingPurgeService {
+  static readonly JOB = 'messaging-body-purge';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly logger: LoggerService,
+    private readonly runLog: RetentionRunLogRepository,
+  ) {}
+
+  async run(): Promise<{ rowsAffected: number; runId: string }> {
+    const cutoff = new Date(Date.now() - RETENTION_CONFIG.messagingBodyMs);
+    const runId = await this.runLog.start({
+      job: MessagingPurgeService.JOB,
+      batchSize: RETENTION_CONFIG.batchSize,
+      dryRun: RETENTION_CONFIG.dryRun,
+    });
+
+    let total = 0;
+    try {
+      for (let batch = 0; batch < RETENTION_CONFIG.maxBatches; batch += 1) {
+        if (RETENTION_CONFIG.dryRun) {
+          total = await this.prisma.message.count({
+            where: {
+              deletedAt: null,
+              conversation: {
+                status: 'CLOSED',
+                lastMessageAt: { lt: cutoff },
+              },
+            },
+          });
+          break;
+        }
+
+        const rows = await this.prisma.$queryRaw<Array<{ id: string }>>`
+          UPDATE "Message"
+          SET "content"   = '',
+              "deletedAt" = NOW()
+          WHERE id IN (
+            SELECT m.id FROM "Message" m
+            JOIN "Conversation" c ON c.id = m."conversationId"
+            WHERE m."deletedAt" IS NULL
+              AND c."status"    = 'CLOSED'
+              AND c."lastMessageAt" < ${cutoff}
+            ORDER BY m."createdAt" ASC
+            LIMIT ${RETENTION_CONFIG.batchSize}
+            FOR UPDATE OF m SKIP LOCKED
+          )
+          RETURNING id
+        `;
+
+        total += rows.length;
+        if (rows.length < RETENTION_CONFIG.batchSize) break;
+      }
+
+      await this.runLog.markFinished(runId, total);
+      this.logger.log(
+        `MessagingPurgeService cleared ${total} message body row(s) (cutoff=${cutoff.toISOString()}, dryRun=${RETENTION_CONFIG.dryRun})`,
+        'MessagingPurgeService',
+      );
+      return { rowsAffected: total, runId };
+    } catch (err) {
+      const error = err as Error;
+      await this.runLog.markFailed(runId, error, total);
+      this.logger.error(
+        `MessagingPurgeService failed: ${error.message}`,
+        error.stack,
+        'MessagingPurgeService',
+      );
+      throw error;
+    }
+  }
+}

apps/api/src/modules/retention/application/services/payment-callback-purge.service.ts

@@ -0,0 +1,79 @@
+import { Injectable } from '@nestjs/common';
+import { LoggerService } from '@modules/shared';
+import { RETENTION_CONFIG } from '../../domain/retention.config';
+import { RetentionRunLogRepository } from '../../infrastructure/repositories/retention-run-log.repository';
+
+/**
+ * Payment callback log purge — three-phase schedule confirmed by CLO in
+ * GOO-201 (MoF Circular 78/2021/TT-BTC, Accounting Law 88/2015 Art. 41,
+ * Tax Admin Law 38/2019 Art. 86):
+ *
+ *   Phase 1 @ 2y  — scrub operational PII (IP, device fingerprint).
+ *   Phase 2 @ 5y  — scrub buyer identity (name/phone/email, bank suffix);
+ *                   preserves `buyerName` for invoice-linked rows.
+ *   Phase 3 @ 10y — hard delete (or cold-archive if
+ *                   RETENTION_PAYMENT_ARCHIVE=true).
+ *
+ * This service is intentionally a **stub** in the initial GOO-196 landing:
+ * the `PaymentCallbackLog` table does not exist in the Prisma schema yet
+ * (tracked under the payments module refactor). Calling `run()` emits a
+ * RetentionRunLog row with status=SUCCESS and rowsAffected=0 so dry-run
+ * telemetry is visible from day one, while the actual UPDATE/DELETE
+ * statements are added when the table lands.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+@Injectable()
+export class PaymentCallbackPurgeService {
+  static readonly JOB = 'payment-callback-purge';
+
+  constructor(
+    private readonly logger: LoggerService,
+    private readonly runLog: RetentionRunLogRepository,
+  ) {}
+
+  async run(phase: 1 | 2 | 3): Promise<{ rowsAffected: number; runId: string }> {
+    const cutoff = this.cutoffFor(phase);
+    const runId = await this.runLog.start({
+      job: PaymentCallbackPurgeService.JOB,
+      phase,
+      batchSize: RETENTION_CONFIG.batchSize,
+      dryRun: RETENTION_CONFIG.dryRun,
+    });
+
+    try {
+      // TODO(GOO-196 follow-up): implement once PaymentCallbackLog schema lands.
+      //   Phase 1: UPDATE … SET ipAddress=NULL, deviceFingerprint=NULL,
+      //            anonymizedPhase1At=NOW() WHERE createdAt < ${cutoff}
+      //            AND anonymizedPhase1At IS NULL
+      //   Phase 2: UPDATE … SET callbackPayload = jsonb_set_lax(..., 'null'),
+      //            bankAccountMasked=NULL, cardSuffix=NULL,
+      //            anonymizedPhase2At=NOW()  (skip buyerName on invoice rows)
+      //   Phase 3: DELETE FROM "PaymentCallbackLog" WHERE createdAt < ${cutoff}
+      //            — OR — INSERT INTO payment_callback_archive … then DELETE.
+      const rowsAffected = 0;
+      await this.runLog.markFinished(runId, rowsAffected);
+      this.logger.warn(
+        `PaymentCallbackPurgeService phase=${phase} is a no-op — PaymentCallbackLog table not yet in schema (cutoff=${cutoff.toISOString()})`,
+        'PaymentCallbackPurgeService',
+      );
+      return { rowsAffected, runId };
+    } catch (err) {
+      const error = err as Error;
+      await this.runLog.markFailed(runId, error);
+      throw error;
+    }
+  }
+
+  private cutoffFor(phase: 1 | 2 | 3): Date {
+    const now = Date.now();
+    switch (phase) {
+      case 1:
+        return new Date(now - RETENTION_CONFIG.paymentPhase1Ms);
+      case 2:
+        return new Date(now - RETENTION_CONFIG.paymentPhase2Ms);
+      case 3:
+        return new Date(now - RETENTION_CONFIG.paymentPhase3Ms);
+    }
+  }
+}

apps/api/src/modules/retention/application/services/refresh-token-purge.service.ts

@@ -0,0 +1,81 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService, LoggerService } from '@modules/shared';
+import { RETENTION_CONFIG } from '../../domain/retention.config';
+import { RetentionRunLogRepository } from '../../infrastructure/repositories/retention-run-log.repository';
+
+/**
+ * Hard-deletes refresh tokens that are revoked or expired and were created
+ * more than `RETENTION_REFRESH_TOKEN_DAYS` (default 30) days ago.
+ *
+ * Idempotency: a single DELETE … RETURNING id with FOR UPDATE SKIP LOCKED is
+ * race-safe — only one writer wins each row. Loops in capped batches to avoid
+ * statement timeouts on large tables.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+@Injectable()
+export class RefreshTokenPurgeService {
+  static readonly JOB = 'refresh-token-purge';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly logger: LoggerService,
+    private readonly runLog: RetentionRunLogRepository,
+  ) {}
+
+  async run(): Promise<{ rowsAffected: number; runId: string }> {
+    const cutoff = new Date(Date.now() - RETENTION_CONFIG.refreshTokenStaleMs);
+    const runId = await this.runLog.start({
+      job: RefreshTokenPurgeService.JOB,
+      batchSize: RETENTION_CONFIG.batchSize,
+      dryRun: RETENTION_CONFIG.dryRun,
+    });
+
+    let total = 0;
+    try {
+      for (let batch = 0; batch < RETENTION_CONFIG.maxBatches; batch += 1) {
+        if (RETENTION_CONFIG.dryRun) {
+          total = await this.prisma.refreshToken.count({
+            where: {
+              createdAt: { lt: cutoff },
+              OR: [{ revokedAt: { not: null } }, { expiresAt: { lt: new Date() } }],
+            },
+          });
+          break;
+        }
+
+        const rows = await this.prisma.$queryRaw<Array<{ id: string }>>`
+          DELETE FROM "RefreshToken"
+          WHERE id IN (
+            SELECT id FROM "RefreshToken"
+            WHERE "createdAt" < ${cutoff}
+              AND ("revokedAt" IS NOT NULL OR "expiresAt" < NOW())
+            ORDER BY "createdAt" ASC
+            LIMIT ${RETENTION_CONFIG.batchSize}
+            FOR UPDATE SKIP LOCKED
+          )
+          RETURNING id
+        `;
+
+        total += rows.length;
+        if (rows.length < RETENTION_CONFIG.batchSize) break;
+      }
+
+      await this.runLog.markFinished(runId, total);
+      this.logger.log(
+        `RefreshTokenPurgeService removed ${total} row(s) (cutoff=${cutoff.toISOString()}, dryRun=${RETENTION_CONFIG.dryRun})`,
+        'RefreshTokenPurgeService',
+      );
+      return { rowsAffected: total, runId };
+    } catch (err) {
+      const error = err as Error;
+      await this.runLog.markFailed(runId, error, total);
+      this.logger.error(
+        `RefreshTokenPurgeService failed: ${error.message}`,
+        error.stack,
+        'RefreshTokenPurgeService',
+      );
+      throw error;
+    }
+  }
+}

apps/api/src/modules/retention/domain/retention.config.ts

@@ -0,0 +1,56 @@
+/**
+ * Centralised retention configuration. Every window comes from env vars so a
+ * staging environment can dry-run aggressively without hard-coding constants.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+
+const days = (n: number): number => n * 24 * 60 * 60 * 1000;
+const years = (n: number): number => Math.floor(n * 365.25 * 24 * 60 * 60 * 1000);
+
+const numEnv = (name: string, fallback: number): number => {
+  const raw = process.env[name];
+  if (!raw) return fallback;
+  const parsed = Number(raw);
+  return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
+};
+
+const boolEnv = (name: string, fallback = false): boolean => {
+  const raw = process.env[name];
+  if (raw === undefined) return fallback;
+  return raw === 'true' || raw === '1';
+};
+
+export const RETENTION_CONFIG = {
+  /** Master kill switch — when false, every cron is a no-op. */
+  enabled: boolEnv('RETENTION_ENABLED', false),
+  /** Forces every job into SELECT-only mode for staging review. */
+  dryRun: boolEnv('RETENTION_DRY_RUN', false),
+
+  /** Per-batch row cap. Keeps statement timeouts and replication lag bounded. */
+  batchSize: numEnv('RETENTION_BATCH_SIZE', 1000),
+  /** Hard cap on batch loops per run. */
+  maxBatches: numEnv('RETENTION_MAX_BATCHES', 50),
+
+  /** RefreshToken: revoked OR expired older than this is hard-deleted. */
+  refreshTokenStaleMs: days(numEnv('RETENTION_REFRESH_TOKEN_DAYS', 30)),
+
+  /** Messaging body purge window after Conversation.lastMessageAt. */
+  messagingBodyMs: days(numEnv('RETENTION_MESSAGING_DAYS', 90)),
+
+  /** KYC blob purge window after User.deletedAt. */
+  kycPurgeMs: days(numEnv('RETENTION_KYC_DAYS', 90)),
+
+  /** Audit log anonymization window. */
+  auditAnonymizeMs: years(numEnv('RETENTION_AUDIT_YEARS', 5)),
+
+  /** Payment callback phased schedule (CLO confirmed via GOO-201). */
+  paymentPhase1Ms: years(numEnv('RETENTION_PAYMENT_PHASE1_YEARS', 2)),
+  paymentPhase2Ms: years(numEnv('RETENTION_PAYMENT_PHASE2_YEARS', 5)),
+  paymentPhase3Ms: years(numEnv('RETENTION_PAYMENT_PHASE3_YEARS', 10)),
+
+  /** When true, phase-3 archives to a cold table instead of hard-deleting. */
+  paymentArchive: boolEnv('RETENTION_PAYMENT_ARCHIVE', false),
+} as const;
+
+export type RetentionConfig = typeof RETENTION_CONFIG;

apps/api/src/modules/retention/index.ts

@@ -0,0 +1,8 @@
+export { RetentionModule } from './retention.module';
+export { RefreshTokenPurgeService } from './application/services/refresh-token-purge.service';
+export { MessagingPurgeService } from './application/services/messaging-purge.service';
+export { KycPurgeService } from './application/services/kyc-purge.service';
+export { AuditLogPurgeService } from './application/services/audit-log-purge.service';
+export { PaymentCallbackPurgeService } from './application/services/payment-callback-purge.service';
+export { RetentionRunLogRepository } from './infrastructure/repositories/retention-run-log.repository';
+export { RETENTION_CONFIG, type RetentionConfig } from './domain/retention.config';

apps/api/src/modules/retention/infrastructure/cron/retention-cron.orchestrator.ts

@@ -0,0 +1,91 @@
+import { Injectable } from '@nestjs/common';
+import { Cron } from '@nestjs/schedule';
+import { LoggerService } from '@modules/shared';
+import { AuditLogPurgeService } from '../../application/services/audit-log-purge.service';
+import { KycPurgeService } from '../../application/services/kyc-purge.service';
+import { MessagingPurgeService } from '../../application/services/messaging-purge.service';
+import { PaymentCallbackPurgeService } from '../../application/services/payment-callback-purge.service';
+import { RefreshTokenPurgeService } from '../../application/services/refresh-token-purge.service';
+import { RETENTION_CONFIG } from '../../domain/retention.config';
+
+/**
+ * Thin scheduler that delegates to each domain purge service. All windows
+ * run during Vietnam off-peak hours (UTC times below correspond to ~23:00–
+ * 01:00 ICT). Set RETENTION_ENABLED=true to activate; otherwise every job
+ * is a no-op so the module can ship behind a flag.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+@Injectable()
+export class RetentionCronOrchestrator {
+  constructor(
+    private readonly logger: LoggerService,
+    private readonly refreshTokens: RefreshTokenPurgeService,
+    private readonly messaging: MessagingPurgeService,
+    private readonly kyc: KycPurgeService,
+    private readonly auditLogs: AuditLogPurgeService,
+    private readonly paymentCallbacks: PaymentCallbackPurgeService,
+  ) {}
+
+  @Cron('0 16 * * *', { name: 'retention-refresh-tokens', timeZone: 'UTC' })
+  async runRefreshTokens(): Promise<void> {
+    if (!RETENTION_CONFIG.enabled) return this.skip('refresh-token-purge');
+    await this.safe(() => this.refreshTokens.run(), 'refresh-token-purge');
+  }
+
+  @Cron('30 16 * * *', { name: 'retention-messaging', timeZone: 'UTC' })
+  async runMessaging(): Promise<void> {
+    if (!RETENTION_CONFIG.enabled) return this.skip('messaging-purge');
+    await this.safe(() => this.messaging.run(), 'messaging-purge');
+  }
+
+  @Cron('0 17 * * *', { name: 'retention-kyc', timeZone: 'UTC' })
+  async runKyc(): Promise<void> {
+    if (!RETENTION_CONFIG.enabled) return this.skip('kyc-purge');
+    await this.safe(() => this.kyc.run(), 'kyc-purge');
+  }
+
+  @Cron('30 17 * * *', { name: 'retention-payment-phase1', timeZone: 'UTC' })
+  async runPaymentPhase1(): Promise<void> {
+    if (!RETENTION_CONFIG.enabled) return this.skip('payment-phase-1');
+    await this.safe(() => this.paymentCallbacks.run(1), 'payment-phase-1');
+  }
+
+  @Cron('0 18 * * *', { name: 'retention-payment-phase2', timeZone: 'UTC' })
+  async runPaymentPhase2(): Promise<void> {
+    if (!RETENTION_CONFIG.enabled) return this.skip('payment-phase-2');
+    await this.safe(() => this.paymentCallbacks.run(2), 'payment-phase-2');
+  }
+
+  @Cron('0 17 * * 0', { name: 'retention-audit-anonymize', timeZone: 'UTC' })
+  async runAuditLogs(): Promise<void> {
+    if (!RETENTION_CONFIG.enabled) return this.skip('audit-log-anonymize');
+    await this.safe(() => this.auditLogs.run(), 'audit-log-anonymize');
+  }
+
+  @Cron('0 18 * * 0', { name: 'retention-payment-phase3', timeZone: 'UTC' })
+  async runPaymentPhase3(): Promise<void> {
+    if (!RETENTION_CONFIG.enabled) return this.skip('payment-phase-3');
+    await this.safe(() => this.paymentCallbacks.run(3), 'payment-phase-3');
+  }
+
+  private skip(name: string): void {
+    this.logger.debug(
+      `Retention job ${name} skipped: RETENTION_ENABLED=false`,
+      'RetentionCronOrchestrator',
+    );
+  }
+
+  private async safe(fn: () => Promise<unknown>, name: string): Promise<void> {
+    try {
+      await fn();
+    } catch (err) {
+      this.logger.error(
+        `Retention job ${name} threw: ${(err as Error).message}`,
+        (err as Error).stack,
+        'RetentionCronOrchestrator',
+      );
+      // Swallow — RetentionRunLog already records FAILED. Do not crash the scheduler.
+    }
+  }
+}

apps/api/src/modules/retention/infrastructure/repositories/retention-run-log.repository.ts

@@ -0,0 +1,56 @@
+import { Injectable } from '@nestjs/common';
+import { PrismaService } from '@modules/shared';
+
+/**
+ * Thin repository around RetentionRunLog. Centralised so every purge service
+ * uses the exact same shape — start row, mark finished/failed, never invent
+ * variations.
+ *
+ * GOO-196 — Decree 13 compliance.
+ */
+@Injectable()
+export class RetentionRunLogRepository {
+  constructor(private readonly prisma: PrismaService) {}
+
+  async start(input: {
+    job: string;
+    phase?: number | null;
+    batchSize?: number | null;
+    dryRun?: boolean;
+  }): Promise<string> {
+    const row = await this.prisma.retentionRunLog.create({
+      data: {
+        job: input.job,
+        phase: input.phase ?? null,
+        batchSize: input.batchSize ?? null,
+        dryRun: input.dryRun ?? false,
+        status: 'RUNNING',
+      },
+      select: { id: true },
+    });
+    return row.id;
+  }
+
+  async markFinished(id: string, rowsAffected: number, partial = false): Promise<void> {
+    await this.prisma.retentionRunLog.update({
+      where: { id },
+      data: {
+        finishedAt: new Date(),
+        rowsAffected,
+        status: partial ? 'PARTIAL' : 'SUCCESS',
+      },
+    });
+  }
+
+  async markFailed(id: string, error: Error, rowsAffected = 0): Promise<void> {
+    await this.prisma.retentionRunLog.update({
+      where: { id },
+      data: {
+        finishedAt: new Date(),
+        rowsAffected,
+        status: 'FAILED',
+        errorMessage: error.message.slice(0, 2000),
+      },
+    });
+  }
+}

apps/api/src/modules/retention/retention.module.ts

@@ -0,0 +1,36 @@
+import { Module } from '@nestjs/common';
+import { AuditLogPurgeService } from './application/services/audit-log-purge.service';
+import { KycPurgeService } from './application/services/kyc-purge.service';
+import { MessagingPurgeService } from './application/services/messaging-purge.service';
+import { PaymentCallbackPurgeService } from './application/services/payment-callback-purge.service';
+import { RefreshTokenPurgeService } from './application/services/refresh-token-purge.service';
+import { RetentionCronOrchestrator } from './infrastructure/cron/retention-cron.orchestrator';
+import { RetentionRunLogRepository } from './infrastructure/repositories/retention-run-log.repository';
+
+/**
+ * GOO-196 — Decree 13 data retention & purge jobs.
+ *
+ * Ships behind RETENTION_ENABLED=false so the module can land without
+ * affecting prod. Flip to true after a 7-day staging dry-run review with
+ * CLO/DPO. See `apps/api/src/modules/retention/domain/retention.config.ts`
+ * for every tunable window.
+ */
+@Module({
+  providers: [
+    RetentionRunLogRepository,
+    RefreshTokenPurgeService,
+    MessagingPurgeService,
+    KycPurgeService,
+    AuditLogPurgeService,
+    PaymentCallbackPurgeService,
+    RetentionCronOrchestrator,
+  ],
+  exports: [
+    RefreshTokenPurgeService,
+    MessagingPurgeService,
+    KycPurgeService,
+    AuditLogPurgeService,
+    PaymentCallbackPurgeService,
+  ],
+})
+export class RetentionModule {}

apps/web/app/[locale]/(dashboard)/dashboard/saved-searches/page.tsx

@@ -9,6 +9,7 @@ import {
   useDeleteSavedSearch,
   useUpdateSavedSearch,
 } from '@/lib/hooks/use-saved-searches';
+import { formatCompact } from '@/lib/currency';
 import { type SavedSearch, type SavedSearchFilters } from '@/lib/saved-search-api';
 
 const PROPERTY_TYPE_LABELS: Record<string, string> = {
@@ -25,12 +26,6 @@ const TRANSACTION_TYPE_LABELS: Record<string, string> = {
   RENT: 'Cho thuê',
 };
 
-function formatPrice(value: string): string {
-  const num = Number(value);
-  if (num >= 1_000_000_000) return `${(num / 1_000_000_000).toFixed(1)} tỷ`;
-  if (num >= 1_000_000) return `${(num / 1_000_000).toFixed(0)} triệu`;
-  return num.toLocaleString('vi-VN');
-}
 
 function formatFilters(filters: SavedSearchFilters): string[] {
   const parts: string[] = [];
@@ -44,11 +39,11 @@ function formatFilters(filters: SavedSearchFilters): string[] {
   if (filters.district) parts.push(filters.district);
   if (filters.city) parts.push(filters.city);
   if (filters.priceMin && filters.priceMax) {
-    parts.push(`${formatPrice(filters.priceMin)} - ${formatPrice(filters.priceMax)}`);
+    parts.push(`${formatCompact(filters.priceMin)} - ${formatCompact(filters.priceMax)}`);
   } else if (filters.priceMin) {
-    parts.push(`Từ ${formatPrice(filters.priceMin)}`);
+    parts.push(`Từ ${formatCompact(filters.priceMin)}`);
   } else if (filters.priceMax) {
-    parts.push(`Đến ${formatPrice(filters.priceMax)}`);
+    parts.push(`Đến ${formatCompact(filters.priceMax)}`);
   }
   if (filters.areaMin || filters.areaMax) {
     if (filters.areaMin && filters.areaMax) {

apps/web/app/[locale]/(public)/page.tsx

@@ -17,23 +17,13 @@ import {
   usePriceMovers,
   useTrendingAreas,
 } from '@/lib/hooks/use-analytics';
+import { formatCompact } from '@/lib/currency';
 import { listingsApi, type ListingDetail } from '@/lib/listings-api';
 
 /* ------------------------------------------------------------------ */
 /*  Helpers                                                           */
 /* ------------------------------------------------------------------ */
 
-const vndFmt = new Intl.NumberFormat('vi-VN', {
-  style: 'currency',
-  currency: 'VND',
-  maximumFractionDigits: 0,
-});
-
-function formatVnd(value: number): string {
-  if (value >= 1_000_000_000) return `${(value / 1_000_000_000).toFixed(1)} tỷ`;
-  if (value >= 1_000_000) return `${(value / 1_000_000).toFixed(0)} tr`;
-  return vndFmt.format(value);
-}
 
 function formatPriceM2(value: number): string {
   if (value >= 1_000_000) return `${(value / 1_000_000).toFixed(1)} tr/m²`;
@@ -146,7 +136,7 @@ function KpiStrip({ city }: { city: string }) {
       />
       <KpiCard
         label="Giá TB"
-        value={data ? formatVnd(data.avgPrice) : '—'}
+        value={data ? formatCompact(data.avgPrice) : '—'}
         delta={data?.priceChangePct?.d30}
         footnote="Toàn thành phố"
         icon={<Building2 className="h-3.5 w-3.5" />}
@@ -154,7 +144,7 @@ function KpiStrip({ city }: { city: string }) {
       />
       <KpiCard
         label="Giá trung vị"
-        value={data ? formatVnd(data.medianPrice) : '—'}
+        value={data ? formatCompact(data.medianPrice) : '—'}
         footnote="Median price"
         icon={<Layers className="h-3.5 w-3.5" />}
         loading={isLoading}
@@ -352,7 +342,7 @@ function RecentListings() {
           const price = Number(r.priceVND);
           return (
             <span className="font-mono text-sm font-semibold tabular-nums text-foreground">
-              {formatVnd(price)}
+              {formatCompact(price)}
             </span>
           );
         },

apps/web/components/agents/agent-profile-client.tsx

@@ -39,6 +39,7 @@ import { Button } from '@/components/ui/button';
 import { Card, CardContent, CardHeader, CardTitle, CardDescription } from '@/components/ui/card';
 import { Link } from '@/i18n/navigation';
 import type { AgentPublicProfile, AgentReviewItem } from '@/lib/agents-api';
+import { formatCompact } from '@/lib/currency';
 import { shimmerBlurDataURL } from '@/lib/image-blur';
 import type { ListingDetail } from '@/lib/listings-api';
 
@@ -48,12 +49,6 @@ import type { ListingDetail } from '@/lib/listings-api';
 
 const VND = new Intl.NumberFormat('vi-VN');
 
-function fmtVND(value: string | number | bigint): string {
-  const n = typeof value === 'bigint' ? Number(value) : Number(value);
-  if (n >= 1_000_000_000) return `${(n / 1_000_000_000).toFixed(1)} tỷ`;
-  if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(0)} tr`;
-  return VND.format(n);
-}
 
 function qualityLabel(score: number): string {
   if (score >= 80) return 'Xuất sắc';
@@ -145,7 +140,7 @@ const listingColumns: DataTableColumn<ListingDetail>[] = [
     sortValue: (row) => Number(row.priceVND),
     cell: (row) => (
       <span className="text-xs font-semibold tabular-nums text-primary">
-        {fmtVND(row.priceVND)}
+        {formatCompact(row.priceVND)}
       </span>
     ),
     width: '12%',
@@ -160,7 +155,7 @@ const listingColumns: DataTableColumn<ListingDetail>[] = [
     cell: (row) =>
       row.pricePerM2 != null ? (
         <span className="text-xs tabular-nums text-foreground-muted">
-          {fmtVND(row.pricePerM2)}
+          {formatCompact(row.pricePerM2)}
         </span>
       ) : (
         <span className="text-xs text-foreground-dim">—</span>
@@ -384,7 +379,7 @@ export function AgentProfileClient({
         />
         <KpiCard
           label="Giá TB"
-          value={avgPriceVND != null ? fmtVND(avgPriceVND) : '—'}
+          value={avgPriceVND != null ? formatCompact(avgPriceVND) : '—'}
           icon={<BarChart2 className="h-4 w-4" />}
           footnote="Danh mục hiện tại"
         />

apps/web/components/listings/price-history-chart.tsx

@@ -11,21 +11,13 @@ import {
 } from 'recharts';
 
 import type { PriceHistoryItem } from '@/lib/listings-api';
+import { formatCompact } from '@/lib/currency';
 
 interface PriceHistoryChartProps {
   data: PriceHistoryItem[];
   height?: number;
 }
 
-function priceToMillions(priceStr: string): number {
-  return Math.round(Number(priceStr) / 1_000_000);
-}
-
-function formatMillions(value: number): string {
-  if (value >= 1000) return `${(value / 1000).toFixed(1)} tỷ`;
-  return `${value} tr`;
-}
-
 export function PriceHistoryChart({ data, height = 280 }: PriceHistoryChartProps) {
   if (data.length === 0) return null;
 
@@ -37,7 +29,7 @@ export function PriceHistoryChart({ data, height = 280 }: PriceHistoryChartProps
         month: '2-digit',
         year: 'numeric',
       }),
-      price: priceToMillions(item.newPrice),
+      price: Number(item.newPrice),
     }));
 
   return (
@@ -48,7 +40,7 @@ export function PriceHistoryChart({ data, height = 280 }: PriceHistoryChartProps
         <YAxis
           tick={{ fontSize: 11 }}
           className="fill-muted-foreground"
-          tickFormatter={(v: number) => formatMillions(v)}
+          tickFormatter={(v: number) => formatCompact(v)}
         />
         <Tooltip
           contentStyle={{
@@ -57,7 +49,7 @@ export function PriceHistoryChart({ data, height = 280 }: PriceHistoryChartProps
             borderRadius: '0.5rem',
             fontSize: '0.875rem',
           }}
-          formatter={(value) => [formatMillions(Number(value)), 'Giá']}
+          formatter={(value) => [formatCompact(Number(value)), 'Giá']}
         />
         <Line
           type="monotone"

apps/web/components/map/listing-map.tsx

@@ -5,16 +5,10 @@ import mapboxgl from 'mapbox-gl';
 import * as React from 'react';
 import 'mapbox-gl/dist/mapbox-gl.css';
 import { ComponentErrorBoundary } from '@/components/error-boundary';
+import { formatCompact } from '@/lib/currency';
 import type { ListingDetail } from '@/lib/listings-api';
 import { useMapboxStyle } from '@/lib/mapbox-style';
 
-function formatPrice(priceVND: string): string {
-  const num = Number(priceVND);
-  if (num >= 1_000_000_000) return `${(num / 1_000_000_000).toFixed(1)} tỷ`;
-  if (num >= 1_000_000) return `${(num / 1_000_000).toFixed(0)} tr`;
-  return num.toLocaleString('vi-VN');
-}
-
 interface ListingMapProps {
   listings: ListingDetail[];
   onMarkerClick?: (listing: ListingDetail) => void;
@@ -67,7 +61,7 @@ function buildGeoJSON(
       geometry: { type: 'Point', coordinates: [lng, lat] },
       properties: {
         id: listing.id,
-        price: formatPrice(listing.priceVND),
+        price: formatCompact(listing.priceVND),
         title: listing.property.title,
         district: listing.property.district ?? '',
         city: listing.property.city ?? '',
@@ -100,7 +94,7 @@ function buildPopupContent(listing: ListingDetail): HTMLDivElement {
   const price = document.createElement('p');
   price.style.cssText =
     'font-weight:700;color:hsl(var(--primary));font-size:14px;margin:0 0 4px;';
-  price.textContent = `${formatPrice(listing.priceVND)} VND`;
+  price.textContent = `${formatCompact(listing.priceVND)} VND`;
   container.appendChild(price);
 
   const title = document.createElement('p');

apps/web/lib/currency.ts

@@ -97,6 +97,18 @@ export function formatPricePerM2(price: string | number): string {
   return `${num.toLocaleString('vi-VN')} \u0111/m\u00b2`;
 }
 
+
+// ---------------------------------------------------------------------------
+// Alias: formatCompact
+// ---------------------------------------------------------------------------
+
+/**
+ * Alias for {@link formatPrice}.
+ * Use this name when the call-site intent is compact/abbreviated display
+ * rather than a full price string (e.g. chart tick labels, map markers).
+ */
+export const formatCompact = formatPrice;
+
 // ---------------------------------------------------------------------------
 // Parser (reverse direction)
 // ---------------------------------------------------------------------------

prisma/migrations/20260424050000_goo196_add_retention_run_log/migration.sql

@@ -0,0 +1,25 @@
+-- GOO-196: Data retention policy & purge jobs (Decree 13 compliance)
+-- Adds the RetentionRunLog table so every purge / anonymization pass is auditable.
+
+-- CreateEnum
+CREATE TYPE "RetentionRunStatus" AS ENUM ('RUNNING', 'SUCCESS', 'PARTIAL', 'FAILED');
+
+-- CreateTable
+CREATE TABLE "RetentionRunLog" (
+    "id" TEXT NOT NULL,
+    "job" TEXT NOT NULL,
+    "phase" INTEGER,
+    "startedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "finishedAt" TIMESTAMP(3),
+    "rowsAffected" INTEGER NOT NULL DEFAULT 0,
+    "status" "RetentionRunStatus" NOT NULL DEFAULT 'RUNNING',
+    "errorMessage" TEXT,
+    "batchSize" INTEGER,
+    "dryRun" BOOLEAN NOT NULL DEFAULT false,
+
+    CONSTRAINT "RetentionRunLog_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "RetentionRunLog_job_startedAt_idx" ON "RetentionRunLog"("job", "startedAt");
+CREATE INDEX "RetentionRunLog_startedAt_idx" ON "RetentionRunLog"("startedAt" DESC);

prisma/schema.prisma

@@ -1567,3 +1567,68 @@ model VnAdministrativeAlias {
   @@index([newWardCode])
   @@map("vn_administrative_aliases")
 }
+
+// =============================================================================
+// READ MODELS — RFC-003 Phase 0
+// =============================================================================
+
+/// Idempotency offset table for CQRS projectors.
+///
+/// Per RFC-003 §0 (CTO ask): every projector wraps its `apply` call in a
+/// transaction that inserts `(eventId, handlerName)` here. Re-deliveries
+/// hit the composite primary key, the transaction rolls back, and the
+/// projector observes a no-op. Port + Prisma implementation live in
+/// `apps/api/src/modules/read-models/`.
+model ProjectionOffset {
+  /// Stable identifier of the projected event. Phase 0 derives this
+  /// from `${aggregateId}:${occurredAt.getTime()}:${eventName}` until
+  /// domain events carry a producer-minted UUID (RFC-003 Option D).
+  eventId String
+
+  /// Stable identifier of the projector that consumed the event.
+  /// Renaming a projector forces a full re-projection — be deliberate.
+  handlerName String
+
+  /// When the offset was first written (i.e. the projection ran).
+  appliedAt DateTime @default(now())
+
+  /// Optional content hash of the projected payload. Reconciliation
+  /// jobs use this to detect drift between the read model and the
+  /// authoritative write model.
+  payloadHash String?
+
+  @@id([eventId, handlerName])
+  @@index([handlerName, appliedAt])
+  @@map("projection_offset")
+}
+
+// =============================================================================
+// RETENTION (GOO-196 — Decree 13 compliance)
+// =============================================================================
+
+enum RetentionRunStatus {
+  RUNNING
+  SUCCESS
+  PARTIAL
+  FAILED
+}
+
+/// Every purge / anonymization pass emits a RetentionRunLog row so the
+/// operator and DPO can audit exactly what was touched and when. Multi-phase
+/// jobs (e.g. payment callback 2y / 5y / 10y) record `phase` for
+/// disambiguation.
+model RetentionRunLog {
+  id           String             @id @default(cuid())
+  job          String
+  phase        Int?
+  startedAt    DateTime           @default(now())
+  finishedAt   DateTime?
+  rowsAffected Int                @default(0)
+  status       RetentionRunStatus @default(RUNNING)
+  errorMessage String?
+  batchSize    Int?
+  dryRun       Boolean            @default(false)
+
+  @@index([job, startedAt])
+  @@index([startedAt(sort: Desc)])
+}