upd

packages/canyon-backend/src/tasks/services/cleaning-up-outdated-data.service.ts

@@ -0,0 +1,34 @@
+import { Cron } from '@nestjs/schedule';
+import { PrismaService } from '../../prisma/prisma.service';
+import { Injectable } from '@nestjs/common';
+import { deleteSpecificCoverageData } from '../../adapter/coverage-data.adapter';
+
+@Injectable()
+export class CleaningUpOutdatedDataService {
+  constructor(private readonly prisma: PrismaService) {}
+  // 每过一天清理一下coverage集合，删除超过3天的normal数据
+  @Cron('30 2 * * *') // 每天凌晨2点半执行
+  async cleaningUpOutdatedData() {
+    const conditions = {
+      where: {
+        covType: 'normal',
+        createdAt: {
+          lt: new Date(new Date().valueOf() - 3 * 24 * 60 * 60 * 1000),
+        },
+      },
+    };
+
+    const willDelCoverages = await this.prisma.coverage.findMany(conditions);
+
+    const coverageDeleteManyRes =
+      await this.prisma.coverage.deleteMany(conditions);
+
+    const ids = willDelCoverages.map(({ relationID }) => relationID);
+
+    for (let i = 0; i < ids.length; i++) {
+      await deleteSpecificCoverageData(ids[i]);
+    }
+
+    console.log(coverageDeleteManyRes, 'coverageDeleteManyRes');
+  }
+}

packages/canyon-backend/src/tasks/services/consumer-coverage.service.ts

@@ -15,37 +15,165 @@ import {
   mapToLowerCamelCase,
   removeNullKeys,
 } from '../../utils/utils';
-// import { PullChangeCodeAndInsertDbService } from './pull-change-code-and-insert-db.service';
+import { PullChangeCodeAndInsertDbService } from './pull-change-code-and-insert-db.service';
 const randomInteger = Math.floor(Math.random() * 1000) + 1000;
 
 @Injectable()
 export class ConsumerCoverageService {
   constructor(
     private readonly prisma: PrismaService,
-    // private readonly pullChangeCodeAndInsertDbService: PullChangeCodeAndInsertDbService,
+    private readonly pullChangeCodeAndInsertDbService: PullChangeCodeAndInsertDbService,
   ) {}
   public consumptionQueue = [];
 
+  async updateEmployeeSalary() {
+    return await this.prisma.$transaction(async (tx) => {
+      // 1.第一步：先把covType为normal的数据，consumer=1的数据，取出来，然后更新consumer=2，每次只消费一条。
+      const normalCoverage = await tx.$queryRaw`
+    SELECT * FROM coverage WHERE cov_type = 'normal' AND consumer = 1 ORDER BY created_at ASC LIMIT 1;
+    `.then((res: any) =>
+        res?.length > 0 ? mapToLowerCamelCase(res[0]) : null,
+      );
+
+      if (!normalCoverage) {
+        return null;
+      } else {
+        await tx.coverage.update({
+          where: {
+            id: normalCoverage.id,
+          },
+          data: {
+            consumer: 2,
+          },
+        });
+        return normalCoverage;
+      }
+    });
+  }
+  async pullChangeCode(coverage) {
+    if (coverage.sha !== coverage.compareTarget) {
+      await this.pullChangeCodeAndInsertDbService.invoke(
+        coverage.projectID,
+        coverage.sha,
+        coverage.compareTarget,
+        'accessToken',
+        this.prisma,
+      );
+    }
+  }
   @Interval(randomInteger) //添加随机数，防止分布式服务同时执行，务必确保原子性，1-2s
   async consumerCoverage() {
     if (this.consumptionQueue.length > 10) {
       return;
     }
-
-    const normalCoverage = {
-      id: Math.random(),
-    };
+    const normalCoverage = await this.updateEmployeeSalary();
+    if (!normalCoverage) {
+      return;
+    }
 
     this.consumptionQueue.push(normalCoverage.id);
 
     // 开个定时器，超时了就删除
-    setTimeout(
-      () => {
-        this.consumptionQueue = this.consumptionQueue.filter(
-          (item) => item !== normalCoverage.id,
-        );
-      },
-      1000 * 60 * 5,
+    setTimeout(() => {
+      this.consumptionQueue = this.consumptionQueue.filter(
+        (item) => item !== normalCoverage.id,
+      );
+    }, 1000 * 60);
+    // 获取与之对应的覆盖率数据
+    const normalCoverageData = await getSpecificCoverageData(
+      normalCoverage.relationID,
+    );
+
+    // 拉取变更代码
+    await this.pullChangeCode(normalCoverage);
+    // ************** 重要逻辑 **************
+    // 关键逻辑，这里需要把normal的数据，分别聚合成agg和all的数据
+    // ************** 重要逻辑 **************
+    const covTypes = ['agg', 'all'];
+    for (let i = 0; i < covTypes.length; i++) {
+      const covType = covTypes[i];
+      const covTypeCoverage = await this.prisma.coverage.findFirst({
+        where: removeNullKeys({
+          covType: covType,
+          sha: normalCoverage.sha,
+          reportID: covType === 'agg' ? normalCoverage.reportID : null,
+        }),
+      });
+
+      // covTypeCoverage为空，说明第一次，把CoverageData设置为{}
+      const covTypeCoverageData = covTypeCoverage
+        ? await getSpecificCoverageData(covTypeCoverage.relationID)
+        : {};
+
+      // normalCoverageData来自循环外部，covTypeCoverageData来自循环内部
+      const mainCov = mergeCoverageMap(normalCoverageData, covTypeCoverageData);
+      // 判断是否需要拉取变更代码，对比sha和compareTarget
+      const codechanges =
+        normalCoverage.sha === normalCoverage.compareTarget
+          ? []
+          : await this.prisma.codechange.findMany({
+              where: {
+                sha: normalCoverage.sha,
+                compareTarget: normalCoverage.compareTarget,
+              },
+            });
+
+      // 删除老的summary数据
+      await this.prisma.summary.deleteMany({
+        where: removeNullKeys({
+          reportID: covType === 'agg' ? normalCoverage.reportID : null,
+          sha: normalCoverage.sha,
+          covType: covType,
+        }),
+      });
+
+      // 生成覆盖率概览数据
+      const coverageSummaryMap = genSummaryMapByCoverageMap(
+        mainCov,
+        codechanges,
+      );
+      const allSummary = getSummaryByPath('~', coverageSummaryMap);
+      for (const allSummaryKey in allSummary) {
+        // 落库数据
+        const { total, skipped, covered } = allSummary[allSummaryKey] as any;
+        await this.prisma.summary.create({
+          data: {
+            reportID: normalCoverage.reportID,
+            metricType: allSummaryKey,
+            sha: normalCoverage.sha,
+            total,
+            skipped,
+            covered,
+            covType: covType,
+          },
+        });
+      }
+      // 创建新的覆盖率数据
+      const { key: newKey } = await createNewCoverageData(mainCov);
+
+      if (covTypeCoverage) {
+        await this.prisma.coverage.update({
+          where: {
+            id: covTypeCoverage.id,
+          },
+          data: {
+            relationID: newKey,
+            createdAt: new Date(),
+          },
+        });
+      } else {
+        await this.prisma.coverage.create({
+          data: {
+            ...deleteID(normalCoverage),
+            covType: covType,
+            relationID: normalCoverage.relationID,
+          },
+        });
+      }
+    }
+
+    this.consumptionQueue = this.consumptionQueue.filter(
+      (item) => item !== normalCoverage.id,
     );
   }
 }

packages/canyon-backend/src/tasks/services/pull-change-code-and-insert-db.service.ts

@@ -0,0 +1,36 @@
+import { diffLine } from '../../utils/diffline';
+
+export class PullChangeCodeAndInsertDbService {
+  async invoke(projectID, commitSha, compareTarget, accessToken, prisma) {
+    const codechanges = await prisma.codechange.findMany({
+      where: {
+        projectID,
+        commitSha,
+        compareTarget,
+      },
+    });
+    if (codechanges.length === 0) {
+      const diffLineData = await diffLine({
+        repoID: projectID,
+        baseCommitSha: compareTarget,
+        compareCommitSha: commitSha,
+        token: accessToken,
+        gitlabUrl: process.env.GITLAB_URL,
+      });
+      const data = diffLineData.map(({ path, additions, deletions }) => {
+        return {
+          projectID,
+          commitSha,
+          compareTarget,
+          path,
+          additions,
+          deletions,
+        };
+      });
+
+      await prisma.codechange.createMany({
+        data: data,
+      });
+    }
+  }
+}

packages/canyon-backend/src/tasks/tasks.module.ts

@@ -6,8 +6,17 @@ import { PrismaModule } from '../prisma/prisma.module';
 // import { AggregationCoverageService } from './services/aggregation-coverage.service';
 // import { PullChangeCodeAndInsertDbService } from './services/pull-change-code-and-insert-db.service';
 import { ConsumerCoverageService } from './services/consumer-coverage.service';
+import { CleaningUpOutdatedDataService } from './services/cleaning-up-outdated-data.service';
+import { PullChangeCodeAndInsertDbService } from './services/pull-change-code-and-insert-db.service';
 @Module({
   imports: [PrismaModule],
-  providers: [ConsumerCoverageService],
+  providers: [
+    // CoverageAggTaskService,
+    // RerunService,
+    CleaningUpOutdatedDataService,
+    // AggregationCoverageService,
+    PullChangeCodeAndInsertDbService,
+    ConsumerCoverageService,
+  ],
 })
 export class TasksModule {}