增加新建订阅

back/services/subscription.ts

@@ -0,0 +1,254 @@
+import { Service, Inject } from 'typedi';
+import winston from 'winston';
+import config from '../config';
+import {
+  Subscription,
+  SubscriptionModel,
+  SubscriptionStatus,
+} from '../data/subscription';
+import { exec, execSync, spawn } from 'child_process';
+import fs from 'fs';
+import cron_parser from 'cron-parser';
+import { getFileContentByName, concurrentRun, fileExist } from '../config/util';
+import { promises, existsSync } from 'fs';
+import { promisify } from 'util';
+import { Op } from 'sequelize';
+import path from 'path';
+
+@Service()
+export default class SubscriptionService {
+  constructor(@Inject('logger') private logger: winston.Logger) {}
+
+  public async create(payload: Subscription): Promise<Subscription> {
+    const tab = new Subscription(payload);
+    const doc = await this.insert(tab);
+    return doc;
+  }
+
+  public async insert(payload: Subscription): Promise<Subscription> {
+    return await SubscriptionModel.create(payload, { returning: true });
+  }
+
+  public async update(payload: Subscription): Promise<Subscription> {
+    const newDoc = await this.updateDb(payload);
+    return newDoc;
+  }
+
+  public async updateDb(payload: Subscription): Promise<Subscription> {
+    await SubscriptionModel.update(payload, { where: { id: payload.id } });
+    return await this.getDb({ id: payload.id });
+  }
+
+  public async status({
+    ids,
+    status,
+    pid,
+    log_path,
+    last_running_time = 0,
+    last_execution_time = 0,
+  }: {
+    ids: number[];
+    status: SubscriptionStatus;
+    pid: number;
+    log_path: string;
+    last_running_time: number;
+    last_execution_time: number;
+  }) {
+    const options: any = {
+      status,
+      pid,
+      log_path,
+      last_execution_time,
+    };
+    if (last_running_time > 0) {
+      options.last_running_time = last_running_time;
+    }
+
+    return await SubscriptionModel.update(
+      { ...options },
+      { where: { id: ids } },
+    );
+  }
+
+  public async remove(ids: number[]) {
+    await SubscriptionModel.destroy({ where: { id: ids } });
+  }
+
+  public async getDb(query: any): Promise<Subscription> {
+    const doc: any = await SubscriptionModel.findOne({ where: { ...query } });
+    return doc && (doc.get({ plain: true }) as Subscription);
+  }
+
+  public async run(ids: number[]) {
+    await SubscriptionModel.update(
+      { status: SubscriptionStatus.queued },
+      { where: { id: ids } },
+    );
+    concurrentRun(
+      ids.map((id) => async () => await this.runSingle(id)),
+      10,
+    );
+  }
+
+  public async stop(ids: number[]) {
+    const docs = await SubscriptionModel.findAll({ where: { id: ids } });
+    for (const doc of docs) {
+      if (doc.pid) {
+        try {
+          process.kill(-doc.pid);
+        } catch (error) {
+          this.logger.silly(error);
+        }
+      }
+      const err = await this.killTask(doc.command);
+      const absolutePath = path.resolve(config.logPath, `${doc.log_path}`);
+      const logFileExist = doc.log_path && (await fileExist(absolutePath));
+      if (logFileExist) {
+        const str = err ? `\n${err}` : '';
+        fs.appendFileSync(
+          `${absolutePath}`,
+          `${str}\n## 执行结束...  ${new Date()
+            .toLocaleString('zh', { hour12: false })
+            .replace(' 24:', ' 00:')} `,
+        );
+      }
+    }
+
+    await SubscriptionModel.update(
+      { status: SubscriptionStatus.idle, pid: undefined },
+      { where: { id: ids } },
+    );
+  }
+
+  public async killTask(name: string) {
+    let taskCommand = `ps -ef | grep "${name}" | grep -v grep | awk '{print $1}'`;
+    const execAsync = promisify(exec);
+    try {
+      let pid = (await execAsync(taskCommand)).stdout;
+      if (pid) {
+        pid = (await execAsync(`pstree -p ${pid}`)).stdout;
+      } else {
+        return;
+      }
+      let pids = pid.match(/\(\d+/g);
+      const killLogs = [];
+      if (pids && pids.length > 0) {
+        // node 执行脚本时还会有10个子进程，但是ps -ef中不存在，所以截取前三个
+        for (const id of pids) {
+          const c = `kill -9 ${id.slice(1)}`;
+          try {
+            const { stdout, stderr } = await execAsync(c);
+            if (stderr) {
+              killLogs.push(stderr);
+            }
+            if (stdout) {
+              killLogs.push(stdout);
+            }
+          } catch (error: any) {
+            killLogs.push(error.message);
+          }
+        }
+      }
+      return killLogs.length > 0 ? JSON.stringify(killLogs) : '';
+    } catch (e) {
+      return JSON.stringify(e);
+    }
+  }
+
+  private async runSingle(cronId: number): Promise<number> {
+    return new Promise(async (resolve: any) => {
+      const cron = await this.getDb({ id: cronId });
+      if (cron.status !== SubscriptionStatus.queued) {
+        resolve();
+        return;
+      }
+
+      let { id, log_path } = cron;
+      const absolutePath = path.resolve(config.logPath, `${log_path}`);
+      const logFileExist = log_path && (await fileExist(absolutePath));
+
+      this.logger.silly('Running job');
+      this.logger.silly('ID: ' + id);
+      this.logger.silly('Original command: ');
+
+      let cmdStr = '';
+
+      const cp = spawn(cmdStr, { shell: '/bin/bash' });
+
+      await SubscriptionModel.update(
+        { status: SubscriptionStatus.running, pid: cp.pid },
+        { where: { id } },
+      );
+      cp.stderr.on('data', (data) => {
+        if (logFileExist) {
+          fs.appendFileSync(`${absolutePath}`, `${data}`);
+        }
+      });
+      cp.on('error', (err) => {
+        if (logFileExist) {
+          fs.appendFileSync(`${absolutePath}`, `${JSON.stringify(err)}`);
+        }
+      });
+
+      cp.on('exit', async (code, signal) => {
+        this.logger.info(`${''} pid: ${cp.pid} exit ${code} signal ${signal}`);
+        await SubscriptionModel.update(
+          { status: SubscriptionStatus.idle, pid: undefined },
+          { where: { id } },
+        );
+        resolve();
+      });
+      cp.on('close', async (code) => {
+        this.logger.info(`${''} pid: ${cp.pid} closed ${code}`);
+        await SubscriptionModel.update(
+          { status: SubscriptionStatus.idle, pid: undefined },
+          { where: { id } },
+        );
+        resolve();
+      });
+    });
+  }
+
+  public async disabled(ids: number[]) {
+    await SubscriptionModel.update({ isDisabled: 1 }, { where: { id: ids } });
+  }
+
+  public async enabled(ids: number[]) {
+    await SubscriptionModel.update({ isDisabled: 0 }, { where: { id: ids } });
+  }
+
+  public async log(id: number) {
+    const doc = await this.getDb({ id });
+    if (!doc) {
+      return '';
+    }
+
+    const absolutePath = path.resolve(config.logPath, `${doc.log_path}`);
+    const logFileExist = doc.log_path && (await fileExist(absolutePath));
+    if (logFileExist) {
+      return getFileContentByName(`${absolutePath}`);
+    }
+  }
+
+  public async logs(id: number) {
+    const doc = await this.getDb({ id });
+    if (!doc) {
+      return [];
+    }
+
+    if (doc.log_path) {
+      const relativeDir = path.dirname(`${doc.log_path}`);
+      const dir = path.resolve(config.logPath, relativeDir);
+      if (existsSync(dir)) {
+        let files = await promises.readdir(dir);
+        return files
+          .map((x) => ({
+            filename: x,
+            directory: relativeDir.replace(config.logPath, ''),
+            time: fs.statSync(`${dir}/${x}`).mtime.getTime(),
+          }))
+          .sort((a, b) => b.time - a.time);
+      }
+    }
+  }
+}