From b48bd4f1f0c8f44ba13aafc451e982a74951595a Mon Sep 17 00:00:00 2001 From: Amr Mohammed El-Sheraey <141947355+AmrMsCLL@users.noreply.github.com> Date: Fri, 16 Oct 2026 14:13:39 +0300 Subject: [PATCH] fix(jobs): bound outbound fetches and keep webhook deliveries behind other jobs --- .../src/bookmarks/bookmarks.controller.ts | 3 +++ .../calendar-subscriptions.service.ts | 2 +- app/server/src/calendar/ical.ts | 2 +- app/server/src/jobs/jobs.service.ts | 10 +++++----- app/server/src/webhooks/webhooks.service.ts | 2 +- app/server/test/ical.spec.ts | 10 ++++++++++ app/server/test/integration/records.spec.ts | 19 +++++++++++++++++++ 7 files changed, 40 insertions(+), 8 deletions(-) diff --git a/app/server/src/bookmarks/bookmarks.controller.ts b/app/server/src/bookmarks/bookmarks.controller.ts index ee86e14e..ca3390d2 100644 --- a/app/server/src/bookmarks/bookmarks.controller.ts +++ b/app/server/src/bookmarks/bookmarks.controller.ts @@ -3,6 +3,7 @@ import { Scope } from '../access-tokens/scopes'; import { AuthenticatedUser } from '../auth/auth.types'; import { CurrentUser } from '../auth/current-user.decorator'; import { JwtAuthGuard } from '../auth/jwt-auth.guard'; +import { RateLimit } from '../redis/rate-limit.guard'; import { UuidPipe } from '../validation/uuid.pipe'; import { BookmarksService } from './bookmarks.service'; import { CreateBookmarkDto, ListBookmarksDto, SaveArticleDto, UpdateBookmarkDto } from './dto/bookmarks.dto'; @@ -20,6 +21,7 @@ export class BookmarksController { @Post('spaces/:spaceId/bookmarks') @Scope('items:write') + @RateLimit('bookmark-fetch', 60, 3_600) create(@CurrentUser() user: AuthenticatedUser, @Param('spaceId', UuidPipe) spaceId: string, @Body() input: CreateBookmarkDto) { return this.bookmarks.create(user.id, spaceId, input); } @@ -38,6 +40,7 @@ export class BookmarksController { @Post('bookmarks/:id/fetch') @Scope('items:write') + @RateLimit('bookmark-fetch', 60, 3_600) fetchAgain(@CurrentUser() user: AuthenticatedUser, @Param('id', UuidPipe) id: string) { return this.bookmarks.fetchAgain(user.id, id); } diff --git a/app/server/src/calendar/calendar-subscriptions.service.ts b/app/server/src/calendar/calendar-subscriptions.service.ts index 9d777a58..a66696b6 100644 --- a/app/server/src/calendar/calendar-subscriptions.service.ts +++ b/app/server/src/calendar/calendar-subscriptions.service.ts @@ -114,7 +114,7 @@ export class CalendarSubscriptionsService implements OnModuleInit { } async refreshMine(userId: string, now: Date) { - return { refreshed: await this.refreshStale({ space: this.access.spacesOf(userId) }, freshForMs, now) }; + return { refreshed: await this.refreshStale({ space: this.access.spacesOf(userId) }, freshForMs, now, refreshesPerRun) }; } async refreshDue(now: Date) { diff --git a/app/server/src/calendar/ical.ts b/app/server/src/calendar/ical.ts index cd3f0e3d..927fe0af 100644 --- a/app/server/src/calendar/ical.ts +++ b/app/server/src/calendar/ical.ts @@ -16,7 +16,7 @@ export interface FeedEvent { }; } -const escapeText = (text: string) => text.replace(/\\/g, '\\\\').replace(/;/g, '\\;').replace(/,/g, '\\,').replace(/\r?\n/g, '\\n'); +const escapeText = (text: string) => text.replace(/\\/g, '\\\\').replace(/;/g, '\\;').replace(/,/g, '\\,').replace(/\r\n|\r|\n/g, '\\n'); const compactDate = (date: Date) => dateText(date).replaceAll('-', ''); const clock = (time: string) => `${time.replace(':', '')}00`; diff --git a/app/server/src/jobs/jobs.service.ts b/app/server/src/jobs/jobs.service.ts index b70f6222..b7de1c33 100644 --- a/app/server/src/jobs/jobs.service.ts +++ b/app/server/src/jobs/jobs.service.ts @@ -20,6 +20,7 @@ interface ClaimedJob { export class JobsService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(JobsService.name); private readonly handlers = new Map(); + private readonly lastKinds = new Set(); private timer?: NodeJS.Timeout; private running = false; @@ -34,8 +35,9 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { clearInterval(this.timer); } - handle(kind: string, handler: JobHandler) { + handle(kind: string, handler: JobHandler, { runLast = false } = {}) { this.handlers.set(kind, handler); + if (runLast) this.lastKinds.add(kind); } schedule( @@ -51,9 +53,7 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { if (this.running) return; this.running = true; try { - for (const job of await this.claim(now)) { - await this.run(job); - } + await Promise.all((await this.claim(now)).map((job) => this.run(job))); } catch (error) { this.logger.error('Claiming due jobs failed; retrying on the next poll', error); } finally { @@ -73,7 +73,7 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { AND "run_at" <= ${at}::timestamptz AT TIME ZONE 'UTC' AND ("claimed_at" IS NULL OR "claimed_at" < (${at}::timestamptz AT TIME ZONE 'UTC') - make_interval(mins => ${claimTimeoutMinutes})) - ORDER BY "run_at" + ORDER BY "kind" = ANY(${[...this.lastKinds]}::text[]), "run_at" LIMIT ${batchSize} FOR UPDATE SKIP LOCKED ) diff --git a/app/server/src/webhooks/webhooks.service.ts b/app/server/src/webhooks/webhooks.service.ts index 96a3356a..c6018b11 100644 --- a/app/server/src/webhooks/webhooks.service.ts +++ b/app/server/src/webhooks/webhooks.service.ts @@ -49,7 +49,7 @@ export class WebhooksService implements OnModuleInit, OnModuleDestroy { ) {} onModuleInit() { - this.jobs.handle(deliverKind, (payload) => this.deliver((payload as { deliveryId: string }).deliveryId)); + this.jobs.handle(deliverKind, (payload) => this.deliver((payload as { deliveryId: string }).deliveryId), { runLast: true }); this.maintenance.register((now) => this.prisma.webhookDelivery.deleteMany({ where: { createdAt: { lt: new Date(now.getTime() - keepDays * dayMs) } } })); this.timer = setInterval(() => void this.scan(), scanEveryMs); this.timer.unref(); diff --git a/app/server/test/ical.spec.ts b/app/server/test/ical.spec.ts index 4cae6d10..8b2d668e 100644 --- a/app/server/test/ical.spec.ts +++ b/app/server/test/ical.spec.ts @@ -50,4 +50,14 @@ describe('calendarText', () => { expect(lines.every((line) => Buffer.byteLength(line) <= 75)).toBe(true); expect(text.replace(/\r\n /g, '')).toContain(`SUMMARY:${longTitle}\r\n`); }); + + it('escapes every kind of line break so text cannot add properties', () => { + const text = calendarText( + 'Agenda', + [{ id: 'sneaky', title: 'Lunch\rX-A:1\r\nX-B:2\nX-C:3', updatedAt: new Date('2026-10-07T10:00:00Z'), event }], + (path) => `https://strata.example${path}`, + ); + + expect(text.split(/\r\n|\r|\n/)).toContain('SUMMARY:Lunch\\nX-A:1\\nX-B:2\\nX-C:3'); + }); }); diff --git a/app/server/test/integration/records.spec.ts b/app/server/test/integration/records.spec.ts index 8404c0cb..eecac3c4 100644 --- a/app/server/test/integration/records.spec.ts +++ b/app/server/test/integration/records.spec.ts @@ -121,6 +121,25 @@ describe('Activity, the security log, the inbox and background work against Post await strata.prisma.scheduledJob.deleteMany({ where: { id: { in: [ok.id, later.id, failing.id] } } }); }); + it('runs jobs registered to run last only after every other due job', async () => { + const jobs = strata.service(JobsService); + const ran: string[] = []; + jobs.handle('integration.first', async () => { + ran.push('first'); + }); + jobs.handle('integration.last', async () => { + ran.push('last'); + }, { runLast: true }); + const now = new Date(); + const last = await Promise.all(Array.from({ length: 25 }, (_, index) => jobs.schedule('integration.last', new Date(now.getTime() - 60_000 + index)))); + const first = await jobs.schedule('integration.first', new Date(now.getTime() - 1_000)); + + await jobs.runDue(now); + expect(ran.filter((kind) => kind === 'first')).toEqual(['first']); + expect(ran).toHaveLength(20); + await strata.prisma.scheduledJob.deleteMany({ where: { id: { in: [first.id, ...last.map((job) => job.id)] } } }); + }); + it('empties the trash after 30 days and keeps newer items', async () => { const owner = await member('tidier'); const old = await strata.item(owner.personalSpaceId, 'Old');