使用主动查询的方法替换企业微信的通知

src/api.server.ts

@@ -9,6 +9,8 @@ import { mongoose } from '@typegoose/typegoose'
 import logger from 'logger/logger'
 import BlocknumSchedule from 'schedule/blocknum.schedule'
 import path from 'path'
+import { RedisClient } from 'redis/RedisClient'
+import WxTaskSchedule from 'schedule/wxtask.schedule'
 
 const zReqParserPlugin = require('plugins/zReqParser')
 
@@ -109,11 +111,12 @@ export class ApiServer {
       logger.log(`DB Connection Error: ${err.message}`)
     }
     let opts = { url: process.env.REDIS }
-    // new RedisClient(opts)
+    new RedisClient(opts)
-    // logger.log('REDIS Connected')
+    logger.log('REDIS Connected')
   }
   private initSchedules() {
     new BlocknumSchedule().scheduleAll()
+    new WxTaskSchedule().scheduleAll()
   }
   private restoreChainQueue() {}
   private setErrHandler() {
@@ -165,7 +168,7 @@ export class ApiServer {
       self.registerRouter()
       self.setErrHandler()
       self.setFormatSend()
-      // self.initSchedules()
+      self.initSchedules()
       // restartAllUnFinishedTask()
       this.server.listen({ port: parseInt(process.env.API_PORT) }, (err: any, address: any) => {
         if (err) {

src/controllers/workflow.controller.ts

@@ -36,16 +36,16 @@ class WorkFlowController extends BaseController {
     if (msg_signature !== signature) {
       throw new ZError(10, 'sign check failed')
     }
-    const { message, id } = decrypt(aesKey, xml.Encrypt)
+    // const { message, id } = decrypt(aesKey, xml.Encrypt)
-    let parser = new XMLParser()
+    // let parser = new XMLParser()
-    let jsonData = parser.parse(message)
+    // let jsonData = parser.parse(message)
-    let spStatus = jsonData.xml?.ApprovalInfo?.SpStatus
+    // let spStatus = jsonData.xml?.ApprovalInfo?.SpStatus
-    if (spStatus === TaskStatus.PASS) {
+    // if (spStatus === TaskStatus.PASS) {
-      let spNo = jsonData.xml?.ApprovalInfo?.SpNo
+    //   let spNo = jsonData.xml?.ApprovalInfo?.SpNo
-      if (spNo) {
+    //   if (spNo) {
-        new TaskQueue().addTaskToQueue(spNo)
+    //     new TaskQueue().addTaskToQueue(spNo)
-      }
+    //   }
-    }
+    // }
     res.send('success')
   }
 

src/queue/confirm.queue.ts

@@ -23,7 +23,7 @@ export class ConfirmQueue {
     this.queue.push(async () => {
       try {
         let receipt = await waitTransaction(this.web3, task.txHash)
-        logger.info(`receipt confirmed: ${task.txHash}`)
+        logger.info(`receipt confirmed: status: ${receipt.statue}, txhash: ${task.txHash}`)
         if (isSuccessfulTransaction(receipt)) {
           if (task.status === ReqTaskStatus.WAIT_CONFIRM) {
             task.status = ReqTaskStatus.WAIT_EXEC

src/schedule/wxtask.schedule.ts

@@ -0,0 +1,31 @@
+import { singleton } from 'decorators/singleton'
+import logger from 'logger/logger'
+import { ChainTask } from 'models/ChainTask'
+import * as schedule from 'node-schedule'
+import { TaskQueue } from 'queue/task.queue'
+import { WechatWorkService } from 'service/wechatwork.service'
+
+@singleton
+export default class WxTaskSchedule {
+  async parseAllRecord() {
+    let detail: any = await new WechatWorkService().queryTasks()
+    if (detail.errcode) {
+      logger.info('approval list error, code: ' + detail.errcode + ' errmsg: ' + detail.errmsg)
+      return
+    }
+    const { sp_no_list } = detail
+    for (let spNo of sp_no_list) {
+      let record = await ChainTask.findOne({ taskId: spNo })
+      if (record) {
+        continue
+      }
+      logger.info('got one task: ' + spNo)
+      new TaskQueue().addTaskToQueue(spNo)
+    }
+  }
+  scheduleAll() {
+    const job = schedule.scheduleJob(' */1  * * * *', async () => {
+      await this.parseAllRecord()
+    })
+  }
+}

src/service/mail.service.ts

@@ -1,4 +1,5 @@
 import { singleton } from 'decorators/singleton'
+import logger from 'logger/logger'
 import { createTransport, Transporter } from 'nodemailer'
 import Mail from 'nodemailer/lib/mailer'
 
@@ -21,6 +22,8 @@ export class MailService {
   }
 
   public async send(message: Mail.Options) {
+    logger.info('begin send mail: ')
+    logger.info(JSON.stringify(message))
     await this.transporter.verify()
     return this.transporter.sendMail(message)
   }

src/service/wechatwork.service.ts

@@ -3,6 +3,7 @@ import { singleton } from 'decorators/singleton'
 import fs from 'fs'
 import os from 'os'
 import path from 'path'
+import { RedisClient } from 'redis/RedisClient'
 import { excelToJson } from 'utils/excel.util'
 
 // 1-审批中；2-已通过；3-已驳回；4-已撤销；6-通过后撤销；7-已删除；10-已支付
@@ -25,6 +26,7 @@ export class WechatWorkService {
   private wxAesKey: string
   private wxCorpId: string
   private wxCorpSecret: string
+  private timePre: number
 
   constructor() {
     this.wxToken = process.env.WX_TOKEN
@@ -148,4 +150,47 @@ export class WechatWorkService {
     fs.writeFileSync(filePath, res.data)
     return filePath
   }
+  // 查询审批列表
+  public async queryTasks() {
+    const url = `${WX_API_HOST}/cgi-bin/oa/getapprovalinfo`
+    const access_token = await this.getAccessToken()
+    let starttime = this.timePre
+    if (!this.timePre) {
+      let timeStr = await new RedisClient().get('qywx_time_cache')
+      if (timeStr) {
+        starttime = parseInt(timeStr)
+      }
+    }
+    starttime = starttime || 1681401600
+    let endtime = (Date.now() / 1000) | 0
+    let config: AxiosRequestConfig = {
+      method: 'post',
+      url,
+      params: {
+        access_token,
+      },
+      data: {
+        starttime,
+        endtime,
+        cursor: 0,
+        size: 100,
+        filters: [
+          {
+            key: 'template_id',
+            value: process.env.WX_TEMPLATE_ID,
+          },
+          {
+            key: 'sp_status',
+            value: '2',
+          },
+        ],
+      },
+    }
+    let response = await axios.request(config).then(response => {
+      return response.data
+    })
+    this.timePre = starttime
+    await new RedisClient().set('qywx_time_cache', starttime + '')
+    return response
+  }
 }