diff --git a/src/db/migrations/025_create_distribution_schedules.sql b/src/db/migrations/025_create_distribution_schedules.sql new file mode 100644 index 00000000..975a0dc8 --- /dev/null +++ b/src/db/migrations/025_create_distribution_schedules.sql @@ -0,0 +1,43 @@ +-- Migration: Create distribution_schedules with cron column +-- Description: Persist per-offering deferred distribution cron-expression window +-- definitions so treasury operators can tune settlement cadence +-- without redeploys. Validated by CronWindowValidator before insert. +-- +-- Issue: #661 +-- +-- Columns +-- id UUID PK +-- offering_id UUID FK → offerings(id) UNIQUE +-- cron VARCHAR(100) NOT NULL +-- Standard 5-field cron expression (minute hour dom month dow). +-- timezone VARCHAR(100) NOT NULL DEFAULT 'UTC' +-- IANA timezone for wall-clock evaluation. +-- created_at / updated_at +-- +-- Note: offerings.cron_expression / offerings.distribution_timezone (migration 020) +-- remain as a denormalised mirror for scheduler joins. Application code should +-- write through OfferingRepository.updateCronSchedule which keeps both in sync +-- when a distribution_schedules row is present. +-- +-- DOWN Migration (manual rollback): +-- DROP TABLE IF EXISTS distribution_schedules; + +CREATE TABLE IF NOT EXISTS distribution_schedules ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + offering_id UUID NOT NULL REFERENCES offerings(id) ON DELETE CASCADE, + cron VARCHAR(100) NOT NULL, + timezone VARCHAR(100) NOT NULL DEFAULT 'UTC', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT uq_distribution_schedules_offering UNIQUE (offering_id), + CONSTRAINT chk_distribution_schedules_cron_nonempty CHECK (char_length(trim(cron)) > 0) +); + +CREATE INDEX IF NOT EXISTS idx_distribution_schedules_cron + ON distribution_schedules (cron); + +COMMENT ON TABLE distribution_schedules IS + 'Per-offering deferred distribution cron windows. Validated by CronWindowValidator before persistence.'; + +COMMENT ON COLUMN distribution_schedules.cron IS + 'Standard 5-field cron expression (minute hour dom month dow). Rejected if overlapping Stellar maintenance.'; diff --git a/src/db/repositories/offeringRepository.ts b/src/db/repositories/offeringRepository.ts index d0422975..9055537b 100644 --- a/src/db/repositories/offeringRepository.ts +++ b/src/db/repositories/offeringRepository.ts @@ -229,6 +229,48 @@ export class OfferingRepository { return this.mapOffering(result.rows[0]); } + /** + * Persist a deferred cron-expression schedule for an offering. + * Callers MUST run CronWindowValidator before invoking this method. + */ + async updateCronSchedule( + id: string, + cronExpression: string | null, + distributionTimezone: string = 'UTC' + ): Promise { + const query = ` + UPDATE offerings + SET cron_expression = $1, + distribution_timezone = $2, + updated_at = NOW() + WHERE id = $3 + RETURNING * + `; + const result: QueryResult = await this.db.query(query, [ + cronExpression, + distributionTimezone, + id, + ]); + if (result.rows.length === 0) { + return null; + } + return this.mapOffering(result.rows[0]); + } + + /** + * List all offerings that have a persisted cron_expression (for overlap checks). + */ + async listWithCronSchedules(): Promise { + const query = ` + SELECT * + FROM offerings + WHERE cron_expression IS NOT NULL + ORDER BY created_at ASC + `; + const result: QueryResult = await this.db.query(query); + return result.rows.map((row) => this.mapOffering(row)); + } + async updateStatus(id: string, status: OfferingStatus): Promise { const query = ` UPDATE offerings diff --git a/src/db/repositories/revenueReportRepository.ts b/src/db/repositories/revenueReportRepository.ts index 4cf92e69..8d4ecac4 100644 --- a/src/db/repositories/revenueReportRepository.ts +++ b/src/db/repositories/revenueReportRepository.ts @@ -188,9 +188,14 @@ export class RevenueReportRepository { */ async findApprovedWithoutDistribution(): Promise { const query = ` - SELECT r.* + SELECT + r.*, + o.cron_expression, + o.distribution_timezone, + COALESCE(o.distribution_timezone, o.timezone) AS offering_timezone FROM revenue_reports r LEFT JOIN distributions d ON d.period_id = r.id + LEFT JOIN offerings o ON o.id = r.offering_id WHERE r.status = 'approved' AND (d.id IS NULL OR d.status != 'completed') AND ( diff --git a/src/services/__tests__/distributionScheduler.test.ts b/src/services/__tests__/distributionScheduler.test.ts index e990c73d..e4a9c891 100644 --- a/src/services/__tests__/distributionScheduler.test.ts +++ b/src/services/__tests__/distributionScheduler.test.ts @@ -157,7 +157,8 @@ describe('deduplicateWindowKey', () => { timezone: 'UTC', }; const key = deduplicateWindowKey(w); - expect(key).toBe(`${utcStart.getTime()}:${utcEnd.getTime()}`); + expect(key).toBe(`:${utcStart.getTime()}:${utcEnd.getTime()}`); + expect(deduplicateWindowKey(w, 'off-1')).toBe(`off-1:${utcStart.getTime()}:${utcEnd.getTime()}`); }); }); diff --git a/src/services/cronScheduleService.test.ts b/src/services/cronScheduleService.test.ts new file mode 100644 index 00000000..b861ebf3 --- /dev/null +++ b/src/services/cronScheduleService.test.ts @@ -0,0 +1,87 @@ +import { CronScheduleService } from './cronScheduleService'; +import { Errors } from '../lib/errors'; + +describe('CronScheduleService', () => { + const offering = { + id: 'off-1', + cron_expression: '0 3 * * 2', + distribution_timezone: 'UTC', + }; + + function makeService(overrides: { + listWithCronSchedules?: jest.Mock; + updateCronSchedule?: jest.Mock; + query?: jest.Mock; + } = {}) { + const offeringRepo = { + listWithCronSchedules: overrides.listWithCronSchedules ?? jest.fn().mockResolvedValue([]), + updateCronSchedule: + overrides.updateCronSchedule ?? jest.fn().mockResolvedValue(offering), + }; + const pool = { + query: overrides.query ?? jest.fn().mockResolvedValue({ rows: [] }), + }; + return { + service: new CronScheduleService(offeringRepo as any, pool as any), + offeringRepo, + pool, + }; + } + + it('persists a valid schedule to offerings and distribution_schedules', async () => { + const { service, offeringRepo, pool } = makeService(); + const result = await service.persistSchedule({ + offeringId: 'off-1', + expression: '0 3 * * 2', + timezone: 'UTC', + }); + + expect(result.validation.valid).toBe(true); + expect(offeringRepo.updateCronSchedule).toHaveBeenCalledWith('off-1', '0 3 * * 2', 'UTC'); + expect(pool.query).toHaveBeenCalledWith( + expect.stringContaining('INSERT INTO distribution_schedules'), + ['off-1', '0 3 * * 2', 'UTC'] + ); + }); + + it('rejects overlapping Stellar maintenance before persistence', async () => { + const { service, offeringRepo } = makeService(); + await expect( + service.persistSchedule({ + offeringId: 'off-1', + expression: '0 6 * * 0', // Sunday 06:00 UTC maintenance + timezone: 'UTC', + }) + ).rejects.toMatchObject({ statusCode: 400 }); + + expect(offeringRepo.updateCronSchedule).not.toHaveBeenCalled(); + }); + + it('rejects overlap with an existing offering window', async () => { + const { service, offeringRepo } = makeService({ + listWithCronSchedules: jest.fn().mockResolvedValue([ + { id: 'off-other', cron_expression: '0 3 * * 2', distribution_timezone: 'UTC' }, + ]), + }); + + await expect( + service.persistSchedule({ + offeringId: 'off-1', + expression: '0 3 * * 2', + timezone: 'UTC', + }) + ).rejects.toMatchObject({ statusCode: 400 }); + + expect(offeringRepo.updateCronSchedule).not.toHaveBeenCalled(); + }); + + it('clears schedule from both stores', async () => { + const { service, offeringRepo, pool } = makeService(); + await service.clearSchedule('off-1'); + expect(offeringRepo.updateCronSchedule).toHaveBeenCalledWith('off-1', null, 'UTC'); + expect(pool.query).toHaveBeenCalledWith( + expect.stringContaining('DELETE FROM distribution_schedules'), + ['off-1'] + ); + }); +}); diff --git a/src/services/cronScheduleService.ts b/src/services/cronScheduleService.ts new file mode 100644 index 00000000..0d5b47e9 --- /dev/null +++ b/src/services/cronScheduleService.ts @@ -0,0 +1,121 @@ +/** + * CronScheduleService — validates and persists deferred distribution cron windows. + * + * @see ../docs/distribution-cron-window-definitions.md + * @see ./distributionScheduler.ts (CronWindowValidator) + * + * Security assumptions: + * - Cron expressions are untrusted operator input; syntax + Stellar maintenance + * + overlap checks run before any persistence. + * - Overlap diffs are logged (expression only — no secrets/PII). + */ + +import { Logger, globalLogger } from '../lib/logger'; +import { Errors } from '../lib/errors'; +import { MetricsCollector } from '../lib/metrics'; +import { Offering, OfferingRepository } from '../db/repositories/offeringRepository'; +import { + CronWindowDefinition, + CronWindowValidator, + CronWindowValidationResult, +} from './distributionScheduler'; +import { Pool } from 'pg'; + +export interface PersistCronScheduleInput { + offeringId: string; + expression: string; + timezone?: string; +} + +export interface PersistCronScheduleResult { + offering: Offering; + validation: CronWindowValidationResult; +} + +export class CronScheduleService { + private readonly validator: CronWindowValidator; + private readonly logger: Logger; + + constructor( + private readonly offeringRepo: OfferingRepository, + private readonly pool: Pool, + options: { metrics?: MetricsCollector; logger?: Logger; lookaheadDays?: number } = {} + ) { + this.logger = options.logger ?? globalLogger; + this.validator = new CronWindowValidator({ + lookaheadDays: options.lookaheadDays ?? 60, + metrics: options.metrics, + logger: this.logger, + }); + } + + /** + * Validate `expression` against Stellar maintenance + existing offering windows, + * then persist to offerings + distribution_schedules. Rejects before write. + */ + async persistSchedule(input: PersistCronScheduleInput): Promise { + const timezone = input.timezone ?? 'UTC'; + const incoming: CronWindowDefinition = { + offeringId: input.offeringId, + expression: input.expression, + timezone, + }; + + const existingOfferings = await this.offeringRepo.listWithCronSchedules(); + const existing: CronWindowDefinition[] = existingOfferings + .filter((o) => o.id !== input.offeringId && typeof o.cron_expression === 'string') + .map((o) => ({ + offeringId: o.id, + expression: String(o.cron_expression), + timezone: String(o.distribution_timezone ?? o.timezone ?? 'UTC'), + })); + + const validation = this.validator.validateAgainstExisting(incoming, existing); + if (!validation.valid) { + throw Errors.validationError( + `Cron window rejected: ${validation.reasons.join('; ')}` + ); + } + + const offering = await this.offeringRepo.updateCronSchedule( + input.offeringId, + input.expression, + timezone + ); + if (!offering) { + throw Errors.notFound(`Offering ${input.offeringId} not found`); + } + + // Upsert into distribution_schedules (issue #661 persistence target). + await this.pool.query( + ` + INSERT INTO distribution_schedules (offering_id, cron, timezone, updated_at) + VALUES ($1, $2, $3, NOW()) + ON CONFLICT (offering_id) DO UPDATE + SET cron = EXCLUDED.cron, + timezone = EXCLUDED.timezone, + updated_at = NOW() + `, + [input.offeringId, input.expression, timezone] + ); + + this.logger.info('scheduler.window.persisted', { + offeringId: input.offeringId, + expression: input.expression, + timezone, + }); + + return { offering, validation }; + } + + /** + * Clear a deferred schedule (falls back to fixed-interval processing). + */ + async clearSchedule(offeringId: string): Promise { + const offering = await this.offeringRepo.updateCronSchedule(offeringId, null, 'UTC'); + await this.pool.query(`DELETE FROM distribution_schedules WHERE offering_id = $1`, [ + offeringId, + ]); + return offering; + } +} diff --git a/src/services/distributionScheduler.test.ts b/src/services/distributionScheduler.test.ts index 2463ef81..78dc0862 100644 --- a/src/services/distributionScheduler.test.ts +++ b/src/services/distributionScheduler.test.ts @@ -1,4 +1,15 @@ -import { DistributionScheduler } from './distributionScheduler'; +import { + DistributionScheduler, + CronWindowValidator, + CronWindowDefinition, + validateCronSyntax, + STELLAR_MAINTENANCE_WINDOWS, + normalizeScheduleTimezone, + assertValidScheduleTimezone, + findNextCronWindow, + computeTimezoneWindow, + deduplicateWindowKey, +} from './distributionScheduler'; import { HolidayCalendarService } from './holidayCalendarService'; import { MetricsCollector } from '../lib/metrics'; import { InMemorySecurityAuditRepository } from '../security/audit'; @@ -457,25 +468,6 @@ describe('DistributionScheduler', () => { })) as any ); - it('does not trigger red-alert when backlog equals the threshold', async () => { - revenueReportRepo.findApprovedWithoutDistribution.mockResolvedValueOnce( - Array.from({ length: 10 }, (_, i) => ({ - id: `r-${i}`, - offering_id: 'off-1', - })) as any - ); - - const s = new DistributionScheduler(engine, revenueReportRepo, { - catchupMax: 10, - catchupBacklogAlertThreshold: 10, - }); - - const result = await s.catchUpMissedWindows(); - - expect(result.totalMissed).toBe(10); - expect(result.backlogExceededCeiling).toBe(false); -}); - const mockLogger = { info: jest.fn(), error: jest.fn(), warn: jest.fn() }; const s = new DistributionScheduler(engine, revenueReportRepo, { catchupMax: 5, @@ -492,6 +484,25 @@ describe('DistributionScheduler', () => { ); }); + it('does not trigger red-alert when backlog equals the threshold', async () => { + revenueReportRepo.findApprovedWithoutDistribution.mockResolvedValueOnce( + Array.from({ length: 10 }, (_, i) => ({ + id: `r-${i}`, + offering_id: 'off-1', + })) as any + ); + + const s = new DistributionScheduler(engine, revenueReportRepo, { + catchupMax: 10, + catchupBacklogAlertThreshold: 10, + }); + + const result = await s.catchUpMissedWindows(); + + expect(result.totalMissed).toBe(10); + expect(result.backlogExceededCeiling).toBe(false); + }); + it('works correctly without metrics collector (no gauge emitted)', async () => { revenueReportRepo.findApprovedWithoutDistribution.mockResolvedValueOnce( Array.from({ length: 3 }, (_, i) => ({ @@ -896,13 +907,20 @@ describe('CronWindowValidator', () => { expect(result.stellarConflict).toBeDefined(); }); - it('rejects an expression that fires at the monthly upgrade window (Mon 02:00 UTC)', () => { + it('rejects an expression that fires at the monthly upgrade window (1st Monday 02:00 UTC)', () => { const result = validator.validate({ ...validDef(), expression: '0 2 * * 1' }); expect(result.valid).toBe(false); expect(result.stellarConflict).toBeDefined(); expect(result.stellarConflict!.windowLabel).toMatch(/monthly upgrade/); }); + it('accepts Mondays after the first week (DOM 15–31) at 02:00 UTC', () => { + // firstWeekdayOfMonth constraint must NOT reject 2nd/3rd/4th Mondays + const result = validator.validate({ ...validDef(), expression: '0 2 15-31 * 1' }); + expect(result.valid).toBe(true); + expect(result.stellarConflict).toBeUndefined(); + }); + it('accepts an expression that fires outside all maintenance windows (Tue 03:00 UTC)', () => { const result = validator.validate({ ...validDef(), expression: '0 3 * * 2' }); expect(result.valid).toBe(true); @@ -1055,6 +1073,18 @@ describe('CronWindowValidator', () => { const result = validator.validate({ ...validDef(), expression: '*/0 3 * * 2' }); expect(result.valid).toBe(false); }); + + it('handles an expression that skips a year (annual Jan 1 fire past the date)', () => { + // Evaluated after Jan 1 within the default 60-day horizon → no fire → still valid + // (no Stellar conflict possible if it never fires in the horizon) + const result = validator.validate({ + expression: '0 0 1 1 *', // Jan 1 00:00 only + timezone: 'UTC', + offeringId: 'off-annual', + }); + expect(typeof result.valid).toBe('boolean'); + expect(result.valid).toBe(true); + }); }); }); @@ -1119,6 +1149,19 @@ describe('findNextCronWindow', () => { expect(result).toBeNull(); }); + it('finds a year-skipping annual expression when lookahead covers the next year', () => { + // After Jan 2 2026, next Jan 1 fire is 2027 — requires ≥365 day lookahead + const after = new Date('2026-01-02T00:00:00Z'); + const missed = findNextCronWindow({ expression: '0 0 1 1 *', timezone: 'UTC' }, after, 60); + expect(missed).toBeNull(); + + const found = findNextCronWindow({ expression: '0 0 1 1 *', timezone: 'UTC' }, after, 400); + expect(found).not.toBeNull(); + expect(found!.start.getUTCFullYear()).toBe(2027); + expect(found!.start.getUTCMonth()).toBe(0); + expect(found!.start.getUTCDate()).toBe(1); + }); + it('evaluates expression in the provided IANA timezone', () => { // 0 3 * * 3 in America/New_York = 07:00 or 08:00 UTC depending on DST const after = new Date('2026-07-28T00:00:00Z'); diff --git a/src/services/distributionScheduler.ts b/src/services/distributionScheduler.ts index 2dc8192c..5880f74f 100644 --- a/src/services/distributionScheduler.ts +++ b/src/services/distributionScheduler.ts @@ -143,11 +143,15 @@ export const STELLAR_MAINTENANCE_WINDOWS = [ /** min hr dom month dow(0=Sun) */ cron: '0 6 * * 0', durationMinutes: 60, + /** When true, only the first weekday occurrence of the month matches (DOM 1–7). */ + firstWeekdayOfMonth: false, }, { label: 'Stellar monthly upgrade window (1st Monday 02:00–04:00 UTC)', + /** Monday 02:00 UTC — constrained to DOM 1–7 via firstWeekdayOfMonth. */ cron: '0 2 * * 1', durationMinutes: 120, + firstWeekdayOfMonth: true, }, ] as const; @@ -168,6 +172,9 @@ function normalizeTimezone(tz: string): string { PST: 'America/Los_Angeles', PDT: 'America/Los_Angeles', GMT: 'UTC', + 'Etc/UTC': 'UTC', + 'Etc/GMT': 'UTC', + Z: 'UTC', }; return aliases[tz.trim()] ?? tz.trim(); } @@ -302,27 +309,27 @@ export class CronWindowValidator { to: Date ): { windowLabel: string; conflictAt: string } | null { const tz = normalizeScheduleTimezone(def.timezone); + const stepMs = cronScanStepMs(def.expression, tz); let cursor = new Date(from); + // Align to minute boundary + cursor.setUTCSeconds(0, 0); while (cursor <= to) { - if (evaluateCronAt(def.expression, cursor, tz)) { + const fireAt = snapCronCandidate(def.expression, cursor, tz) ?? cursor; + if (fireAt >= from && fireAt <= to && evaluateCronAt(def.expression, fireAt, tz)) { for (const maint of STELLAR_MAINTENANCE_WINDOWS) { const maintCron = maint.cron.replace(/\s+/g, ' ').trim(); - // Check if the firing minute falls inside the maintenance window - const maintStart = new Date(cursor); - if (evaluateCronAt(maintCron, cursor, 'UTC')) { - return { windowLabel: maint.label, conflictAt: cursor.toISOString() }; + if (matchesMaintenanceAt(maintCron, fireAt, maint)) { + return { windowLabel: maint.label, conflictAt: fireAt.toISOString() }; } - // Also check if cursor falls within an already-started maintenance window - // by scanning backwards up to durationMinutes for (let back = 1; back <= maint.durationMinutes; back++) { - const candidate = new Date(cursor.getTime() - back * 60_000); - if (evaluateCronAt(maintCron, candidate, 'UTC')) { - return { windowLabel: maint.label, conflictAt: cursor.toISOString() }; + const candidate = new Date(fireAt.getTime() - back * 60_000); + if (matchesMaintenanceAt(maintCron, candidate, maint)) { + return { windowLabel: maint.label, conflictAt: fireAt.toISOString() }; } } } } - cursor = new Date(cursor.getTime() + 60_000); + cursor = new Date((snapCronCandidate(def.expression, cursor, tz) ?? cursor).getTime() + stepMs); } return null; } @@ -335,15 +342,23 @@ export class CronWindowValidator { ): string | null { const tzA = normalizeScheduleTimezone(a.timezone); const tzB = normalizeScheduleTimezone(b.timezone); + const stepMs = Math.min(cronScanStepMs(a.expression, tzA), cronScanStepMs(b.expression, tzB)); let cursor = new Date(from); + cursor.setUTCSeconds(0, 0); while (cursor <= to) { + const candidate = + snapCronCandidate(a.expression, cursor, tzA) ?? + snapCronCandidate(b.expression, cursor, tzB) ?? + cursor; if ( - evaluateCronAt(a.expression, cursor, tzA) && - evaluateCronAt(b.expression, cursor, tzB) + candidate >= from && + candidate <= to && + evaluateCronAt(a.expression, candidate, tzA) && + evaluateCronAt(b.expression, candidate, tzB) ) { - return cursor.toISOString(); + return candidate.toISOString(); } - cursor = new Date(cursor.getTime() + 60_000); + cursor = new Date((snapCronCandidate(a.expression, cursor, tzA) ?? cursor).getTime() + stepMs); } return null; } @@ -473,6 +488,69 @@ function matchesCronField(value: number, field: string): boolean { return false; } +/** + * Choose a scan step for cron horizon walks. + * - Concrete minute+hour in UTC → 1 day + * - Concrete minute (any tz) → 1 hour (caller should snap to :MM) + * - Otherwise → 1 minute + */ +function cronScanStepMs(expression: string, tz: string): number { + const fields = expression.trim().split(/\s+/); + if (fields.length !== 5) return 60_000; + const concreteMin = /^\d+$/.test(fields[0]!); + const concreteHour = /^\d+$/.test(fields[1]!); + if (tz === 'UTC' && concreteMin && concreteHour) return 24 * 60 * 60 * 1000; + if (concreteMin) return 60 * 60 * 1000; + return 60_000; +} + +/** + * Snap `cursor` toward the next plausible fire candidate for concrete fields. + * UTC + concrete HH:MM → that UTC instant on the cursor's day. + * Concrete minute only → same hour with that minute. + */ +function snapCronCandidate(expression: string, cursor: Date, tz: string): Date | null { + const fields = expression.trim().split(/\s+/); + if (fields.length !== 5) return null; + const minConcrete = /^\d+$/.test(fields[0]!) ? parseInt(fields[0]!, 10) : null; + const hourConcrete = /^\d+$/.test(fields[1]!) ? parseInt(fields[1]!, 10) : null; + + if (tz === 'UTC' && minConcrete !== null && hourConcrete !== null) { + return new Date(Date.UTC( + cursor.getUTCFullYear(), + cursor.getUTCMonth(), + cursor.getUTCDate(), + hourConcrete, + minConcrete, + 0, + 0 + )); + } + + if (minConcrete !== null) { + const snapped = new Date(cursor); + snapped.setUTCSeconds(0, 0); + snapped.setUTCMinutes(minConcrete); + return snapped; + } + + return null; +} + +/** + * Returns true when `date` (UTC) falls on a maintenance start matching `maintCron`, + * honouring the optional first-weekday-of-month constraint. + */ +function matchesMaintenanceAt( + maintCron: string, + date: Date, + maint: { firstWeekdayOfMonth?: boolean } +): boolean { + if (!evaluateCronAt(maintCron, date, 'UTC')) return false; + if (maint.firstWeekdayOfMonth && date.getUTCDate() > 7) return false; + return true; +} + function evaluateCronAt(cron: string, date: Date, tz: string): boolean { const fields = cron.trim().split(/\s+/); if (fields.length !== 5) return false; @@ -546,8 +624,8 @@ export function computeTimezoneWindow( }; } -export function deduplicateWindowKey(window: TimezoneWindow): string { - return `${window.utcStart.getTime()}:${window.utcEnd.getTime()}`; +export function deduplicateWindowKey(window: TimezoneWindow, offeringId?: string): string { + return `${offeringId ?? ''}:${window.utcStart.getTime()}:${window.utcEnd.getTime()}`; } export function formatWindowForAudit(window: TimezoneWindow): Record { @@ -560,23 +638,90 @@ export function formatWindowForAudit(window: TimezoneWindow): Record 0 ? lookaheadDays : 60; + const lookaheadMs = days * 24 * 60 * 60 * 1000; const horizon = new Date(afterDate.getTime() + lookaheadMs); - let cursor = new Date(afterDate); + const fields = schedule.expression.trim().split(/\s+/); + const concreteMin = fields.length === 5 && /^\d+$/.test(fields[0]!) ? parseInt(fields[0]!, 10) : null; + const concreteHour = fields.length === 5 && /^\d+$/.test(fields[1]!) ? parseInt(fields[1]!, 10) : null; + // Daily snap is only safe for UTC (local HH:MM == UTC HH:MM). Other zones + // keep the minute scan so DST offsets are handled by evaluateCronAt. + const dailyFastPath = + tz === 'UTC' && concreteMin !== null && concreteHour !== null; + + let cursor: Date; + if (dailyFastPath) { + cursor = new Date(afterDate.getTime()); + cursor.setUTCSeconds(0, 0); + } else { + cursor = new Date(afterDate.getTime()); + cursor.setUTCSeconds(0, 0); + } + while (cursor <= horizon) { + if (dailyFastPath) { + // Snap to concrete HH:MM on the current UTC day, then step days. + const candidate = new Date(Date.UTC( + cursor.getUTCFullYear(), + cursor.getUTCMonth(), + cursor.getUTCDate(), + concreteHour!, + concreteMin!, + 0, + 0 + )); + if (candidate >= afterDate && candidate <= horizon && evaluateCronAt(schedule.expression, candidate, tz)) { + return { start: candidate, end: new Date(candidate.getTime() + 24 * 60 * 60 * 1000) }; + } + // Advance to next day after the candidate (or cursor if candidate is before afterDate) + const base = candidate < afterDate ? afterDate : candidate; + cursor = new Date(Date.UTC( + base.getUTCFullYear(), + base.getUTCMonth(), + base.getUTCDate() + 1, + concreteHour!, + concreteMin!, + 0, + 0 + )); + continue; + } + if (evaluateCronAt(schedule.expression, cursor, tz)) { - const windowEnd = new Date(cursor.getTime() + 24 * 60 * 60 * 1000); - return { start: cursor, end: windowEnd }; + return { start: cursor, end: new Date(cursor.getTime() + 24 * 60 * 60 * 1000) }; } cursor = new Date(cursor.getTime() + 60_000); } + + // Only warn for long-horizon searches (year-skip / annual expressions). Short + // lookbacks used as deferred-gate probes are expected to miss frequently. + if (days >= 60) { + globalLogger.warn('findNextCronWindow: expression never fires within lookahead', { + expression: schedule.expression, + timezone: tz, + afterDate: afterDate.toISOString(), + lookaheadDays: days, + }); + } return null; } @@ -667,6 +812,31 @@ export class DistributionScheduler { let claim: typeof report | null = null; try { + const timezone = this.resolveOfferingTimezone( + (report.distribution_timezone as string | undefined) ?? + (report.offering_timezone as string | undefined) + ); + const cronExpression = report.cron_expression as string | undefined | null; + + // Deferred cron gate (pre-claim): skip until a fire window is open. + // Uses a 24h lookback so a delayed scheduler tick still processes the day-of fire. + if (cronExpression) { + const schedule: CronSchedule = { expression: cronExpression, timezone }; + const now = new Date(); + const lookback = new Date(now.getTime() - 24 * 60 * 60 * 1000); + const openWindow = findNextCronWindow(schedule, lookback, 2); + if (!openWindow || now < openWindow.start) { + this.logger.info('Deferring distribution until cron window opens', { + reportId: report.id, + offeringId: report.offering_id, + nextFireAt: openWindow?.start.toISOString() ?? null, + expression: cronExpression, + timezone, + }); + continue; + } + } + claim = await this.revenueReportRepo.claimApprovedReportForDistribution(report.id); if (!claim) { @@ -680,8 +850,6 @@ export class DistributionScheduler { throw Errors.badRequest(`Report ${claim.id} is missing critical data (period or amount)`); } - const timezone = this.resolveOfferingTimezone(claim.offering_timezone as string | undefined); - const { window, dstTransition } = computeTimezoneWindow( claim.offering_id, claim.period_start, @@ -689,11 +857,11 @@ export class DistributionScheduler { timezone ); - if (this.isWindowAlreadyCompleted(window)) { + if (this.isWindowAlreadyCompleted(window, claim.offering_id)) { this.logger.info('Skipping already-completed timezone window', { reportId: claim.id, offeringId: claim.offering_id, - windowKey: deduplicateWindowKey(window), + windowKey: deduplicateWindowKey(window, claim.offering_id), }); continue; } @@ -704,6 +872,7 @@ export class DistributionScheduler { amount: claim.amount, dstTransition, window: formatWindowForAudit(window), + cronExpression: cronExpression ?? null, }); let periodEnd = claim.period_end; @@ -737,7 +906,7 @@ export class DistributionScheduler { ); await this.revenueReportRepo.markReportDistributionCompleted(claim.id); - this.markWindowCompleted(window); + this.markWindowCompleted(window, claim.offering_id); summary.successful++; this.logger.info('Automated distribution successful', { @@ -857,19 +1026,31 @@ export class DistributionScheduler { // ── Window de-duplication ────────────────────────────────────────────────── - private isWindowAlreadyCompleted(window: TimezoneWindow): boolean { - return this.completedWindows.has(deduplicateWindowKey(window)); + /** @notice Returns true when this UTC window was already processed in-process. */ + isWindowAlreadyCompleted(window: TimezoneWindow, offeringId?: string): boolean { + return this.completedWindows.has(deduplicateWindowKey(window, offeringId)); } - private markWindowCompleted(window: TimezoneWindow): void { - this.completedWindows.add(deduplicateWindowKey(window)); + /** @notice Mark a UTC window as completed so fall-back DST ticks are idempotent. */ + markWindowCompleted(window: TimezoneWindow, offeringId?: string): void { + this.completedWindows.add(deduplicateWindowKey(window, offeringId)); } - // ── Timezone resolution ──────────────────────────────────────────────────── + // ── Timezone / cron helpers ──────────────────────────────────────────────── - private resolveOfferingTimezone(tz: string | undefined): string { + /** @notice Resolve an offering timezone, falling back to UTC for invalid values. */ + resolveOfferingTimezone(tz: string | undefined): string { return normalizeScheduleTimezone(tz); } + + /** + * @notice Evaluate whether `expression` matches `date` in `timezone`. + * @dev Returns false for syntactically invalid expressions (never throws). + */ + evaluateCron(expression: string, date: Date, timezone: string): boolean { + if (validateCronSyntax(expression)) return false; + return evaluateCronAt(expression, date, normalizeScheduleTimezone(timezone)); + } } // ─── DistributionStateManager ─────────────────────────────────────────────────