perf: 数据库备份支持oss

packages/plugins/plugin-lib/src/oss/alioss.ts

@@ -1,25 +0,0 @@
-import { AliyunAccess } from "../aliyun";
-import { HttpClient, ILogger } from "@certd/basic";
-import { IOssClient, OssClientDeleteReq } from "./api";
-
-export class AliossClient implements IOssClient {
-  access: AliyunAccess;
-  logger: ILogger;
-  http: HttpClient;
-
-  constructor(opts: { access: AliyunAccess; http: HttpClient; logger: ILogger }) {
-    this.access = opts.access;
-    this.http = opts.http;
-    this.logger = opts.logger;
-  }
-
-  upload(key: string, content: Buffer | string): Promise<void> {
-    throw new Error("Method not implemented.");
-  }
-  download(key: string, savePath?: string): Promise<void> {
-    throw new Error("Method not implemented.");
-  }
-  delete(opts: OssClientDeleteReq): Promise<void> {
-    throw new Error("Method not implemented.");
-  }
-}

packages/plugins/plugin-lib/src/oss/api.ts

@@ -1,12 +1,89 @@
-export type OssClientDeleteReq = {
-  key: string;
+import { IAccessService } from "@certd/pipeline";
+import { ILogger, utils } from "@certd/basic";
+import dayjs from "dayjs";
+
+export type OssClientRemoveByOpts = {
+  dir?: string;
+  //删除多少天前的文件
   beforeDays?: number;
 };
 
-export interface IOssClient {
-  upload(key: string, content: Buffer | string): Promise<void>;
+export type OssFileItem = {
+  name: string;
+  path: string;
+  size: number;
+  //毫秒时间戳
+  lastModified: number;
+};
 
-  download(key: string, savePath?: string): Promise<void>;
+export type IOssClient = {
+  upload: (fileName: string, fileContent: Buffer) => Promise<void>;
+  remove: (fileName: string) => Promise<void>;
 
-  delete(opts: OssClientDeleteReq): Promise<void>;
+  download: (fileName: string, savePath: string) => Promise<void>;
+
+  removeBy: (removeByOpts: OssClientRemoveByOpts) => Promise<void>;
+
+  listDir: (dir: string) => Promise<OssFileItem[]>;
+};
+
+export type OssClientContext = {
+  accessService: IAccessService;
+  logger: ILogger;
+  utils: typeof utils;
+};
+
+export abstract class BaseOssClient<A> implements IOssClient {
+  rootDir: string = "";
+  access: A = null;
+  logger: ILogger;
+  utils: typeof utils;
+  ctx: OssClientContext;
+
+  protected constructor(opts: { rootDir?: string; access: A }) {
+    this.rootDir = opts.rootDir || "";
+    this.access = opts.access;
+  }
+
+  join(...strs: string[]) {
+    let res = "";
+    for (const item of strs) {
+      if (item) {
+        if (!res) {
+          res = item;
+        } else {
+          res += "/" + item;
+        }
+      }
+    }
+    res = res.replace(/[\\/]+/g, "/");
+    return res;
+  }
+
+  async setCtx(ctx: any) {
+    // set context
+    this.ctx = ctx;
+    this.logger = ctx.logger;
+    this.utils = ctx.utils;
+    await this.init();
+  }
+
+  async init() {
+    // do nothing
+  }
+
+  abstract remove(fileName: string): Promise<void>;
+  abstract upload(fileName: string, fileContent: Buffer): Promise<void>;
+  abstract download(fileName: string, savePath: string): Promise<void>;
+  abstract listDir(dir: string): Promise<OssFileItem[]>;
+
+  async removeBy(removeByOpts: OssClientRemoveByOpts): Promise<void> {
+    const list = await this.listDir(removeByOpts.dir);
+    const beforeDate = dayjs().subtract(removeByOpts.beforeDays, "day");
+    for (const item of list) {
+      if (item.lastModified && item.lastModified < beforeDate.valueOf()) {
+        await this.remove(item.path);
+      }
+    }
+  }
 }

packages/plugins/plugin-lib/src/oss/factory.ts

@@ -0,0 +1,41 @@
+import { OssClientContext } from "./api";
+
+export class OssClientFactory {
+  async getClassByType(type: string) {
+    if (type === "alioss") {
+      const module = await import("./impls/alioss.js");
+      return module.default;
+    } else if (type === "ssh") {
+      const module = await import("./impls/ssh.js");
+      return module.default;
+    } else if (type === "sftp") {
+      const module = await import("./impls/sftp.js");
+      return module.default;
+    } else if (type === "ftp") {
+      const module = await import("./impls/ftp.js");
+      return module.default;
+    } else if (type === "tencentcos") {
+      const module = await import("./impls/tencentcos.js");
+      return module.default;
+    } else if (type === "qiniuoss") {
+      const module = await import("./impls/qiniuoss.js");
+      return module.default;
+    } else if (type === "s3") {
+      const module = await import("./impls/s3.js");
+      return module.default;
+    } else {
+      throw new Error(`暂不支持此文件上传方式: ${type}`);
+    }
+  }
+  async createOssClientByType(type: string, opts: { rootDir: string; access: any; ctx: OssClientContext }) {
+    const cls = await this.getClassByType(type);
+    if (cls) {
+      // @ts-ignore
+      const instance = new cls(opts);
+      await instance.setCtx(opts.ctx);
+      return instance;
+    }
+  }
+}
+
+export const ossClientFactory = new OssClientFactory();

packages/plugins/plugin-lib/src/oss/impls/alioss.ts

@@ -0,0 +1,54 @@
+import { BaseOssClient, OssFileItem } from "../api.js";
+import { AliossAccess, AliossClient, AliyunAccess } from "../../aliyun/index.js";
+import dayjs from "dayjs";
+
+export default class AliOssClientImpl extends BaseOssClient<AliossAccess> {
+  client: AliossClient;
+  join(...strs: string[]) {
+    const str = super.join(...strs);
+    if (str.startsWith("/")) {
+      return str.substring(1);
+    }
+    return str;
+  }
+  async init() {
+    const aliyunAccess = await this.ctx.accessService.getById<AliyunAccess>(this.access.accessId);
+    const client = new AliossClient({
+      access: aliyunAccess,
+      bucket: this.access.bucket,
+      region: this.access.region,
+    });
+    await client.init();
+    this.client = client;
+  }
+  async download(filePath: string, savePath: string): Promise<void> {
+    const key = this.join(this.rootDir, filePath);
+    await this.client.downloadFile(key, savePath);
+  }
+  async listDir(dir: string): Promise<OssFileItem[]> {
+    const dirKey = this.join(this.rootDir, dir) + "/";
+    const list = await this.client.listDir(dirKey);
+    this.logger.info(`列出目录: ${dirKey},文件数：${list.length}`);
+    return list.map(item => {
+      return {
+        path: item.name,
+        lastModified: dayjs(item.lastModified).valueOf(),
+        size: item.size,
+      };
+    });
+  }
+  async upload(filePath: string, fileContent: Buffer) {
+    const key = this.join(this.rootDir, filePath);
+    this.logger.info(`开始上传文件: ${key}`);
+    await this.client.uploadFile(key, fileContent);
+
+    this.logger.info(`文件上传成功: ${filePath}`);
+  }
+
+  async remove(filePath: string) {
+    const key = this.join(this.rootDir, filePath);
+    // remove file from alioss
+    await this.client.removeFile(key);
+    this.logger.info(`文件删除成功: ${key}`);
+  }
+}

packages/plugins/plugin-lib/src/oss/impls/ftp.ts

@@ -0,0 +1,46 @@
+import { BaseOssClient } from "../api.js";
+import path from "path";
+import os from "os";
+import fs from "fs";
+import { FtpAccess, FtpClient } from "../../ftp/index.js";
+
+export default class FtpOssClientImpl extends BaseOssClient<FtpAccess> {
+  async download(fileName: string, savePath: string) {}
+  async listDir(dir: string) {
+    return [];
+  }
+  async upload(filePath: string, fileContent: Buffer) {
+    const client = this.getFtpClient();
+    await client.connect(async client => {
+      const tmpFilePath = path.join(os.tmpdir(), "cert", "http", filePath);
+      const dir = path.dirname(tmpFilePath);
+      if (!fs.existsSync(dir)) {
+        fs.mkdirSync(dir, { recursive: true });
+      }
+      fs.writeFileSync(tmpFilePath, fileContent);
+      try {
+        // Write file to temp path
+        const path = this.join(this.rootDir, filePath);
+        await client.upload(path, tmpFilePath);
+      } finally {
+        // Remove temp file
+        fs.unlinkSync(tmpFilePath);
+      }
+    });
+  }
+
+  private getFtpClient() {
+    return new FtpClient({
+      access: this.access,
+      logger: this.logger,
+    });
+  }
+
+  async remove(filePath: string) {
+    const client = this.getFtpClient();
+    await client.connect(async client => {
+      const path = this.join(this.rootDir, filePath);
+      await client.client.remove(path);
+    });
+  }
+}

packages/plugins/plugin-lib/src/oss/impls/qiniuoss.ts

@@ -0,0 +1,33 @@
+import { QiniuAccess, QiniuClient, QiniuOssAccess } from "../../qiniu/index.js";
+import { BaseOssClient, OssClientRemoveByOpts, OssFileItem } from "../api.js";
+
+export default class QiniuOssClientImpl extends BaseOssClient<QiniuOssAccess> {
+  client: QiniuClient;
+  async init() {
+    const qiniuAccess = await this.ctx.accessService.getById<QiniuAccess>(this.access.accessId);
+    this.client = new QiniuClient({
+      access: qiniuAccess,
+      logger: this.logger,
+      http: this.ctx.utils.http,
+    });
+  }
+
+  download(fileName: string, savePath: string): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  removeBy(removeByOpts: OssClientRemoveByOpts): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  listDir(dir: string): Promise<OssFileItem[]> {
+    throw new Error("Method not implemented.");
+  }
+  async upload(filePath: string, fileContent: Buffer) {
+    const path = this.join(this.rootDir, filePath);
+    await this.client.uploadFile(this.access.bucket, path, fileContent);
+  }
+
+  async remove(filePath: string) {
+    const path = this.join(this.rootDir, filePath);
+    await this.client.removeFile(this.access.bucket, path);
+  }
+}

packages/plugins/plugin-lib/src/oss/impls/s3.ts

@@ -0,0 +1,55 @@
+import { BaseOssClient, OssClientRemoveByOpts, OssFileItem } from "../api.js";
+import path from "node:path";
+import { S3Access } from "../../s3/access.js";
+export default class S3OssClientImpl extends BaseOssClient<S3Access> {
+  client: any;
+
+  async init() {
+    // import { S3Client } from "@aws-sdk/client-s3";
+    const { S3Client } = await import("@aws-sdk/client-s3");
+    this.client = new S3Client({
+      forcePathStyle: true,
+      credentials: {
+        accessKeyId: this.access.accessKeyId, // 默认 MinIO 访问密钥
+        secretAccessKey: this.access.secretAccessKey, // 默认 MinIO 秘密密钥
+      },
+      region: "us-east-1",
+      endpoint: this.access.endpoint,
+    });
+  }
+
+  download(fileName: string, savePath: string): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  removeBy(removeByOpts: OssClientRemoveByOpts): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  listDir(dir: string): Promise<OssFileItem[]> {
+    throw new Error("Method not implemented.");
+  }
+  async upload(filePath: string, fileContent: Buffer) {
+    const { PutObjectCommand } = await import("@aws-sdk/client-s3");
+    const key = path.join(this.rootDir, filePath);
+    this.logger.info(`开始上传文件: ${key}`);
+    const params = {
+      Bucket: this.access.bucket, // The name of the bucket. For example, 'sample_bucket_101'.
+      Key: key, // The name of the object. For example, 'sample_upload.txt'.
+    };
+    await this.client.send(new PutObjectCommand({ Body: fileContent, ...params }));
+
+    this.logger.info(`文件上传成功: ${filePath}`);
+  }
+
+  async remove(filePath: string) {
+    const key = path.join(this.rootDir, filePath);
+    const { DeleteObjectCommand } = await import("@aws-sdk/client-s3");
+    await this.client.send(
+      new DeleteObjectCommand({
+        Bucket: this.access.bucket,
+        Key: key,
+      })
+    );
+
+    this.logger.info(`文件删除成功: ${key}`);
+  }
+}

packages/plugins/plugin-lib/src/oss/impls/sftp.ts

@@ -0,0 +1,59 @@
+import { BaseOssClient, OssClientRemoveByOpts, OssFileItem } from "../api.js";
+import path from "path";
+import os from "os";
+import fs from "fs";
+import { SftpAccess, SshAccess, SshClient } from "../../ssh/index.js";
+
+export default class SftpOssClientImpl extends BaseOssClient<SftpAccess> {
+  download(fileName: string, savePath: string): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  removeBy(removeByOpts: OssClientRemoveByOpts): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  listDir(dir: string): Promise<OssFileItem[]> {
+    throw new Error("Method not implemented.");
+  }
+  async upload(filePath: string, fileContent: Buffer) {
+    const tmpFilePath = path.join(os.tmpdir(), "cert", "http", filePath);
+
+    // Write file to temp path
+    const dir = path.dirname(tmpFilePath);
+    if (!fs.existsSync(dir)) {
+      fs.mkdirSync(dir, { recursive: true });
+    }
+    fs.writeFileSync(tmpFilePath, fileContent);
+
+    const access = await this.ctx.accessService.getById<SshAccess>(this.access.sshAccess);
+    const key = this.rootDir + filePath;
+    try {
+      const client = new SshClient(this.logger);
+      await client.uploadFiles({
+        connectConf: access,
+        mkdirs: true,
+        transports: [
+          {
+            localPath: tmpFilePath,
+            remotePath: key,
+          },
+        ],
+        opts: {
+          mode: this.access?.fileMode ?? undefined,
+        },
+      });
+    } finally {
+      // Remove temp file
+      fs.unlinkSync(tmpFilePath);
+    }
+  }
+
+  async remove(filePath: string) {
+    const access = await this.ctx.accessService.getById<SshAccess>(this.access.sshAccess);
+    const client = new SshClient(this.logger);
+    const key = this.rootDir + filePath;
+    await client.removeFiles({
+      connectConf: access,
+      files: [key],
+    });
+  }
+}

packages/plugins/plugin-lib/src/oss/impls/ssh.ts

@@ -0,0 +1,54 @@
+import { BaseOssClient, OssClientRemoveByOpts, OssFileItem } from "../api.js";
+import path from "path";
+import os from "os";
+import fs from "fs";
+import { SshAccess, SshClient } from "../../ssh/index.js";
+
+export default class SshOssClientImpl extends BaseOssClient<SshAccess> {
+  download(fileName: string, savePath: string): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  removeBy(removeByOpts: OssClientRemoveByOpts): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  listDir(dir: string): Promise<OssFileItem[]> {
+    throw new Error("Method not implemented.");
+  }
+  async upload(filePath: string, fileContent: Buffer) {
+    const tmpFilePath = path.join(os.tmpdir(), "cert", "http", filePath);
+
+    // Write file to temp path
+    const dir = path.dirname(tmpFilePath);
+    if (!fs.existsSync(dir)) {
+      fs.mkdirSync(dir, { recursive: true });
+    }
+    fs.writeFileSync(tmpFilePath, fileContent);
+
+    const key = this.rootDir + filePath;
+    try {
+      const client = new SshClient(this.logger);
+      await client.uploadFiles({
+        connectConf: this.access,
+        mkdirs: true,
+        transports: [
+          {
+            localPath: tmpFilePath,
+            remotePath: key,
+          },
+        ],
+      });
+    } finally {
+      // Remove temp file
+      fs.unlinkSync(tmpFilePath);
+    }
+  }
+
+  async remove(filePath: string) {
+    const client = new SshClient(this.logger);
+    const key = this.rootDir + filePath;
+    await client.removeFiles({
+      connectConf: this.access,
+      files: [key],
+    });
+  }
+}

packages/plugins/plugin-lib/src/oss/impls/tencentcos.ts

@@ -0,0 +1,37 @@
+import { TencentAccess, TencentCosAccess, TencentCosClient } from "../../tencent/index.js";
+import { BaseOssClient, OssClientRemoveByOpts, OssFileItem } from "../api.js";
+
+export default class TencentOssClientImpl extends BaseOssClient<TencentCosAccess> {
+  download(fileName: string, savePath: string): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  removeBy(removeByOpts: OssClientRemoveByOpts): Promise<void> {
+    throw new Error("Method not implemented.");
+  }
+  listDir(dir: string): Promise<OssFileItem[]> {
+    throw new Error("Method not implemented.");
+  }
+  async upload(filePath: string, fileContent: Buffer) {
+    const access = await this.ctx.accessService.getById<TencentAccess>(this.access.accessId);
+    const client = new TencentCosClient({
+      access: access,
+      logger: this.logger,
+      region: this.access.region,
+      bucket: this.access.bucket,
+    });
+    const key = this.rootDir + filePath;
+    await client.uploadFile(key, fileContent);
+  }
+
+  async remove(filePath: string) {
+    const access = await this.ctx.accessService.getById<TencentAccess>(this.access.accessId);
+    const client = new TencentCosClient({
+      access: access,
+      logger: this.logger,
+      region: this.access.region,
+      bucket: this.access.bucket,
+    });
+    const key = this.rootDir + filePath;
+    await client.removeFile(key);
+  }
+}

packages/plugins/plugin-lib/src/oss/index.ts

@@ -0,0 +1,2 @@
+export * from "./factory.js";
+export * from "./api.js";