stack/apps/gateway/src/gc/session-gc.service.ts

import { Inject, Injectable, Logger, type OnModuleInit } from '@nestjs/common';
import type { QueueHandle } from '@mosaicstack/queue';
import type { LogService } from '@mosaicstack/log';
import { LOG_SERVICE } from '../log/log.tokens.js';
import { REDIS } from './gc.tokens.js';

export interface GCResult {
  sessionId: string;
  cleaned: {
    valkeyKeys?: number;
    logsDemoted?: number;
    tempFilesRemoved?: number;
  };
}

export interface GCSweepResult {
  orphanedSessions: number;
  totalCleaned: GCResult[];
  duration: number;
}

export interface FullGCResult {
  valkeyKeys: number;
  logsDemoted: number;
  jobsPurged: number;
  tempFilesRemoved: number;
  duration: number;
}

@Injectable()
export class SessionGCService implements OnModuleInit {
  private readonly logger = new Logger(SessionGCService.name);

  constructor(
    @Inject(REDIS) private readonly redis: QueueHandle['redis'],
    @Inject(LOG_SERVICE) private readonly logService: LogService,
  ) {}

  onModuleInit(): void {
    // Fire-and-forget: run full GC asynchronously so it does not block the
    // NestJS bootstrap chain.  Cold-start GC typically takes 100–500 ms
    // depending on Valkey key count; deferring it removes that latency from
    // the TTFB of the first HTTP request.
    this.fullCollect()
      .then((result) => {
        this.logger.log(
          `Full GC complete: ${result.valkeyKeys} Valkey keys, ` +
            `${result.logsDemoted} logs demoted, ` +
            `${result.jobsPurged} jobs purged, ` +
            `${result.tempFilesRemoved} temp dirs removed ` +
            `(${result.duration}ms)`,
        );
      })
      .catch((err: unknown) => {
        this.logger.error('Cold-start GC failed', err instanceof Error ? err.stack : String(err));
      });
  }

  /**
   * Scan Valkey for all keys matching a pattern using SCAN (non-blocking).
   * KEYS is avoided because it blocks the Valkey event loop for the full scan
   * duration, which can cause latency spikes under production key volumes.
   */
  private async scanKeys(pattern: string): Promise<string[]> {
    const collected: string[] = [];
    let cursor = '0';
    do {
      const [nextCursor, keys] = await this.redis.scan(cursor, 'MATCH', pattern, 'COUNT', 100);
      cursor = nextCursor;
      collected.push(...keys);
    } while (cursor !== '0');
    return collected;
  }

  /**
   * Immediate cleanup for a single session (call from destroySession).
   */
  async collect(sessionId: string): Promise<GCResult> {
    const result: GCResult = { sessionId, cleaned: {} };

    // 1. Valkey: delete all session-scoped keys
    const pattern = `mosaic:session:${sessionId}:*`;
    const valkeyKeys = await this.scanKeys(pattern);
    if (valkeyKeys.length > 0) {
      await this.redis.del(...valkeyKeys);
      result.cleaned.valkeyKeys = valkeyKeys.length;
    }

    // 2. PG: demote hot-tier agent_logs for this session to warm
    const cutoff = new Date(); // demote all hot logs for this session
    const logsDemoted = await this.logService.logs.promoteToWarm(cutoff);
    if (logsDemoted > 0) {
      result.cleaned.logsDemoted = logsDemoted;
    }

    return result;
  }

  /**
   * Sweep GC — find orphaned artifacts from dead sessions.
   * System-wide operation: only call from admin-authorized paths or internal
   * scheduled jobs.  Individual session cleanup is handled by collect().
   */
  async sweepOrphans(): Promise<GCSweepResult> {
    const start = Date.now();
    const cleaned: GCResult[] = [];

    // 1. Find all session-scoped Valkey keys (non-blocking SCAN)
    const allSessionKeys = await this.scanKeys('mosaic:session:*');

    // Extract unique session IDs from keys
    const sessionIds = new Set<string>();
    for (const key of allSessionKeys) {
      const match = key.match(/^mosaic:session:([^:]+):/);
      if (match) sessionIds.add(match[1]!);
    }

    // 2. For each session ID, collect stale keys
    for (const sessionId of sessionIds) {
      const gcResult = await this.collect(sessionId);
      if (Object.keys(gcResult.cleaned).length > 0) {
        cleaned.push(gcResult);
      }
    }

    return {
      orphanedSessions: cleaned.length,
      totalCleaned: cleaned,
      duration: Date.now() - start,
    };
  }

  /**
   * Full GC — aggressive collection for cold start.
   * Assumes no sessions survived the restart.
   */
  async fullCollect(): Promise<FullGCResult> {
    const start = Date.now();

    // 1. Valkey: delete ALL session-scoped keys (non-blocking SCAN)
    const sessionKeys = await this.scanKeys('mosaic:session:*');
    if (sessionKeys.length > 0) {
      await this.redis.del(...sessionKeys);
    }

    // 2. NOTE: channel keys are NOT collected on cold start
    //    (discord/telegram plugins may reconnect and resume)

    // 3. PG: demote stale hot-tier logs older than 24h to warm
    const hotCutoff = new Date(Date.now() - 24 * 60 * 60 * 1000);
    const logsDemoted = await this.logService.logs.promoteToWarm(hotCutoff);

    // 4. No summarization job purge API available yet
    const jobsPurged = 0;

    return {
      valkeyKeys: sessionKeys.length,
      logsDemoted,
      jobsPurged,
      tempFilesRemoved: 0,
      duration: Date.now() - start,
    };
  }
}