feat: add cron-expression deferred distribution windows

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.';

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<Offering | null> {
+    const query = `
+      UPDATE offerings
+      SET cron_expression = $1,
+          distribution_timezone = $2,
+          updated_at = NOW()
+      WHERE id = $3
+      RETURNING *
+    `;
+    const result: QueryResult<Offering> = 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<Offering[]> {
+    const query = `
+      SELECT *
+      FROM offerings
+      WHERE cron_expression IS NOT NULL
+      ORDER BY created_at ASC
+    `;
+    const result: QueryResult<Offering> = await this.db.query(query);
+    return result.rows.map((row) => this.mapOffering(row));
+  }
+
   async updateStatus(id: string, status: OfferingStatus): Promise<Offering | null> {
     const query = `
       UPDATE offerings

src/db/repositories/revenueReportRepository.ts

@@ -188,9 +188,14 @@ export class RevenueReportRepository {
    */
   async findApprovedWithoutDistribution(): Promise<RevenueReport[]> {
     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 (

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()}`);
   });
 });
 

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']
+    );
+  });
+});

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<PersistCronScheduleResult> {
+    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<Offering | null> {
+    const offering = await this.offeringRepo.updateCronSchedule(offeringId, null, 'UTC');
+    await this.pool.query(`DELETE FROM distribution_schedules WHERE offering_id = $1`, [
+      offeringId,
+    ]);
+    return offering;
+  }
+}