Skip to content

Commit 88bc2e7

Browse files
kkartunovkkartunov
authored
Merge pull request #184 from topcoder-platform/develop
[PROD] - Security & Bug Fixes
2 parents 7a0b4cc + 5fdbf5f commit 88bc2e7

File tree

11 files changed

+2649
-2748
lines changed

11 files changed

+2649
-2748
lines changed

‎.circleci/config.yml‎

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -74,11 +74,7 @@ workflows:
7474
branches:
7575
only:
7676
- develop
77-
- feat/ai-workflows
78-
- pm-1955_2
79-
- re-try-failed-jobs
80-
- pm-2539
81-
- pm-2177_fixes
77+
- pm-2660
8278

8379

8480
- 'build-prod':

‎docs/ai-workflow-flow.full-diagram.png‎

583 KB
Loading

‎docs/ai-workflow-flow.pgboss-diagram.png‎

483 KB
Loading

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@
4949
"pg-boss": "^11.0.5",
5050
"reflect-metadata": "^0.2.2",
5151
"rxjs": "^7.8.1",
52-
"tc-core-library-js": "appirio-tech/tc-core-library-js.git#security"
52+
"tc-core-library-js": "topcoder-platform/tc-core-library-js#master"
5353
},
5454
"devDependencies": {
5555
"@eslint/eslintrc": "^3.2.0",

‎pnpm-lock.yaml‎

Lines changed: 2518 additions & 2690 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎prisma/migrations/20251203163619_cascade_delete_workflowrun_deps/migration.sql‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
-- DropForeignKey
2+
ALTER TABLE "aiWorkflowRunItem" DROP CONSTRAINT "aiWorkflowRunItem_workflowRunId_fkey";
3+
4+
-- DropForeignKey
5+
ALTER TABLE "aiWorkflowRunItemComment" DROP CONSTRAINT "aiWorkflowRunItemComment_workflowRunItemId_fkey";
6+
7+
-- AddForeignKey
8+
ALTER TABLE "aiWorkflowRunItem" ADD CONSTRAINT "aiWorkflowRunItem_workflowRunId_fkey" FOREIGN KEY ("workflowRunId") REFERENCES "aiWorkflowRun"("id") ON DELETE CASCADE ON UPDATE CASCADE;
9+
10+
-- AddForeignKey
11+
ALTER TABLE "aiWorkflowRunItemComment" ADD CONSTRAINT "aiWorkflowRunItemComment_workflowRunItemId_fkey" FOREIGN KEY ("workflowRunItemId") REFERENCES "aiWorkflowRunItem"("id") ON DELETE CASCADE ON UPDATE CASCADE;

‎src/api/appeal/appeal.service.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -267,6 +267,11 @@ export class AppealService {
267267
);
268268
}
269269

270+
// Check if the appeal phase is open while updating appeals
271+
if (!isPrivileged && challengeId) {
272+
await this.challengeApiService.validateAppealSubmission(challengeId);
273+
}
274+
270275
if (!isPrivileged) {
271276
await this.ensureChallengeAllowsAppealChange(challengeId, {
272277
logContext: 'updateAppeal',

‎src/api/review-application/reviewApplication.service.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ export class ReviewApplicationService {
7171
}
7272
// make sure application role matches
7373
if (
74+
// eslint-disable-next-line @typescript-eslint/no-unsafe-enum-comparison
7475
ReviewApplicationRoleOpportunityTypeMap[dto.role] !== opportunity.type
7576
) {
7677
throw new BadRequestException(

‎src/api/submission/submission.service.ts‎

Lines changed: 35 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -576,6 +576,7 @@ export class SubmissionService {
576576
let isReviewer = false;
577577
let isCopilot = false;
578578
let isSubmitter = false;
579+
let isManager = false;
579580
if (!isOwner && submission.challengeId && uid) {
580581
try {
581582
const resources =
@@ -612,7 +613,10 @@ export class SubmissionService {
612613
if (rn.includes('submitter')) {
613614
isSubmitter = true;
614615
}
615-
if (isReviewer && isCopilot && isSubmitter) {
616+
if (rn.includes('manager')) {
617+
isManager = true;
618+
}
619+
if (isReviewer && isCopilot && isSubmitter && isManager) {
616620
break;
617621
}
618622
}
@@ -624,7 +628,7 @@ export class SubmissionService {
624628
}
625629
}
626630

627-
let canDownload = isOwner || isReviewer || isCopilot;
631+
let canDownload = isOwner || isReviewer || isCopilot || isManager;
628632

629633
if (!canDownload && isSubmitter && submission.challengeId && uid) {
630634
try {
@@ -667,7 +671,7 @@ export class SubmissionService {
667671
if (!canDownload) {
668672
throw new ForbiddenException({
669673
message:
670-
'Only the submission owner, a challenge reviewer/copilot, or an admin can download the submission',
674+
'Only the submission owner, a challenge reviewer/copilot/manager, or an admin can download the submission',
671675
code: 'FORBIDDEN_SUBMISSION_DOWNLOAD',
672676
details: {
673677
submissionId,
@@ -2287,6 +2291,34 @@ export class SubmissionService {
22872291
});
22882292
}
22892293
}
2294+
const TERMINAL_STATUSES = [
2295+
'COMPLETED',
2296+
'FAILURE',
2297+
'CANCELLED',
2298+
'SUCCESS',
2299+
];
2300+
2301+
const runs = await this.prisma.aiWorkflowRun.findMany({
2302+
where: { submissionId: id },
2303+
select: { id: true, status: true },
2304+
});
2305+
2306+
if (runs.length > 0) {
2307+
const nonTerminal = runs.filter(
2308+
(r) => !TERMINAL_STATUSES.includes(r.status),
2309+
);
2310+
2311+
if (nonTerminal.length > 0) {
2312+
throw new Error(
2313+
`Cannot delete submission: ${nonTerminal.length} workflow run(s) still active.`,
2314+
);
2315+
}
2316+
2317+
await this.prisma.aiWorkflowRun.deleteMany({
2318+
where: { submissionId: id },
2319+
});
2320+
}
2321+
22902322
await this.prisma.submission.delete({
22912323
where: { id },
22922324
});

‎src/shared/modules/global/queue-scheduler.service.ts‎

Lines changed: 77 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,10 @@ import {
77
import * as PgBoss from 'pg-boss';
88
import { policies, Queue } from 'pg-boss';
99

10+
const PGBOSS_JOB_POLLING_INTERVAL_SEC = parseFloat(
11+
process.env.PGBOSS_JOB_POLLING_INTERVAL_SEC || '10',
12+
);
13+
1014
/**
1115
* QueueSchedulerService
1216
*/
@@ -16,11 +20,6 @@ export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
1620
private boss: PgBoss;
1721
private $start;
1822

19-
private jobsHandlersMap = new Map<
20-
string,
21-
(resolution?: 'fail' | 'complete', result?: any) => void
22-
>();
23-
2423
get isEnabled() {
2524
return String(process.env.DISPATCH_AI_REVIEW_WORKFLOWS) === 'true';
2625
}
@@ -115,26 +114,7 @@ export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
115114
return;
116115
}
117116

118-
if (resolution === 'fail') {
119-
// IMPORTANT!
120-
// thes 4 operations will update the cache for the active singletons in the database
121-
// and will allow the jobs queue to go next or retry
122-
await this.boss.cancel(queueName, jobId);
123-
await this.boss.getQueueStats(queueName);
124-
await this.boss.supervise(queueName);
125-
await this.boss.resume(queueName, jobId);
126-
}
127-
128-
if (this.jobsHandlersMap.has(jobId)) {
129-
this.logger.log(
130-
`Found job handler for ${jobId}. Calling with '${resolution}' resolution.`,
131-
);
132-
this.jobsHandlersMap.get(jobId)?.call(null, resolution);
133-
this.jobsHandlersMap.delete(jobId);
134-
this.logger.log('JobHandlers left:', [...this.jobsHandlersMap.keys()]);
135-
} else {
136-
await this.boss[resolution](queueName, jobId);
137-
}
117+
await this.boss[resolution](queueName, jobId);
138118

139119
this.logger.log(`Job ${jobId} ${resolution} called.`);
140120

@@ -168,16 +148,79 @@ export class QueueSchedulerService implements OnModuleInit, OnModuleDestroy {
168148
await this.createQueue(queueName);
169149
}
170150

171-
return this.boss.work(queueName, handlerFn);
151+
/**
152+
* Continuously polls a job queue and processes jobs one at a time.
153+
*
154+
* @typeParam T - The type of the job payload/data.
155+
*
156+
* @remarks
157+
* - Fetches a single job from the queue (batchSize = 1) via `this.boss.fetch<T>(queueName, { batchSize: 1 })`.
158+
* - If a job is returned, logs the job and invokes `handlerFn([job])`, awaiting its completion.
159+
* - After each iteration (whether a job was found or not), schedules the next poll using
160+
* `setTimeout(..., PGBOSS_JOB_POLLING_INTERVAL)` to avoid deep recursion and to yield to the event loop.
161+
* - The scheduled, recursive invocation calls `poll().catch(() => {})` so that errors from those future
162+
* invocations are swallowed and do not produce unhandled promise rejections. Note that errors thrown
163+
* by the current invocation (for example from `handlerFn`) will propagate to the caller of this invocation
164+
* unless the caller handles them.
165+
*
166+
* @returns A Promise that resolves when this single poll iteration completes.
167+
*/
168+
const poll = async () => {
169+
try {
170+
// guard: ensure boss still exists and service still enabled
171+
if (!this.boss || !this.isEnabled) {
172+
this.logger.warn(
173+
`Polling for queue "${queueName}" stopped: boss not available or service disabled.`,
174+
);
175+
return;
176+
}
177+
178+
const [job] = await this.boss.fetch<T>(queueName, { batchSize: 1 });
179+
if (job) {
180+
// avoid throwing from here to keep the loop alive
181+
try {
182+
this.logger.log(
183+
`Starting job processing for job ${job.id} from queue "${queueName}"`,
184+
);
185+
this.logger.debug(
186+
`Job ${job.id} payload: ${JSON.stringify(job.data)}`,
187+
);
188+
} catch {
189+
// ignore stringify errors
190+
}
191+
192+
try {
193+
await handlerFn([job]);
194+
} catch (err) {
195+
this.logger.error(
196+
`Handler error while processing job ${job.id} from "${queueName}": ${
197+
(err && (err as Error).message) || err
198+
}`,
199+
err,
200+
);
201+
// don't rethrow so the scheduled next poll still runs
202+
}
203+
}
204+
} catch (err) {
205+
this.logger.error(
206+
`Error fetching job from queue "${queueName}": ${
207+
(err && (err as Error).message) || err
208+
}`,
209+
err,
210+
);
211+
// swallow to avoid unhandled promise rejection; next poll still scheduled
212+
} finally {
213+
// schedule next poll (non-blocking). Any errors from the scheduled invocation are logged.
214+
setTimeout(() => {
215+
poll().catch((err) =>
216+
this.logger.error('Unhandled poll error', err),
217+
);
218+
}, PGBOSS_JOB_POLLING_INTERVAL_SEC * 1000);
219+
}
220+
};
221+
222+
await poll();
172223
}),
173224
);
174225
}
175-
176-
registerJobHandler(
177-
jobId: string,
178-
handler: (resolution?: string, result?: any) => void,
179-
) {
180-
this.logger.log(`Registering job handler for job ${jobId}.`);
181-
this.jobsHandlersMap.set(jobId, handler);
182-
}
183226
}

0 commit comments

Comments
 (0)