diff --git a/backend/src/analytics/analytics.service.ts b/backend/src/analytics/analytics.service.ts index 4372479c..b66cf7d5 100644 --- a/backend/src/analytics/analytics.service.ts +++ b/backend/src/analytics/analytics.service.ts @@ -485,7 +485,6 @@ export class AnalyticsService { to, }; } - } /** * Record market snapshot for historical tracking diff --git a/backend/src/cache/warming.service.spec.ts b/backend/src/cache/warming.service.spec.ts index 23e5410e..6f693b87 100644 --- a/backend/src/cache/warming.service.spec.ts +++ b/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: [ diff --git a/backend/src/cache/warming.service.ts b/backend/src/cache/warming.service.ts index a1308c70..5522f6ad 100644 --- a/backend/src/cache/warming.service.ts +++ b/backend/src/cache/warming.service.ts @@ -25,6 +25,7 @@ type CacheWarmResult = Omit; 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 { + 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 { if (!this.strategy.enabled) { this.logger.debug('Cache warming skipped because it is disabled'); return { warmed: [], failed: [] }; diff --git a/backend/src/notifications/digest.service.spec.ts b/backend/src/notifications/digest.service.spec.ts index 7ae14e97..571055dc 100644 --- a/backend/src/notifications/digest.service.spec.ts +++ b/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', () => { diff --git a/backend/src/notifications/digest.service.ts b/backend/src/notifications/digest.service.ts index f1a3a0de..ed4fb82a 100644 --- a/backend/src/notifications/digest.service.ts +++ b/backend/src/notifications/digest.service.ts @@ -162,7 +162,7 @@ export class DigestService { hour: number, ): Promise { 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 { 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')}`; diff --git a/backend/src/notifications/notification-generator.service.spec.ts b/backend/src/notifications/notification-generator.service.spec.ts index 4c143de2..0527c7c3 100644 --- a/backend/src/notifications/notification-generator.service.spec.ts +++ b/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, diff --git a/backend/src/notifications/notification-generator.service.ts b/backend/src/notifications/notification-generator.service.ts index 31e3aee2..07a4aeb3 100644 --- a/backend/src/notifications/notification-generator.service.ts +++ b/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 = []; + private readonly claimedNotificationKeys = new Set(); 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; + }): 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 { // 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'); diff --git a/contracts/open-market/src/config.rs b/contracts/open-market/src/config.rs index bbb96cfb..59deabd0 100644 --- a/contracts/open-market/src/config.rs +++ b/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; diff --git a/contracts/open-market/src/liquidity.rs b/contracts/open-market/src/liquidity.rs index b322b6e4..9e6e7047 100644 --- a/contracts/open-market/src/liquidity.rs +++ b/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 diff --git a/contracts/open-market/src/market.rs b/contracts/open-market/src/market.rs index 80bf7d7c..3ef0823f 100644 --- a/contracts/open-market/src/market.rs +++ b/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); diff --git a/contracts/open-market/tests/twap_validation_tests.rs b/contracts/open-market/tests/twap_validation_tests.rs index ff6c0963..96c4e8b8 100644 --- a/contracts/open-market/tests/twap_validation_tests.rs +++ b/contracts/open-market/tests/twap_validation_tests.rs @@ -442,6 +442,91 @@ fn test_valid_twap_query_succeeds() { ); } +// ══════════════════════════════════════════════════════════════════════════════ +// SETTLEMENT GUARD: Spot Price vs TWAP Deviation +// ══════════════════════════════════════════════════════════════════════════════ + +/// Creates a funded pool at t=1000 and advances the ledger to the market's +/// resolution time (t=91_000), returning the market id. +fn setup_settlement_pool( + env: &Env, + client: &InsightArenaContractClient<'_>, + admin: &Address, + xlm_token: &Address, +) -> u64 { + let provider = Address::generate(env); + let market_id = client.create_market(admin, &lp_market_params(env)); + + let sa = StellarAssetClient::new(env, xlm_token); + let token = TokenClient::new(env, xlm_token); + let liquidity = 100_000_000_i128; + sa.mint(&provider, &liquidity); + token.approve(&provider, &client.address, &liquidity, &9999); + client.add_liquidity(&provider, &market_id, &liquidity); + + env.ledger().with_mut(|l| l.timestamp = 91_000); + market_id +} + +/// **Test Case**: a spot price pushed far from its TWAP right before settlement +/// must cause `resolve_market` to reject with `InvalidInput`, leaving the market +/// unresolved. +/// +/// **Rationale**: A large swap at settlement moves the spot reserve immediately +/// but contributes nothing to the TWAP until time passes, so the deviation guard +/// must catch it. +#[test] +fn test_manipulated_spot_price_rejected_at_settlement() { + let env = Env::default(); + env.ledger().with_mut(|l| l.timestamp = 1000); + env.mock_all_auths(); + let (client, admin, oracle, xlm_token) = deploy_with_token(&env); + let market_id = setup_settlement_pool(&env, &client, &admin, &xlm_token); + + let trader = Address::generate(&env); + let sa = StellarAssetClient::new(&env, &xlm_token); + let token = TokenClient::new(&env, &xlm_token); + let manipulation = 100_000_000_i128; + sa.mint(&trader, &manipulation); + token.approve(&trader, &client.address, &manipulation, &9999); + client.swap_outcome( + &trader, + &market_id, + &symbol_short!("no"), + &symbol_short!("yes"), + &manipulation, + &0_i128, + &None::, + ); + + let result = client.try_resolve_market(&oracle, &market_id, &symbol_short!("yes")); + assert!( + matches!(result, Err(Ok(InsightArenaError::InvalidInput))), + "Expected settlement to reject manipulated spot price, got: {:?}", + result + ); + assert!(!client.get_market(&market_id).is_resolved); +} + +/// **Test Case**: a spot price that tracks its TWAP passes the settlement guard +/// and the market resolves. +#[test] +fn test_normal_spot_price_passes_settlement() { + let env = Env::default(); + env.ledger().with_mut(|l| l.timestamp = 1000); + env.mock_all_auths(); + let (client, admin, oracle, xlm_token) = deploy_with_token(&env); + let market_id = setup_settlement_pool(&env, &client, &admin, &xlm_token); + + let result = client.try_resolve_market(&oracle, &market_id, &symbol_short!("yes")); + assert!( + result.is_ok(), + "Expected settlement with normal spot price to succeed, got: {:?}", + result + ); + assert!(client.get_market(&market_id).is_resolved); +} + // ══════════════════════════════════════════════════════════════════════════════ // EDGE CASE: Single Observation // ══════════════════════════════════════════════════════════════════════════════