fix(jobs): bound outbound fetches and keep webhook deliveries behind other jobs

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);
   }

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) {

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`;
 

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<string, JobHandler>();
+  private readonly lastKinds = new Set<string>();
   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
       )

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();

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');
+  });
 });

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');