fix(core): resolve P0/P1 audit findings across security, tenant isolation, and scheduler'

prisma/schema.prisma

@@ -209,6 +209,7 @@ model CrawlAsset {
 
   @@index([pageId])
   @@index([assetType])
+  @@index([crawlJobId])
   @@map("crawl_assets")
 }
 
@@ -275,6 +276,9 @@ model AuditLog {
 
   user User? @relation(fields: [userId], references: [id], onDelete: SetNull)
 
+  @@index([userId, createdAt])
+  @@index([action])
+  @@index([createdAt])
   @@map("audit_logs")
 }
 

src/docs/swagger.json

@@ -3615,9 +3615,6 @@
       },
       "CreateCrawlJobRequest": {
         "type": "object",
-        "required": [
-          "startUrl"
-        ],
         "properties": {
           "startUrl": {
             "type": "string",
@@ -3645,6 +3642,18 @@
             "minimum": 1,
             "maximum": 10,
             "example": 3
+          },
+          "urls": {
+            "type": "array",
+            "items": {
+              "type": "string",
+              "format": "uri"
+            },
+            "example": [
+              "https://example.com/1",
+              "https://example.com/2"
+            ],
+            "description": "Bắt buộc khi mode là URL_LIST"
           }
         }
       },

src/docs/swagger.ts

@@ -251,12 +251,17 @@ const rawSchemas = {
   },
   CreateCrawlJobRequest: {
     type: 'object',
-    required: ['startUrl'],
     properties: {
       startUrl: { type: 'string', format: 'uri', example: 'https://example.com' },
       mode: { type: 'string', enum: ['SCRAPE', 'CRAWL', 'SITEMAP', 'URL_LIST'], example: 'CRAWL' },
       maxPages: { type: 'integer', minimum: 1, maximum: 1000, example: 100 },
-      maxDepth: { type: 'integer', minimum: 1, maximum: 10, example: 3 }
+      maxDepth: { type: 'integer', minimum: 1, maximum: 10, example: 3 },
+      urls: {
+        type: 'array',
+        items: { type: 'string', format: 'uri' },
+        example: ['https://example.com/1', 'https://example.com/2'],
+        description: 'Bắt buộc khi mode là URL_LIST'
+      }
     }
   },
   CreateExportRequest: {

src/middlewares/api-key.middleware.ts

 import { Request, Response, NextFunction } from 'express';
 import { ApiKeyService } from '../modules/api-keys/api-key.service';
-import { prisma } from '../database/prisma.client';
+import { UserRepository } from '../modules/users/user.repository';
 import { authMiddleware } from './auth.middleware';
 import { AppError } from '../common/errors/app-error';
 import { ERROR_CODE } from '../common/errors/error-code';
 
 const apiKeyService = new ApiKeyService();
+const userRepository = new UserRepository();
 
 export async function apiKeyOrAuthMiddleware(req: Request, res: Response, next: NextFunction): Promise<void> {
   const apiKey = req.headers['x-api-key'] as string | undefined;
@@ -14,10 +15,7 @@ export async function apiKeyOrAuthMiddleware(req: Request, res: Response, next:
     try {
       const validKeyRecord = await apiKeyService.validate(apiKey);
       
-      const user = await prisma.user.findFirst({
-        where: { id: validKeyRecord.userId, deletedAt: null },
-        select: { id: true, email: true, role: true, isActive: true },
-      });
+      const user = await userRepository.findById(validKeyRecord.userId);
 
       if (!user) {
         next(new AppError('User associated with API key not found', 401, ERROR_CODE.UNAUTHORIZED));

src/middlewares/auth.middleware.ts

@@ -3,9 +3,11 @@ import jwt from 'jsonwebtoken';
 import { jwtConfig } from '../config/jwt.config';
 import { AppError } from '../common/errors/app-error';
 import { ERROR_CODE } from '../common/errors/error-code';
-import { prisma } from '../database/prisma.client';
+import { UserRepository } from '../modules/users/user.repository';
 import { UserRole } from '@prisma/client';
 
+const userRepository = new UserRepository();
+
 export async function authMiddleware(req: Request, res: Response, next: NextFunction): Promise<void> {
   let token: string | undefined = req.cookies?.accessToken;
 
@@ -28,10 +30,7 @@ export async function authMiddleware(req: Request, res: Response, next: NextFunc
       role: string;
     };
 
-    const user = await prisma.user.findFirst({
-      where: { id: payload.id, deletedAt: null },
-      select: { isActive: true },
-    });
+    const user = await userRepository.findById(payload.id);
 
     if (!user) {
       next(new AppError("User not found", 401, ERROR_CODE.UNAUTHORIZED));

src/modules/crawl-jobs/__tests__/crawl-job.service.test.ts

+jest.mock('../../../database/prisma.client', () => ({
+  prisma: {},
+}));
+
+jest.mock('../crawl-job.repository');
+jest.mock('../../users/user.repository');
+jest.mock('../../crawl-exports/crawl-export.repository');
+jest.mock('../../../common/helpers/url.helper');
+jest.mock('../../../queues/crawl.queue', () => ({
+  crawlQueue: {
+    add: jest.fn().mockResolvedValue({ id: 'bull-job-1' }),
+  },
+}));
+
+import { CrawlJobService } from '../crawl-job.service';
+import { CrawlJobRepository } from '../crawl-job.repository';
+import { UserRepository } from '../../users/user.repository';
+import * as urlHelper from '../../../common/helpers/url.helper';
+
+describe('CrawlJobService', () => {
+  let service: CrawlJobService;
+  let mockJobRepo: jest.Mocked<CrawlJobRepository>;
+  let mockUserRepo: jest.Mocked<UserRepository>;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+
+    mockJobRepo = {
+      create: jest.fn().mockResolvedValue({ id: 'job-1', status: 'PENDING' }),
+      countJobsSince: jest.fn().mockResolvedValue(0),
+      countConcurrentJobs: jest.fn().mockResolvedValue(0),
+      findById: jest.fn(),
+    } as any;
+
+    mockUserRepo = {
+      findById: jest.fn().mockResolvedValue({
+        id: 'user-1',
+        email: 'user@example.com',
+        role: 'CRAWLER_USER',
+        maxPagesLimit: 50,
+        maxJobsPerDayLimit: 10,
+        maxConcurrentJobsLimit: 3,
+        isActive: true,
+      }),
+    } as any;
+
+    (CrawlJobRepository as jest.Mock).mockReturnValue(mockJobRepo);
+    (UserRepository as jest.Mock).mockReturnValue(mockUserRepo);
+    (urlHelper.validateUrl as jest.Mock).mockImplementation((url: string) => new URL(url));
+    (urlHelper.extractDomain as jest.Mock).mockReturnValue('example.com');
+    (urlHelper.validateUrlAsync as jest.Mock).mockResolvedValue(undefined);
+
+    service = new CrawlJobService();
+  });
+
+  describe('create', () => {
+    it('creates a job when within quota and valid startUrl', async () => {
+      const result = await service.create('user-1', {
+        startUrl: 'https://example.com',
+        mode: 'SCRAPE',
+        maxPages: 10,
+      });
+
+      expect(mockUserRepo.findById).toHaveBeenCalledWith('user-1');
+      expect(mockJobRepo.countJobsSince).toHaveBeenCalled();
+      expect(mockJobRepo.create).toHaveBeenCalledWith(
+        expect.objectContaining({
+          userId: 'user-1',
+          startUrl: 'https://example.com/',
+          mode: 'SCRAPE',
+        }),
+      );
+      expect(result.id).toBe('job-1');
+    });
+
+    it('throws error when requested pages exceed user maxPagesLimit', async () => {
+      await expect(
+        service.create('user-1', {
+          startUrl: 'https://example.com',
+          mode: 'CRAWL',
+          maxPages: 100,
+        }),
+      ).rejects.toThrow('exceeds quota limit');
+    });
+
+    it('throws error when daily job quota is reached', async () => {
+      mockJobRepo.countJobsSince.mockResolvedValue(10);
+
+      await expect(
+        service.create('user-1', {
+          startUrl: 'https://example.com',
+          mode: 'SCRAPE',
+          maxPages: 5,
+        }),
+      ).rejects.toThrow('Daily job quota of 10 exceeded');
+    });
+
+    it('throws error when concurrent job quota is reached', async () => {
+      mockJobRepo.countConcurrentJobs.mockResolvedValue(3);
+
+      await expect(
+        service.create('user-1', {
+          startUrl: 'https://example.com',
+          mode: 'SCRAPE',
+          maxPages: 5,
+        }),
+      ).rejects.toThrow('Concurrent jobs quota of 3 exceeded');
+    });
+
+    it('deduplicates URLs in URL_LIST mode and validates them', async () => {
+      const urls = [
+        'https://example.com/1',
+        'https://example.com/2',
+        'https://example.com/1',
+      ];
+
+      await service.create('user-1', {
+        mode: 'URL_LIST',
+        urls,
+      });
+
+      expect(urlHelper.validateUrlAsync).toHaveBeenCalledTimes(2);
+      expect(mockJobRepo.create).toHaveBeenCalledWith(
+        expect.objectContaining({
+          mode: 'URL_LIST',
+          startUrl: 'https://example.com/1',
+          urls: ['https://example.com/1', 'https://example.com/2'],
+        }),
+      );
+    });
+  });
+});

src/modules/crawl-jobs/crawl-job.dto.ts

 import { CrawlJobStatus, CrawlMode } from '@prisma/client';
 
 export interface CreateCrawlJobDto {
-  startUrl: string;
+  startUrl?: string;
   mode?: CrawlMode;
   maxPages?: number;
   maxDepth?: number;

src/modules/crawl-jobs/crawl-job.repository.ts

@@ -239,4 +239,23 @@ export class CrawlJobRepository {
       prisma.crawlJob.count({ where: { scheduleId } }),
     ]);
   }
+
+  countJobsSince(userId: string, sinceDate: Date): Promise<number> {
+    return prisma.crawlJob.count({
+      where: {
+        userId,
+        createdAt: { gte: sinceDate },
+      },
+    });
+  }
+
+  countConcurrentJobs(userId: string, activeStatuses: CrawlJobStatus[], sinceDate?: Date): Promise<number> {
+    return prisma.crawlJob.count({
+      where: {
+        userId,
+        status: { in: activeStatuses },
+        ...(sinceDate ? { createdAt: { gte: sinceDate } } : {}),
+      },
+    });
+  }
 }
\ No newline at end of file

src/modules/crawl-jobs/crawl-job.service.ts

-import { prisma } from '../../database/prisma.client';
 import { CrawlJobRepository } from './crawl-job.repository';
 import { CrawlExportRepository } from '../crawl-exports/crawl-export.repository';
+import { UserRepository } from '../users/user.repository';
 import { AppError } from '../../common/errors/app-error';
 import { ERROR_CODE } from '../../common/errors/error-code';
 import { validateUrl, extractDomain } from '../../common/helpers/url.helper';
+import {
+  getZonedDateParts,
+  createUtcDateFromZonedParts,
+} from '../../common/helpers/schedule-calculator.helper';
 import { crawlQueue } from '../../queues/crawl.queue';
 import { ROLES } from '../../common/constants/role.constant';
 import { JOB_STATUS } from '../../common/constants/job-status.constant';
@@ -12,11 +16,12 @@ import { StorageFactory } from '../../common/storage/storage.factory';
 
 export class CrawlJobService {
   private readonly repository = new CrawlJobRepository();
+  private readonly userRepository = new UserRepository();
 
   async create(userId: string, payload: CreateCrawlJobDto) {
     const isUrlList = payload.mode === 'URL_LIST';
 
-    // Fix #3: Deduplicate URLs before anything else
+    // Deduplicate URLs before anything else
     const deduplicatedUrls = isUrlList
       ? [...new Set(payload.urls!.map((u) => u.trim()))]
       : [];
@@ -26,17 +31,21 @@ export class CrawlJobService {
       ? new URL(deduplicatedUrls[0]).hostname
       : extractDomain(payload.startUrl!);
 
-    const user = await prisma.user.findUnique({ where: { id: userId } });
+    const user = await this.userRepository.findById(userId);
 
     if (!user) {
       throw new AppError('User not found', 404, ERROR_CODE.NOT_FOUND);
     }
 
-    // Fix #1: SSRF validation for ALL roles for URL_LIST
+    // SSRF validation with bounded concurrency for URL_LIST
     if (isUrlList) {
       const { validateUrlAsync } =
         await import('../../common/helpers/url.helper');
-      for (const url of deduplicatedUrls) {
+      const chunkSize = 10;
+      for (let i = 0; i < deduplicatedUrls.length; i += chunkSize) {
+        const chunk = deduplicatedUrls.slice(i, i + chunkSize);
+        await Promise.all(
+          chunk.map(async (url) => {
             try {
               await validateUrlAsync(url);
             } catch (err: any) {
@@ -46,11 +55,12 @@ export class CrawlJobService {
                 ERROR_CODE.INVALID_URL,
               );
             }
+          }),
+        );
       }
     }
 
     if (user.role !== ROLES.ADMIN) {
-      // Fix #2: For URL_LIST, quota check uses deduplicated urls.length
       const requestedPages = isUrlList
         ? deduplicatedUrls.length
         : (payload.maxPages ?? 20);
@@ -63,11 +73,18 @@ export class CrawlJobService {
         );
       }
 
-      const startOfDay = new Date();
-      startOfDay.setHours(0, 0, 0, 0);
-      const jobsTodayCount = await prisma.crawlJob.count({
-        where: { userId, createdAt: { gte: startOfDay } },
-      });
+      // Timezone UTC+7 start of day calculation
+      const nowZoned = getZonedDateParts(new Date(), 'Asia/Ho_Chi_Minh');
+      const startOfDay = createUtcDateFromZonedParts(
+        nowZoned.year,
+        nowZoned.month,
+        nowZoned.day,
+        0,
+        0,
+        'Asia/Ho_Chi_Minh',
+      );
+
+      const jobsTodayCount = await this.repository.countJobsSince(userId, startOfDay);
 
       if (jobsTodayCount >= user.maxJobsPerDayLimit) {
         throw new AppError(
@@ -79,20 +96,17 @@ export class CrawlJobService {
 
       const twoHoursAgo = new Date();
       twoHoursAgo.setHours(twoHoursAgo.getHours() - 2);
-      const concurrentJobsCount = await prisma.crawlJob.count({
-        where: {
-          userId,
-          status: {
-            in: [
+      const activeStatuses = [
         JOB_STATUS.PENDING,
         JOB_STATUS.QUEUED,
         JOB_STATUS.RUNNING,
         JOB_STATUS.PROCESSING_EXPORT,
-            ],
-          },
-          createdAt: { gte: twoHoursAgo },
-        },
-      });
+      ];
+      const concurrentJobsCount = await this.repository.countConcurrentJobs(
+        userId,
+        activeStatuses,
+        twoHoursAgo,
+      );
 
       if (concurrentJobsCount >= user.maxConcurrentJobsLimit) {
         throw new AppError(

src/modules/crawl-schedules/__tests__/crawl-schedule.service.test.ts

@@ -33,6 +33,7 @@ describe('CrawlScheduleService', () => {
       findAll: jest.fn(),
       findDueSchedules: jest.fn(),
       updateNextRun: jest.fn(),
+      claimDueSchedule: jest.fn().mockResolvedValue(true),
     } as any;
 
     mockJobRepo = {
@@ -203,20 +204,46 @@ describe('CrawlScheduleService', () => {
 
   describe('processDueSchedules', () => {
     it('finds and triggers all due active schedules', async () => {
-      mockScheduleRepo.findDueSchedules.mockResolvedValue([mockSchedule] as any);
+      mockScheduleRepo.findDueSchedules.mockResolvedValue([
+        { ...mockSchedule, user: { isActive: true, deletedAt: null } },
+      ] as any);
       mockJobRepo.create.mockResolvedValue({ id: 'job-due-1' } as any);
-      mockScheduleRepo.updateNextRun.mockResolvedValue({} as any);
 
       const count = await service.processDueSchedules();
 
       expect(count).toBe(1);
+      expect(mockScheduleRepo.claimDueSchedule).toHaveBeenCalled();
       expect(mockJobRepo.create).toHaveBeenCalledWith(
         expect.objectContaining({
           scheduleId: 'schedule-1',
         }),
       );
       expect(crawlQueue?.add).toHaveBeenCalledWith('crawl-job', { jobId: 'job-due-1' });
-      expect(mockScheduleRepo.updateNextRun).toHaveBeenCalled();
+    });
+
+    it('skips schedule when user is inactive or deleted', async () => {
+      mockScheduleRepo.findDueSchedules.mockResolvedValue([
+        { ...mockSchedule, user: { isActive: false, deletedAt: null } },
+        { ...mockSchedule, id: 'schedule-2', user: { isActive: true, deletedAt: new Date() } },
+      ] as any);
+
+      const count = await service.processDueSchedules();
+
+      expect(count).toBe(0);
+      expect(mockScheduleRepo.claimDueSchedule).not.toHaveBeenCalled();
+      expect(mockJobRepo.create).not.toHaveBeenCalled();
+    });
+
+    it('skips schedule when another worker instance already claimed it', async () => {
+      mockScheduleRepo.findDueSchedules.mockResolvedValue([
+        { ...mockSchedule, user: { isActive: true, deletedAt: null } },
+      ] as any);
+      mockScheduleRepo.claimDueSchedule.mockResolvedValue(false);
+
+      const count = await service.processDueSchedules();
+
+      expect(count).toBe(0);
+      expect(mockJobRepo.create).not.toHaveBeenCalled();
     });
   });
 });

src/modules/crawl-schedules/crawl-schedule.repository.ts

@@ -176,4 +176,19 @@ export class CrawlScheduleRepository {
       },
     });
   }
+
+  async claimDueSchedule(id: string, now: Date, nextRunAt: Date): Promise<boolean> {
+    const result = await prisma.crawlSchedule.updateMany({
+      where: {
+        id,
+        isActive: true,
+        nextRunAt: { lte: now },
+      },
+      data: {
+        lastRunAt: now,
+        nextRunAt,
+      },
+    });
+    return result.count > 0;
+  }
 }

src/modules/crawl-schedules/crawl-schedule.service.ts

@@ -255,18 +255,12 @@ export class CrawlScheduleService {
 
     for (const schedule of dueSchedules) {
       try {
-        const job = await this.jobRepository.create({
-          userId: schedule.userId,
-          startUrl: schedule.startUrl,
-          domain: schedule.domain ?? undefined,
-          mode: schedule.mode,
-          maxPages: schedule.maxPages,
-          maxDepth: schedule.maxDepth,
-          urls: schedule.urls,
-          scheduleId: schedule.id,
-        });
-
-        await crawlQueue.add('crawl-job', { jobId: job.id });
+        // Skip if user is inactive or deleted
+        const user = (schedule as any).user;
+        if (user && (!user.isActive || user.deletedAt)) {
+          console.warn(`[Schedule Service] Skipping schedule ${schedule.id}: user is inactive or deleted`);
+          continue;
+        }
 
         const nextRunAt = calculateNextRun({
           frequency: schedule.frequency,
@@ -279,7 +273,25 @@ export class CrawlScheduleService {
           fromDate: now,
         });
 
-        await this.repository.updateNextRun(schedule.id, now, nextRunAt);
+        // Atomic claim: only proceed if this instance successfully updated nextRunAt
+        const claimed = await this.repository.claimDueSchedule(schedule.id, now, nextRunAt);
+        if (!claimed) {
+          // Another worker instance already claimed and triggered this schedule
+          continue;
+        }
+
+        const job = await this.jobRepository.create({
+          userId: schedule.userId,
+          startUrl: schedule.startUrl,
+          domain: schedule.domain ?? undefined,
+          mode: schedule.mode,
+          maxPages: schedule.maxPages,
+          maxDepth: schedule.maxDepth,
+          urls: schedule.urls,
+          scheduleId: schedule.id,
+        });
+
+        await crawlQueue.add('crawl-job', { jobId: job.id });
         triggeredCount++;
       } catch (err: any) {
         console.error(`[Schedule Service] Failed to trigger due schedule ${schedule.id}: ${err.message}`);

src/modules/exports/__tests__/csv-export.service.test.ts

+import fs from 'fs';
+import { CsvExportService } from '../csv-export.service';
+
+jest.mock('../../../database/prisma.client', () => ({
+  prisma: {},
+}));
+
+jest.mock('../../crawl-assets/crawl-asset.repository', () => ({
+  CrawlAssetRepository: jest.fn().mockImplementation(() => ({
+    findByJobId: jest.fn().mockResolvedValue([]),
+  })),
+}));
+
+jest.mock('../../../common/helpers/file.helper', () => ({
+  buildJobDataFilePath: jest.fn((jobId: string, fileName: string) => ({
+    fileName,
+    filePath: `test/${fileName}`,
+  })),
+  ensureJobExportStructure: jest.fn(),
+}));
+
+jest.mock('fs');
+
+describe('CsvExportService', () => {
+  let service: CsvExportService;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    service = new CsvExportService();
+  });
+
+  it('neutralizes formula injection characters (=, +, -, @) in CSV export', async () => {
+    const writtenFiles: Record<string, string> = {};
+    (fs.writeFileSync as jest.Mock).mockImplementation((filePath, content) => {
+      writtenFiles[filePath] = content;
+    });
+
+    const mockJob: any = {
+      id: 'job-1',
+      startUrl: 'https://example.com',
+      domain: 'example.com',
+      pages: [
+        {
+          id: 'page-1',
+          url: 'https://example.com/test',
+          title: '=cmd|\'/C calc\'!A0',
+          description: '@SUM(1,2)',
+          status: 'COMPLETED',
+          statusCode: 200,
+          markdownContent: '+12345',
+          crawledAt: new Date('2026-09-02T12:00:00Z'),
+        },
+      ],
+    };
+
+    await (service as any).executeExport(mockJob);
+
+    const pagesCsv = writtenFiles['test/pages.csv'];
+    expect(pagesCsv).toBeDefined();
+    expect(pagesCsv).toContain("'=cmd|'/C calc'!A0");
+    expect(pagesCsv).toContain("'@SUM(1,2)");
+    expect(pagesCsv).toContain("'+12345");
+  });
+});

src/modules/exports/csv-export.service.ts

@@ -151,9 +151,18 @@ export class CsvExportService extends BaseExportService {
   }
 
   private escapeCsv(value: string): string {
-    if (value.includes(',') || value.includes('"') || value.includes('\n')) {
-      return `"${value.replace(/"/g, '""')}"`;
+    let sanitized = value;
+    if (/^[=+\-@\t\r]/.test(sanitized)) {
+      sanitized = `'${sanitized}`;
     }
-    return value;
+    if (
+      sanitized.includes(',') ||
+      sanitized.includes('"') ||
+      sanitized.includes('\n') ||
+      sanitized.includes('\r')
+    ) {
+      return `"${sanitized.replace(/"/g, '""')}"`;
+    }
+    return sanitized;
   }
 }
\ No newline at end of file

src/modules/extraction-templates/extraction-runner.ts

@@ -53,18 +53,21 @@ export async function runExtractionIfTemplate(
   pageId: string,
   pageUrl: string,
   item: FirecrawlPageResult,
+  userId?: string,
 ): Promise<void> {
   // Extraction requires raw HTML — Firecrawl returns it via the html field
   // which is not currently surfaced in FirecrawlPageResult. We fall back to
-  // markdownContent if html is unavailable. A future task should add html
-  // to FirecrawlPageResult and pass it through normalizePage().
+  // markdownContent if html is unavailable.
   const html = (item as any).html ?? item.markdown ?? '';
   if (!html) return;
 
   const domain = extractDomainFromUrl(pageUrl);
   if (!domain) return;
 
-  const template = await getTemplateRepository().findByDomain(domain);
+  const repository = getTemplateRepository();
+  const template = userId
+    ? await repository.findByUserAndDomain(userId, domain)
+    : await repository.findByDomain(domain);
   if (!template) return;
 
   const fields = template.fields as unknown as ExtractionFieldDto[];

src/modules/extraction-templates/extraction-template.repository.ts

@@ -28,6 +28,20 @@ export class ExtractionTemplateRepository {
     return prisma.extractionTemplate.findFirst({ where: { domain } });
   }
 
+  findByUserAndDomain(userId: string, domain: string) {
+    const isUuid = /^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/.test(userId);
+    if (!isUuid) return null;
+
+    return prisma.extractionTemplate.findUnique({
+      where: {
+        userId_domain: {
+          userId,
+          domain,
+        },
+      },
+    });
+  }
+
   update(id: string, data: UpdateExtractionTemplateDto) {
     return prisma.extractionTemplate.update({
       where: { id },

src/modules/webhooks/__tests__/webhook.service.test.ts

+jest.mock('../../../database/prisma.client', () => ({
+  prisma: {},
+}));
+
+jest.mock('../webhook.repository');
+jest.mock('../../../common/helpers/url.helper');
+jest.mock('../../../queues/webhook.queue', () => ({
+  webhookQueue: {
+    add: jest.fn().mockResolvedValue({ id: 'webhook-job-1' }),
+  },
+}));
+
+import { WebhookConfigService } from '../webhook-config.service';
+import { WebhookDeliveryService } from '../webhook-delivery.service';
+import { WebhookRepository } from '../webhook.repository';
+import * as urlHelper from '../../../common/helpers/url.helper';
+import { webhookQueue } from '../../../queues/webhook.queue';
+
+import { encrypt } from '../webhook-crypto.helper';
+
+describe('Webhook Services', () => {
+  let configService: WebhookConfigService;
+  let deliveryService: WebhookDeliveryService;
+  let mockWebhookRepo: jest.Mocked<WebhookRepository>;
+  let mockSecureAxiosPost: jest.Mock;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+
+    mockWebhookRepo = {
+      createConfig: jest.fn(),
+      listConfigsByUser: jest.fn(),
+      findConfigById: jest.fn(),
+      deleteConfig: jest.fn(),
+      findActiveConfigsByEvent: jest.fn(),
+      createDelivery: jest.fn(),
+      findDeliveryById: jest.fn(),
+      updateDelivery: jest.fn(),
+      listDeliveries: jest.fn(),
+    } as any;
+
+    (WebhookRepository as jest.Mock).mockReturnValue(mockWebhookRepo);
+
+    mockSecureAxiosPost = jest.fn().mockResolvedValue({ status: 200, data: 'OK' });
+    (urlHelper.getSecureAxios as jest.Mock).mockReturnValue({
+      post: mockSecureAxiosPost,
+    });
+
+    configService = new WebhookConfigService();
+    deliveryService = new WebhookDeliveryService();
+  });
+
+  describe('WebhookConfigService', () => {
+    it('creates webhook config and strips encryptedSecret from return value', async () => {
+      mockWebhookRepo.createConfig.mockResolvedValue({
+        id: 'config-1',
+        userId: 'user-1',
+        url: 'https://webhook.site/test',
+        encryptedSecret: 'enc:secret',
+        events: ['job.completed'],
+        isActive: true,
+        createdAt: new Date(),
+        updatedAt: new Date(),
+      });
+
+      const result = await configService.create(
+        'user-1',
+        'https://webhook.site/test',
+        'plain-secret-123',
+        ['job.completed'],
+      );
+
+      expect(mockWebhookRepo.createConfig).toHaveBeenCalledWith(
+        expect.objectContaining({
+          userId: 'user-1',
+          url: 'https://webhook.site/test',
+          events: ['job.completed'],
+        }),
+      );
+      expect((result as any).encryptedSecret).toBeUndefined();
+      expect(result.id).toBe('config-1');
+    });
+
+    it('throws 404 when deleting a non-existent or other user config', async () => {
+      mockWebhookRepo.findConfigById.mockResolvedValue({
+        id: 'config-1',
+        userId: 'other-user',
+      } as any);
+
+      await expect(
+        configService.delete('config-1', 'user-1'),
+      ).rejects.toThrow('Webhook configuration not found');
+    });
+  });
+
+  describe('WebhookDeliveryService', () => {
+    it('dispatches deliveries and enqueues to webhook queue', async () => {
+      mockWebhookRepo.findActiveConfigsByEvent.mockResolvedValue([
+        { id: 'config-1', userId: 'user-1' } as any,
+      ]);
+      mockWebhookRepo.createDelivery.mockResolvedValue({
+        id: 'delivery-1',
+      } as any);
+
+      await deliveryService.dispatch('job-1', 'user-1', 'job.completed', { pages: 10 });
+
+      expect(mockWebhookRepo.createDelivery).toHaveBeenCalledWith(
+        expect.objectContaining({
+          webhookConfigId: 'config-1',
+          crawlJobId: 'job-1',
+          event: 'job.completed',
+        }),
+      );
+      expect(webhookQueue?.add).toHaveBeenCalledWith(
+        'send-webhook',
+        { deliveryId: 'delivery-1' },
+        expect.any(Object),
+      );
+    });
+
+    it('sends delivery using getSecureAxios to prevent SSRF', async () => {
+      mockWebhookRepo.findDeliveryById.mockResolvedValue({
+        id: 'delivery-1',
+        event: 'job.completed',
+        payload: { test: true },
+        webhookConfig: {
+          url: 'https://webhook.site/callback',
+          encryptedSecret: encrypt('my-secret-123'),
+        },
+      } as any);
+
+      await deliveryService.send('delivery-1', 1);
+
+      expect(urlHelper.getSecureAxios).toHaveBeenCalled();
+      expect(mockSecureAxiosPost).toHaveBeenCalledWith(
+        'https://webhook.site/callback',
+        expect.any(String),
+        expect.objectContaining({
+          headers: expect.objectContaining({
+            'X-Webhook-Event': 'job.completed',
+          }),
+        }),
+      );
+      expect(mockWebhookRepo.updateDelivery).toHaveBeenCalledWith(
+        'delivery-1',
+        expect.objectContaining({
+          status: 'SUCCESS',
+          statusCode: 200,
+        }),
+      );
+    });
+  });
+});

src/modules/webhooks/webhook-config.service.ts

-import { prisma } from '../../database/prisma.client';
 import { WebhookConfig } from '@prisma/client';
+import { WebhookRepository } from './webhook.repository';
 import { encrypt } from './webhook-crypto.helper';
 import { AppError } from '../../common/errors/app-error';
 import { ERROR_CODE } from '../../common/errors/error-code';
 
 export class WebhookConfigService {
+  private readonly repository = new WebhookRepository();
+
   async create(
     userId: string,
     url: string,
     plainSecret: string,
-    events: string[]
+    events: string[],
   ): Promise<Omit<WebhookConfig, 'encryptedSecret'>> {
     const encryptedSecret = encrypt(plainSecret);
 
-    const config = await prisma.webhookConfig.create({
-      data: {
+    const config = await this.repository.createConfig({
       userId,
       url,
       encryptedSecret,
       events,
-      },
     });
 
     const { encryptedSecret: _, ...rest } = config;
@@ -27,31 +27,18 @@ export class WebhookConfigService {
   }
 
   async list(userId: string): Promise<Omit<WebhookConfig, 'encryptedSecret'>[]> {
-    const configs = await prisma.webhookConfig.findMany({
-      where: {
-        userId,
-      },
-      orderBy: {
-        createdAt: 'desc',
-      },
-    });
-
-    return configs.map(({ encryptedSecret, ...rest }) => rest);
+    const configs = await this.repository.listConfigsByUser(userId);
+    return configs.map(({ encryptedSecret: _, ...rest }) => rest);
   }
 
   async delete(configId: string, userId: string): Promise<Omit<WebhookConfig, 'encryptedSecret'>> {
-    const config = await prisma.webhookConfig.findUnique({
-      where: { id: configId },
-    });
+    const config = await this.repository.findConfigById(configId);
 
     if (!config || config.userId !== userId) {
       throw new AppError('Webhook configuration not found', 404, ERROR_CODE.WEBHOOK_CONFIG_NOT_FOUND);
     }
 
-    const deleted = await prisma.webhookConfig.delete({
-      where: { id: configId },
-    });
-
+    const deleted = await this.repository.deleteConfig(configId);
     const { encryptedSecret: _, ...rest } = deleted;
     return rest;
   }

src/modules/webhooks/webhook-delivery.service.ts

-import { prisma } from '../../database/prisma.client';
-import { WebhookDelivery, WebhookConfig } from '@prisma/client';
+import { WebhookDelivery } from '@prisma/client';
+import { WebhookRepository } from './webhook.repository';
 import { decrypt, signPayload } from './webhook-crypto.helper';
 import { webhookQueue } from '../../queues/webhook.queue';
-import axios from 'axios';
+import { getSecureAxios } from '../../common/helpers/url.helper';
 
 export class WebhookDeliveryService {
+  private readonly repository = new WebhookRepository();
+
   async dispatch(crawlJobId: string, userId: string, event: string, jobData: any): Promise<void> {
     try {
-      const configs = await prisma.webhookConfig.findMany({
-        where: {
-          userId,
-          isActive: true,
-          events: {
-            has: event,
-          },
-        },
-      });
+      const configs = await this.repository.findActiveConfigsByEvent(userId, event);
 
       if (configs.length === 0) {
         return;
@@ -29,15 +23,13 @@ export class WebhookDeliveryService {
       };
 
       for (const config of configs) {
-        const delivery = await prisma.webhookDelivery.create({
-          data: {
+        const delivery = await this.repository.createDelivery({
           webhookConfigId: config.id,
           crawlJobId,
           event,
           payload: payload as any,
           status: 'PENDING',
           attempt: 1,
-          },
         });
 
         if (webhookQueue) {
@@ -50,7 +42,7 @@ export class WebhookDeliveryService {
                 type: 'exponential',
                 delay: 5000, // 5s, 25s, 125s
               },
-            }
+            },
           );
         } else {
           console.error('[Webhook] Redis/BullMQ is not initialized. Webhook could not be enqueued.');
@@ -62,18 +54,14 @@ export class WebhookDeliveryService {
   }
 
   async send(deliveryId: string, attemptNumber: number): Promise<void> {
-    const delivery = await prisma.webhookDelivery.findUnique({
-      where: { id: deliveryId },
-      include: { webhookConfig: true },
-    });
+    const delivery = await this.repository.findDeliveryById(deliveryId);
 
     if (!delivery) {
       throw new Error(`WebhookDelivery ${deliveryId} not found`);
     }
 
-    await prisma.webhookDelivery.update({
-      where: { id: deliveryId },
-      data: { attempt: attemptNumber },
+    await this.repository.updateDelivery(deliveryId, {
+      attempt: attemptNumber,
     });
 
     const config = delivery.webhookConfig;
@@ -82,7 +70,7 @@ export class WebhookDeliveryService {
     const signature = signPayload(secret, payloadStr);
 
     try {
-      const response = await axios.post(config.url, payloadStr, {
+      const response = await getSecureAxios().post(config.url, payloadStr, {
         headers: {
           'Content-Type': 'application/json',
           'X-Webhook-Signature': `sha256=${signature}`,
@@ -96,21 +84,18 @@ export class WebhookDeliveryService {
         ? response.data
         : JSON.stringify(response.data);
 
-      await prisma.webhookDelivery.update({
-        where: { id: deliveryId },
-        data: {
+      await this.repository.updateDelivery(deliveryId, {
         status: 'SUCCESS',
         statusCode: response.status,
         responseBody: responseBody.substring(0, 2000), // Limit size stored in DB
         deliveredAt: new Date(),
         errorMessage: null,
-        },
       });
 
     } catch (error: any) {
       let statusCode: number | null = null;
       let responseBody: string | null = null;
-      let errorMessage = error.message || 'Unknown network error';
+      const errorMessage = error.message || 'Unknown network error';
 
       if (error.response) {
         statusCode = error.response.status;
@@ -119,13 +104,10 @@ export class WebhookDeliveryService {
           : JSON.stringify(error.response.data);
       }
 
-      await prisma.webhookDelivery.update({
-        where: { id: deliveryId },
-        data: {
+      await this.repository.updateDelivery(deliveryId, {
         statusCode,
         responseBody: responseBody ? responseBody.substring(0, 2000) : null,
         errorMessage: errorMessage.substring(0, 1000),
-        },
       });
 
       // Throw error to trigger BullMQ retry
@@ -138,45 +120,16 @@ export class WebhookDeliveryService {
    * Called by BullMQ worker when job fails after max attempts.
    */
   async markFailed(deliveryId: string, errorReason: string): Promise<void> {
-    await prisma.webhookDelivery.update({
-      where: { id: deliveryId },
-      data: {
+    await this.repository.updateDelivery(deliveryId, {
       status: 'FAILED',
       errorMessage: `Max attempts exhausted. Last error: ${errorReason}`.substring(0, 1000),
-      },
     });
   }
 
   async listDeliveries(
     userId: string,
-    query: { jobId?: string; status?: string }
+    query: { jobId?: string; status?: string },
   ): Promise<WebhookDelivery[]> {
-    const where: any = {
-      webhookConfig: {
-        userId,
-      },
-    };
-
-    if (query.jobId) {
-      where.crawlJobId = query.jobId;
-    }
-
-    if (query.status) {
-      where.status = query.status;
-    }
-
-    return prisma.webhookDelivery.findMany({
-      where,
-      include: {
-        webhookConfig: {
-          select: {
-            url: true,
-          },
-        },
-      },
-      orderBy: {
-        createdAt: 'desc',
-      },
-    });
+    return this.repository.listDeliveries(userId, query);
   }
 }

src/modules/webhooks/webhook.repository.ts

+import { prisma } from '../../database/prisma.client';
+import { WebhookConfig, WebhookDelivery, Prisma } from '@prisma/client';
+
+export class WebhookRepository {
+  createConfig(data: {
+    userId: string;
+    url: string;
+    encryptedSecret: string;
+    events: string[];
+  }): Promise<WebhookConfig> {
+    return prisma.webhookConfig.create({
+      data: {
+        userId: data.userId,
+        url: data.url,
+        encryptedSecret: data.encryptedSecret,
+        events: data.events,
+      },
+    });
+  }
+
+  listConfigsByUser(userId: string): Promise<WebhookConfig[]> {
+    return prisma.webhookConfig.findMany({
+      where: { userId },
+      orderBy: { createdAt: 'desc' },
+    });
+  }
+
+  findConfigById(id: string): Promise<WebhookConfig | null> {
+    return prisma.webhookConfig.findUnique({
+      where: { id },
+    });
+  }
+
+  deleteConfig(id: string): Promise<WebhookConfig> {
+    return prisma.webhookConfig.delete({
+      where: { id },
+    });
+  }
+
+  findActiveConfigsByEvent(userId: string, event: string): Promise<WebhookConfig[]> {
+    return prisma.webhookConfig.findMany({
+      where: {
+        userId,
+        isActive: true,
+        events: {
+          has: event,
+        },
+      },
+    });
+  }
+
+  createDelivery(data: {
+    webhookConfigId: string;
+    crawlJobId: string;
+    event: string;
+    payload: any;
+    status: string;
+    attempt: number;
+  }): Promise<WebhookDelivery> {
+    return prisma.webhookDelivery.create({
+      data: {
+        webhookConfigId: data.webhookConfigId,
+        crawlJobId: data.crawlJobId,
+        event: data.event,
+        payload: data.payload,
+        status: data.status,
+        attempt: data.attempt,
+      },
+    });
+  }
+
+  findDeliveryById(id: string) {
+    return prisma.webhookDelivery.findUnique({
+      where: { id },
+      include: { webhookConfig: true },
+    });
+  }
+
+  updateDelivery(id: string, data: Prisma.WebhookDeliveryUpdateInput): Promise<WebhookDelivery> {
+    return prisma.webhookDelivery.update({
+      where: { id },
+      data,
+    });
+  }
+
+  listDeliveries(
+    userId: string,
+    query: { jobId?: string; status?: string },
+  ) {
+    const where: Prisma.WebhookDeliveryWhereInput = {
+      webhookConfig: {
+        userId,
+      },
+    };
+
+    if (query.jobId) {
+      where.crawlJobId = query.jobId;
+    }
+
+    if (query.status) {
+      where.status = query.status;
+    }
+
+    return prisma.webhookDelivery.findMany({
+      where,
+      include: {
+        webhookConfig: {
+          select: {
+            url: true,
+          },
+        },
+      },
+      orderBy: {
+        createdAt: 'desc',
+      },
+    });
+  }
+}

src/queues/crawl.worker.processor.ts

@@ -108,6 +108,7 @@ export async function scanAndFlagPage(pageId: string, ...texts: (string | undefi
 export async function persistBatchResults(
   jobId: string,
   result: CrawlStatusResult,
+  userId?: string,
 ): Promise<{ successCount: number; failedCount: number; saveErrors: number; totalPages: number }> {
   let successCount = 0;
   let failedCount = 0;
@@ -123,7 +124,7 @@ export async function persistBatchResults(
       const page = await getPageRepository().upsert(normalized);
       await savePageAssets(jobId, page.id, item);
       await scanAndFlagPage(page.id, normalized.markdownContent, normalized.title, normalized.description);
-      await runExtractionIfTemplate(jobId, page.id, item.url, item);
+      await runExtractionIfTemplate(jobId, page.id, item.url, item, userId);
       if (item.success) successCount++;
       else failedCount++;
     } catch (err: any) {
@@ -233,7 +234,7 @@ export async function processCrawlJob(job: Job<{ jobId: string }>) {
         const page = await getPageRepository().upsert(normalized);
         await savePageAssets(jobId, page.id, result);
         await scanAndFlagPage(page.id, normalized.markdownContent, normalized.title, normalized.description);
-        await runExtractionIfTemplate(jobId, page.id, crawlJob.startUrl, result);
+        await runExtractionIfTemplate(jobId, page.id, crawlJob.startUrl, result, crawlJob.userId);
         await getJobRepository().updateStatus(jobId, 'COMPLETED', {
           finishedAt: new Date(),
           totalPages: 1,
@@ -307,7 +308,7 @@ export async function processCrawlJob(job: Job<{ jobId: string }>) {
           return;
         }
 
-        const { successCount, failedCount, saveErrors, totalPages } = await persistBatchResults(jobId, result);
+        const { successCount, failedCount, saveErrors, totalPages } = await persistBatchResults(jobId, result, crawlJob.userId);
         console.log(`[Worker] Job ${jobId} completed: ${successCount} success, ${failedCount} failed, ${saveErrors} save errors, ${totalPages} total`);
         await getJobRepository().updateStatus(jobId, 'COMPLETED', {
           finishedAt: new Date(),
@@ -362,7 +363,7 @@ export async function processCrawlJob(job: Job<{ jobId: string }>) {
           return;
         }
 
-        const { successCount, failedCount, saveErrors, totalPages } = await persistBatchResults(jobId, result);
+        const { successCount, failedCount, saveErrors, totalPages } = await persistBatchResults(jobId, result, crawlJob.userId);
         console.log(`[Worker] Job ${jobId} completed: ${successCount} success, ${failedCount} failed, ${saveErrors} save errors, ${totalPages} total`);
         await getJobRepository().updateStatus(jobId, 'COMPLETED', {
           finishedAt: new Date(),
@@ -424,7 +425,7 @@ export async function processCrawlJob(job: Job<{ jobId: string }>) {
           return;
         }
 
-        const { successCount, failedCount, saveErrors, totalPages } = await persistBatchResults(jobId, result);
+        const { successCount, failedCount, saveErrors, totalPages } = await persistBatchResults(jobId, result, crawlJob.userId);
         console.log(`[Worker] Job ${jobId} completed: ${successCount} success, ${failedCount} failed, ${saveErrors} save errors, ${totalPages} total`);
         await getJobRepository().updateStatus(jobId, 'COMPLETED', {
           finishedAt: new Date(),