dorfteich/apps/api/src/read-trail/read-trail-maintenance.service.ts

import { Injectable } from '@nestjs/common';
import { PinoLogger } from 'nestjs-pino';

import { AuditService } from '../audit/audit.service';
import { ClockService } from '../common/clock.service';
import { PrismaService } from '../prisma/prisma.service';
import { InstanceSettingsService } from '../settings/instance-settings.service';

const MS_PER_DAY = 24 * 60 * 60 * 1000;

/** How many months ahead of "now" a partition must exist. Two keeps a
 * multi-week job outage from ever reaching an uncovered month (the DEFAULT
 * partition would still catch it — reads never fail on a missing month). */
const MONTHS_AHEAD = 2;

/**
 * Read-trail storage maintenance (issue #224, ADR 0023), one daily job with
 * two duties:
 *
 * 1. **Partition upkeep** — `read_events` is RANGE-partitioned by month
 *    (migration `20260731170000`); this creates the next {@link MONTHS_AHEAD}
 *    monthly partitions, each with its per-partition dedup unique index
 *    (#223 — the partitioned parent cannot carry it). A `db push` database
 *    (tests) has a plain table; partition work skips itself there.
 * 2. **Retention** — events older than `readTrail.retentionDays` are
 *    removed: whole months by dropping their partition (no scan), the
 *    remainder (default partition, plain tables) by a ranged delete. The
 *    deletion is audited (`read_trail.pruned`) so a gap in the evidence is
 *    always explainable — same principle as `audit.pruned` (#196), but a
 *    deliberately separate period.
 */
@Injectable()
export class ReadTrailMaintenanceService {
  constructor(
    private readonly prisma: PrismaService,
    private readonly settings: InstanceSettingsService,
    private readonly audit: AuditService,
    private readonly clock: ClockService,
    private readonly logger: PinoLogger,
  ) {
    this.logger.setContext(ReadTrailMaintenanceService.name);
  }

  async run(): Promise<void> {
    await this.ensurePartitions();
    await this.pruneExpired();
  }

  /** True when read_events is a partitioned parent (relkind `p`). */
  private async isPartitioned(): Promise<boolean> {
    const rows = await this.prisma.$queryRaw<{ relkind: string }[]>`
      SELECT relkind::text FROM pg_class
       WHERE relname = 'read_events' AND relnamespace = 'public'::regnamespace`;
    return rows[0]?.relkind === 'p';
  }

  /** `read_events_y2026m08` for 2026-08. */
  private partitionName(month: Date): string {
    const y = month.getUTCFullYear();
    const m = String(month.getUTCMonth() + 1).padStart(2, '0');
    return `read_events_y${y}m${m}`;
  }

  private monthStart(base: Date, offsetMonths: number): Date {
    return new Date(Date.UTC(base.getUTCFullYear(), base.getUTCMonth() + offsetMonths, 1));
  }

  async ensurePartitions(): Promise<void> {
    if (!(await this.isPartitioned())) {
      this.logger.debug('read_events is not partitioned here; skipping partition upkeep');
      return;
    }
    const now = this.clock.now();
    for (let offset = 0; offset <= MONTHS_AHEAD; offset += 1) {
      const from = this.monthStart(now, offset);
      const to = this.monthStart(now, offset + 1);
      const name = this.partitionName(from);
      try {
        await this.prisma.$executeRawUnsafe(
          `CREATE TABLE IF NOT EXISTS "${name}" PARTITION OF "read_events"
             FOR VALUES FROM ('${from.toISOString()}') TO ('${to.toISOString()}')`,
        );
        await this.prisma.$executeRawUnsafe(
          `CREATE UNIQUE INDEX IF NOT EXISTS "${name}_dedup_key"
             ON "${name}" ("dedup_key", "window_bucket")`,
        );
      } catch (error) {
        // Most likely: the DEFAULT partition already holds rows of this month
        // (the job lagged past a month boundary). Nothing is lost — those
        // rows live in the default partition and age out through the ranged
        // delete below; the month just cannot get its own partition anymore.
        this.logger.warn({ partition: name, err: error }, 'read-trail partition not created');
      }
    }
  }

  async pruneExpired(): Promise<number> {
    const retentionDays = await this.settings.get('readTrail.retentionDays');
    const cutoff = new Date(this.clock.now().getTime() - retentionDays * MS_PER_DAY);
    let dropped = 0;

    if (await this.isPartitioned()) {
      // Whole months strictly before the cutoff month go by DROP — no scan,
      // and the dropped range is exact (every row in them is < cutoff).
      const partitions = await this.prisma.$queryRaw<{ relname: string }[]>`
        SELECT c.relname::text
          FROM pg_inherits i
          JOIN pg_class c ON c.oid = i.inhrelid
         WHERE i.inhparent = 'read_events'::regclass
           AND c.relname ~ '^read_events_y[0-9]{4}m[0-9]{2}$'`;
      const cutoffMonth = this.monthStart(cutoff, 0);
      for (const { relname } of partitions) {
        const match = /^read_events_y(\d{4})m(\d{2})$/.exec(relname);
        if (!match) continue;
        const monthEnd = new Date(Date.UTC(Number(match[1]), Number(match[2]), 1));
        if (monthEnd.getTime() > cutoffMonth.getTime()) continue;
        const counted = await this.prisma.$queryRawUnsafe<{ count: bigint }[]>(
          `SELECT count(*)::bigint AS count FROM "${relname}"`,
        );
        dropped += Number(counted[0]?.count ?? 0n);
        await this.prisma.$executeRawUnsafe(`DROP TABLE "${relname}"`);
      }
    }

    // The remainder: rows before the cutoff inside surviving partitions,
    // the default partition, or a plain (test) table.
    const deleted = await this.prisma.readEvent.deleteMany({
      where: { occurredAt: { lt: cutoff } },
    });
    const count = dropped + deleted.count;
    if (count > 0) {
      await this.audit.record({
        action: 'read_trail.pruned',
        details: { count, cutoff: cutoff.toISOString(), retentionDays },
      });
    }
    return count;
  }
}