Commit 3a9b2bea authored by ThinhNC's avatar ThinhNC

fix(crawl-jobs): implement soft delete for quota retention and auto-complete finished jobs

parent 24ac3314
-- AlterTable
ALTER TABLE "crawl_jobs" ADD COLUMN "deleted_at" TIMESTAMP(3),
ADD COLUMN "deleted_by" UUID;
-- CreateIndex
CREATE INDEX "crawl_jobs_deleted_at_idx" ON "crawl_jobs"("deleted_at");
-- CreateIndex
CREATE INDEX "crawl_jobs_user_id_deleted_at_idx" ON "crawl_jobs"("user_id", "deleted_at");
...@@ -141,6 +141,9 @@ model CrawlJob { ...@@ -141,6 +141,9 @@ model CrawlJob {
createdAt DateTime @default(now()) @map("created_at") createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @updatedAt @map("updated_at") updatedAt DateTime @updatedAt @map("updated_at")
deletedAt DateTime? @map("deleted_at")
deletedBy String? @map("deleted_by") @db.Uuid
user User @relation(fields: [userId], references: [id]) user User @relation(fields: [userId], references: [id])
schedule CrawlSchedule? @relation(fields: [scheduleId], references: [id], onDelete: SetNull) schedule CrawlSchedule? @relation(fields: [scheduleId], references: [id], onDelete: SetNull)
pages CrawlPage[] pages CrawlPage[]
...@@ -154,6 +157,8 @@ model CrawlJob { ...@@ -154,6 +157,8 @@ model CrawlJob {
@@index([userId, status]) @@index([userId, status])
@@index([userId, createdAt]) @@index([userId, createdAt])
@@index([scheduleId]) @@index([scheduleId])
@@index([deletedAt])
@@index([userId, deletedAt])
@@map("crawl_jobs") @@map("crawl_jobs")
} }
......
...@@ -335,16 +335,17 @@ async function seedCrawlSchedules(userId: string) { ...@@ -335,16 +335,17 @@ async function seedCrawlSchedules(userId: string) {
async function seedCrawlJobsAndPages(userId: string) { async function seedCrawlJobsAndPages(userId: string) {
console.log("Seeding realistic sample Crawl Jobs, Pages, and Logs..."); console.log("Seeding realistic sample Crawl Jobs, Pages, and Logs...");
// Job 1: RUNNING // Job 1: COMPLETED (52 pages crawled, 49 success, 3 failed)
const job1Id = "088f635c-9c3a-4467-93bb-e58f001bf001"; const job1Id = "088f635c-9c3a-4467-93bb-e58f001bf001";
const job1 = await prisma.crawlJob.upsert({ const job1 = await prisma.crawlJob.upsert({
where: { id: job1Id }, where: { id: job1Id },
update: { update: {
status: CrawlJobStatus.RUNNING, status: CrawlJobStatus.COMPLETED,
totalPages: 52, totalPages: 52,
successPages: 49, successPages: 49,
failedPages: 3, failedPages: 3,
startedAt: new Date(Date.now() - 3600000 * 1.5), startedAt: new Date(Date.now() - 3600000 * 1.5),
finishedAt: new Date(Date.now() - 3600000 * 0.5),
}, },
create: { create: {
id: job1Id, id: job1Id,
...@@ -352,7 +353,7 @@ async function seedCrawlJobsAndPages(userId: string) { ...@@ -352,7 +353,7 @@ async function seedCrawlJobsAndPages(userId: string) {
startUrl: "https://vnexpress.net/so-hoa/cong-nghe", startUrl: "https://vnexpress.net/so-hoa/cong-nghe",
domain: "vnexpress.net", domain: "vnexpress.net",
mode: CrawlMode.CRAWL, mode: CrawlMode.CRAWL,
status: CrawlJobStatus.RUNNING, status: CrawlJobStatus.COMPLETED,
maxPages: 100, maxPages: 100,
maxDepth: 3, maxDepth: 3,
urls: [], urls: [],
...@@ -365,6 +366,7 @@ async function seedCrawlJobsAndPages(userId: string) { ...@@ -365,6 +366,7 @@ async function seedCrawlJobsAndPages(userId: string) {
userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", userAgent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
delayMs: 1000, delayMs: 1000,
startedAt: new Date(Date.now() - 3600000 * 1.5), startedAt: new Date(Date.now() - 3600000 * 1.5),
finishedAt: new Date(Date.now() - 3600000 * 0.5),
diffSummary: { diffSummary: {
totalCurrentPages: 52, totalCurrentPages: 52,
totalPreviousPages: 45, totalPreviousPages: 45,
......
...@@ -52,6 +52,8 @@ describe("ChangeDetectionService", () => { ...@@ -52,6 +52,8 @@ describe("ChangeDetectionService", () => {
finishedAt: new Date(), finishedAt: new Date(),
createdAt: new Date(), createdAt: new Date(),
updatedAt: new Date(), updatedAt: new Date(),
deletedAt: null,
deletedBy: null,
}; };
const makePage = ( const makePage = (
......
...@@ -61,6 +61,23 @@ export class CrawlExportService { ...@@ -61,6 +61,23 @@ export class CrawlExportService {
); );
} }
const processedPages = (job.successPages ?? 0) + (job.failedPages ?? 0);
const targetPages =
job.totalPages > 0
? Math.min(job.maxPages, job.totalPages)
: job.maxPages;
const isFinished =
job.totalPages > 0 &&
processedPages >= targetPages &&
(job.successPages ?? 0) > 0;
if (job.status === JOB_STATUS.RUNNING && isFinished) {
await this.jobRepository.updateStatus(job.id, JOB_STATUS.COMPLETED, {
finishedAt: job.finishedAt || new Date(),
});
job.status = JOB_STATUS.COMPLETED;
}
const isExportable = const isExportable =
job.status === JOB_STATUS.COMPLETED || job.status === JOB_STATUS.COMPLETED ||
(job.status === JOB_STATUS.CANCELED && (job.successPages ?? 0) > 0) || (job.status === JOB_STATUS.CANCELED && (job.successPages ?? 0) > 0) ||
......
...@@ -54,7 +54,10 @@ describe("CrawlJobService delete, rerun, and getLogs", () => { ...@@ -54,7 +54,10 @@ describe("CrawlJobService delete, rerun, and getLogs", () => {
expect(result.success).toBe(true); expect(result.success).toBe(true);
expect(mockStorage.deleteFile).toHaveBeenCalledWith("exports/exp-1.zip"); expect(mockStorage.deleteFile).toHaveBeenCalledWith("exports/exp-1.zip");
expect(mockStorage.deleteFile).toHaveBeenCalledWith("diffs/job-123.json"); expect(mockStorage.deleteFile).toHaveBeenCalledWith("diffs/job-123.json");
expect(CrawlJobRepository.prototype.delete).toHaveBeenCalledWith("job-123"); expect(CrawlJobRepository.prototype.delete).toHaveBeenCalledWith(
"job-123",
"user-1",
);
}); });
it("blocks deletion of an actively running job", async () => { it("blocks deletion of an actively running job", async () => {
......
import { CrawlJobRepository } from "../crawl-job.repository";
import { prisma } from "../../../database/prisma.client";
import { JOB_STATUS } from "../../../common/constants/job-status.constant";
jest.mock("../../../database/prisma.client", () => ({
prisma: {
$transaction: jest.fn(),
crawlAsset: { deleteMany: jest.fn() },
crawlJobLog: { deleteMany: jest.fn() },
crawlExport: { deleteMany: jest.fn() },
crawlPage: { deleteMany: jest.fn() },
crawlJob: {
update: jest.fn(),
findUnique: jest.fn(),
findFirst: jest.fn(),
findMany: jest.fn(),
count: jest.fn(),
aggregate: jest.fn(),
},
},
}));
describe("CrawlJobRepository soft-delete and quota retention", () => {
let repository: CrawlJobRepository;
beforeEach(() => {
jest.clearAllMocks();
repository = new CrawlJobRepository();
});
it("delete() soft-deletes the job and cleans up child resources in a transaction", async () => {
const mockTx = {
crawlAsset: { deleteMany: jest.fn().mockResolvedValue({ count: 1 }) },
crawlJobLog: { deleteMany: jest.fn().mockResolvedValue({ count: 2 }) },
crawlExport: { deleteMany: jest.fn().mockResolvedValue({ count: 1 }) },
crawlPage: { deleteMany: jest.fn().mockResolvedValue({ count: 5 }) },
crawlJob: {
update: jest.fn().mockResolvedValue({ id: "job-1", deletedAt: new Date(), deletedBy: "user-1" }),
},
};
(prisma.$transaction as jest.Mock).mockImplementation(async (cb: any) => cb(mockTx));
const result = await repository.delete("job-1", "user-1");
expect(prisma.$transaction).toHaveBeenCalled();
expect(mockTx.crawlAsset.deleteMany).toHaveBeenCalledWith({ where: { crawlJobId: "job-1" } });
expect(mockTx.crawlJobLog.deleteMany).toHaveBeenCalledWith({ where: { jobId: "job-1" } });
expect(mockTx.crawlExport.deleteMany).toHaveBeenCalledWith({ where: { jobId: "job-1" } });
expect(mockTx.crawlPage.deleteMany).toHaveBeenCalledWith({ where: { jobId: "job-1" } });
expect(mockTx.crawlJob.update).toHaveBeenCalledWith(
expect.objectContaining({
where: { id: "job-1" },
data: expect.objectContaining({
deletedAt: expect.any(Date),
deletedBy: "user-1",
}),
}),
);
expect(result.deletedBy).toBe("user-1");
});
it("findById() excludes soft-deleted jobs by returning null when deletedAt is present", async () => {
(prisma.crawlJob.findUnique as jest.Mock).mockResolvedValue({
id: "deleted-job-id",
deletedAt: new Date(),
});
const result = await repository.findById("deleted-job-id");
expect(prisma.crawlJob.findUnique).toHaveBeenCalledWith(
expect.objectContaining({
where: { id: "deleted-job-id" },
}),
);
expect(result).toBeNull();
});
it("countJobsSince() preserves daily quota count by including soft-deleted jobs", async () => {
const sinceDate = new Date("2026-09-07T00:00:00Z");
(prisma.crawlJob.count as jest.Mock).mockResolvedValue(5);
const count = await repository.countJobsSince("user-1", sinceDate);
expect(count).toBe(5);
// Notice: deletedAt is NOT filtered out, so daily job quota usage is retained
expect(prisma.crawlJob.count).toHaveBeenCalledWith({
where: {
userId: "user-1",
createdAt: { gte: sinceDate },
},
});
});
it("sumPagesCrawledByUser() preserves crawled pages usage by including soft-deleted jobs", async () => {
(prisma.crawlJob.aggregate as jest.Mock).mockResolvedValue({
_sum: { totalPages: 150 },
});
const pages = await repository.sumPagesCrawledByUser("user-1");
expect(pages).toBe(150);
// Notice: deletedAt is NOT filtered out, so totalPages crawled history is retained
expect(prisma.crawlJob.aggregate).toHaveBeenCalledWith({
where: { userId: "user-1" },
_sum: { totalPages: true },
});
});
it("countConcurrentJobs() excludes soft-deleted jobs", async () => {
(prisma.crawlJob.count as jest.Mock).mockResolvedValue(1);
await repository.countConcurrentJobs("user-1", [JOB_STATUS.RUNNING]);
expect(prisma.crawlJob.count).toHaveBeenCalledWith(
expect.objectContaining({
where: expect.objectContaining({
userId: "user-1",
status: { in: [JOB_STATUS.RUNNING] },
deletedAt: null,
}),
}),
);
});
});
...@@ -38,7 +38,7 @@ export class CrawlJobRepository { ...@@ -38,7 +38,7 @@ export class CrawlJobRepository {
} }
private async find(query: CrawlJobQueryDto, userId?: string) { private async find(query: CrawlJobQueryDto, userId?: string) {
const where: Prisma.CrawlJobWhereInput = {}; const where: Prisma.CrawlJobWhereInput = { deletedAt: null };
if (userId) { if (userId) {
where.userId = userId; where.userId = userId;
} }
...@@ -152,17 +152,21 @@ export class CrawlJobRepository { ...@@ -152,17 +152,21 @@ export class CrawlJobRepository {
}; };
} }
findById(id: string) { async findById(id: string) {
return prisma.crawlJob.findUnique({ const job = await prisma.crawlJob.findUnique({
where: { id }, where: { id },
include: { include: {
exports: true, exports: true,
}, },
}); });
if (!job || job.deletedAt) {
return null;
}
return job;
} }
findByIdWithPages(id: string) { async findByIdWithPages(id: string) {
return prisma.crawlJob.findUnique({ const job = await prisma.crawlJob.findUnique({
where: { id }, where: { id },
include: { include: {
pages: { pages: {
...@@ -180,6 +184,10 @@ export class CrawlJobRepository { ...@@ -180,6 +184,10 @@ export class CrawlJobRepository {
}, },
}, },
}); });
if (!job || job.deletedAt) {
return null;
}
return job;
} }
async updateStatus( async updateStatus(
...@@ -243,6 +251,7 @@ export class CrawlJobRepository { ...@@ -243,6 +251,7 @@ export class CrawlJobRepository {
scheduleId, scheduleId,
id: { not: currentJobId }, id: { not: currentJobId },
status: JOB_STATUS.COMPLETED, status: JOB_STATUS.COMPLETED,
deletedAt: null,
}, },
orderBy: { createdAt: "desc" }, orderBy: { createdAt: "desc" },
include: { include: {
...@@ -274,6 +283,7 @@ export class CrawlJobRepository { ...@@ -274,6 +283,7 @@ export class CrawlJobRepository {
userId, userId,
id: { not: currentJobId }, id: { not: currentJobId },
status: JOB_STATUS.COMPLETED, status: JOB_STATUS.COMPLETED,
deletedAt: null,
OR: [...(domain ? [{ domain }] : []), { startUrl }], OR: [...(domain ? [{ domain }] : []), { startUrl }],
}, },
orderBy: { createdAt: "desc" }, orderBy: { createdAt: "desc" },
...@@ -313,12 +323,12 @@ export class CrawlJobRepository { ...@@ -313,12 +323,12 @@ export class CrawlJobRepository {
const skip = (Math.max(1, page) - 1) * limit; const skip = (Math.max(1, page) - 1) * limit;
return Promise.all([ return Promise.all([
prisma.crawlJob.findMany({ prisma.crawlJob.findMany({
where: { scheduleId }, where: { scheduleId, deletedAt: null },
orderBy: { createdAt: "desc" }, orderBy: { createdAt: "desc" },
skip, skip,
take: limit, take: limit,
}), }),
prisma.crawlJob.count({ where: { scheduleId } }), prisma.crawlJob.count({ where: { scheduleId, deletedAt: null } }),
]); ]);
} }
...@@ -340,6 +350,7 @@ export class CrawlJobRepository { ...@@ -340,6 +350,7 @@ export class CrawlJobRepository {
where: { where: {
userId, userId,
status: { in: activeStatuses }, status: { in: activeStatuses },
deletedAt: null,
...(sinceDate ? { createdAt: { gte: sinceDate } } : {}), ...(sinceDate ? { createdAt: { gte: sinceDate } } : {}),
}, },
}); });
...@@ -353,13 +364,19 @@ export class CrawlJobRepository { ...@@ -353,13 +364,19 @@ export class CrawlJobRepository {
return aggregate._sum.totalPages ?? 0; return aggregate._sum.totalPages ?? 0;
} }
async delete(id: string) { async delete(id: string, deletedBy?: string) {
return prisma.$transaction(async (tx) => { return prisma.$transaction(async (tx) => {
await tx.crawlAsset.deleteMany({ where: { crawlJobId: id } }); await tx.crawlAsset.deleteMany({ where: { crawlJobId: id } });
await tx.crawlJobLog.deleteMany({ where: { jobId: id } }); await tx.crawlJobLog.deleteMany({ where: { jobId: id } });
await tx.crawlExport.deleteMany({ where: { jobId: id } }); await tx.crawlExport.deleteMany({ where: { jobId: id } });
await tx.crawlPage.deleteMany({ where: { jobId: id } }); await tx.crawlPage.deleteMany({ where: { jobId: id } });
return tx.crawlJob.delete({ where: { id } }); return tx.crawlJob.update({
where: { id },
data: {
deletedAt: new Date(),
...(deletedBy ? { deletedBy } : {}),
},
});
}); });
} }
...@@ -399,6 +416,7 @@ export class CrawlJobRepository { ...@@ -399,6 +416,7 @@ export class CrawlJobRepository {
status: { status: {
in: [JOB_STATUS.PENDING, JOB_STATUS.QUEUED, JOB_STATUS.RUNNING], in: [JOB_STATUS.PENDING, JOB_STATUS.QUEUED, JOB_STATUS.RUNNING],
}, },
deletedAt: null,
createdAt: { gte: since }, createdAt: { gte: since },
}, },
orderBy: { createdAt: "desc" }, orderBy: { createdAt: "desc" },
......
...@@ -171,10 +171,23 @@ export class CrawlJobService { ...@@ -171,10 +171,23 @@ export class CrawlJobService {
} }
async findAllByUser(userId: string, role: string, query: CrawlJobQueryDto) { async findAllByUser(userId: string, role: string, query: CrawlJobQueryDto) {
if (role === ROLES.ADMIN) { const result = role === ROLES.ADMIN
return this.repository.findAll(query); ? await this.repository.findAll(query)
: await this.repository.findAllByUser(userId, query);
// Auto-complete any jobs that reached all target pages but were left in RUNNING
for (const job of result.items) {
const processed = (job.successPages ?? 0) + (job.failedPages ?? 0);
const target = job.totalPages > 0 ? Math.min(job.maxPages, job.totalPages) : job.maxPages;
if (job.status === JOB_STATUS.RUNNING && job.totalPages > 0 && processed >= target) {
job.status = JOB_STATUS.COMPLETED;
void this.repository.updateStatus(job.id, JOB_STATUS.COMPLETED, {
finishedAt: job.finishedAt || new Date(),
});
}
} }
return this.repository.findAllByUser(userId, query);
return result;
} }
async findById(userId: string, role: string, jobId: string) { async findById(userId: string, role: string, jobId: string) {
...@@ -196,6 +209,24 @@ export class CrawlJobService { ...@@ -196,6 +209,24 @@ export class CrawlJobService {
); );
} }
// Auto-complete if job is in RUNNING but all target pages have already been crawled
const processedPages = (job.successPages ?? 0) + (job.failedPages ?? 0);
const targetPages = job.totalPages > 0 ? Math.min(job.maxPages, job.totalPages) : job.maxPages;
if (
job.status === JOB_STATUS.RUNNING &&
job.totalPages > 0 &&
processedPages >= targetPages
) {
const completedJob = await this.repository.updateStatus(
job.id,
JOB_STATUS.COMPLETED,
{
finishedAt: job.finishedAt || new Date(),
},
);
return completedJob || { ...job, status: JOB_STATUS.COMPLETED };
}
return job; return job;
} }
...@@ -325,7 +356,7 @@ export class CrawlJobService { ...@@ -325,7 +356,7 @@ export class CrawlJobService {
} }
} }
await this.repository.delete(jobId); await this.repository.delete(jobId, userId);
return { success: true, message: "Crawl job deleted successfully" }; return { success: true, message: "Crawl job deleted successfully" };
} }
......
...@@ -6,10 +6,23 @@ import { CRAWL_PAGE_STATUS } from "../../common/constants/crawl-page-status.cons ...@@ -6,10 +6,23 @@ import { CRAWL_PAGE_STATUS } from "../../common/constants/crawl-page-status.cons
export class DashboardRepository { export class DashboardRepository {
async getStats(userId: string, role: string) { async getStats(userId: string, role: string) {
const isGlobal = role === ROLES.ADMIN; const isGlobal = role === ROLES.ADMIN;
const jobWhere = isGlobal ? {} : { userId }; const jobWhere = {
const pageWhere = isGlobal ? {} : { job: { userId } }; deletedAt: null,
...(isGlobal ? {} : { userId }),
};
const pageWhere = {
job: {
deletedAt: null,
...(isGlobal ? {} : { userId }),
},
};
const scheduleWhere = isGlobal ? {} : { userId }; const scheduleWhere = isGlobal ? {} : { userId };
const exportWhere = isGlobal ? {} : { job: { userId } }; const exportWhere = {
job: {
deletedAt: null,
...(isGlobal ? {} : { userId }),
},
};
const [ const [
jobStatusGroups, jobStatusGroups,
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment