From 924039f449cd80407f14ef3b1dc0da935039db9b Mon Sep 17 00:00:00 2001 From: Abdulrasaq1515 Date: Tue, 26 May 2026 18:41:57 +0100 Subject: [PATCH 1/6] =?UTF-8?q?feat:=20scheduler=20hardening=20=E2=80=94?= =?UTF-8?q?=20idempotent=20jobs,=20pg=20advisory=20locking,=20run=20histor?= =?UTF-8?q?y?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add SchedulerModule with JobLockService (pg_try_advisory_lock) and JobHistoryService - Add JobRun entity and migration 1772000000000-CreateJobRuns - Harden ReconciliationScheduler, SnapshotScheduler, ModelRetrainingScheduler, NewsService.fetchAndSaveArticles, NewsSentimentService.updateMissingSentiments, OutboxService.pollAndDispatch with acquire/release lock pattern - All schedulers write RUNNING/COMPLETED/FAILED/SKIPPED rows to job_runs table - Lock is session-scoped: auto-released on process crash, no stuck locks --- .../migrations/1772000000000-CreateJobRuns.ts | 41 ++++++++++ .../model-retraining.module.ts | 2 + .../model-retraining.scheduler.ts | 30 ++++++- .../src/news/news-sentiment.services.ts | 57 ++++++++++--- apps/backend/src/news/news.module.ts | 2 + apps/backend/src/news/news.service.ts | 22 ++++++ apps/backend/src/outbox/outbox.module.ts | 3 +- apps/backend/src/outbox/outbox.service.ts | 36 ++++++--- .../reconciliation/reconciliation.module.ts | 2 + .../reconciliation.scheduler.ts | 30 ++++++- .../src/scheduler/entities/job-run.entity.ts | 55 +++++++++++++ .../src/scheduler/job-history.service.ts | 67 ++++++++++++++++ .../backend/src/scheduler/job-lock.service.ts | 79 +++++++++++++++++++ .../backend/src/scheduler/scheduler.module.ts | 20 +++++ apps/backend/src/snapshot/snapshot.module.ts | 3 +- .../src/snapshot/snapshot.scheduler.ts | 27 +++++-- 16 files changed, 439 insertions(+), 37 deletions(-) create mode 100644 apps/backend/src/database/migrations/1772000000000-CreateJobRuns.ts create mode 100644 apps/backend/src/scheduler/entities/job-run.entity.ts create mode 100644 apps/backend/src/scheduler/job-history.service.ts create mode 100644 apps/backend/src/scheduler/job-lock.service.ts create mode 100644 apps/backend/src/scheduler/scheduler.module.ts diff --git a/apps/backend/src/database/migrations/1772000000000-CreateJobRuns.ts b/apps/backend/src/database/migrations/1772000000000-CreateJobRuns.ts new file mode 100644 index 00000000..d712c983 --- /dev/null +++ b/apps/backend/src/database/migrations/1772000000000-CreateJobRuns.ts @@ -0,0 +1,41 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class CreateJobRuns1772000000000 implements MigrationInterface { + name = 'CreateJobRuns1772000000000'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query( + `CREATE TYPE "job_runs_status_enum" AS ENUM ('running', 'completed', 'failed', 'skipped')`, + ); + + await queryRunner.query(` + CREATE TABLE "job_runs" ( + "id" uuid NOT NULL DEFAULT uuid_generate_v4(), + "jobName" character varying(100) NOT NULL, + "status" "job_runs_status_enum" NOT NULL DEFAULT 'running', + "triggeredBy" character varying(50) NOT NULL DEFAULT 'scheduled', + "result" jsonb, + "errorMessage" text, + "startedAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), + "finishedAt" TIMESTAMP WITH TIME ZONE, + "durationMs" integer, + CONSTRAINT "PK_job_runs" PRIMARY KEY ("id") + ) + `); + + await queryRunner.query( + `CREATE INDEX "IDX_job_runs_name_started" ON "job_runs" ("jobName", "startedAt" DESC)`, + ); + + await queryRunner.query( + `CREATE INDEX "IDX_job_runs_status" ON "job_runs" ("status")`, + ); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX "IDX_job_runs_status"`); + await queryRunner.query(`DROP INDEX "IDX_job_runs_name_started"`); + await queryRunner.query(`DROP TABLE "job_runs"`); + await queryRunner.query(`DROP TYPE "job_runs_status_enum"`); + } +} diff --git a/apps/backend/src/model-retraining/model-retraining.module.ts b/apps/backend/src/model-retraining/model-retraining.module.ts index f86e3d64..e4cd629e 100644 --- a/apps/backend/src/model-retraining/model-retraining.module.ts +++ b/apps/backend/src/model-retraining/model-retraining.module.ts @@ -4,6 +4,7 @@ import { ConfigModule } from '@nestjs/config'; import { ModelRetrainingService } from './model-retraining.service'; import { ModelRetrainingScheduler } from './model-retraining.scheduler'; import { ModelRetrainingController } from './model-retraining.controller'; +import { SchedulerModule } from '../scheduler/scheduler.module'; @Module({ imports: [ @@ -14,6 +15,7 @@ import { ModelRetrainingController } from './model-retraining.controller'; }), }), ConfigModule, + SchedulerModule, ], providers: [ModelRetrainingService, ModelRetrainingScheduler], controllers: [ModelRetrainingController], diff --git a/apps/backend/src/model-retraining/model-retraining.scheduler.ts b/apps/backend/src/model-retraining/model-retraining.scheduler.ts index fd22d280..776ec81c 100644 --- a/apps/backend/src/model-retraining/model-retraining.scheduler.ts +++ b/apps/backend/src/model-retraining/model-retraining.scheduler.ts @@ -1,6 +1,10 @@ import { Injectable, Logger } from '@nestjs/common'; import { Cron } from '@nestjs/schedule'; import { ModelRetrainingService } from './model-retraining.service'; +import { JobLockService } from '../scheduler/job-lock.service'; +import { JobHistoryService } from '../scheduler/job-history.service'; + +const JOB_NAME = 'model-retraining-daily'; /** * NestJS-side scheduled trigger for model retraining. @@ -10,29 +14,49 @@ import { ModelRetrainingService } from './model-retraining.service'; * missed its window (e.g. restart, cold start). * * The Python service itself deduplicates concurrent runs via a threading lock, - * so double-triggering is safe. + * so double-triggering is safe. The advisory lock here prevents multiple + * NestJS instances from all firing the fallback simultaneously. */ @Injectable() export class ModelRetrainingScheduler { private readonly logger = new Logger(ModelRetrainingScheduler.name); - constructor(private readonly retrainingService: ModelRetrainingService) {} + constructor( + private readonly retrainingService: ModelRetrainingService, + private readonly jobLock: JobLockService, + private readonly jobHistory: JobHistoryService, + ) {} - @Cron('30 2 * * *', { timeZone: 'UTC', name: 'model-retraining-daily' }) + @Cron('30 2 * * *', { timeZone: 'UTC', name: JOB_NAME }) async handleDailyRetraining(): Promise { this.logger.log('Daily model retraining job triggered (NestJS scheduler)'); + + const acquired = await this.jobLock.tryAcquire(JOB_NAME); + if (!acquired) { + await this.jobHistory.markSkipped(JOB_NAME); + return; + } + + const run = await this.jobHistory.start(JOB_NAME); try { const result = await this.retrainingService.triggerRetraining(); + await this.jobHistory.complete(run, { + status: result.status, + durationSeconds: result.duration_seconds, + }); this.logger.log( `Daily retraining finished: status=${result.status} ` + `duration=${result.duration_seconds?.toFixed(1)}s`, ); } catch (err) { // Never crash the process — log and move on + await this.jobHistory.fail(run, err); this.logger.error( 'Daily model retraining job failed', err instanceof Error ? err.stack : String(err), ); + } finally { + await this.jobLock.release(JOB_NAME); } } } diff --git a/apps/backend/src/news/news-sentiment.services.ts b/apps/backend/src/news/news-sentiment.services.ts index e6689ba4..e5488222 100644 --- a/apps/backend/src/news/news-sentiment.services.ts +++ b/apps/backend/src/news/news-sentiment.services.ts @@ -5,6 +5,10 @@ import { AxiosResponse } from 'axios'; import { firstValueFrom } from 'rxjs'; import { NewsService } from './news.service'; import { Cron, CronExpression } from '@nestjs/schedule'; +import { JobLockService } from '../scheduler/job-lock.service'; +import { JobHistoryService } from '../scheduler/job-history.service'; + +const SENTIMENT_JOB_NAME = 'news-sentiment-update'; interface SentimentApiResponse { sentiment: number; @@ -18,6 +22,8 @@ export class NewsSentimentService { private readonly httpService: HttpService, private readonly newsService: NewsService, private readonly configService: ConfigService, + private readonly jobLock: JobLockService, + private readonly jobHistory: JobHistoryService, ) {} async analyzeSentiment(text: string): Promise { @@ -42,21 +48,48 @@ export class NewsSentimentService { async updateMissingSentiments(): Promise { this.logger.log('Running sentiment update job...'); - const articlesWithoutSentiment = - await this.newsService.findUnscoredArticles(); + const acquired = await this.jobLock.tryAcquire(SENTIMENT_JOB_NAME); + if (!acquired) { + await this.jobHistory.markSkipped(SENTIMENT_JOB_NAME); + return; + } + + const run = await this.jobHistory.start(SENTIMENT_JOB_NAME); + let scored = 0; + let failed = 0; + + try { + const articlesWithoutSentiment = + await this.newsService.findUnscoredArticles(); - for (const article of articlesWithoutSentiment) { - const score = await this.analyzeSentiment(article.title); + for (const article of articlesWithoutSentiment) { + const score = await this.analyzeSentiment(article.title); - if (score !== null) { - await this.newsService.update(article.id, { sentimentScore: score }); - } else { - this.logger.warn(`Failed to score article: ${article.id}`); + if (score !== null) { + await this.newsService.update(article.id, { sentimentScore: score }); + scored++; + } else { + this.logger.warn(`Failed to score article: ${article.id}`); + failed++; + } } - } - this.logger.log( - `Sentiment update done. Processed: ${articlesWithoutSentiment.length} articles.`, - ); + await this.jobHistory.complete(run, { + processed: articlesWithoutSentiment.length, + scored, + failed, + }); + + this.logger.log( + `Sentiment update done. Processed: ${articlesWithoutSentiment.length} articles.`, + ); + } catch (err) { + await this.jobHistory.fail(run, err); + this.logger.error( + `Sentiment update job failed: ${err instanceof Error ? err.message : String(err)}`, + ); + } finally { + await this.jobLock.release(SENTIMENT_JOB_NAME); + } } } diff --git a/apps/backend/src/news/news.module.ts b/apps/backend/src/news/news.module.ts index 6060548b..48f74206 100644 --- a/apps/backend/src/news/news.module.ts +++ b/apps/backend/src/news/news.module.ts @@ -9,6 +9,7 @@ import { News } from './news.entity'; import { NewsSentimentService } from './news-sentiment.services'; import { AppCacheModule } from '../cache/cache.module'; import { ProfilingModule } from '../common/profiling/profiling.module'; +import { SchedulerModule } from '../scheduler/scheduler.module'; @Module({ imports: [ @@ -19,6 +20,7 @@ import { ProfilingModule } from '../common/profiling/profiling.module'; TypeOrmModule.forFeature([News]), AppCacheModule, ProfilingModule, + SchedulerModule, ], controllers: [NewsController], providers: [NewsProviderService, NewsService, NewsSentimentService], diff --git a/apps/backend/src/news/news.service.ts b/apps/backend/src/news/news.service.ts index a587486e..1650ee8d 100644 --- a/apps/backend/src/news/news.service.ts +++ b/apps/backend/src/news/news.service.ts @@ -9,6 +9,10 @@ import { NewsProviderService } from './news-provider.service'; import { NewsArticleDto } from './dto/news-article.dto'; import { CacheService } from '../cache/cache.service'; import { QueryProfilerService } from '../common/profiling/query-profiler.service'; +import { JobLockService } from '../scheduler/job-lock.service'; +import { JobHistoryService } from '../scheduler/job-history.service'; + +const FETCH_JOB_NAME = 'news-fetch'; interface RawOverallResult { average: string | null; @@ -31,6 +35,8 @@ export class NewsService { private readonly newsProviderService: NewsProviderService, private readonly cacheService: CacheService, private readonly profiler: QueryProfilerService, + private readonly jobLock: JobLockService, + private readonly jobHistory: JobHistoryService, ) {} async create(createArticleDto: CreateArticleDto): Promise { @@ -193,6 +199,13 @@ export class NewsService { async fetchAndSaveArticles(): Promise { this.logger.log('Running scheduled news fetch job...'); + const acquired = await this.jobLock.tryAcquire(FETCH_JOB_NAME); + if (!acquired) { + await this.jobHistory.markSkipped(FETCH_JOB_NAME); + return; + } + + const run = await this.jobHistory.start(FETCH_JOB_NAME); try { // Fetch latest articles from provider const response = await this.newsProviderService.getLatestArticles({ @@ -214,6 +227,12 @@ export class NewsService { } } + await this.jobHistory.complete(run, { + fetched: articles.length, + newArticles: newCount, + duplicatesSkipped: skippedCount, + }); + this.logger.log( `News fetch completed. Fetched ${articles.length} articles, ${newCount} new, ${skippedCount} duplicates skipped.`, ); @@ -222,9 +241,12 @@ export class NewsService { await this.cacheService.invalidateNewsCache(); } } catch (error) { + await this.jobHistory.fail(run, error); this.logger.error( `Failed to fetch and save articles: ${error instanceof Error ? error.message : 'Unknown error'}`, ); + } finally { + await this.jobLock.release(FETCH_JOB_NAME); } } } diff --git a/apps/backend/src/outbox/outbox.module.ts b/apps/backend/src/outbox/outbox.module.ts index 429c63be..623ceadf 100644 --- a/apps/backend/src/outbox/outbox.module.ts +++ b/apps/backend/src/outbox/outbox.module.ts @@ -2,9 +2,10 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; import { OutboxEvent } from './outbox-event.entity'; import { OutboxService } from './outbox.service'; +import { SchedulerModule } from '../scheduler/scheduler.module'; @Module({ - imports: [TypeOrmModule.forFeature([OutboxEvent])], + imports: [TypeOrmModule.forFeature([OutboxEvent]), SchedulerModule], providers: [OutboxService], exports: [OutboxService], }) diff --git a/apps/backend/src/outbox/outbox.service.ts b/apps/backend/src/outbox/outbox.service.ts index feb0c519..0e323185 100644 --- a/apps/backend/src/outbox/outbox.service.ts +++ b/apps/backend/src/outbox/outbox.service.ts @@ -3,6 +3,7 @@ import { InjectRepository } from '@nestjs/typeorm'; import { EntityManager, LessThan, Repository } from 'typeorm'; import { Cron, CronExpression } from '@nestjs/schedule'; import { OutboxEvent, OutboxEventStatus } from './outbox-event.entity'; +import { JobLockService } from '../scheduler/job-lock.service'; /** Maximum dispatch attempts before an event is permanently marked failed */ const MAX_ATTEMPTS = 5; @@ -10,6 +11,8 @@ const MAX_ATTEMPTS = 5; /** How many pending events to process per poll cycle */ const BATCH_SIZE = 50; +const OUTBOX_LOCK = 'outbox-poll'; + export type OutboxEventHandler = ( eventType: string, payload: Record, @@ -23,6 +26,7 @@ export class OutboxService { constructor( @InjectRepository(OutboxEvent) private readonly outboxRepo: Repository, + private readonly jobLock: JobLockService, ) {} /** @@ -65,22 +69,30 @@ export class OutboxService { /** * Poll for pending events and dispatch them to all registered handlers. * Runs every 5 seconds. Events exceeding MAX_ATTEMPTS are marked failed. + * Advisory lock prevents two instances from processing the same batch. */ @Cron(CronExpression.EVERY_5_SECONDS) async pollAndDispatch(): Promise { - const events = await this.outboxRepo.find({ - where: { - status: OutboxEventStatus.PENDING, - attempts: LessThan(MAX_ATTEMPTS), - }, - order: { createdAt: 'ASC' }, - take: BATCH_SIZE, - }); - - if (events.length === 0) return; + const acquired = await this.jobLock.tryAcquire(OUTBOX_LOCK); + if (!acquired) return; // another instance is already polling - for (const event of events) { - await this.dispatch(event); + try { + const events = await this.outboxRepo.find({ + where: { + status: OutboxEventStatus.PENDING, + attempts: LessThan(MAX_ATTEMPTS), + }, + order: { createdAt: 'ASC' }, + take: BATCH_SIZE, + }); + + if (events.length === 0) return; + + for (const event of events) { + await this.dispatch(event); + } + } finally { + await this.jobLock.release(OUTBOX_LOCK); } } diff --git a/apps/backend/src/reconciliation/reconciliation.module.ts b/apps/backend/src/reconciliation/reconciliation.module.ts index 543671cd..280481a3 100644 --- a/apps/backend/src/reconciliation/reconciliation.module.ts +++ b/apps/backend/src/reconciliation/reconciliation.module.ts @@ -8,12 +8,14 @@ import { PortfolioAsset } from '../portfolio/portfolio-asset.entity'; import { User } from '../users/entities/user.entity'; import { PortfolioModule } from '../portfolio/portfolio.module'; import { ProfilingModule } from '../common/profiling/profiling.module'; +import { SchedulerModule } from '../scheduler/scheduler.module'; @Module({ imports: [ TypeOrmModule.forFeature([ReconciliationJob, PortfolioAsset, User]), PortfolioModule, ProfilingModule, + SchedulerModule, ], providers: [ReconciliationService, ReconciliationScheduler], controllers: [ReconciliationController], diff --git a/apps/backend/src/reconciliation/reconciliation.scheduler.ts b/apps/backend/src/reconciliation/reconciliation.scheduler.ts index 5b4395b1..386ab5bc 100644 --- a/apps/backend/src/reconciliation/reconciliation.scheduler.ts +++ b/apps/backend/src/reconciliation/reconciliation.scheduler.ts @@ -1,27 +1,51 @@ import { Injectable, Logger } from '@nestjs/common'; import { Cron } from '@nestjs/schedule'; import { ReconciliationService } from './reconciliation.service'; +import { JobLockService } from '../scheduler/job-lock.service'; +import { JobHistoryService } from '../scheduler/job-history.service'; + +const JOB_NAME = 'reconciliation'; @Injectable() export class ReconciliationScheduler { private readonly logger = new Logger(ReconciliationScheduler.name); - constructor(private readonly reconciliationService: ReconciliationService) {} + constructor( + private readonly reconciliationService: ReconciliationService, + private readonly jobLock: JobLockService, + private readonly jobHistory: JobHistoryService, + ) {} /** Run reconciliation every 6 hours */ @Cron('0 */6 * * *') async handleScheduledReconciliation(): Promise { this.logger.log('Scheduled reconciliation triggered'); + + const acquired = await this.jobLock.tryAcquire(JOB_NAME); + if (!acquired) { + await this.jobHistory.markSkipped(JOB_NAME); + return; + } + + const run = await this.jobHistory.start(JOB_NAME); try { - const job = - await this.reconciliationService.runReconciliation('scheduled'); + const job = await this.reconciliationService.runReconciliation('scheduled'); + await this.jobHistory.complete(run, { + reconciliationJobId: job.id, + driftsDetected: job.driftsDetected, + driftsRepaired: job.driftsRepaired, + usersProcessed: job.usersProcessed, + }); this.logger.log( `Scheduled reconciliation complete — jobId=${job.id} drifts=${job.driftsDetected} repaired=${job.driftsRepaired}`, ); } catch (err) { + await this.jobHistory.fail(run, err); this.logger.error( `Scheduled reconciliation failed: ${err instanceof Error ? err.message : String(err)}`, ); + } finally { + await this.jobLock.release(JOB_NAME); } } } diff --git a/apps/backend/src/scheduler/entities/job-run.entity.ts b/apps/backend/src/scheduler/entities/job-run.entity.ts new file mode 100644 index 00000000..727fe37a --- /dev/null +++ b/apps/backend/src/scheduler/entities/job-run.entity.ts @@ -0,0 +1,55 @@ +import { + Column, + CreateDateColumn, + Entity, + Index, + PrimaryGeneratedColumn, +} from 'typeorm'; + +export enum JobRunStatus { + RUNNING = 'running', + COMPLETED = 'completed', + FAILED = 'failed', + SKIPPED = 'skipped', // lock was held by another instance +} + +@Entity('job_runs') +@Index(['jobName', 'startedAt']) +@Index(['status']) +export class JobRun { + @PrimaryGeneratedColumn('uuid') + id: string; + + /** Logical job name, e.g. "reconciliation", "daily-snapshot" */ + @Column({ type: 'varchar', length: 100 }) + jobName: string; + + @Column({ + type: 'enum', + enum: JobRunStatus, + default: JobRunStatus.RUNNING, + }) + status: JobRunStatus; + + /** How the job was triggered: "scheduled" | "manual" */ + @Column({ type: 'varchar', length: 50, default: 'scheduled' }) + triggeredBy: string; + + /** Arbitrary result payload (counts, cursors, etc.) */ + @Column({ type: 'jsonb', nullable: true, default: null }) + result: Record | null; + + /** Last error message when status = FAILED */ + @Column({ type: 'text', nullable: true, default: null }) + errorMessage: string | null; + + @CreateDateColumn({ type: 'timestamptz' }) + startedAt: Date; + + @Column({ type: 'timestamptz', nullable: true, default: null }) + finishedAt: Date | null; + + /** Duration in milliseconds */ + @Column({ type: 'integer', nullable: true, default: null }) + durationMs: number | null; +} diff --git a/apps/backend/src/scheduler/job-history.service.ts b/apps/backend/src/scheduler/job-history.service.ts new file mode 100644 index 00000000..c7426f65 --- /dev/null +++ b/apps/backend/src/scheduler/job-history.service.ts @@ -0,0 +1,67 @@ +import { Injectable } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { JobRun, JobRunStatus } from './entities/job-run.entity'; + +@Injectable() +export class JobHistoryService { + constructor( + @InjectRepository(JobRun) + private readonly repo: Repository, + ) {} + + /** Create a RUNNING record and return it so callers can update it later. */ + async start(jobName: string, triggeredBy = 'scheduled'): Promise { + const run = this.repo.create({ jobName, triggeredBy, status: JobRunStatus.RUNNING }); + return this.repo.save(run); + } + + /** Mark a run as SKIPPED (lock was held). */ + async markSkipped(jobName: string): Promise { + const run = this.repo.create({ + jobName, + status: JobRunStatus.SKIPPED, + finishedAt: new Date(), + durationMs: 0, + }); + await this.repo.save(run); + } + + /** Mark an existing run as COMPLETED with an optional result payload. */ + async complete( + run: JobRun, + result?: Record, + ): Promise { + run.status = JobRunStatus.COMPLETED; + run.result = result ?? null; + run.finishedAt = new Date(); + run.durationMs = run.finishedAt.getTime() - run.startedAt.getTime(); + await this.repo.save(run); + } + + /** Mark an existing run as FAILED with an error message. */ + async fail(run: JobRun, error: unknown): Promise { + run.status = JobRunStatus.FAILED; + run.errorMessage = error instanceof Error ? error.message : String(error); + run.finishedAt = new Date(); + run.durationMs = run.finishedAt.getTime() - run.startedAt.getTime(); + await this.repo.save(run); + } + + /** Fetch the N most recent runs for a given job name. */ + async getHistory(jobName: string, limit = 20): Promise { + return this.repo.find({ + where: { jobName }, + order: { startedAt: 'DESC' }, + take: limit, + }); + } + + /** Fetch the last run for a given job name. */ + async getLastRun(jobName: string): Promise { + return this.repo.findOne({ + where: { jobName }, + order: { startedAt: 'DESC' }, + }); + } +} diff --git a/apps/backend/src/scheduler/job-lock.service.ts b/apps/backend/src/scheduler/job-lock.service.ts new file mode 100644 index 00000000..33ea978c --- /dev/null +++ b/apps/backend/src/scheduler/job-lock.service.ts @@ -0,0 +1,79 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectDataSource } from '@nestjs/typeorm'; +import { DataSource } from 'typeorm'; + +/** + * Distributed job locking via PostgreSQL advisory locks. + * + * pg_try_advisory_lock(key) is session-scoped: the lock is automatically + * released when the DB connection closes, so a crashed process can never + * leave a lock permanently held. + * + * Keys are derived from a stable hash of the job name so they are + * consistent across restarts and deployments. + */ +@Injectable() +export class JobLockService { + private readonly logger = new Logger(JobLockService.name); + + constructor(@InjectDataSource() private readonly dataSource: DataSource) {} + + /** + * Try to acquire an advisory lock for the given job name. + * Returns true if the lock was acquired (safe to run), false if another + * instance already holds it (skip this run). + */ + async tryAcquire(jobName: string): Promise { + const key = this.nameToKey(jobName); + const result = await this.dataSource.query<[{ pg_try_advisory_lock: boolean }]>( + 'SELECT pg_try_advisory_lock($1) AS pg_try_advisory_lock', + [key], + ); + const acquired = result[0]?.pg_try_advisory_lock === true; + if (!acquired) { + this.logger.warn(`Job "${jobName}" skipped — lock held by another instance`); + } + return acquired; + } + + /** + * Release the advisory lock for the given job name. + * Safe to call even if the lock is not held. + */ + async release(jobName: string): Promise { + const key = this.nameToKey(jobName); + await this.dataSource.query('SELECT pg_advisory_unlock($1)', [key]); + } + + /** + * Convenience wrapper: acquire → run fn → release. + * Returns null when the lock could not be acquired (another instance running). + */ + async withLock( + jobName: string, + fn: () => Promise, + ): Promise { + const acquired = await this.tryAcquire(jobName); + if (!acquired) return null; + try { + return await fn(); + } finally { + await this.release(jobName); + } + } + + /** + * Convert a job name string to a stable 32-bit integer key. + * Uses a simple djb2-style hash — collision probability is negligible + * for the small number of job names in this application. + */ + private nameToKey(jobName: string): number { + let hash = 5381; + for (let i = 0; i < jobName.length; i++) { + hash = ((hash << 5) + hash) ^ jobName.charCodeAt(i); + hash = hash >>> 0; // keep unsigned 32-bit + } + // pg advisory lock keys are bigint; cast to signed 32-bit to stay safe + return hash | 0; + } +} diff --git a/apps/backend/src/scheduler/scheduler.module.ts b/apps/backend/src/scheduler/scheduler.module.ts new file mode 100644 index 00000000..f1e6ceed --- /dev/null +++ b/apps/backend/src/scheduler/scheduler.module.ts @@ -0,0 +1,20 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { JobRun } from './entities/job-run.entity'; +import { JobLockService } from './job-lock.service'; +import { JobHistoryService } from './job-history.service'; + +/** + * Shared module that provides distributed job locking (PostgreSQL advisory + * locks) and a unified job-run history store. + * + * Import into any feature module whose scheduler needs hardening: + * + * imports: [SchedulerModule] + */ +@Module({ + imports: [TypeOrmModule.forFeature([JobRun])], + providers: [JobLockService, JobHistoryService], + exports: [JobLockService, JobHistoryService], +}) +export class SchedulerModule {} diff --git a/apps/backend/src/snapshot/snapshot.module.ts b/apps/backend/src/snapshot/snapshot.module.ts index 7c076a7b..cf775b16 100644 --- a/apps/backend/src/snapshot/snapshot.module.ts +++ b/apps/backend/src/snapshot/snapshot.module.ts @@ -4,6 +4,7 @@ import { SnapshotRepository } from './snapshot.repository'; import { DailySnapshot } from './entities/daily-snapshot.entity'; import { SnapshotScheduler } from './snapshot.scheduler'; import { SnapshotGenerator } from './snapshot.generator'; +import { SchedulerModule } from '../scheduler/scheduler.module'; /** * Self-contained module for daily snapshot generation. @@ -20,7 +21,7 @@ import { SnapshotGenerator } from './snapshot.generator'; * ``` */ @Module({ - imports: [TypeOrmModule.forFeature([DailySnapshot])], + imports: [TypeOrmModule.forFeature([DailySnapshot]), SchedulerModule], providers: [SnapshotRepository, SnapshotGenerator, SnapshotScheduler], exports: [SnapshotGenerator], }) diff --git a/apps/backend/src/snapshot/snapshot.scheduler.ts b/apps/backend/src/snapshot/snapshot.scheduler.ts index 2deec8f5..21fd3b0f 100644 --- a/apps/backend/src/snapshot/snapshot.scheduler.ts +++ b/apps/backend/src/snapshot/snapshot.scheduler.ts @@ -1,6 +1,10 @@ import { Injectable, Logger } from '@nestjs/common'; import { Cron } from '@nestjs/schedule'; import { SnapshotGenerator } from './snapshot.generator'; +import { JobLockService } from '../scheduler/job-lock.service'; +import { JobHistoryService } from '../scheduler/job-history.service'; + +const JOB_NAME = 'daily-snapshot'; /** * Hooks `SnapshotGenerator` into NestJS's built-in task scheduler @@ -17,7 +21,11 @@ import { SnapshotGenerator } from './snapshot.generator'; export class SnapshotScheduler { private readonly logger = new Logger(SnapshotScheduler.name); - constructor(private readonly generator: SnapshotGenerator) {} + constructor( + private readonly generator: SnapshotGenerator, + private readonly jobLock: JobLockService, + private readonly jobHistory: JobHistoryService, + ) {} /** * Nightly snapshot job — fires at 01:00 UTC. @@ -27,18 +35,27 @@ export class SnapshotScheduler { * '0 2 * * *' = 02:00 every day * CronExpression.EVERY_DAY_AT_1AM (same as above, named constant) */ - @Cron('0 1 * * *', { timeZone: 'UTC', name: 'daily-snapshot' }) + @Cron('0 1 * * *', { timeZone: 'UTC', name: JOB_NAME }) async handleDailySnapshot(): Promise { this.logger.log('Nightly snapshot job triggered'); + const acquired = await this.jobLock.tryAcquire(JOB_NAME); + if (!acquired) { + await this.jobHistory.markSkipped(JOB_NAME); + return; + } + + const run = await this.jobHistory.start(JOB_NAME); try { const result = await this.generator.generateForYesterday(); - this.logger.log( - `Nightly snapshot job finished: ${JSON.stringify(result)}`, - ); + await this.jobHistory.complete(run, result as Record); + this.logger.log(`Nightly snapshot job finished: ${JSON.stringify(result)}`); } catch (err) { // Log but don't rethrow — a failed snapshot job must not crash the process. + await this.jobHistory.fail(run, err); this.logger.error('Nightly snapshot job failed', (err as Error).stack); + } finally { + await this.jobLock.release(JOB_NAME); } } } From 390560d10674e6f07590723ad730ec4e31a525cd Mon Sep 17 00:00:00 2001 From: Umeokonkwo Samuel Date: Tue, 26 May 2026 19:28:14 +0100 Subject: [PATCH 2/6] Finalize grant analytics UI and fix mobile Expo tsconfig --- apps/backend/src/grants/dto/grants.dto.ts | 28 ++- apps/backend/src/grants/grants.controller.ts | 5 + .../backend/src/grants/grants.service.spec.ts | 50 +++++ apps/backend/src/grants/grants.service.ts | 146 ++++++++++++-- apps/backend/tsconfig.json | 2 +- apps/mobile/app/(tabs)/grants/[id].tsx | 184 +++++------------- apps/mobile/expo/tsconfig.base.json | 22 +++ apps/mobile/lib/grants.ts | 8 + apps/mobile/tsconfig.json | 2 +- 9 files changed, 292 insertions(+), 155 deletions(-) create mode 100644 apps/mobile/expo/tsconfig.base.json diff --git a/apps/backend/src/grants/dto/grants.dto.ts b/apps/backend/src/grants/dto/grants.dto.ts index 5b235020..c5a36d29 100644 --- a/apps/backend/src/grants/dto/grants.dto.ts +++ b/apps/backend/src/grants/dto/grants.dto.ts @@ -98,8 +98,34 @@ export interface ProjectQfDto { estimatedMatch: string; } +export interface ProjectAllocationDto extends ProjectQfDto { + contributionPercentage: string; + qfPercentage: string; + allocationPercentage: string; +} + +export interface RoundParticipationMetricsDto { + totalContributors: number; + totalContributionAmount: string; + totalContributionRecords: number; + totalProjectsWithContributions: number; + averageContributionPerContributor: string; + averageContributionPerProject: string; +} + +export interface ContributionRecordDto { + projectId: number; + contributorPublicKey: string; + amount: string; +} + export interface RoundSummaryDto { round: RoundDto; poolBalance: string; - projects: ProjectQfDto[]; + participationMetrics: RoundParticipationMetricsDto; + projects: ProjectAllocationDto[]; +} + +export interface RoundExportDto extends RoundSummaryDto { + contributions: ContributionRecordDto[]; } diff --git a/apps/backend/src/grants/grants.controller.ts b/apps/backend/src/grants/grants.controller.ts index 1e894660..8d5b06bf 100644 --- a/apps/backend/src/grants/grants.controller.ts +++ b/apps/backend/src/grants/grants.controller.ts @@ -42,6 +42,11 @@ export class GrantsController { return this.grantsService.getRoundSummary(id); } + @Get('rounds/:id/export') + getRoundExport(@Param('id', ParseIntPipe) id: number) { + return this.grantsService.getRoundExport(id); + } + @Post('rounds') @UseGuards(JwtAuthGuard, RolesGuard) @Roles(UserRole.ADMIN) diff --git a/apps/backend/src/grants/grants.service.spec.ts b/apps/backend/src/grants/grants.service.spec.ts index 799a8239..859c67b4 100644 --- a/apps/backend/src/grants/grants.service.spec.ts +++ b/apps/backend/src/grants/grants.service.spec.ts @@ -382,4 +382,54 @@ describe('GrantsService', () => { ); expect(total).toBe(BigInt(summary.poolBalance)); }); + + it('adds participation metrics and provides an exportable round dataset', () => { + const round = service.createRound({ + name: 'R', + tokenAddress: 'GT', + startTime: past, + endTime: future, + }); + service.fundPool({ + roundId: round.id, + funderPublicKey: 'GF', + amount: '1000', + }); + service.approveProject({ roundId: round.id, projectId: 1 }); + service.approveProject({ roundId: round.id, projectId: 2 }); + + service.recordContribution({ + roundId: round.id, + projectId: 1, + contributorPublicKey: 'GA', + amount: '100', + }); + service.recordContribution({ + roundId: round.id, + projectId: 1, + contributorPublicKey: 'GB', + amount: '100', + }); + service.recordContribution({ + roundId: round.id, + projectId: 2, + contributorPublicKey: 'GC', + amount: '200', + }); + + const summary = service.getRoundSummary(round.id); + expect(summary.participationMetrics.totalContributors).toBe(3); + expect(summary.participationMetrics.totalContributionAmount).toBe('400'); + expect(summary.participationMetrics.totalContributionRecords).toBe(3); + expect(summary.participationMetrics.totalProjectsWithContributions).toBe(2); + expect(summary.projects.every((project) => project.contributionPercentage !== undefined)).toBe(true); + + const exportData = service.getRoundExport(round.id); + expect(exportData.contributions).toHaveLength(3); + expect(exportData.contributions[0]).toMatchObject({ + projectId: 1, + contributorPublicKey: 'GA', + amount: '100', + }); + }); }); diff --git a/apps/backend/src/grants/grants.service.ts b/apps/backend/src/grants/grants.service.ts index bb23ab3a..5ffdc7ab 100644 --- a/apps/backend/src/grants/grants.service.ts +++ b/apps/backend/src/grants/grants.service.ts @@ -8,7 +8,11 @@ import { ConfigService } from '@nestjs/config'; import { RoundDto, ProjectQfDto, + ProjectAllocationDto, + RoundParticipationMetricsDto, + ContributionRecordDto, RoundSummaryDto, + RoundExportDto, CreateRoundDto, FundPoolDto, ApproveProjectDto, @@ -210,58 +214,166 @@ export class GrantsService { return intPart + remainder; } - getRoundSummary(roundId: number): RoundSummaryDto { - const record = this.getRecord(roundId); + private formatPercentage( + part: bigint, + total: bigint, + decimals = 2, + ): string { + if (total === 0n) return `0.${'0'.repeat(decimals)}`; + const scale = 10n ** BigInt(decimals); + const value = (part * 100n * scale + total / 2n) / total; + const integer = value / scale; + const fraction = (value % scale).toString().padStart(decimals, '0'); + return `${integer}.${fraction}`; + } - // Compute all QF scores - const scores = new Map(); - let totalQf = 0n; + private computeParticipationMetrics( + record: RoundRecord, + ): RoundParticipationMetricsDto { + let totalContributionAmount = 0n; + let totalContributionRecords = 0; + let totalProjectsWithContributions = 0; + const uniqueContributors = new Set(); for (const pid of record.eligibleProjects) { - const contribs = - record.contributions.get(pid) ?? new Map(); - const score = this.computeQfScore(contribs); - scores.set(pid, score); - totalQf += score; + const contribs = record.contributions.get(pid) ?? new Map(); + const projectTotal = [...contribs.values()].reduce( + (a: bigint, b: bigint) => a + b, + 0n, + ); + if (projectTotal > 0n) { + totalProjectsWithContributions += 1; + } + totalContributionAmount += projectTotal; + totalContributionRecords += contribs.size; + for (const contributor of contribs.keys()) { + uniqueContributors.add(contributor); + } } - const projects: ProjectQfDto[] = []; + return { + totalContributors: uniqueContributors.size, + totalContributionAmount: totalContributionAmount.toString(), + totalContributionRecords, + totalProjectsWithContributions, + averageContributionPerContributor: + uniqueContributors.size > 0 + ? (totalContributionAmount / BigInt(uniqueContributors.size)).toString() + : '0', + averageContributionPerProject: + totalProjectsWithContributions > 0 + ? (totalContributionAmount / + BigInt(totalProjectsWithContributions)).toString() + : '0', + }; + } + + private buildProjectAllocations( + record: RoundRecord, + scores: Map, + totalQf: bigint, + totalContributionAmount: bigint, + ): ProjectAllocationDto[] { + const allocations: ProjectAllocationDto[] = []; for (const pid of record.eligibleProjects) { - const contribs = - record.contributions.get(pid) ?? new Map(); + const contribs = record.contributions.get(pid) ?? new Map(); const score = scores.get(pid) ?? 0n; const totalContribs = [...contribs.values()].reduce( (a: bigint, b: bigint) => a + b, 0n, ); - const estimatedMatch = totalQf > 0n && record.totalPool > 0n ? (record.totalPool * score) / totalQf : 0n; - projects.push({ + allocations.push({ projectId: pid, qfScore: score.toString(), totalContributions: totalContribs.toString(), contributorCount: contribs.size, estimatedMatch: estimatedMatch.toString(), + contributionPercentage: this.formatPercentage( + totalContribs, + totalContributionAmount, + ), + qfPercentage: this.formatPercentage(score, totalQf), + allocationPercentage: this.formatPercentage( + estimatedMatch, + record.totalPool, + ), }); } - // Sort by estimated match descending - projects.sort((a, b) => + allocations.sort((a, b) => Number(BigInt(b.estimatedMatch) - BigInt(a.estimatedMatch)), ); + return allocations; + } + + private buildContributionRecords( + record: RoundRecord, + ): ContributionRecordDto[] { + const contributions: ContributionRecordDto[] = []; + + for (const pid of record.eligibleProjects) { + const contribs = record.contributions.get(pid) ?? new Map(); + for (const [contributorPublicKey, amount] of contribs.entries()) { + contributions.push({ + projectId: pid, + contributorPublicKey, + amount: amount.toString(), + }); + } + } + + contributions.sort((a, b) => + a.projectId - b.projectId || a.contributorPublicKey.localeCompare(b.contributorPublicKey), + ); + return contributions; + } + + getRoundSummary(roundId: number): RoundSummaryDto { + const record = this.getRecord(roundId); + + const scores = new Map(); + let totalQf = 0n; + + for (const pid of record.eligibleProjects) { + const contribs = + record.contributions.get(pid) ?? new Map(); + const score = this.computeQfScore(contribs); + scores.set(pid, score); + totalQf += score; + } + + const participationMetrics = this.computeParticipationMetrics(record); + const totalContributionAmount = BigInt(participationMetrics.totalContributionAmount); + const projects = this.buildProjectAllocations( + record, + scores, + totalQf, + totalContributionAmount, + ); return { round: this.toRoundDto(record), poolBalance: record.totalPool.toString(), + participationMetrics, projects, }; } + getRoundExport(roundId: number): RoundExportDto { + const summary = this.getRoundSummary(roundId); + const record = this.getRecord(roundId); + return { + ...summary, + contributions: this.buildContributionRecords(record), + }; + } + distribute(dto: DistributeDto): { totalDistributed: string; allocations: { projectId: number; owner: string; amount: string }[]; diff --git a/apps/backend/tsconfig.json b/apps/backend/tsconfig.json index 418471fd..f5971197 100644 --- a/apps/backend/tsconfig.json +++ b/apps/backend/tsconfig.json @@ -1,7 +1,7 @@ { "compilerOptions": { "module": "nodenext", - "moduleResolution": "nodenext", + "moduleResolution": "nodenext", "resolvePackageJsonExports": true, "esModuleInterop": true, "isolatedModules": true, diff --git a/apps/mobile/app/(tabs)/grants/[id].tsx b/apps/mobile/app/(tabs)/grants/[id].tsx index c1221c97..396490a9 100644 --- a/apps/mobile/app/(tabs)/grants/[id].tsx +++ b/apps/mobile/app/(tabs)/grants/[id].tsx @@ -42,7 +42,13 @@ function InfoRow({ ); } -function QfBar({ share, colors }: { share: number; colors: ReturnType['colors'] }) { +function QfBar({ + share, + colors, +}: { + share: number; + colors: ReturnType['colors']; +}) { return ( - + {t('grants.project')} #{item.projectId} @@ -178,132 +188,11 @@ export default function GrantRoundDetailScreen() { accessible accessibilityLabel={t('errors.error')} /> - - {error ?? t('errors.couldnt_load', { item: 'round' })} - - void fetchSummary()} - activeOpacity={0.8} - accessibilityRole="button" - accessibilityLabel={t('common.retry')} - > - {t('common.retry')} - - - ); - } - - const { round, poolBalance, projects } = summary; - const endDate = new Date(round.endTime * 1000).toLocaleDateString(); - const startDate = new Date(round.startTime * 1000).toLocaleDateString(); - - return ( - - - - {round.name} - - - - {roundStatusLabel(round.status, t)} - - - - - - {t('grants.matching_pool')} - - - {formatTokenAmount(poolBalance)} XLM - - - {t('grants.qf_explanation')} - - - - - - - - - - - {t('grants.qf_explanation')} - - - - - {t('grant_detail.estimated_allocations')} - - - {projects.length === 0 ? ( - - {t('grants.no_rounds')} - - ) : ( - projects.map((p, idx) => ( - - )) - )} - - - ); -} - - } catch { - setError(t('errors.something_went_wrong')); - } finally { - setIsLoading(false); - } - }, [roundId, t]); - - useEffect(() => { - void fetchSummary(); - }, [fetchSummary]); - - if (isLoading) { - return ( - - - - ); - } - - if (error || !summary) { - return ( - - - {error ?? t('errors.couldnt_load', { item: 'round' })} - {t('common.retry')} + + {t('common.retry')} + ); @@ -326,7 +217,6 @@ export default function GrantRoundDetailScreen() { return ( - {/* Header */} {round.name} @@ -336,7 +226,6 @@ export default function GrantRoundDetailScreen() { - {/* Pool card */} - {/* Round info */} - + + + + + - {/* QF explanation */} - {/* Project allocations */} - + {t('grant_detail.estimated_allocations')} {projects.length === 0 ? ( - {t('grants.no_projects')} + {t('grants.no_rounds')} ) : ( projects.map((p, idx) => ( diff --git a/apps/mobile/expo/tsconfig.base.json b/apps/mobile/expo/tsconfig.base.json new file mode 100644 index 00000000..12658f8c --- /dev/null +++ b/apps/mobile/expo/tsconfig.base.json @@ -0,0 +1,22 @@ +{ + "compilerOptions": { + "target": "ES2020", + "lib": ["es2020", "dom"], + "jsx": "react-native", + "module": "esnext", + "moduleResolution": "bundler", + "allowJs": true, + "noEmit": true, + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "resolveJsonModule": true, + "isolatedModules": true, + "allowSyntheticDefaultImports": true, + "forceConsistentCasingInFileNames": true, + "baseUrl": ".", + "paths": { + "@/*": ["./*"] + } + } +} diff --git a/apps/mobile/lib/grants.ts b/apps/mobile/lib/grants.ts index e5e75917..2f4feb9f 100644 --- a/apps/mobile/lib/grants.ts +++ b/apps/mobile/lib/grants.ts @@ -25,6 +25,14 @@ export interface ProjectQf { export interface RoundSummary { round: GrantRound; poolBalance: string; + participationMetrics: { + totalContributors: number; + totalContributionAmount: string; + totalContributionRecords: number; + totalProjectsWithContributions: number; + averageContributionPerContributor: string; + averageContributionPerProject: string; + }; projects: ProjectQf[]; } diff --git a/apps/mobile/tsconfig.json b/apps/mobile/tsconfig.json index 4143f351..b4020f28 100644 --- a/apps/mobile/tsconfig.json +++ b/apps/mobile/tsconfig.json @@ -21,5 +21,5 @@ }, "include": ["**/*.ts", "**/*.tsx"], "exclude": ["node_modules"], - "extends": "expo/tsconfig.base.json" + "extends": "./expo/tsconfig.base.json" } From 005bfd91a67fc25f6795bc0c4d932aab8773dfee Mon Sep 17 00:00:00 2001 From: Umeokonkwo Samuel Date: Wed, 27 May 2026 00:19:15 +0100 Subject: [PATCH 3/6] issue #735: add matching round analytics (participation metrics, allocation breakdown, export); comment unused import --- apps/backend/src/grants/grants.service.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/apps/backend/src/grants/grants.service.ts b/apps/backend/src/grants/grants.service.ts index 5ffdc7ab..5f656828 100644 --- a/apps/backend/src/grants/grants.service.ts +++ b/apps/backend/src/grants/grants.service.ts @@ -7,7 +7,7 @@ import { import { ConfigService } from '@nestjs/config'; import { RoundDto, - ProjectQfDto, + // ProjectQfDto, ProjectAllocationDto, RoundParticipationMetricsDto, ContributionRecordDto, @@ -213,7 +213,6 @@ export class GrantsService { low > 0n ? ((value - low * low) * scale) / (2n * low) : 0n; return intPart + remainder; } - private formatPercentage( part: bigint, total: bigint, From eaadc456c1e174ddc939b129e78d03d8612fdaf5 Mon Sep 17 00:00:00 2001 From: Timi Date: Wed, 27 May 2026 00:26:55 +0100 Subject: [PATCH 4/6] feat: integrate price fetching functionality with new PriceFetcher module --- .../data-processing/src/ingestion/__init__.py | 2 + .../src/ingestion/price_fetcher.py | 223 ++++++++++++++++++ apps/data-processing/src/main.py | 31 ++- .../tests/test_price_fetcher.py | 100 ++++++++ 4 files changed, 349 insertions(+), 7 deletions(-) create mode 100644 apps/data-processing/src/ingestion/price_fetcher.py create mode 100644 apps/data-processing/tests/test_price_fetcher.py diff --git a/apps/data-processing/src/ingestion/__init__.py b/apps/data-processing/src/ingestion/__init__.py index a649c9a2..2bff5394 100644 --- a/apps/data-processing/src/ingestion/__init__.py +++ b/apps/data-processing/src/ingestion/__init__.py @@ -10,6 +10,7 @@ get_asset_volume, get_network_overview, ) +from .price_fetcher import PriceFetcher from .social_fetcher import ( SocialFetcher, SocialPost, @@ -29,6 +30,7 @@ "TransactionRecord", "get_asset_volume", "get_network_overview", + "PriceFetcher", # Social media fetchers "SocialFetcher", "SocialPost", diff --git a/apps/data-processing/src/ingestion/price_fetcher.py b/apps/data-processing/src/ingestion/price_fetcher.py new file mode 100644 index 00000000..1ad533ef --- /dev/null +++ b/apps/data-processing/src/ingestion/price_fetcher.py @@ -0,0 +1,223 @@ +""" +Off-chain price fetcher for Soroban pricing adapter feeds. + +This module supports fetching USD prices for supported Stellar assets, +scaling them to the pricing adapter base decimals, and handling failures +with stale cache fallback. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +import requests +from requests.exceptions import RequestException + +logger = logging.getLogger(__name__) + +BASE_DECIMALS = 7 +DEFAULT_CACHE_TTL_SECONDS = 300 +DEFAULT_STALE_TTL_SECONDS = 600 +DEFAULT_REQUEST_TIMEOUT = 10 + +COINGECKO_URL = "https://api.coingecko.com/api/v3/simple/price" +COINCAP_URL = "https://api.coincap.io/v2/assets" + +SUPPORTED_ASSETS: Dict[str, Dict[str, Any]] = { + "XLM": { + "coingecko_id": "stellar", + "coincap_id": "stellar", + "asset_decimals": 7, + "asset_issuer": None, + }, + "USDC": { + "coingecko_id": "usd-coin", + "coincap_id": "usd-coin", + "asset_decimals": 6, + "asset_issuer": None, + }, +} + + +class PriceFetcher: + """Fetch current asset prices and prepare adapter-ready payloads.""" + + def __init__( + self, + cache_ttl_seconds: int = DEFAULT_CACHE_TTL_SECONDS, + stale_ttl_seconds: int = DEFAULT_STALE_TTL_SECONDS, + request_timeout: int = DEFAULT_REQUEST_TIMEOUT, + ): + self.cache_ttl_seconds = cache_ttl_seconds + self.stale_ttl_seconds = stale_ttl_seconds + self.request_timeout = request_timeout + self.cache: Dict[str, Dict[str, Any]] = {} + + def fetch_all_prices(self, asset_codes: Optional[List[str]] = None) -> List[Dict[str, Any]]: + """Fetch prices for supported assets and return adapter-ready values.""" + asset_codes = asset_codes or list(SUPPORTED_ASSETS.keys()) + now = datetime.now(timezone.utc) + source = "coingecko" + + try: + price_map = self._fetch_coingecko(asset_codes) + except Exception as primary_error: + logger.warning( + "Primary price source failed: %s; trying fallback endpoint.", + primary_error, + ) + source = "coincap" + try: + price_map = self._fetch_coincap(asset_codes) + except Exception as fallback_error: + logger.warning( + "Fallback price source failed: %s; using cached stale values if available.", + fallback_error, + ) + price_map = {} + source = "cache" + + results: List[Dict[str, Any]] = [] + for asset_code in asset_codes: + asset_config = SUPPORTED_ASSETS.get(asset_code) + if not asset_config: + logger.warning("Skipping unsupported asset code: %s", asset_code) + continue + + coingecko_id = asset_config["coingecko_id"] + price_usd = price_map.get(coingecko_id) + + if price_usd is not None: + scaled_price = self._scale_price(price_usd) + payload = self._build_price_payload( + asset_code=asset_code, + asset_issuer=asset_config.get("asset_issuer"), + price_usd=price_usd, + scaled_price=scaled_price, + asset_decimals=asset_config["asset_decimals"], + source=source, + timestamp=now, + is_stale=False, + ) + self.cache[asset_code] = { + "payload": payload, + "cached_at": now, + } + results.append(payload) + continue + + stale_payload = self._get_stale_payload(asset_code, now) + if stale_payload is not None: + results.append(stale_payload) + continue + + results.append( + { + "asset_code": asset_code, + "asset_issuer": asset_config.get("asset_issuer"), + "success": False, + "error": "price_unavailable", + "source": source, + "is_stale": False, + "timestamp": now.isoformat(), + } + ) + + return results + + def fetch_price(self, asset_code: str) -> Dict[str, Any]: + """Fetch the current price for a single asset.""" + return self.fetch_all_prices([asset_code])[0] + + def _fetch_coingecko(self, asset_codes: List[str]) -> Dict[str, float]: + """Fetch usd prices from CoinGecko.""" + asset_ids = self._asset_ids(asset_codes, key="coingecko_id") + response = requests.get( + COINGECKO_URL, + params={"ids": ",".join(asset_ids), "vs_currencies": "usd"}, + timeout=self.request_timeout, + ) + response.raise_for_status() + data = response.json() + prices: Dict[str, float] = {} + for asset_code in asset_codes: + asset_id = SUPPORTED_ASSETS[asset_code]["coingecko_id"] + asset_data = data.get(asset_id, {}) + usd_value = asset_data.get("usd") + if usd_value is not None: + prices[asset_id] = float(usd_value) + if not prices: + raise RequestException("CoinGecko returned no valid prices") + return prices + + def _fetch_coincap(self, asset_codes: List[str]) -> Dict[str, float]: + """Fetch usd prices from CoinCap as a fallback.""" + asset_ids = self._asset_ids(asset_codes, key="coincap_id") + response = requests.get( + COINCAP_URL, + params={"ids": ",".join(asset_ids)}, + timeout=self.request_timeout, + ) + response.raise_for_status() + data = response.json() + prices: Dict[str, float] = {} + for item in data.get("data", []): + asset_id = item.get("id") + price_usd = item.get("priceUsd") + if asset_id and price_usd: + prices[asset_id] = float(price_usd) + if not prices: + raise RequestException("CoinCap returned no valid prices") + return prices + + def _scale_price(self, price_usd: float) -> int: + return int(round(price_usd * (10 ** BASE_DECIMALS))) + + def _build_price_payload( + self, + asset_code: str, + asset_issuer: Optional[str], + price_usd: float, + scaled_price: int, + asset_decimals: int, + source: str, + timestamp: datetime, + is_stale: bool, + ) -> Dict[str, Any]: + return { + "asset_code": asset_code, + "asset_issuer": asset_issuer, + "price_usd": price_usd, + "price": scaled_price, + "asset_decimals": asset_decimals, + "base_decimals": BASE_DECIMALS, + "source": source, + "timestamp": timestamp.isoformat(), + "is_stale": is_stale, + "success": True, + } + + def _asset_ids(self, asset_codes: List[str], key: str) -> List[str]: + return [SUPPORTED_ASSETS[asset_code][key] for asset_code in asset_codes] + + def _get_stale_payload( + self, asset_code: str, now: datetime + ) -> Optional[Dict[str, Any]]: + cached = self.cache.get(asset_code) + if not cached: + return None + age = (now - cached["cached_at"]).total_seconds() + if age > self.stale_ttl_seconds: + logger.warning( + "Cached price for %s is stale (%.0fs old), discarding.", + asset_code, + age, + ) + return None + payload = cached["payload"].copy() + payload["is_stale"] = True + payload["source"] = "cache" + payload["timestamp"] = now.isoformat() + return payload diff --git a/apps/data-processing/src/main.py b/apps/data-processing/src/main.py index 8b3949a9..233b7293 100644 --- a/apps/data-processing/src/main.py +++ b/apps/data-processing/src/main.py @@ -16,6 +16,7 @@ # Import both pipeline and scheduler from src.ingestion.news_fetcher import fetch_news +from src.ingestion.price_fetcher import PriceFetcher from src.ingestion.stellar_fetcher import get_asset_volume, get_network_overview from src.validators import validate_news_article, validate_onchain_metric from src.analytics.market_analyzer import MarketAnalyzer, MarketData @@ -77,16 +78,19 @@ def run_data_pipeline(): print("1. FETCHING DATA (news + on-chain in parallel)") print("-" * 40) - with ThreadPoolExecutor(max_workers=4) as io_pool: + price_fetcher = PriceFetcher() + with ThreadPoolExecutor(max_workers=5) as io_pool: news_future = io_pool.submit(fetch_news, limit=5) vol_24h_future = io_pool.submit(get_asset_volume, "XLM", hours=24) vol_48h_future = io_pool.submit(get_asset_volume, "XLM", hours=48) network_future = io_pool.submit(get_network_overview) + price_future = io_pool.submit(price_fetcher.fetch_all_prices, ["XLM", "USDC"]) raw_news_articles = news_future.result() raw_volume_24h = vol_24h_future.result() raw_volume_48h = vol_48h_future.result() network_stats = network_future.result() + raw_price_feed = price_future.result() fetch_elapsed = time.perf_counter() - pipeline_start print(f"All fetches completed in {fetch_elapsed:.2f}s (parallel)") @@ -102,8 +106,20 @@ def run_data_pipeline(): print(f"Fetched {len(raw_news_articles)} raw → {len(news_articles)} validated articles") + print("\n2. PRICE FEED") + print("-" * 40) + if raw_price_feed: + for price_point in raw_price_feed: + status = "stale" if price_point.get("is_stale") else "fresh" + print( + f"{price_point['asset_code']}: ${price_point['price_usd']:.7f} " + f"({price_point['price']} scaled, decimals={price_point['asset_decimals']}, {status})" + ) + else: + print("Price feed unavailable") + # ── Sentiment analysis (parallel for large batches) ────────── - print("\n2. SENTIMENT ANALYSIS") + print("\n3. SENTIMENT ANALYSIS") print("-" * 40) sentiment_analyzer = SentimentAnalyzer() @@ -125,7 +141,7 @@ def run_data_pipeline(): print("No valid articles, using neutral sentiment") # ── Validate on-chain metrics ──────────────────────────────── - print("\n3. STELLAR ON-CHAIN DATA") + print("\n4. STELLAR ON-CHAIN DATA") print("-" * 40) validated_volume_24h = validate_onchain_metric({ @@ -171,8 +187,8 @@ def run_data_pipeline(): print(f"Latest Ledger: {network_stats.get('latest_ledger', 'N/A')}") print(f"Transaction Count: {network_stats.get('transaction_count', 0)}") - # Step 4: Market Analysis - print("\n4. MARKET ANALYSIS") + # Step 5: Market Analysis + print("\n5. MARKET ANALYSIS") print("-" * 40) # Create market data @@ -192,8 +208,8 @@ def run_data_pipeline(): explanation = get_explanation(score, trend) print(f"\nAnalysis: {explanation}") - # Step 5: Anomaly Detection - print("\n5. ANOMALY DETECTION") + # Step 6: Anomaly Detection + print("\n6. ANOMALY DETECTION") print("-" * 40) current_volume = float(volume_24h["total_volume"]) @@ -255,6 +271,7 @@ def run_data_pipeline(): "success": True, "news_count": len(news_articles), "volume_xlm": volume_24h["total_volume"], + "price_feed": raw_price_feed, "market_trend": trend.value, "health_score": score, "anomalies": anomalies_found, diff --git a/apps/data-processing/tests/test_price_fetcher.py b/apps/data-processing/tests/test_price_fetcher.py new file mode 100644 index 00000000..49263dc8 --- /dev/null +++ b/apps/data-processing/tests/test_price_fetcher.py @@ -0,0 +1,100 @@ +"""Unit tests for the new price fetcher module.""" + +from datetime import datetime, timezone + +import pytest + +from src.ingestion.price_fetcher import PriceFetcher, SUPPORTED_ASSETS + + +class DummyResponse: + def __init__(self, status_code: int, json_data): + self.status_code = status_code + self._json_data = json_data + + def raise_for_status(self): + if self.status_code >= 400: + raise Exception(f"HTTP {self.status_code}") + + def json(self): + return self._json_data + + +def test_fetch_all_prices_success(monkeypatch): + fetcher = PriceFetcher() + + def mock_get(url, params=None, timeout=None): + assert timeout == fetcher.request_timeout + if "coingecko" in url: + return DummyResponse( + 200, + { + "stellar": {"usd": 0.1234567}, + "usd-coin": {"usd": 1.0}, + }, + ) + raise AssertionError("Unexpected URL: %s" % url) + + monkeypatch.setattr("src.ingestion.price_fetcher.requests.get", mock_get) + prices = fetcher.fetch_all_prices(["XLM", "USDC"]) + + assert len(prices) == 2 + xlm = next(item for item in prices if item["asset_code"] == "XLM") + usdc = next(item for item in prices if item["asset_code"] == "USDC") + + assert xlm["success"] is True + assert xlm["price"] == 1234567 + assert xlm["asset_decimals"] == SUPPORTED_ASSETS["XLM"]["asset_decimals"] + assert xlm["is_stale"] is False + + assert usdc["success"] is True + assert usdc["price"] == 10000000 + assert usdc["asset_decimals"] == SUPPORTED_ASSETS["USDC"]["asset_decimals"] + assert usdc["is_stale"] is False + + +def test_fetch_all_prices_uses_cache_when_source_fails(monkeypatch): + fetcher = PriceFetcher() + now = datetime.now(timezone.utc) + stale_payload = { + "asset_code": "XLM", + "asset_issuer": None, + "price_usd": 0.12, + "price": 1200000, + "asset_decimals": SUPPORTED_ASSETS["XLM"]["asset_decimals"], + "base_decimals": 7, + "source": "coingecko", + "timestamp": now.isoformat(), + "is_stale": False, + "success": True, + } + fetcher.cache["XLM"] = {"payload": stale_payload, "cached_at": now} + + def mock_get(url, params=None, timeout=None): + raise Exception("Source unreachable") + + monkeypatch.setattr("src.ingestion.price_fetcher.requests.get", mock_get) + prices = fetcher.fetch_all_prices(["XLM"]) + + assert len(prices) == 1 + xlm = prices[0] + assert xlm["success"] is True + assert xlm["is_stale"] is True + assert xlm["source"] == "cache" + assert xlm["price"] == 1200000 + + +def test_fetch_all_prices_returns_error_when_no_source_and_no_cache(monkeypatch): + fetcher = PriceFetcher() + + def mock_get(url, params=None, timeout=None): + raise Exception("Source unreachable") + + monkeypatch.setattr("src.ingestion.price_fetcher.requests.get", mock_get) + prices = fetcher.fetch_all_prices(["XLM"]) + + assert len(prices) == 1 + xlm = prices[0] + assert xlm["success"] is False + assert xlm["error"] == "price_unavailable" + assert xlm["source"] == "cache" From 794d009f51a3a4ae42ebc1fd800680068691f1f4 Mon Sep 17 00:00:00 2001 From: Abdulrasaq1515 Date: Wed, 27 May 2026 00:28:51 +0100 Subject: [PATCH 5/6] fix: add JobLockService and JobHistoryService mocks to failing unit tests --- apps/backend/src/news/news-sentiment.spec.ts | 18 ++++++++++++++++++ apps/backend/src/outbox/outbox.service.spec.ts | 5 +++++ 2 files changed, 23 insertions(+) diff --git a/apps/backend/src/news/news-sentiment.spec.ts b/apps/backend/src/news/news-sentiment.spec.ts index 51de2771..756d24e4 100644 --- a/apps/backend/src/news/news-sentiment.spec.ts +++ b/apps/backend/src/news/news-sentiment.spec.ts @@ -10,6 +10,8 @@ import { NewsSentimentService } from './news-sentiment.services'; import { NewsProviderService } from './news-provider.service'; import { CacheService } from '../cache/cache.service'; import { QueryProfilerService } from '../common/profiling/query-profiler.service'; +import { JobLockService } from '../scheduler/job-lock.service'; +import { JobHistoryService } from '../scheduler/job-history.service'; // ─── Types ─────────────────────────────────────────────────────────────────── @@ -86,6 +88,14 @@ describe('NewsSentimentService', () => { update: jest.fn(), }, }, + { + provide: JobLockService, + useValue: { tryAcquire: jest.fn().mockResolvedValue(true), release: jest.fn().mockResolvedValue(undefined) }, + }, + { + provide: JobHistoryService, + useValue: { start: jest.fn().mockResolvedValue({ startedAt: new Date() }), complete: jest.fn().mockResolvedValue(undefined), fail: jest.fn().mockResolvedValue(undefined), markSkipped: jest.fn().mockResolvedValue(undefined) }, + }, ], }).compile(); @@ -267,6 +277,14 @@ describe('NewsService - sentiment methods', () => { provide: QueryProfilerService, useValue: mockQueryProfilerService, }, + { + provide: JobLockService, + useValue: { tryAcquire: jest.fn().mockResolvedValue(true), release: jest.fn().mockResolvedValue(undefined) }, + }, + { + provide: JobHistoryService, + useValue: { start: jest.fn().mockResolvedValue({ startedAt: new Date() }), complete: jest.fn().mockResolvedValue(undefined), fail: jest.fn().mockResolvedValue(undefined), markSkipped: jest.fn().mockResolvedValue(undefined) }, + }, ], }).compile(); diff --git a/apps/backend/src/outbox/outbox.service.spec.ts b/apps/backend/src/outbox/outbox.service.spec.ts index c06d6ea5..e927c929 100644 --- a/apps/backend/src/outbox/outbox.service.spec.ts +++ b/apps/backend/src/outbox/outbox.service.spec.ts @@ -2,6 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing'; import { getRepositoryToken } from '@nestjs/typeorm'; import { OutboxService } from './outbox.service'; import { OutboxEvent, OutboxEventStatus } from './outbox-event.entity'; +import { JobLockService } from '../scheduler/job-lock.service'; const mockRepo = () => ({ create: jest.fn(), @@ -20,6 +21,10 @@ describe('OutboxService', () => { providers: [ OutboxService, { provide: getRepositoryToken(OutboxEvent), useFactory: mockRepo }, + { + provide: JobLockService, + useValue: { tryAcquire: jest.fn().mockResolvedValue(true), release: jest.fn().mockResolvedValue(undefined) }, + }, ], }).compile(); From cd49ed980a80f4ac119acadfdde4d1289b9eb7c3 Mon Sep 17 00:00:00 2001 From: Abdulrasaq1515 Date: Wed, 27 May 2026 00:38:19 +0100 Subject: [PATCH 6/6] fix: replace unsafe SnapshotRunResult cast with explicit serialization in snapshot scheduler --- apps/backend/src/snapshot/snapshot.scheduler.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/backend/src/snapshot/snapshot.scheduler.ts b/apps/backend/src/snapshot/snapshot.scheduler.ts index 21fd3b0f..25349bbc 100644 --- a/apps/backend/src/snapshot/snapshot.scheduler.ts +++ b/apps/backend/src/snapshot/snapshot.scheduler.ts @@ -48,7 +48,12 @@ export class SnapshotScheduler { const run = await this.jobHistory.start(JOB_NAME); try { const result = await this.generator.generateForYesterday(); - await this.jobHistory.complete(run, result as Record); + await this.jobHistory.complete(run, { + date: result.date.toISOString(), + assetRowsWritten: result.assetRowsWritten, + globalRowWritten: result.globalRowWritten, + durationMs: result.durationMs, + }); this.logger.log(`Nightly snapshot job finished: ${JSON.stringify(result)}`); } catch (err) { // Log but don't rethrow — a failed snapshot job must not crash the process.