Fix/issues 1762 1936 1934 1933

backend/src/analytics/analytics.service.ts

@@ -60,7 +60,7 @@
 export interface TimeBucketAggregationOptions {
   from: Date;
   to: Date;
   interval?: 'hour' | 'day' | 'week' | 'month' | string;
   intervalMs?: number;
   fillEmptyBuckets?: boolean;
   timestampField?: string;
@@ -485,7 +485,6 @@
       to,
     };
   }
-  }
 
   /**
    * Record market snapshot for historical tracking

backend/src/cache/warming.service.spec.ts

@@ -141,7 +141,8 @@ describe('CacheWarmingService', () => {
     // Remaining sub-warmers still ran despite the thrown error.
     expect(result.warmed).toContain(CACHE_WARMING_KEYS.activeEvents);
     expect(result.warmed).toContain(CACHE_WARMING_KEYS.platformStatistics);
-    expect(result.warmed).toContain(
+    // Popular event details are derived from trending results, so none are warmed.
+    expect(result.warmed).not.toContain(
       CACHE_WARMING_KEYS.popularEventDetail('popular-1'),
     );
     expect(result.warmed).toContain(CACHE_WARMING_KEYS.leaderboardTopN(10, null));
@@ -154,6 +155,31 @@ describe('CacheWarmingService', () => {
     expect(result.warmed.length).toBeGreaterThan(0);
   });
 
+  it('does not overlap warm cycles; a call during a running cycle is skipped', async () => {
+    let releaseFirstCycle: () => void = () => undefined;
+    marketsService.findAllFiltered.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          releaseFirstCycle = () => resolve({ data: [], total: 0 });
+        }),
+    );
+
+    const first = service.warmFrequentlyAccessedData();
+    const overlapping = await service.warmFrequentlyAccessedData();
+
+    expect(overlapping).toEqual({ warmed: [], failed: [] });
+    // Only the first cycle ran its loaders.
+    expect(marketsService.findAllFiltered).toHaveBeenCalledTimes(1);
+
+    releaseFirstCycle();
+    const firstResult = await first;
+    expect(firstResult.failed).toEqual([]);
+
+    // Lock is released once the running cycle finishes.
+    const next = await service.warmFrequentlyAccessedData();
+    expect(next.warmed.length).toBeGreaterThan(0);
+  });
+
   it('skips warming when disabled by config', async () => {
     const module: TestingModule = await Test.createTestingModule({
       providers: [

backend/src/cache/warming.service.ts

@@ -25,6 +25,7 @@ type CacheWarmResult = Omit<WarmMetrics, 'duration_ms'>;
 export class CacheWarmingService {
   private readonly logger = new Logger(CacheWarmingService.name);
   private readonly strategy: CacheWarmingStrategy;
+  private isWarming = false;
 
   constructor(
     @Inject(CACHE_MANAGER) private readonly cacheManager: Cache,
@@ -38,6 +39,20 @@ export class CacheWarmingService {
 
   @Cron('0 */10 * * * *')
   async warmFrequentlyAccessedData(): Promise<CacheWarmResult> {
+    if (this.isWarming) {
+      this.logger.warn('Cache warming skipped: previous cycle still running');
+      return { warmed: [], failed: [] };
+    }
+
+    this.isWarming = true;
+    try {
+      return await this.runWarmCycle();
+    } finally {
+      this.isWarming = false;
+    }
+  }
+
+  private async runWarmCycle(): Promise<CacheWarmResult> {
     if (!this.strategy.enabled) {
       this.logger.debug('Cache warming skipped because it is disabled');
       return { warmed: [], failed: [] };

backend/src/notifications/digest.service.spec.ts

@@ -192,6 +192,44 @@ describe('DigestService', () => {
     });
   });
 
+  // ── Timezone-correct windowing ──────────────────────────────────────────
+
+  describe('digest window start', () => {
+    it('starts at the previous local digest hour, not a fixed 24h earlier, across DST', async () => {
+      const runDigests = jest
+        .spyOn(service as any, 'runDigests')
+        .mockResolvedValue(undefined);
+
+      // 2024-03-10 08:00 EDT (after spring-forward). The previous digest was
+      // 2024-03-09 08:00 EST = 13:00 UTC, which is 25h earlier, not 24h.
+      await service.sendDailyDigests(
+        new Date('2024-03-10T12:00:00Z'),
+        'America/New_York',
+        8,
+      );
+
+      expect(runDigests.mock.calls[0][2]).toEqual(
+        new Date('2024-03-09T13:00:00Z'),
+      );
+    });
+
+    it('matches a fixed 24h window in UTC', async () => {
+      const runDigests = jest
+        .spyOn(service as any, 'runDigests')
+        .mockResolvedValue(undefined);
+
+      await service.sendDailyDigests(
+        new Date('2024-01-15T08:00:00Z'),
+        'UTC',
+        8,
+      );
+
+      expect(runDigests.mock.calls[0][2]).toEqual(
+        new Date('2024-01-14T08:00:00Z'),
+      );
+    });
+  });
+
   // ── Frequency selection & timezone bucketing ────────────────────────────
 
   describe('handleHourlyCheck', () => {

backend/src/notifications/digest.service.ts

@@ -162,7 +162,7 @@ export class DigestService {
     hour: number,
   ): Promise<void> {
     const periodKey = this.getLocalDailyPeriodKey(now, timezone);
-    const windowStart = new Date(now.getTime() - 24 * 60 * 60 * 1000);
+    const windowStart = this.getPreviousLocalDigestTime(now, timezone, hour, 1);
     this.logger.log(
       `Running daily digest for ${timezone} @ ${hour}:00 local, period ${periodKey}`,
     );
@@ -175,7 +175,7 @@ export class DigestService {
     hour: number,
   ): Promise<void> {
     const periodKey = this.getLocalWeeklyPeriodKey(now, timezone);
-    const windowStart = new Date(now.getTime() - 7 * 24 * 60 * 60 * 1000);
+    const windowStart = this.getPreviousLocalDigestTime(now, timezone, hour, 7);
     this.logger.log(
       `Running weekly digest for ${timezone} @ ${hour}:00 local, period ${periodKey}`,
     );
@@ -355,6 +355,71 @@ export class DigestService {
     return { year: get('year'), month: get('month'), day: get('day') };
   }
 
+  /**
+   * The instant of `hour`:00 local time, `daysBack` local calendar days before
+   * `date`. Used as the window start so each digest covers exactly the span
+   * since the previous scheduled digest, even across DST transitions (a fixed
+   * 24h/7d offset would be an hour off on those days).
+   */
+  private getPreviousLocalDigestTime(
+    date: Date,
+    timezone: string,
+    hour: number,
+    daysBack: number,
+  ): Date {
+    const { year, month, day } = this.getLocalDateParts(date, timezone);
+    const prev = new Date(Date.UTC(year, month - 1, day - daysBack));
+    return this.localWallTimeToUtc(
+      prev.getUTCFullYear(),
+      prev.getUTCMonth() + 1,
+      prev.getUTCDate(),
+      hour,
+      timezone,
+    );
+  }
+
+  /** The instant at which the local wall clock in `timezone` reads y-m-d hh:00. */
+  private localWallTimeToUtc(
+    year: number,
+    month: number,
+    day: number,
+    hour: number,
+    timezone: string,
+  ): Date {
+    const wallAsUtc = Date.UTC(year, month - 1, day, hour);
+    const offset = this.getTimeZoneOffsetMs(new Date(wallAsUtc), timezone);
+    let utc = wallAsUtc - offset;
+    // Re-check the offset at the candidate instant in case it differs (DST boundary).
+    const corrected = this.getTimeZoneOffsetMs(new Date(utc), timezone);
+    if (corrected !== offset) utc = wallAsUtc - corrected;
+    return new Date(utc);
+  }
+
+  /** The UTC offset of `timezone` at instant `date`, in milliseconds. */
+  private getTimeZoneOffsetMs(date: Date, timezone: string): number {
+    const parts = new Intl.DateTimeFormat('en-US', {
+      timeZone: timezone,
+      hourCycle: 'h23',
+      year: 'numeric',
+      month: 'numeric',
+      day: 'numeric',
+      hour: 'numeric',
+      minute: 'numeric',
+      second: 'numeric',
+    }).formatToParts(date);
+    const get = (type: string) =>
+      parseInt(parts.find((p) => p.type === type)?.value ?? '0', 10);
+    const wallAsUtc = Date.UTC(
+      get('year'),
+      get('month') - 1,
+      get('day'),
+      get('hour'),
+      get('minute'),
+      get('second'),
+    );
+    return wallAsUtc - Math.floor(date.getTime() / 1000) * 1000;
+  }
+
   private getLocalDailyPeriodKey(date: Date, timezone: string): string {
     const { year, month, day } = this.getLocalDateParts(date, timezone);
     return `${year}-${String(month).padStart(2, '0')}-${String(day).padStart(2, '0')}`;

backend/src/notifications/notification-generator.service.spec.ts

@@ -155,6 +155,24 @@ describe('NotificationGeneratorService', () => {
       expect(notificationsRepository.create).toHaveBeenCalled();
     });
 
+    it('creates only one notification when the same event is delivered twice', async () => {
+      const data = {
+        event_id: 1,
+        creator: 'GABC123',
+        title: 'Test Event',
+      };
+
+      jest.spyOn(userRepository, 'findOne').mockResolvedValue(mockUser as any);
+      jest.spyOn(notificationsRepository, 'create').mockReturnValue({} as any);
+      jest.spyOn(notificationsRepository, 'save').mockResolvedValue({} as any);
+
+      await service.handleEventCreated(data);
+      await service.handleEventCreated(data);
+      await service.flushQueue();
+
+      expect(notificationsRepository.create).toHaveBeenCalledTimes(1);
+    });
+
     it('should skip notification if user has disabled event_created_notifications', async () => {
       const data = {
         event_id: 1,

backend/src/notifications/notification-generator.service.ts

@@ -41,6 +41,7 @@ export interface OracleDivergenceNotificationInput {
 export class NotificationGeneratorService implements OnModuleDestroy {
   private readonly logger = new Logger(NotificationGeneratorService.name);
   private readonly notificationQueue: Array<NotificationBatch> = [];
+  private readonly claimedNotificationKeys = new Set<string>();
   private isProcessing = false;
   private readonly BATCH_SIZE = 50;
   private readonly FLUSH_INTERVAL = 5000; // 5 seconds
@@ -434,9 +435,27 @@ export class NotificationGeneratorService implements OnModuleDestroy {
       return;
     }
 
+    if (!this.claimNotificationKey(notification)) return;
+
     this.notificationQueue.push({ notifications: [notification] });
   }
 
+  /**
+   * Returns true the first time a given notification (type, recipient, and
+   * event data) is seen, false for any repeat. Keeps re-delivered events from
+   * creating duplicate notification rows.
+   */
+  private claimNotificationKey(notification: {
+    userAddress: string;
+    type: NotificationType;
+    data?: Record<string, unknown>;
+  }): boolean {
+    const key = `${notification.type}:${notification.userAddress}:${JSON.stringify(notification.data ?? {})}`;
+    if (this.claimedNotificationKeys.has(key)) return false;
+    this.claimedNotificationKeys.add(key);
+    return true;
+  }
+
   /**
    * Queue a batch of notifications after filtering by category preferences.
    * Only notifications for enabled categories are queued.
@@ -451,9 +470,9 @@ export class NotificationGeneratorService implements OnModuleDestroy {
     }>,
   ): Promise<void> {
     // Filter notifications based on category preferences
-    const filteredNotifications = await this.filterNotificationsByPreferences(
-      notifications,
-    );
+    const filteredNotifications = (
+      await this.filterNotificationsByPreferences(notifications)
+    ).filter((notification) => this.claimNotificationKey(notification));
 
     if (filteredNotifications.length === 0) {
       this.logger.debug('All notifications filtered out by category preferences');

contracts/open-market/src/config.rs

@@ -161,6 +161,11 @@ pub const MIN_EARLY_EXIT_FEE_BPS: u32 = 0;
 /// Maximum early exit fee in basis points (100%). Inclusive: `0..=10_000`.
 pub const MAX_EARLY_EXIT_FEE_BPS: u32 = 10_000;
 
+/// Trailing window (seconds) of the TWAP used to sanity-check the spot price at settlement.
+pub const TWAP_SETTLEMENT_WINDOW_SECONDS: u64 = 3_600;
+/// Max allowed |spot - TWAP| / TWAP at settlement, in basis points (20%).
+pub const TWAP_SETTLEMENT_MAX_DEVIATION_BPS: u32 = 2_000;
+
 // ── Storage-specific TTL constants (merged from ttl.rs) ───────────────────────
 // ~30 days at ~6s/ledger for frequently accessed market state.
 pub const LEDGER_BUMP_MARKET: u32 = 432_000;

contracts/open-market/src/liquidity.rs

@@ -532,6 +532,55 @@ pub fn get_market_twap(
     get_twap(env, market_id, outcome, window_seconds)
 }
 
+/// Settlement guard against spot-price manipulation.
+///
+/// Rejects when the outcome's spot reserve deviates from its TWAP over
+/// `TWAP_SETTLEMENT_WINDOW_SECONDS` by more than `TWAP_SETTLEMENT_MAX_DEVIATION_BPS`.
+/// A large trade right before settlement moves the spot immediately but barely
+/// moves the TWAP, so the mismatch is caught here. Markets without a pool, or
+/// without retained history covering the window, are not checked.
+pub fn validate_settlement_price(
+    env: &Env,
+    market_id: u64,
+    outcome: Symbol,
+) -> Result<(), InsightArenaError> {
+    if !env
+        .storage()
+        .persistent()
+        .has(&DataKey::LiquidityPool(market_id))
+    {
+        return Ok(());
+    }
+
+    let spot = get_outcome_price(env, market_id, outcome.clone())?;
+    let twap = match get_twap(
+        env,
+        market_id,
+        outcome,
+        config::TWAP_SETTLEMENT_WINDOW_SECONDS,
+    ) {
+        Ok(twap) => twap,
+        Err(InsightArenaError::TwapInsufficientHistory) => return Ok(()),
+        Err(err) => return Err(err),
+    };
+
+    let deviation = spot
+        .checked_sub(twap)
+        .and_then(|d| d.checked_abs())
+        .ok_or(InsightArenaError::Overflow)?;
+    let lhs = deviation
+        .checked_mul(10_000)
+        .ok_or(InsightArenaError::Overflow)?;
+    let rhs = twap
+        .checked_mul(config::TWAP_SETTLEMENT_MAX_DEVIATION_BPS as i128)
+        .ok_or(InsightArenaError::Overflow)?;
+
+    if lhs > rhs {
+        return Err(InsightArenaError::InvalidInput);
+    }
+    Ok(())
+}
+
 // ── Impermanent Loss Accounting ───────────────────────────────────────────────
 //
 // Scope: this contract's AMM pool is generalized to N outcomes

contracts/open-market/src/market.rs

@@ -901,6 +901,9 @@ pub fn resolve_market(
         return Err(InsightArenaError::InvalidOutcome);
     }
 
+    // Reject settlement if the spot price has been pushed far from its TWAP.
+    crate::liquidity::validate_settlement_price(&env, market_id, resolved_outcome.clone())?;
+
     market.is_resolved = true;
     market.resolved_outcome = Some(resolved_outcome.clone());
     market.resolved_at = Some(now);