mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-18 21:16:47 +00:00
refactor(server): move the cache sweeps, idempotency purge and airtrail poll in-container
This commit is contained in:
@@ -9,8 +9,6 @@ import fs from 'node:fs';
|
||||
import http from 'node:http';
|
||||
import type { INestApplication } from '@nestjs/common';
|
||||
import { buildApp } from './bootstrap';
|
||||
import { PlacePhotoCacheService } from './nest/place-photos/place-photo-cache.service';
|
||||
import { AirtrailSyncService } from './nest/integrations/airtrail-sync.service';
|
||||
|
||||
// Create upload and data directories on startup.
|
||||
// Every uploads subdir the app writes to must be listed here (#1762): a dir
|
||||
@@ -89,22 +87,6 @@ const onListen = () => {
|
||||
if (env.demo.enabled && env.app.isProduction) {
|
||||
sLogWarn('SECURITY WARNING: DEMO_MODE is enabled in production!');
|
||||
}
|
||||
// The scheduler runs outside the container but needs things inside it. It is
|
||||
// handed them here, from the app buildApp() already initialised, rather than
|
||||
// reaching for src/services at call time.
|
||||
scheduler.setSchedulerDeps({
|
||||
// The container singleton, not a fresh instance: the in-flight dedup and the
|
||||
// known-on-disk set only work if the whole process shares one.
|
||||
placePhotos: nestApp.get(PlacePhotoCacheService),
|
||||
// The last lazy require() in the cron path. It used to fail inside the tick,
|
||||
// logged as "AirTrail sync tick failed", where a boot-time wiring mistake
|
||||
// belongs at boot.
|
||||
airtrail: nestApp.get(AirtrailSyncService),
|
||||
});
|
||||
scheduler.startIdempotencyCleanup();
|
||||
scheduler.startTrekPhotoCacheCleanup();
|
||||
scheduler.startPlacePhotoCacheCleanup();
|
||||
scheduler.startAirTrailSync();
|
||||
import('./websocket').then(({ setupWebSocket }) => {
|
||||
setupWebSocket(server);
|
||||
});
|
||||
|
||||
@@ -60,6 +60,7 @@ import { trekMcpAccessPolicy, trekMcpValidateAccess } from '../mcp/nest-mcp-poli
|
||||
import { TrekExceptionFilter } from './common/trek-exception.filter';
|
||||
import { SpaFallbackFilter } from './platform/spa-fallback.filter';
|
||||
import { IdempotencyInterceptor } from './common/idempotency.interceptor';
|
||||
import { IdempotencyCleanupJob } from './common/idempotency-cleanup.job';
|
||||
|
||||
/**
|
||||
* Root NestJS module for the incremental migration. Domain modules
|
||||
@@ -87,6 +88,8 @@ import { IdempotencyInterceptor } from './common/idempotency.interceptor';
|
||||
// Replays the X-Idempotency-Key the client sends on every write, so retried
|
||||
// mutations don't double-apply.
|
||||
{ provide: APP_INTERCEPTOR, useClass: IdempotencyInterceptor },
|
||||
// Its nightly TTL purge — a provider here because common/ has no module.
|
||||
IdempotencyCleanupJob,
|
||||
// Global Zod validation: any parameter typed with a createZodDto class
|
||||
// (the <domain>.dto.ts wrappers over @trek/shared schemas) is validated;
|
||||
// everything else passes through untouched. Paired with the boot gate in
|
||||
|
||||
@@ -0,0 +1,34 @@
|
||||
import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
|
||||
import { logInfo, logError } from '../audit/audit-log.logger';
|
||||
import { DatabaseService } from '../database/database.service';
|
||||
import { CronRegistrarService } from '../scheduling/cron-registrar.service';
|
||||
import { purgeExpiredIdempotencyKeys } from './idempotency-cleanup';
|
||||
|
||||
/**
|
||||
* Nightly 3 AM purge of expired idempotency keys (moved from src/scheduler.ts).
|
||||
* Registered as an AppModule provider beside the global IdempotencyInterceptor
|
||||
* that writes the keys — common/ has no module of its own.
|
||||
*/
|
||||
@Injectable()
|
||||
export class IdempotencyCleanupJob implements OnApplicationBootstrap {
|
||||
constructor(
|
||||
private readonly db: DatabaseService,
|
||||
private readonly registrar: CronRegistrarService,
|
||||
) {}
|
||||
|
||||
onApplicationBootstrap(): void {
|
||||
if (!this.registrar.isEnabled()) return;
|
||||
this.registrar.register('idempotency-cleanup', '0 3 * * *', () => this.tick());
|
||||
}
|
||||
|
||||
tick(): void {
|
||||
try {
|
||||
const removed = purgeExpiredIdempotencyKeys(undefined, undefined, this.db);
|
||||
if (removed > 0) {
|
||||
logInfo(`Idempotency cleanup: removed ${removed} expired key(s)`);
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
logError(`Idempotency cleanup: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
import { readEnv } from '../../app-config';
|
||||
|
||||
/**
|
||||
* Idempotency-key TTL purge (moved from src/scheduler.ts; the interceptor that
|
||||
* writes the keys lives next door in idempotency.interceptor.ts).
|
||||
*
|
||||
* The TTL must exceed any realistic offline window: the TREK client replays
|
||||
* queued mutations with their X-Idempotency-Key when it reconnects, so a key
|
||||
* GC'd before the device comes back online would let the replay create a
|
||||
* duplicate. 24h was far too short for a multi-day offline trip; default 30d,
|
||||
* overridable via IDEMPOTENCY_TTL_SECONDS (default lives in app-config).
|
||||
*/
|
||||
|
||||
export interface PurgeDb {
|
||||
prepare(sql: string): { run(...args: unknown[]): { changes: number } };
|
||||
}
|
||||
|
||||
/** Delete idempotency keys older than the configured TTL. Returns rows removed.
|
||||
* The db is a required parameter now (DatabaseService satisfies PurgeDb
|
||||
* structurally) — the old lazy-require default died with the scheduler. */
|
||||
export function purgeExpiredIdempotencyKeys(
|
||||
now: number = Date.now(),
|
||||
ttlSeconds: number = readEnv().session.idempotencyTtlSeconds,
|
||||
database: PurgeDb,
|
||||
): number {
|
||||
const cutoff = Math.floor(now / 1000) - ttlSeconds;
|
||||
const result = database.prepare('DELETE FROM idempotency_keys WHERE created_at < ?').run(cutoff);
|
||||
return result.changes;
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
|
||||
import { logInfo, logError } from '../audit/audit-log.logger';
|
||||
import { DatabaseService } from '../database/database.service';
|
||||
import { CronRegistrarService } from '../scheduling/cron-registrar.service';
|
||||
import { AirtrailSyncService } from './airtrail-sync.service';
|
||||
|
||||
/**
|
||||
* AirTrail sync: poll connected instances on an interval and reconcile linked
|
||||
* flights both ways (#214). Moved from src/scheduler.ts. The per-tick enable
|
||||
* gate (addon + setting) lives in runAirtrailSync, so toggling the addon takes
|
||||
* effect without a restart; the poll interval is read once at bootstrap
|
||||
* (parity — changing it still needs a restart).
|
||||
*/
|
||||
@Injectable()
|
||||
export class AirtrailSyncJob implements OnApplicationBootstrap {
|
||||
constructor(
|
||||
private readonly db: DatabaseService,
|
||||
private readonly airtrail: AirtrailSyncService,
|
||||
private readonly registrar: CronRegistrarService,
|
||||
) {}
|
||||
|
||||
onApplicationBootstrap(): void {
|
||||
if (!this.registrar.isEnabled()) return;
|
||||
const value = this.db.get<{ value: string }>('SELECT value FROM app_settings WHERE key = ?', 'airtrail_poll_interval_minutes')?.value;
|
||||
const raw = parseInt(value || '5', 10);
|
||||
const minutes = Number.isFinite(raw) && raw >= 1 && raw <= 59 ? raw : 5;
|
||||
logInfo(`AirTrail sync: scheduled every ${minutes}m`);
|
||||
this.registrar.register('airtrail-sync', `*/${minutes} * * * *`, () => this.tick());
|
||||
}
|
||||
|
||||
async tick(): Promise<void> {
|
||||
try {
|
||||
await this.airtrail.runAirtrailSync();
|
||||
} catch (err: unknown) {
|
||||
logError(`AirTrail sync tick failed: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,9 @@ import { AirtrailController } from './airtrail.controller';
|
||||
import { AirtrailClient } from './airtrail.client';
|
||||
import { AirtrailService } from './airtrail.service';
|
||||
import { AirtrailSyncService } from './airtrail-sync.service';
|
||||
import { AirtrailSyncJob } from './airtrail-sync.job';
|
||||
import { AirtrailImportService } from './airtrail-import.service';
|
||||
import { SchedulingModule } from '../scheduling/scheduling.module';
|
||||
import { PermissionsModule } from '../permissions/permissions.module';
|
||||
import { AddonsModule } from '../addons/addons.module';
|
||||
import { AuditModule } from '../audit/audit.module';
|
||||
@@ -24,9 +26,9 @@ import { ReservationsModule } from '../reservations/reservations.module';
|
||||
* seam on purpose - see airtrail.bridge.ts.
|
||||
*/
|
||||
@Module({
|
||||
imports: [PermissionsModule, AddonsModule, AuditModule, ReservationsModule],
|
||||
imports: [PermissionsModule, AddonsModule, AuditModule, ReservationsModule, SchedulingModule],
|
||||
controllers: [AirtrailController],
|
||||
providers: [AirtrailClient, AirtrailService, AirtrailSyncService, AirtrailImportService],
|
||||
providers: [AirtrailClient, AirtrailService, AirtrailSyncService, AirtrailSyncJob, AirtrailImportService],
|
||||
exports: [AirtrailSyncService, AirtrailImportService],
|
||||
})
|
||||
export class AirtrailModule {}
|
||||
|
||||
@@ -7,6 +7,8 @@ import { UnifiedMemoriesService } from './unified-memories.service';
|
||||
import { PhotoResolverService } from './photo-resolver.service';
|
||||
import { ThumbnailService } from './thumbnail.service';
|
||||
import { TrekPhotoCacheService } from './trek-photo-cache.service';
|
||||
import { TrekPhotoCacheJob } from './trek-photo-cache.job';
|
||||
import { SchedulingModule } from '../scheduling/scheduling.module';
|
||||
import { UnifiedMemoriesController } from './unified.controller';
|
||||
import { ImmichMemoriesController } from './immich.controller';
|
||||
import { SynologyMemoriesController } from './synology.controller';
|
||||
@@ -41,7 +43,7 @@ import { NotificationsModule } from '../notifications/notifications.module';
|
||||
* adapters themselves so the set is readable in one place.
|
||||
*/
|
||||
@Module({
|
||||
imports: [NotificationsModule, AddonsModule, AuditModule, TrekPhotosModule, RealtimeModule],
|
||||
imports: [NotificationsModule, AddonsModule, AuditModule, TrekPhotosModule, RealtimeModule, SchedulingModule],
|
||||
controllers: [UnifiedMemoriesController, ImmichMemoriesController, SynologyMemoriesController],
|
||||
providers: [
|
||||
MemoriesService,
|
||||
@@ -52,6 +54,7 @@ import { NotificationsModule } from '../notifications/notifications.module';
|
||||
PhotoResolverService,
|
||||
ThumbnailService,
|
||||
TrekPhotoCacheService,
|
||||
TrekPhotoCacheJob,
|
||||
ImmichPhotoProvider,
|
||||
SynologyPhotoProvider,
|
||||
PhotoProviderRegistry,
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
|
||||
import { logError } from '../audit/audit-log.logger';
|
||||
import { CronRegistrarService } from '../scheduling/cron-registrar.service';
|
||||
import { TrekPhotoCacheService } from './trek-photo-cache.service';
|
||||
|
||||
/**
|
||||
* Trek photo cache cleanup: every 2 hours — evict disk files and DB rows past
|
||||
* their 1h TTL (moved from src/scheduler.ts, which used to construct its own
|
||||
* TrekPhotoCacheService per sweep; this injects the container singleton).
|
||||
* Runs in the server-local zone — the old cron passed no timezone.
|
||||
*/
|
||||
@Injectable()
|
||||
export class TrekPhotoCacheJob implements OnApplicationBootstrap {
|
||||
constructor(
|
||||
private readonly cache: TrekPhotoCacheService,
|
||||
private readonly registrar: CronRegistrarService,
|
||||
) {}
|
||||
|
||||
onApplicationBootstrap(): void {
|
||||
if (!this.registrar.isEnabled()) return;
|
||||
// Run once immediately on startup to evict any entries left over from a previous run
|
||||
try {
|
||||
this.cache.sweepExpired();
|
||||
} catch { /* cache dir may not exist yet — harmless */ }
|
||||
this.registrar.register('trek-photo-cache', '0 */2 * * *', () => this.tick(), { timezone: 'none' });
|
||||
}
|
||||
|
||||
tick(): void {
|
||||
try {
|
||||
this.cache.sweepExpired();
|
||||
} catch (err: unknown) {
|
||||
logError(`Trek photo cache cleanup: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -14,11 +14,12 @@ export const CACHE_TTL = 60 * 60 * 1000; // 1 hour
|
||||
* In-flight fetches, module-scoped on purpose.
|
||||
*
|
||||
* This is the stampede guard: ten tiles asking for the same uncached asset must
|
||||
* share one upstream fetch. The scheduler drives sweepExpired through a lazy
|
||||
* require outside the container, so a second instance of this service exists by
|
||||
* design - a per-instance Map would hand each of them a private one and the
|
||||
* guard would be silently gone. Same reasoning as oauth/oauth.pending-codes.ts
|
||||
* and the notification channel registry.
|
||||
* share one upstream fetch. The sweep cron injects the container singleton now
|
||||
* (TrekPhotoCacheJob), but the module scope keeps the guard whole even if a
|
||||
* second instance of this service ever exists again — a per-instance Map would
|
||||
* hand each of them a private one and the guard would be silently gone. Same
|
||||
* reasoning as oauth/oauth.pending-codes.ts and the notification channel
|
||||
* registry.
|
||||
*/
|
||||
const inFlight = new Map<string, Promise<Buffer | null>>();
|
||||
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
|
||||
import { logInfo, logError } from '../audit/audit-log.logger';
|
||||
import { CronRegistrarService } from '../scheduling/cron-registrar.service';
|
||||
import { PlacePhotoCacheService } from './place-photo-cache.service';
|
||||
|
||||
/**
|
||||
* Place-photo (Google/Wikimedia) cache cleanup: nightly — reclaim cached files
|
||||
* and meta rows no place references anymore (deleted places/trips, overwritten
|
||||
* image_url). Moved from src/scheduler.ts; injects the container singleton —
|
||||
* the in-flight dedup and known-on-disk set only work if the whole process
|
||||
* shares one.
|
||||
*/
|
||||
@Injectable()
|
||||
export class PlacePhotoCacheJob implements OnApplicationBootstrap {
|
||||
constructor(
|
||||
private readonly cache: PlacePhotoCacheService,
|
||||
private readonly registrar: CronRegistrarService,
|
||||
) {}
|
||||
|
||||
onApplicationBootstrap(): void {
|
||||
if (!this.registrar.isEnabled()) return;
|
||||
// Run once on startup to reclaim orphans left over from before this sweeper existed.
|
||||
this.sweep();
|
||||
this.registrar.register('place-photo-cache', '30 3 * * *', () => this.sweep());
|
||||
}
|
||||
|
||||
sweep(): void {
|
||||
try {
|
||||
const removed = this.cache.sweepOrphans();
|
||||
if (removed > 0) logInfo(`Place-photo cache cleanup: removed ${removed} orphaned file(s)/row(s)`);
|
||||
} catch (err: unknown) {
|
||||
logError(`Place-photo cache cleanup: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -42,8 +42,8 @@ interface CachedPhoto {
|
||||
* three in-memory structures only do their job if there is exactly ONE of them.
|
||||
* A second instance would give the stampede guard its own inFlight map, so two
|
||||
* simultaneous requests for the same uncached place would both call the
|
||||
* provider. That is the reason the scheduler takes this from the container
|
||||
* rather than constructing it: see SchedulerDeps.
|
||||
* provider. That is the reason the nightly sweep (PlacePhotoCacheJob) injects
|
||||
* this container singleton rather than constructing its own.
|
||||
*/
|
||||
@Injectable()
|
||||
export class PlacePhotoCacheService implements OnModuleInit {
|
||||
|
||||
@@ -1,16 +1,18 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { PlacePhotoCacheService } from './place-photo-cache.service';
|
||||
import { PlacePhotoCacheJob } from './place-photo-cache.job';
|
||||
import { AppConfigModule } from '../app-config/app-config.module';
|
||||
import { SchedulingModule } from '../scheduling/scheduling.module';
|
||||
|
||||
/** The marker-photo cache. No controller of its own — maps serves the bytes,
|
||||
* places and share read through it, and the scheduler sweeps it nightly.
|
||||
* places and share read through it, and PlacePhotoCacheJob sweeps it nightly.
|
||||
*
|
||||
* Deliberately NOT @Global (permissions precedent), and AppConfigModule is
|
||||
* imported explicitly because @Global only reaches modules that are in the
|
||||
* graph — which a single-domain e2e TestingModule is not. */
|
||||
@Module({
|
||||
imports: [AppConfigModule],
|
||||
providers: [PlacePhotoCacheService],
|
||||
imports: [AppConfigModule, SchedulingModule],
|
||||
providers: [PlacePhotoCacheService, PlacePhotoCacheJob],
|
||||
exports: [PlacePhotoCacheService],
|
||||
})
|
||||
export class PlacePhotosModule {}
|
||||
|
||||
+13
-167
@@ -1,174 +1,20 @@
|
||||
import cron, { type ScheduledTask } from 'node-cron';
|
||||
import { readEnv } from './app-config';
|
||||
import { logInfo, logError } from './nest/audit/audit-log.logger';
|
||||
|
||||
/**
|
||||
* What the scheduler needs from the Nest container.
|
||||
* Retired. Every cron moved into its owning Nest domain, scheduled through
|
||||
* CronRegistrarService (src/nest/scheduling/):
|
||||
*
|
||||
* The shapes are structural on purpose: the scheduler names the capability it
|
||||
* wants, never the provider class — importing a provider class here would pull
|
||||
* its module graph into a file that must stay loadable without a container
|
||||
* (which is what the unit tests rely on).
|
||||
* auto-backup → nest/backup/auto-backup.job.ts
|
||||
* trip/todo reminders → nest/notifications/reminder-jobs.service.ts
|
||||
* version check → nest/admin/version-check.job.ts
|
||||
* demo reset → nest/admin/demo-reset.job.ts
|
||||
* idempotency purge → nest/common/idempotency-cleanup.job.ts
|
||||
* trek-photo sweep → nest/memories/trek-photo-cache.job.ts
|
||||
* place-photo sweep → nest/place-photos/place-photo-cache.job.ts
|
||||
* airtrail poll → nest/integrations/airtrail-sync.job.ts
|
||||
*
|
||||
* index.ts fills this in from the container after buildApp(), the way
|
||||
* bootstrap.ts hands the /mcp handler its registry. Nothing here is resolved at
|
||||
* import time.
|
||||
* Only this shutdown no-op remains for index.ts; the file is deleted next.
|
||||
*/
|
||||
export interface SchedulerDeps {
|
||||
placePhotos: { sweepOrphans(): number };
|
||||
airtrail: { runAirtrailSync(): Promise<void> };
|
||||
}
|
||||
|
||||
let deps: SchedulerDeps | undefined;
|
||||
|
||||
/** Hand the scheduler its container-resolved dependencies. Call after app.init(). */
|
||||
export function setSchedulerDeps(next: SchedulerDeps): void {
|
||||
deps = next;
|
||||
}
|
||||
|
||||
// Idempotency key cleanup: nightly at 3 AM — delete keys past their TTL.
|
||||
// The TTL must exceed any realistic offline window: the TREK client replays
|
||||
// queued mutations with their X-Idempotency-Key when it reconnects, so a key
|
||||
// GC'd before the device comes back online would let the replay create a
|
||||
// duplicate. 24h was far too short for a multi-day offline trip; default 30d,
|
||||
// overridable via IDEMPOTENCY_TTL_SECONDS (default lives in app-config).
|
||||
let idempotencyCleanupTask: ScheduledTask | null = null;
|
||||
|
||||
function idempotencyTtlSeconds(): number {
|
||||
return readEnv().session.idempotencyTtlSeconds;
|
||||
}
|
||||
|
||||
interface PurgeDb {
|
||||
prepare(sql: string): { run(...args: unknown[]): { changes: number } };
|
||||
}
|
||||
|
||||
/** Delete idempotency keys older than the configured TTL. Returns rows removed.
|
||||
* The db is injectable for testing; the cron job uses the default. */
|
||||
function purgeExpiredIdempotencyKeys(
|
||||
now: number = Date.now(),
|
||||
ttlSeconds: number = idempotencyTtlSeconds(),
|
||||
database: PurgeDb = require('./db/database').db,
|
||||
): number {
|
||||
const cutoff = Math.floor(now / 1000) - ttlSeconds;
|
||||
const result = database.prepare('DELETE FROM idempotency_keys WHERE created_at < ?').run(cutoff);
|
||||
return result.changes;
|
||||
}
|
||||
|
||||
function startIdempotencyCleanup(): void {
|
||||
if (idempotencyCleanupTask) { idempotencyCleanupTask.stop(); idempotencyCleanupTask = null; }
|
||||
|
||||
const tz = readEnv().app.tz || 'UTC';
|
||||
idempotencyCleanupTask = cron.schedule('0 3 * * *', () => {
|
||||
try {
|
||||
const removed = purgeExpiredIdempotencyKeys();
|
||||
if (removed > 0) {
|
||||
logInfo(`Idempotency cleanup: removed ${removed} expired key(s)`);
|
||||
}
|
||||
} catch (err: unknown) {
|
||||
logError(`Idempotency cleanup: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}, { timezone: tz });
|
||||
}
|
||||
|
||||
// Trek photo cache cleanup: every 2 hours — evict disk files and DB rows past their 1h TTL
|
||||
/**
|
||||
* The cache sweep, reached from outside the container.
|
||||
*
|
||||
* Constructed per call rather than injected: the cron runs before/independently
|
||||
* of any request, and the service holds no state of its own — the stampede
|
||||
* guard it does own is a module-scoped Map, deliberately shared with the
|
||||
* container instance. Required lazily so a boot without the nest graph (some
|
||||
* tests) does not pull it in.
|
||||
*/
|
||||
function sweepTrekPhotoCache(): void {
|
||||
// eslint-disable-next-line @typescript-eslint/no-require-imports
|
||||
const { TrekPhotoCacheService } = require('./nest/memories/trek-photo-cache.service');
|
||||
// eslint-disable-next-line @typescript-eslint/no-require-imports
|
||||
const { DatabaseService } = require('./nest/database/database.service');
|
||||
// eslint-disable-next-line @typescript-eslint/no-require-imports
|
||||
const { db } = require('./db/database');
|
||||
new TrekPhotoCacheService(new DatabaseService(db)).sweepExpired();
|
||||
}
|
||||
|
||||
let trekPhotoCacheTask: ScheduledTask | null = null;
|
||||
|
||||
function startTrekPhotoCacheCleanup(): void {
|
||||
if (trekPhotoCacheTask) { trekPhotoCacheTask.stop(); trekPhotoCacheTask = null; }
|
||||
|
||||
// Run once immediately on startup to evict any entries left over from a previous run
|
||||
try {
|
||||
sweepTrekPhotoCache();
|
||||
} catch { /* cache dir may not exist yet — harmless */ }
|
||||
|
||||
trekPhotoCacheTask = cron.schedule('0 */2 * * *', () => {
|
||||
try {
|
||||
sweepTrekPhotoCache();
|
||||
} catch (err: unknown) {
|
||||
logError(`Trek photo cache cleanup: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// Place-photo (Google/Wikimedia) cache cleanup: nightly — reclaim cached files and
|
||||
// meta rows no place references anymore (deleted places/trips, overwritten image_url).
|
||||
let placePhotoCacheTask: ScheduledTask | null = null;
|
||||
|
||||
function startPlacePhotoCacheCleanup(): void {
|
||||
if (placePhotoCacheTask) { placePhotoCacheTask.stop(); placePhotoCacheTask = null; }
|
||||
|
||||
const sweep = () => {
|
||||
try {
|
||||
if (!deps) {
|
||||
logError('Place-photo cache cleanup: skipped, the scheduler was started without its container dependencies');
|
||||
return;
|
||||
}
|
||||
const removed = deps.placePhotos.sweepOrphans();
|
||||
if (removed > 0) logInfo(`Place-photo cache cleanup: removed ${removed} orphaned file(s)/row(s)`);
|
||||
} catch (err: unknown) {
|
||||
logError(`Place-photo cache cleanup: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
};
|
||||
|
||||
// Run once on startup to reclaim orphans left over from before this sweeper existed.
|
||||
sweep();
|
||||
|
||||
const tz = readEnv().app.tz || 'UTC';
|
||||
placePhotoCacheTask = cron.schedule('30 3 * * *', sweep, { timezone: tz });
|
||||
}
|
||||
|
||||
// AirTrail sync: poll connected instances on an interval and reconcile linked
|
||||
// flights both ways (#214). The per-tick enable gate (addon + setting) lives in
|
||||
// runAirtrailSync, so toggling the addon takes effect without a restart.
|
||||
let airtrailSyncTask: ScheduledTask | null = null;
|
||||
|
||||
function startAirTrailSync(): void {
|
||||
if (airtrailSyncTask) { airtrailSyncTask.stop(); airtrailSyncTask = null; }
|
||||
|
||||
const { db } = require('./db/database');
|
||||
const getSetting = (key: string) => (db.prepare('SELECT value FROM app_settings WHERE key = ?').get(key) as { value: string } | undefined)?.value;
|
||||
const raw = parseInt(getSetting('airtrail_poll_interval_minutes') || '5', 10);
|
||||
const minutes = Number.isFinite(raw) && raw >= 1 && raw <= 59 ? raw : 5;
|
||||
const tz = readEnv().app.tz || 'UTC';
|
||||
logInfo(`AirTrail sync: scheduled every ${minutes}m`);
|
||||
|
||||
airtrailSyncTask = cron.schedule(`*/${minutes} * * * *`, async () => {
|
||||
try {
|
||||
if (!deps) {
|
||||
logError('AirTrail sync: skipped, the scheduler was started without its container dependencies');
|
||||
return;
|
||||
}
|
||||
await deps.airtrail.runAirtrailSync();
|
||||
} catch (err: unknown) {
|
||||
logError(`AirTrail sync tick failed: ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}, { timezone: tz });
|
||||
}
|
||||
|
||||
function stop(): void {
|
||||
if (idempotencyCleanupTask) { idempotencyCleanupTask.stop(); idempotencyCleanupTask = null; }
|
||||
if (trekPhotoCacheTask) { trekPhotoCacheTask.stop(); trekPhotoCacheTask = null; }
|
||||
if (placePhotoCacheTask) { placePhotoCacheTask.stop(); placePhotoCacheTask = null; }
|
||||
if (airtrailSyncTask) { airtrailSyncTask.stop(); airtrailSyncTask = null; }
|
||||
// Nest owns every job now — nestApp.close() stops them via the registrar.
|
||||
}
|
||||
|
||||
export { stop, startIdempotencyCleanup, purgeExpiredIdempotencyKeys, startTrekPhotoCacheCleanup, startPlacePhotoCacheCleanup, startAirTrailSync };
|
||||
export { stop };
|
||||
|
||||
@@ -6,9 +6,12 @@
|
||||
* offline window — otherwise a key GC'd before the device returns lets the
|
||||
* replay create a duplicate. The TTL was raised from 24h to 30d (overridable).
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
|
||||
import { db } from '../../src/db/database';
|
||||
import { purgeExpiredIdempotencyKeys } from '../../src/scheduler';
|
||||
import { purgeExpiredIdempotencyKeys } from '../../src/nest/common/idempotency-cleanup';
|
||||
import { IdempotencyCleanupJob } from '../../src/nest/common/idempotency-cleanup.job';
|
||||
import { DatabaseService } from '../../src/nest/database/database.service';
|
||||
import type { CronRegistrarService } from '../../src/nest/scheduling/cron-registrar.service';
|
||||
|
||||
const DAY = 24 * 60 * 60;
|
||||
const NOW = 2_000_000_000_000; // fixed ms so the test is deterministic
|
||||
@@ -56,3 +59,49 @@ describe('purgeExpiredIdempotencyKeys', () => {
|
||||
expect(purgeExpiredIdempotencyKeys(NOW, undefined, db)).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('IdempotencyCleanupJob', () => {
|
||||
function makeJob(enabled = true) {
|
||||
const registrar = {
|
||||
isEnabled: vi.fn(() => enabled),
|
||||
register: vi.fn(() => enabled),
|
||||
unregister: vi.fn(),
|
||||
};
|
||||
const job = new IdempotencyCleanupJob(new DatabaseService(db), registrar as unknown as CronRegistrarService);
|
||||
return { job, registrar };
|
||||
}
|
||||
|
||||
it('registers the nightly 3 AM cron, and stays out of the registry under the test gate', () => {
|
||||
const on = makeJob();
|
||||
on.job.onApplicationBootstrap();
|
||||
expect(on.registrar.register).toHaveBeenCalledWith('idempotency-cleanup', '0 3 * * *', expect.any(Function));
|
||||
|
||||
const off = makeJob(false);
|
||||
off.job.onApplicationBootstrap();
|
||||
expect(off.registrar.register).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('the tick purges through the injected DatabaseService', () => {
|
||||
// The tick uses the live clock, so these fixtures age against Date.now()
|
||||
// (the pure-function cases above pin their own fixed NOW instead).
|
||||
const liveNowSec = Math.floor(Date.now() / 1000);
|
||||
const liveInsert = (key: string, ageSeconds: number) =>
|
||||
db.prepare(
|
||||
`INSERT INTO idempotency_keys (key, user_id, method, path, status_code, response_body, created_at)
|
||||
VALUES (?, 1, 'POST', '/x', 200, '{}', ?)`,
|
||||
).run(key, liveNowSec - ageSeconds);
|
||||
liveInsert('old', 31 * DAY);
|
||||
liveInsert('fresh', 5 * DAY);
|
||||
|
||||
const { job } = makeJob();
|
||||
job.tick();
|
||||
const keys = db.prepare('SELECT key FROM idempotency_keys').all().map((r: { key: string }) => r.key);
|
||||
expect(keys).toEqual(['fresh']);
|
||||
});
|
||||
|
||||
it('a failing purge is contained to the Idempotency cleanup log line', () => {
|
||||
const broken = { prepare: () => { throw new Error('db gone'); } } as unknown as DatabaseService;
|
||||
const job = new IdempotencyCleanupJob(broken, { isEnabled: () => true } as unknown as CronRegistrarService);
|
||||
expect(() => job.tick()).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
/**
|
||||
* The cache-sweep and poll cron providers moved from src/scheduler.ts:
|
||||
* TrekPhotoCacheJob (2-hourly, server-local, boot sweep with a swallow),
|
||||
* PlacePhotoCacheJob (nightly app-tz, boot sweep, removed>0 log condition) and
|
||||
* AirtrailSyncJob (every N minutes, N from the clamped app-setting). The
|
||||
* services they drive have their own suites; these cover the scheduling shell.
|
||||
*/
|
||||
import { describe, it, expect, vi, beforeEach } from 'vitest';
|
||||
|
||||
const logMock = vi.hoisted(() => ({ LOG_LEVEL: 'error', logInfo: vi.fn(), logError: vi.fn(), logWarn: vi.fn(), logDebug: vi.fn() }));
|
||||
vi.mock('../../../src/nest/audit/audit-log.logger', () => logMock);
|
||||
|
||||
import { TrekPhotoCacheJob } from '../../../src/nest/memories/trek-photo-cache.job';
|
||||
import { PlacePhotoCacheJob } from '../../../src/nest/place-photos/place-photo-cache.job';
|
||||
import { AirtrailSyncJob } from '../../../src/nest/integrations/airtrail-sync.job';
|
||||
import type { TrekPhotoCacheService } from '../../../src/nest/memories/trek-photo-cache.service';
|
||||
import type { PlacePhotoCacheService } from '../../../src/nest/place-photos/place-photo-cache.service';
|
||||
import type { AirtrailSyncService } from '../../../src/nest/integrations/airtrail-sync.service';
|
||||
import type { DatabaseService } from '../../../src/nest/database/database.service';
|
||||
import type { CronRegistrarService } from '../../../src/nest/scheduling/cron-registrar.service';
|
||||
|
||||
function registrarStub(enabled = true) {
|
||||
return {
|
||||
isEnabled: vi.fn(() => enabled),
|
||||
register: vi.fn(() => enabled),
|
||||
unregister: vi.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
beforeEach(() => vi.clearAllMocks());
|
||||
|
||||
describe('TrekPhotoCacheJob', () => {
|
||||
function make(sweep: () => void = vi.fn(), enabled = true) {
|
||||
const registrar = registrarStub(enabled);
|
||||
const cache = { sweepExpired: vi.fn(sweep) };
|
||||
const job = new TrekPhotoCacheJob(cache as unknown as TrekPhotoCacheService, registrar as unknown as CronRegistrarService);
|
||||
return { job, registrar, cache };
|
||||
}
|
||||
|
||||
it('CSJOB-001 — boot sweep runs immediately, then the 2-hourly server-local cron registers', () => {
|
||||
const { job, registrar, cache } = make();
|
||||
job.onApplicationBootstrap();
|
||||
expect(cache.sweepExpired).toHaveBeenCalledTimes(1);
|
||||
expect(registrar.register).toHaveBeenCalledWith('trek-photo-cache', '0 */2 * * *', expect.any(Function), { timezone: 'none' });
|
||||
});
|
||||
|
||||
it('CSJOB-002 — a throwing boot sweep is swallowed (cache dir may not exist yet) and the cron still registers', () => {
|
||||
const { job, registrar } = make(() => { throw new Error('ENOENT'); });
|
||||
expect(() => job.onApplicationBootstrap()).not.toThrow();
|
||||
expect(registrar.register).toHaveBeenCalled();
|
||||
expect(logMock.logError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('CSJOB-003 — the test gate skips the boot sweep entirely', () => {
|
||||
const { job, registrar, cache } = make(vi.fn(), false);
|
||||
job.onApplicationBootstrap();
|
||||
expect(cache.sweepExpired).not.toHaveBeenCalled();
|
||||
expect(registrar.register).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('CSJOB-004 — a throwing tick is contained to the cleanup log line', () => {
|
||||
const { job } = make(() => { throw new Error('disk gone'); });
|
||||
expect(() => job.tick()).not.toThrow();
|
||||
expect(logMock.logError).toHaveBeenCalledWith('Trek photo cache cleanup: disk gone');
|
||||
});
|
||||
});
|
||||
|
||||
describe('PlacePhotoCacheJob', () => {
|
||||
function make(removed: number | (() => number) = 0, enabled = true) {
|
||||
const registrar = registrarStub(enabled);
|
||||
const cache = { sweepOrphans: vi.fn(typeof removed === 'function' ? removed : () => removed) };
|
||||
const job = new PlacePhotoCacheJob(cache as unknown as PlacePhotoCacheService, registrar as unknown as CronRegistrarService);
|
||||
return { job, registrar, cache };
|
||||
}
|
||||
|
||||
it('CSJOB-005 — boot sweep runs immediately, then the nightly app-tz cron registers', () => {
|
||||
const { job, registrar, cache } = make();
|
||||
job.onApplicationBootstrap();
|
||||
expect(cache.sweepOrphans).toHaveBeenCalledTimes(1);
|
||||
expect(registrar.register).toHaveBeenCalledWith('place-photo-cache', '30 3 * * *', expect.any(Function));
|
||||
});
|
||||
|
||||
it('CSJOB-006 — logs only when something was removed', () => {
|
||||
make(0).job.sweep();
|
||||
expect(logMock.logInfo).not.toHaveBeenCalled();
|
||||
make(3).job.sweep();
|
||||
expect(logMock.logInfo).toHaveBeenCalledWith('Place-photo cache cleanup: removed 3 orphaned file(s)/row(s)');
|
||||
});
|
||||
|
||||
it('CSJOB-007 — a throwing sweep is contained to the cleanup log line, and the gate skips the boot sweep', () => {
|
||||
const { job } = make(() => { throw new Error('fs down'); });
|
||||
expect(() => job.sweep()).not.toThrow();
|
||||
expect(logMock.logError).toHaveBeenCalledWith('Place-photo cache cleanup: fs down');
|
||||
|
||||
const gated = make(0, false);
|
||||
gated.job.onApplicationBootstrap();
|
||||
expect(gated.cache.sweepOrphans).not.toHaveBeenCalled();
|
||||
expect(gated.registrar.register).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('AirtrailSyncJob', () => {
|
||||
function make(intervalSetting: string | undefined, enabled = true) {
|
||||
const registrar = registrarStub(enabled);
|
||||
const db = {
|
||||
get: vi.fn(() => (intervalSetting === undefined ? undefined : { value: intervalSetting })),
|
||||
};
|
||||
const airtrail = { runAirtrailSync: vi.fn().mockResolvedValue(undefined) };
|
||||
const job = new AirtrailSyncJob(
|
||||
db as unknown as DatabaseService,
|
||||
airtrail as unknown as AirtrailSyncService,
|
||||
registrar as unknown as CronRegistrarService,
|
||||
);
|
||||
return { job, registrar, airtrail };
|
||||
}
|
||||
|
||||
it('CSJOB-008 — registers */N from the app-setting and logs the banner', () => {
|
||||
const { job, registrar } = make('10');
|
||||
job.onApplicationBootstrap();
|
||||
expect(logMock.logInfo).toHaveBeenCalledWith('AirTrail sync: scheduled every 10m');
|
||||
expect(registrar.register).toHaveBeenCalledWith('airtrail-sync', '*/10 * * * *', expect.any(Function));
|
||||
});
|
||||
|
||||
it('CSJOB-009 — clamps the interval to 1–59 and defaults to 5', () => {
|
||||
for (const [setting, minutes] of [
|
||||
[undefined, 5],
|
||||
['0', 5],
|
||||
['60', 5],
|
||||
['not-a-number', 5],
|
||||
['1', 1],
|
||||
['59', 59],
|
||||
] as const) {
|
||||
vi.clearAllMocks();
|
||||
const { job, registrar } = make(setting);
|
||||
job.onApplicationBootstrap();
|
||||
expect(registrar.register).toHaveBeenCalledWith('airtrail-sync', `*/${minutes} * * * *`, expect.any(Function));
|
||||
expect(logMock.logInfo).toHaveBeenCalledWith(`AirTrail sync: scheduled every ${minutes}m`);
|
||||
}
|
||||
});
|
||||
|
||||
it('CSJOB-010 — the gate skips registration and the banner', () => {
|
||||
const { job, registrar } = make('5', false);
|
||||
job.onApplicationBootstrap();
|
||||
expect(registrar.register).not.toHaveBeenCalled();
|
||||
expect(logMock.logInfo).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('CSJOB-011 — the tick delegates to runAirtrailSync and contains a rejection', async () => {
|
||||
const { job, airtrail } = make('5');
|
||||
await job.tick();
|
||||
expect(airtrail.runAirtrailSync).toHaveBeenCalledTimes(1);
|
||||
|
||||
airtrail.runAirtrailSync.mockRejectedValue(new Error('remote 500'));
|
||||
await expect(job.tick()).resolves.toBeUndefined();
|
||||
expect(logMock.logError).toHaveBeenCalledWith('AirTrail sync tick failed: remote 500');
|
||||
});
|
||||
});
|
||||
@@ -147,9 +147,10 @@ describe('the stampede guard', () => {
|
||||
expect(svc.getInFlight(key)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('CACHE-010: the map is shared across instances — the scheduler builds its own', () => {
|
||||
// Load-bearing: a per-instance map would give the cron and the container
|
||||
// separate guards and the dedup would be silently gone.
|
||||
it('CACHE-010: the map is shared across instances (module-scoped stampede guard)', () => {
|
||||
// Load-bearing: a per-instance map would hand any second instance a private
|
||||
// guard and the dedup would be silently gone. The sweep cron injects the
|
||||
// container singleton now, but the invariant stays pinned.
|
||||
const other = new TrekPhotoCacheService(new DatabaseService(testDb));
|
||||
const promise = Promise.resolve<Buffer | null>(null);
|
||||
svc.setInFlight('shared-key', promise);
|
||||
|
||||
Reference in New Issue
Block a user