perf: 支持webhook触发流水线，新增触发类型图标显示

2026-04-24 20:57:26 +08:00 · 2026-01-09 01:10:43 +08:00
parent 2b353094eb
commit 1a29541140
17 changed files with 347 additions and 97 deletions
@@ -1,4 +1,6 @@
 ALTER TABLE cd_site_info ADD COLUMN ip_scan boolean DEFAULT (0);

 ALTER TABLE pi_pipeline ADD COLUMN webhook_key varchar(100);
+ALTER TABLE pi_pipeline ADD COLUMN trigger_count integer DEFAULT (0);

+CREATE INDEX "index_pipeline_webhook_key" ON "pi_pipeline" ("webhook_key");
@@ -0,0 +1,19 @@
+import { BaseController, Constants } from '@certd/lib-server';
+import { Controller, Get, Inject, Param, Post, Provide } from '@midwayjs/core';
+import { PipelineService } from '../../modules/pipeline/service/pipeline-service.js';
+
+/**
+ */
+@Provide()
+@Controller('/api/webhook/')
+export class WebhookController extends BaseController {
+  @Inject()
+  pipelineService: PipelineService;
+
+  @Get('/:webhookKey', { summary: Constants.per.guest })
+  @Post('/:webhookKey', { summary: Constants.per.guest })
+  async webhook(@Param('webhookKey') webhookKey: string): Promise<any> {
+    await this.pipelineService.triggerByWebhook(webhookKey);
+    return this.ok({});
+  }
+}
@@ -35,18 +35,18 @@ export class PipelineController extends CrudController<PipelineService> {
    const publicSettings = await this.sysSettingsService.getPublicSettings();

    let onlyOther = false
-    if(isAdmin){
-      if(publicSettings.managerOtherUserPipeline){
+    if (isAdmin) {
+      if (publicSettings.managerOtherUserPipeline) {
        //管理员管理 其他用户
-        if( body.query.userId === -1){
+        if (body.query.userId === -1) {
          //如果只查询其他用户
          onlyOther = true;
          delete body.query.userId;
        }
-      }else{
+      } else {
        body.query.userId = this.getUserId();
      }
-    }else{
+    } else {
      body.query.userId = this.getUserId();
    }

@@ -57,7 +57,7 @@ export class PipelineController extends CrudController<PipelineService> {
      if (title) {
        qb.andWhere('(title like :title or content like :content)', { title: `%${title}%`, content: `%${title}%` });
      }
-      if(onlyOther){
+      if (onlyOther) {
        qb.andWhere('user_id  != :userId', { userId: this.getUserId() });
      }
    };
@@ -76,7 +76,7 @@ export class PipelineController extends CrudController<PipelineService> {

  @Post('/getSimpleByIds', { summary: Constants.per.authOnly })
  async getSimpleById(@Body(ALL) body) {
-    const ret = await this.getService().getSimplePipelines(body.ids,this.getUserId() );
+    const ret = await this.getService().getSimplePipelines(body.ids, this.getUserId());
    return this.ok(ret);
  }

@@ -95,19 +95,19 @@ export class PipelineController extends CrudController<PipelineService> {
  }

  @Post('/save', { summary: Constants.per.authOnly })
-  async save(@Body(ALL) bean: {addToMonitorEnabled: boolean, addToMonitorDomains: string} & PipelineEntity) {
+  async save(@Body(ALL) bean: { addToMonitorEnabled: boolean, addToMonitorDomains: string } & PipelineEntity) {
    if (bean.id > 0) {
      await this.authService.checkEntityUserId(this.ctx, this.getService(), bean.id);
    } else {
      bean.userId = this.getUserId();
    }

-    if(!this.isAdmin()){
+    if (!this.isAdmin()) {
      // 非管理员用户 不允许设置流水线有效期
      delete bean.validTime
    }

-    const {version} = await this.service.save(bean);
+    const { version } = await this.service.save(bean);
    //是否增加证书监控
    if (bean.addToMonitorEnabled && bean.addToMonitorDomains) {
      const sysPublicSettings = await this.sysSettingsService.getPublicSettings();
@@ -119,7 +119,7 @@ export class PipelineController extends CrudController<PipelineService> {
        });
      }
    }
-    return this.ok({id:bean.id,version:version});
+    return this.ok({ id: bean.id, version: version });
  }

  @Post('/delete', { summary: Constants.per.authOnly })
@@ -147,7 +147,7 @@ export class PipelineController extends CrudController<PipelineService> {
  @Post('/trigger', { summary: Constants.per.authOnly })
  async trigger(@Query('id') id: number, @Query('stepId') stepId?: string) {
    await this.authService.checkEntityUserId(this.ctx, this.getService(), id);
-    await this.service.trigger(id, stepId,true);
+    await this.service.trigger(id, stepId, true);
    return this.ok({});
  }

@@ -167,7 +167,7 @@ export class PipelineController extends CrudController<PipelineService> {
  @Post('/batchDelete', { summary: Constants.per.authOnly })
  async batchDelete(@Body('ids') ids: number[]) {
    const isAdmin = await this.authService.isAdmin(this.ctx);
-    const userId = isAdmin ?  undefined : this.getUserId() ;
+    const userId = isAdmin ? undefined : this.getUserId();
    await this.service.batchDelete(ids, userId);
    return this.ok({});
  }
@@ -177,7 +177,7 @@ export class PipelineController extends CrudController<PipelineService> {
  @Post('/batchUpdateGroup', { summary: Constants.per.authOnly })
  async batchUpdateGroup(@Body('ids') ids: number[], @Body('groupId') groupId: number) {
    const isAdmin = await this.authService.isAdmin(this.ctx);
-    const userId = isAdmin ?  undefined : this.getUserId() ;
+    const userId = isAdmin ? undefined : this.getUserId();
    await this.service.batchUpdateGroup(ids, groupId, userId);
    return this.ok({});
  }
@@ -186,7 +186,7 @@ export class PipelineController extends CrudController<PipelineService> {
  @Post('/batchUpdateTrigger', { summary: Constants.per.authOnly })
  async batchUpdateTrigger(@Body('ids') ids: number[], @Body('trigger') trigger: any) {
    const isAdmin = await this.authService.isAdmin(this.ctx);
-    const userId = isAdmin ?  undefined : this.getUserId() ;
+    const userId = isAdmin ? undefined : this.getUserId();
    await this.service.batchUpdateTrigger(ids, trigger, userId);
    return this.ok({});
  }
@@ -194,7 +194,7 @@ export class PipelineController extends CrudController<PipelineService> {
  @Post('/batchUpdateNotification', { summary: Constants.per.authOnly })
  async batchUpdateNotification(@Body('ids') ids: number[], @Body('notification') notification: any) {
    const isAdmin = await this.authService.isAdmin(this.ctx);
-    const userId = isAdmin ?  undefined : this.getUserId() ;
+    const userId = isAdmin ? undefined : this.getUserId();
    await this.service.batchUpdateNotifications(ids, notification, userId);
    return this.ok({});
  }
@@ -204,4 +204,14 @@ export class PipelineController extends CrudController<PipelineService> {
    await this.service.batchRerun(ids, this.getUserId(), force);
    return this.ok({});
  }
+
+  @Post('/refreshWebhookKey', { summary: Constants.per.authOnly })
+  async refreshWebhookKey(@Body('id') id: number) {
+    await this.authService.checkEntityUserId(this.ctx, this.getService(), id);
+    const res = await this.service.refreshWebhookKey(id);
+    return this.ok({
+      webhookKey: res,
+    });
+  }
+
 }
@@ -33,6 +33,12 @@ export class PipelineEntity {
  @Column({ comment: '类型', nullable: true, default: 'cert' })
  type: string;

+  @Column({ name: 'webhook_key', comment: 'webhookkey', length: 100, nullable: true })
+  webhookKey: string;
+
+  @Column({ name: 'trigger_count', comment: '触发器数量', nullable: true, default: 0 })
+  triggerCount: number;
+
  // custom: 自定义; monitor: 监控;
  @Column({ comment: '来源', nullable: true, default: '' })
  from: string;
@@ -39,7 +39,7 @@ import { PluginConfigGetter } from "../../plugin/service/plugin-config-getter.js
 import dayjs from "dayjs";
 import { DbAdapter } from "../../db/index.js";
 import { isComm, isPlus } from "@certd/plus-core";
-import { logger } from "@certd/basic";
+import { logger, utils } from "@certd/basic";
 import { UrlService } from "./url-service.js";
 import { NotificationService } from "./notification-service.js";
 import { UserSuiteEntity, UserSuiteService } from "@certd/commercial-core";
@@ -177,7 +177,7 @@ export class PipelineService extends BaseService<PipelineEntity> {
    if (!info.disabled) {
      const pipeline = JSON.parse(info.content);
      this.registerTriggers(pipeline, false);
-    }else {
+    } else {
      this.unregisterTriggers(info);
    }
  }
@@ -214,6 +214,10 @@ export class PipelineService extends BaseService<PipelineEntity> {
      //修改
      old = await this.info(bean.id);
    }
+    if (!old || !old.webhookKey ) {
+      bean.webhookKey = await this.genWebhookKey();
+    }
+
    const isUpdate = bean.id > 0 && old != null;


@@ -273,6 +277,9 @@ export class PipelineService extends BaseService<PipelineEntity> {
      pipeline.version = 0;
    }
    pipeline.version++;
+
+    bean.triggerCount = pipeline.triggers?.filter((trigger) => trigger.type === "timer").length || 0;
+
    bean.content = JSON.stringify(pipeline);
    await this.addOrUpdate(bean);
    await this.registerTrigger(bean);
@@ -816,8 +823,8 @@ export class PipelineService extends BaseService<PipelineEntity> {
    if (!isPlus()) {
      throw new NeedVIPException("此功能需要升级专业版");
    }
-    const query :any = {}
-    if(userId && userId>0){
+    const query: any = {}
+    if (userId && userId > 0) {
      query.userId = userId;
    }
    await this.repository.update(
@@ -834,8 +841,8 @@ export class PipelineService extends BaseService<PipelineEntity> {
    if (!isPlus()) {
      throw new NeedVIPException("此功能需要升级专业版");
    }
-    const query :any = {}
-    if(userId && userId>0){
+    const query: any = {}
+    if (userId && userId > 0) {
      query.userId = userId;
    }
    const list = await this.find({
@@ -851,20 +858,20 @@ export class PipelineService extends BaseService<PipelineEntity> {
        //清除trigger
        pipeline.triggers = []
      } else {
-        if(trigger.random === true){
-            //随机时间
-            const start = dayjs().format("YYYY-MM-DD") + " " + trigger.randomRange[0];
-            let end = dayjs().format("YYYY-MM-DD") + " " + trigger.randomRange[1];
-            if(trigger.randomRange[1]<trigger.randomRange[0]){
-              //跨天
-              end = dayjs().add(1, "day").format("YYYY-MM-DD") + " " + trigger.randomRange[1];
-            }
-            const startTime = dayjs(start).valueOf();
-            const endTime = dayjs(end).valueOf();
-            const randomTime = Math.floor(Math.random() * (endTime - startTime)) + startTime;
-            const time = dayjs(randomTime).format(" ss:mm:HH").replaceAll(":", " ").replaceAll(" 0", " ").trim();
-            set(trigger,"props.cron",  `${time} * * *`)
-        } 
+        if (trigger.random === true) {
+          //随机时间
+          const start = dayjs().format("YYYY-MM-DD") + " " + trigger.randomRange[0];
+          let end = dayjs().format("YYYY-MM-DD") + " " + trigger.randomRange[1];
+          if (trigger.randomRange[1] < trigger.randomRange[0]) {
+            //跨天
+            end = dayjs().add(1, "day").format("YYYY-MM-DD") + " " + trigger.randomRange[1];
+          }
+          const startTime = dayjs(start).valueOf();
+          const endTime = dayjs(end).valueOf();
+          const randomTime = Math.floor(Math.random() * (endTime - startTime)) + startTime;
+          const time = dayjs(randomTime).format(" ss:mm:HH").replaceAll(":", " ").replaceAll(" 0", " ").trim();
+          set(trigger, "props.cron", `${time} * * *`)
+        }
        delete trigger.random
        delete trigger.randomRange;
        pipeline.triggers = [{
@@ -883,8 +890,8 @@ export class PipelineService extends BaseService<PipelineEntity> {
    if (!isPlus()) {
      throw new NeedVIPException("此功能需要升级专业版");
    }
-    const query :any = {}
-    if(userId && userId>0){
+    const query: any = {}
+    if (userId && userId > 0) {
      query.userId = userId;
    }
    const list = await this.find({
@@ -935,19 +942,19 @@ export class PipelineService extends BaseService<PipelineEntity> {
    ids = list.map(item => item.id);

    //异步执行
-    this.startBatchRerun(userId,ids, force);
+    this.startBatchRerun(userId, ids, force);
  }

-  startBatchRerun(userId:number, ids: number[], force: boolean) {
+  startBatchRerun(userId: number, ids: number[], force: boolean) {
    for (const id of ids) {
-      executorQueue.addTask(userId,{
-          task: async () => {
-            if (force) {
-              await this.run(id, null, "ALL");
-            } else {
-              await this.run(id, null);
-            }
+      executorQueue.addTask(userId, {
+        task: async () => {
+          if (force) {
+            await this.run(id, null, "ALL");
+          } else {
+            await this.run(id, null);
          }
+        }
      });
    }
  }
@@ -1107,4 +1114,33 @@ export class PipelineService extends BaseService<PipelineEntity> {
    await this.repository.update(id, { disabled });
    await this.registerTriggerById(id);
  }
+
+  async refreshWebhookKey(id: number) {
+    const webhookKey = await this.genWebhookKey();
+    await this.repository.update(id, { webhookKey });
+    return webhookKey;
+  }
+  async genWebhookKey() {
+    return utils.id.simpleNanoId(24);
+  }
+  async triggerByWebhook(webhookKey: string) {
+    const pipelineEntity = await this.findOne({
+      select: {
+        id: true,
+        content: true,
+      },
+      where: {
+        webhookKey
+      }
+    })
+    if (!pipelineEntity) {
+      throw new Error("webhookKey不存在");
+    }
+    const pipeline = JSON.parse(pipelineEntity.content);
+    const trigger = pipeline.triggers.find((trigger: any) => trigger.type === "webhook");
+    if (!trigger) {
+      throw new Error("该流水线的webhook未启用");
+    }
+    await this.run(pipelineEntity.id, trigger.id);
+  }
 }