fix(dependencies): quote and validate Python source installs

2026-10-16 22:35:32 +08:00 · 2026-10-12 00:23:48 +08:00
parent fcf0ee619b 70636900b2
commit 4cd6c6ad2e
494 changed files with 56787 additions and 5398 deletions
@@ -1,9 +1,10 @@
-import { Service, Inject } from 'typedi';
-import path, { join } from 'path';
+import { Service } from 'typedi';
 import config from '../config';
 import { getFileContentByName } from '../config/util';
+import { t } from '../shared/i18n';
 import { Response } from 'express';
 import { request } from 'undici';
+import { resolveFileAccess } from '../shared/fileAccess';

@Service()
 export default class ConfigService {
@@ -11,19 +12,16 @@ export default class ConfigService {

  public async getFile(filePath: string, res: Response) {
    let content = '';
-    const avaliablePath = [config.rootPath, config.configPath].map((x) =>
-      path.resolve(x, filePath),
+    if (!filePath || filePath.startsWith('data/scripts/')) {
+      return res.send({ code: 403, message: t('文件无法访问') });
+    }
+    const resolved = resolveFileAccess(
+      config.configPath,
+      [filePath],
+      config.blackFileList,
    );
-
-    if (
-      config.blackFileList.includes(filePath) ||
-      avaliablePath.every(
-        (x) =>
-          !x.startsWith(config.scriptPath) && !x.startsWith(config.configPath),
-      ) ||
-      !filePath
-    ) {
-      return res.send({ code: 403, message: '文件无法访问' });
+    if (!resolved) {
+      return res.send({ code: 403, message: t('文件无法访问') });
    }

    if (filePath.startsWith('sample/')) {
@@ -31,10 +29,8 @@ export default class ConfigService {
        `https://gitlab.com/whyour/qinglong/-/raw/master/${filePath}`,
      );
      content = await res.body.text();
-    } else if (filePath.startsWith('data/scripts/')) {
-      content = await getFileContentByName(join(config.rootPath, filePath));
    } else {
-      content = await getFileContentByName(join(config.configPath, filePath));
+      content = await getFileContentByName(resolved);
    }

    res.send({ code: 200, data: content });
@@ -1,15 +1,24 @@
+import { randomUUID } from 'crypto';
+import { getInvalidCronSchedules, isValidCronSchedule } from '../shared/cronSchedule';
+import {
+  withSchedulerMutation,
+  schedulerRegistrationError,
+} from '../shared/schedulerMutationLock';
 import { Service, Inject } from 'typedi';
 import winston from 'winston';
 import config from '../config';
+import { resolveLogPath } from '../shared/logPath';
 import { Crontab, CrontabModel, CrontabStatus } from '../data/cron';
+import {
+  RunningInstanceModel,
+  InstanceStatus,
+} from '../data/runningInstance';
 import { exec, execSync } from 'child_process';
 import fs from 'fs/promises';
-import CronExpressionParser from 'cron-parser';
 import {
  getFileContentByName,
  fileExist,
  killTask,
-  killAllTasks,
  getUniqPath,
  safeJSONParse,
  isDemoEnv,
@@ -24,8 +33,13 @@ import dayjs from 'dayjs';
 import pickBy from 'lodash/pickBy';
 import omit from 'lodash/omit';
 import { writeFileWithLock } from '../shared/utils';
+import { t } from '../shared/i18n';
 import { ScheduleType } from '../interface/schedule';
 import { logStreamManager } from '../shared/logStreamManager';
+import { observeChildProcess, asError } from '../shared/childProcess';
+import { isEmpty } from 'lodash';
+import { LogReadOptions, readLogChunk } from '../shared/logReader';
+import { resolveFileAccess } from '../shared/fileAccess';

@Service()
 export default class CronService {
@@ -33,10 +47,40 @@ export default class CronService {

  private isNodeCron(cron: Crontab) {
    const { schedule, extra_schedules } = cron;
-    if (Number(schedule?.split(/ +/).length) > 5 || extra_schedules?.length) {
-      return true;
+    const fields = schedule?.trim().split(/\s+/) || [];
+    // System crontab only receives portable numeric five-field expressions.
+    // Extended syntax, macros and legacy shorthand use the Node scheduler.
+    return (
+      fields.length !== 5 ||
+      /[^\d\s*,/\-]/.test(schedule || '') ||
+      Boolean(extra_schedules?.length) ||
+      // BusyBox treats N/step as a single value and doesn't support Sunday=7.
+      fields.some((field) => field.split(',').some((part) => /^\d+\//.test(part))) ||
+      fields[4].includes('7') ||
+      // BusyBox steps DOM from zero; its full-range day/week wildcard
+      // handling also differs from the legacy Node parser's OR semantics.
+      fields[2].includes('/') ||
+      (fields[2] !== '*' && fields[4] !== '*')
+    );
+  }
+
+  private get schedulerMode(): 'system' | 'node' {
+    const env = process.env.QL_SCHEDULER;
+    if (env === 'system') return 'system';
+    if (env === 'node') return 'node';
+    try {
+      execSync('which crond', { stdio: 'ignore' });
+      return 'system';
+    } catch {
+      return 'node';
    }
-    return false;
+  }
+
+  private shouldUseCronClient(cron: Crontab): boolean {
+    if (this.schedulerMode === 'node') {
+      return !this.isSpecialSchedule(cron.schedule);
+    }
+    return this.isNodeCron(cron) && !this.isSpecialSchedule(cron.schedule);
  }

  private isOnceSchedule(schedule?: string) {
@@ -71,29 +115,45 @@ export default class CronService {
  }

  public async create(payload: Crontab): Promise<Crontab> {
-    const tab = new Crontab(payload);
-    tab.saved = false;
-    tab.log_name = await this.getLogName(tab);
-    const doc = await this.insert(tab);
+    return withSchedulerMutation(async () => {
+      const tab = new Crontab(payload);
+      tab.saved = false;
+      tab.log_name = await this.getLogName(tab);
+      const doc = await this.insert(tab);

-    if (isDemoEnv()) {
+      if (isDemoEnv()) {
+        return doc;
+      }
+
+      if (this.shouldUseCronClient(doc)) {
+        try {
+          await cronClient.addCron([
+            {
+              name: doc.name || '',
+              id: String(doc.id),
+              schedule: doc.schedule!,
+              command: this.makeCommand(doc),
+              extra_schedules: doc.extra_schedules || [],
+            },
+          ]);
+        } catch (error: any) {
+          // gRPC 注册失败时回滚 DB 记录，避免产生"僵尸任务"
+          // （DB 和 crontab.list 有记录但调度器永远不会执行）
+          await CrontabModel.destroy({ where: { id: doc.id } });
+          this.logger.error(
+            '[crontab] Failed to register cron job in scheduler, task creation rolled back:',
+            error?.message || error,
+          );
+          throw schedulerRegistrationError(
+            `${t('调度器注册失败，任务创建已回滚')}: ${(error as any)?.details || error?.message}`,
+            error,
+          );
+        }
+      }
+
+      await this.setCrontab();
      return doc;
-    }
-
-    if (this.isNodeCron(doc) && !this.isSpecialSchedule(doc.schedule)) {
-      await cronClient.addCron([
-        {
-          name: doc.name || '',
-          id: String(doc.id),
-          schedule: doc.schedule!,
-          command: this.makeCommand(doc),
-          extra_schedules: doc.extra_schedules || [],
-        },
-      ]);
-    }
-
-    await this.setCrontab();
-    return doc;
+    });
  }

  public async insert(payload: Crontab): Promise<Crontab> {
@@ -101,34 +161,68 @@ export default class CronService {
  }

  public async update(payload: Partial<Crontab>): Promise<Crontab> {
-    const doc = await this.getDb({ id: payload.id });
-    const tab = new Crontab({ ...doc, ...payload });
-    tab.saved = false;
-    tab.log_name = await this.getLogName(tab);
-    const newDoc = await this.updateDb(tab);
+    return withSchedulerMutation(async () => {
+      const doc = await this.getDb({ id: payload.id });
+      const tab = new Crontab({ ...doc, ...payload });
+      tab.saved = false;
+      tab.log_name = await this.getLogName(tab);
+      if (doc.isDisabled === 1 || isDemoEnv()) {
+        return await this.updateDb(tab);
+      }

-    if (doc.isDisabled === 1 || isDemoEnv()) {
-      return newDoc;
-    }
-
-    if (this.isNodeCron(doc)) {
+      // Keep the DB snapshot unchanged if deletion has an uncertain outcome.
+      // Recovery uses that snapshot after this mutation releases its lock.
      await cronClient.delCron([String(doc.id)]);
-    }
+      const newDoc = await this.updateDb(tab);

-    if (this.isNodeCron(newDoc) && !this.isSpecialSchedule(newDoc.schedule)) {
-      await cronClient.addCron([
-        {
-          name: doc.name || '',
-          id: String(newDoc.id),
-          schedule: newDoc.schedule!,
-          command: this.makeCommand(newDoc),
-          extra_schedules: newDoc.extra_schedules || [],
-        },
-      ]);
-    }
+      if (this.shouldUseCronClient(newDoc)) {
+        try {
+          await cronClient.addCron([
+            {
+              name: doc.name || '',
+              id: String(newDoc.id),
+              schedule: newDoc.schedule!,
+              command: this.makeCommand(newDoc),
+              extra_schedules: newDoc.extra_schedules || [],
+            },
+          ]);
+        } catch (error: any) {
+          // gRPC 注册新任务失败 → 回滚 DB 到旧数据，并尝试恢复旧调度注册
+          await CrontabModel.update(omit(doc, ['queued_token']), {
+            where: { id: doc.id },
+          });
+          if (this.shouldUseCronClient(doc)) {
+            try {
+              await cronClient.addCron([
+                {
+                  name: doc.name || '',
+                  id: String(doc.id),
+                  schedule: doc.schedule!,
+                  command: this.makeCommand(doc),
+                  extra_schedules: doc.extra_schedules || [],
+                },
+              ]);
+            } catch (_recoveryError: any) {
+              this.logger.warn(
+                '[crontab] Failed to restore old cron job in scheduler after rollback:',
+                _recoveryError?.message || _recoveryError,
+              );
+            }
+          }
+          this.logger.error(
+            '[crontab] Failed to register updated cron job in scheduler, update rolled back:',
+            error?.message || error,
+          );
+          throw schedulerRegistrationError(
+            `${t('调度器注册失败，任务更新已回滚')}: ${(error as any)?.details || error?.message}`,
+            error,
+          );
+        }
+      }

-    await this.setCrontab();
-    return newDoc;
+      await this.setCrontab();
+      return newDoc;
+    });
  }

  public async updateDb(payload: Crontab): Promise<Crontab> {
@@ -143,6 +237,7 @@ export default class CronService {
    log_path,
    last_running_time = 0,
    last_execution_time = 0,
+    exit_code,
  }: {
    ids: number[];
    status: CrontabStatus;
@@ -150,7 +245,11 @@ export default class CronService {
    log_path: string;
    last_running_time: number;
    last_execution_time: number;
+    exit_code?: number;
  }) {
+    if (log_path && log_path !== '/dev/null') {
+      resolveLogPath(config.logPath, log_path);
+    }
    let options: any = {
      status,
      pid,
@@ -172,6 +271,40 @@ export default class CronService {
      if (status === CrontabStatus.idle && log_path !== cron.log_path) {
        options = omit(options, ['status', 'log_path', 'pid']);
      }
+
+      // Manage RunningInstance records for status transitions from shell scripts
+      if (status === CrontabStatus.running) {
+        // Create a new running instance record
+        await RunningInstanceModel.create({
+          cron_id: id,
+          pid: pid || undefined,
+          log_path: log_path || undefined,
+          started_at: last_execution_time || dayjs().unix(),
+          status: InstanceStatus.running,
+        });
+      } else if (status === CrontabStatus.idle) {
+        // Mark the matching running instance as finished
+        const finishedAt = dayjs().unix();
+        const instanceStatus =
+          exit_code !== undefined && exit_code !== null && exit_code !== 0
+            ? InstanceStatus.error
+            : InstanceStatus.finished;
+        await RunningInstanceModel.update(
+          {
+            finished_at: finishedAt,
+            status: instanceStatus,
+            exit_code: exit_code ?? undefined,
+          },
+          {
+            where: {
+              cron_id: id,
+              pid: pid || undefined,
+              status: InstanceStatus.running,
+            },
+          },
+        );
+      }
+
      await CrontabModel.update(
        { ...pickBy(options, (v) => v === 0 || !!v) },
        { where: { id } },
@@ -180,9 +313,11 @@ export default class CronService {
  }

  public async remove(ids: number[]) {
-    await CrontabModel.destroy({ where: { id: ids } });
-    await cronClient.delCron(ids.map(String));
-    await this.setCrontab();
+    return withSchedulerMutation(async () => {
+      await cronClient.delCron(ids.map(String));
+      await CrontabModel.destroy({ where: { id: ids } });
+      await this.setCrontab();
+    });
  }

  public async pin(ids: number[]) {
@@ -350,7 +485,7 @@ export default class CronService {
  }

  private formatFilterQuery(query: any, filterQuery: any) {
-    if (filterQuery) {
+    if (!isEmpty(filterQuery)) {
      if (!query[Op.and]) {
        query[Op.and] = [];
      }
@@ -453,175 +588,385 @@ export default class CronService {
  }

  public async run(ids: number[]) {
+    const queuedToken = randomUUID();
    await CrontabModel.update(
-      { status: CrontabStatus.queued },
+      { status: CrontabStatus.queued, queued_token: queuedToken },
      { where: { id: ids } },
    );
    ids.forEach((id) => {
-      this.runSingle(id);
+      this.runSingle(id, queuedToken);
    });
  }

  public async stop(ids: number[]) {
    const docs = await CrontabModel.findAll({ where: { id: ids } });
+    // Cancel the queued snapshot first, so a late spawn cannot claim it.
    for (const doc of docs) {
-      // Kill all running instances of this task
-      try {
-        if (doc.pid) {
-          await killTask(doc.pid);
-        }
-        const command = doc.command.replace(/\s+/g, ' ').trim();
-        await killAllTasks(command);
-        this.logger.info(
-          `[panel][停止所有运行中的任务实例] 任务ID: ${doc.id}, 命令: ${command}`,
+      if (doc.status === CrontabStatus.queued) {
+        const [cancelled] = await CrontabModel.update(
+          { status: CrontabStatus.idle, pid: null, queued_token: null } as any,
+          {
+            where: {
+              id: doc.id,
+              status: CrontabStatus.queued,
+              [Op.and]: [
+                where(colFn('log_path'), { [Op.eq]: doc.log_path ?? null }),
+                where(colFn('queued_token'), {
+                  [Op.eq]: doc.queued_token ?? null,
+                }),
+              ],
+            },
+          }
        );
+        // A concurrent claim may have won; capture its PID before signalling.
+        if (!cancelled) await doc.reload();
+      }
+    }
+    const stoppingInstances = await RunningInstanceModel.findAll({
+      attributes: ['id', 'pid', 'cron_id'],
+      where: { cron_id: ids, status: InstanceStatus.running },
+    });
+    const targets = new Set<number>(
+      [
+        ...stoppingInstances.map((instance) => instance.pid),
+        ...docs
+          .filter((doc) => doc.status === CrontabStatus.running)
+          .map((doc) => doc.pid),
+      ].filter((pid): pid is number => typeof pid === 'number' && pid > 0)
+    );
+    const stopped = new Set<number>();
+    for (const pid of targets) {
+      try {
+        await killTask(pid, true);
+        stopped.add(pid);
      } catch (error) {
        this.logger.error(
-          `[panel][停止任务失败] 任务ID: ${doc.id}, 错误: ${error}`,
+          '[panel][停止任务失败] PID: %s, 错误: %s',
+          pid,
+          asError(error).message
        );
      }
    }
-
-    await CrontabModel.update(
-      { status: CrontabStatus.idle, pid: undefined },
-      { where: { id: ids } },
-    );
+    const stoppedIds = stoppingInstances
+      .filter((instance) => instance.pid && stopped.has(instance.pid))
+      .map((instance) => instance.id!);
+    if (stoppedIds.length) {
+      await RunningInstanceModel.update(
+        { status: InstanceStatus.stopped, finished_at: dayjs().unix() },
+        { where: { id: stoppedIds } }
+      );
+    }
+    for (const doc of docs) {
+      if (
+        doc.status !== CrontabStatus.running ||
+        (doc.pid && !stopped.has(doc.pid))
+      )
+        continue;
+      const remaining = await RunningInstanceModel.count({
+        where: { cron_id: doc.id, status: InstanceStatus.running },
+      });
+      if (remaining) continue;
+      await CrontabModel.update(
+        { status: CrontabStatus.idle, pid: null, queued_token: null } as any,
+        {
+          where: {
+            id: doc.id,
+            status: CrontabStatus.running,
+            [Op.and]: [
+              where(colFn('queued_token'), { [Op.eq]: doc.queued_token ?? null }),
+              where(colFn('pid'), { [Op.eq]: doc.pid ?? null }),
+              where(colFn('log_path'), { [Op.eq]: doc.log_path ?? null }),
+            ],
+          },
+        }
+      );
+    }
  }

-  private async runSingle(cronId: number): Promise<number | void> {
-    return taskLimit.manualRunWithCronLimit(() => {
-      return new Promise(async (resolve: any) => {
-        const cron = await this.getDb({ id: cronId });
-        const params = {
-          name: cron.name,
-          command: cron.command,
-          schedule: cron.schedule,
-          extra_schedules: cron.extra_schedules,
-        };
-        if (cron.status !== CrontabStatus.queued) {
-          resolve(params);
-          return;
-        }
-
-        this.logger.info(
-          `[panel][开始执行任务] 参数: ${JSON.stringify(params)}`,
+  public async stopInstance(instanceId: number) {
+    const instance = await RunningInstanceModel.findOne({
+      where: { id: instanceId, status: InstanceStatus.running },
+    });
+    if (!instance) {
+      return { code: 400, message: t('实例不存在或已停止') };
+    }
+    if (instance.pid) {
+      try {
+        await killTask(instance.pid);
+      } catch (error) {
+        this.logger.error(
+          `[panel][停止实例失败] 实例ID: ${instanceId}, PID: ${instance.pid}, 错误: ${error}`,
        );
+      }
+    }
+    await RunningInstanceModel.update(
+      { status: InstanceStatus.stopped, finished_at: dayjs().unix(), exit_code: 143 },
+      { where: { id: instanceId } },
+    );

-        let { id, command, log_name } = cron;
+    // Check if there are still other running instances for this cron
+    const otherRunning = await RunningInstanceModel.count({
+      where: { cron_id: instance.cron_id, status: InstanceStatus.running },
+    });
+    if (otherRunning === 0) {
+      await CrontabModel.update(
+        { status: CrontabStatus.idle, pid: undefined },
+        { where: { id: instance.cron_id } },
+      );
+    }
+    return { code: 200, message: t('实例已停止') };
+  }

+  private async runSingle(
+    cronId: number,
+    expectedToken?: string,
+  ): Promise<number | void> {
+    return taskLimit.manualRunWithCronLimit(async () => {
+      let absolutePath: string | undefined;
+      let logPath: string | undefined;
+      let queuedLogPath: string | null | undefined;
+      let queuedToken: string | null = null;
+      let claimed = false;
+      try {
+        const cron = await this.getDb({ id: cronId });
+        if (
+          cron.status !== CrontabStatus.queued ||
+          (expectedToken !== undefined && cron.queued_token !== expectedToken)
+        ) return;
+        queuedToken = cron.queued_token ?? null;
+        queuedLogPath = cron.log_path ?? null;
+        const { id, command, log_name } = cron;
        const uniqPath =
          log_name === '/dev/null' || !log_name
            ? await getUniqPath(command, `${id}`)
            : log_name;
        const logTime = dayjs().format('YYYY-MM-DD-HH-mm-ss-SSS');
-        const logDirPath = path.resolve(config.logPath, `${uniqPath}`);
-        await fs.mkdir(logDirPath, { recursive: true });
-        const logPath = `${uniqPath}/${logTime}.log`;
-        const absolutePath = path.resolve(config.logPath, `${logPath}`);
+        logPath = `${uniqPath}/${logTime}.log`;
+        absolutePath = resolveFileAccess(config.logPath, [logPath]);
+        if (!absolutePath)
+          throw new Error('Log path is outside the log directory');
+        await fs.mkdir(path.dirname(absolutePath), { recursive: true });
+        const outputPath = absolutePath;
        const cp = spawn(
          `real_log_path=${logPath} no_delay=true ${this.makeCommand(
            cron,
-            true,
+            true
          )}`,
-          { shell: '/bin/bash' },
+          { shell: '/bin/bash' }
        );
-
-        await CrontabModel.update(
-          { status: CrontabStatus.running, pid: cp.pid, log_path: logPath },
-          { where: { id } },
-        );
-        cp.stdout.on('data', async (data) => {
-          await logStreamManager.write(absolutePath, data.toString());
+        // Install observers before the first await: very short children may already exit.
+        const { completed } = observeChildProcess(cp, {
+          onStart: async () => {
+            try {
+              const [count] = await CrontabModel.update(
+                {
+                  status: CrontabStatus.running,
+                  pid: cp.pid,
+                  log_path: logPath,
+                },
+                {
+                  where: {
+                    id,
+                    status: CrontabStatus.queued,
+                    [Op.and]: [
+                      where(colFn('queued_token'), { [Op.eq]: queuedToken }),
+                      where(colFn('log_path'), { [Op.eq]: queuedLogPath }),
+                    ],
+                  },
+                }
+              );
+              if (count !== 1)
+                throw new Error(
+                  'Task was stopped or superseded before startup'
+                );
+              claimed = true;
+            } catch (error) {
+              if (cp.pid) await killTask(cp.pid, true);
+              throw error;
+            }
+          },
+          onStdout: (message) => logStreamManager.write(outputPath, message),
+          onStderr: (message) => logStreamManager.write(outputPath, message),
        });
-        cp.stderr.on('data', async (data) => {
-          this.logger.info(
-            '[panel][执行任务失败] 命令: %s, 错误信息: %j',
-            command,
-            data.toString(),
-          );
-          await logStreamManager.write(absolutePath, data.toString());
-        });
-        cp.on('error', async (err) => {
+        const result = await completed;
+        if (result.error) {
          this.logger.error(
-            '[panel][创建任务失败] 命令: %s, 错误信息: %j',
-            command,
-            err,
+            '[panel][执行任务失败] 任务ID: %s, 错误: %s',
+            id,
+            result.error.message
          );
-          await logStreamManager.write(absolutePath, JSON.stringify(err));
-        });
-
-        cp.on('exit', async (code) => {
-          this.logger.info(
-            '[panel][执行任务结束] 参数: %s, 退出码: %j',
-            JSON.stringify(params),
-            code,
+        }
+        this.logger.info(
+          '[panel][执行任务结束] 任务ID: %s, 退出码: %j',
+          id,
+          result.code
+        );
+        return { ...cron, pid: cp.pid, ...result } as any;
+      } catch (error) {
+        this.logger.error(
+          '[panel][创建任务失败] 任务ID: %s, 错误: %s',
+          cronId,
+          asError(error).message
+        );
+      } finally {
+        try {
+          if (absolutePath) await logStreamManager.closeStream(absolutePath);
+        } catch (error) {
+          this.logger.error(
+            '[panel][关闭任务日志失败] %s',
+            asError(error).message
          );
-          // Close the stream after task completion
-          await logStreamManager.closeStream(absolutePath);
+        }
+        try {
+          // Do not overwrite a newer run's state or its script-reported exit code.
          await CrontabModel.update(
-            { status: CrontabStatus.idle, pid: undefined },
-            { where: { id } },
+            { status: CrontabStatus.idle, pid: null, queued_token: null } as any,
+            {
+              where: {
+                id: cronId,
+                [Op.and]: where(colFn('queued_token'), { [Op.eq]: queuedToken }),
+                [Op.or]: [
+                  ...(queuedLogPath !== undefined && !claimed
+                    ? [
+                        {
+                          status: CrontabStatus.queued,
+                          [Op.and]: where(colFn('log_path'), {
+                            [Op.eq]: queuedLogPath,
+                          }),
+                        },
+                      ]
+                    : []),
+                  ...(claimed && logPath
+                    ? [{ log_path: logPath, status: CrontabStatus.running }]
+                    : []),
+                ],
+              },
+            }
          );
-          resolve({ ...params, pid: cp.pid, code });
-        });
-      });
+        } catch (error) {
+          this.logger.error(
+            '[panel][清理任务状态失败] %s',
+            asError(error).message
+          );
+        }
+      }
    });
  }

  public async disabled(ids: number[]) {
-    await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
-    await cronClient.delCron(ids.map(String));
-    await this.setCrontab();
+    return withSchedulerMutation(async () => {
+      await cronClient.delCron(ids.map(String));
+      await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
+      await this.setCrontab();
+    });
  }

  public async enabled(ids: number[]) {
-    await CrontabModel.update({ isDisabled: 0 }, { where: { id: ids } });
-    const docs = await CrontabModel.findAll({ where: { id: ids } });
-    const sixCron = docs
-      .filter((x) => this.isNodeCron(x) && !this.isSpecialSchedule(x.schedule))
-      .map((doc) => ({
-        name: doc.name || '',
-        id: String(doc.id),
-        schedule: doc.schedule!,
-        command: this.makeCommand(doc),
-        extra_schedules: doc.extra_schedules || [],
-      }));
+    return withSchedulerMutation(async () => {
+      await CrontabModel.update({ isDisabled: 0 }, { where: { id: ids } });
+      const docs = await CrontabModel.findAll({ where: { id: ids } });
+      const crons = docs
+        .filter((x) => this.shouldUseCronClient(x))
+        .map((doc) => ({
+          name: doc.name || '',
+          id: String(doc.id),
+          schedule: doc.schedule!,
+          command: this.makeCommand(doc),
+          extra_schedules: doc.extra_schedules || [],
+        }));

-    if (isDemoEnv()) {
-      return;
-    }
-    await cronClient.addCron(sixCron);
-    await this.setCrontab();
+      if (isDemoEnv()) {
+        return;
+      }
+
+      try {
+        await cronClient.addCron(crons);
+      } catch (error: any) {
+        // gRPC 注册失败 → 回滚启用状态，避免 DB 显示已启用但调度器未注册
+        await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
+        this.logger.error(
+          '[crontab] Failed to register cron job in scheduler, enable rolled back:',
+          error?.message || error,
+        );
+        throw schedulerRegistrationError(
+          `${t('调度器注册失败，任务启用已回滚')}: ${(error as any)?.details || error?.message}`,
+          error,
+        );
+      }
+      await this.setCrontab();
+    });
  }

-  public async log(id: number) {
+  public async log(
+    id: number,
+    options: LogReadOptions = {},
+  ): Promise<{
+    content: string;
+    status: string;
+    offset: number;
+    nextOffset: number;
+    total: number;
+    truncated: boolean;
+  }> {
    const doc = await this.getDb({ id });
    if (!doc) {
-      return '';
+      return {
+        content: '',
+        status: 'empty',
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
    }
    if (doc.log_name === '/dev/null') {
-      return '日志设置为忽略';
+      return {
+        content: t('日志设置为忽略'),
+        status: 'ignored',
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
    }
-    const absolutePath = path.resolve(config.logPath, `${doc.log_path}`);
+    const absolutePath = doc.log_path
+      ? resolveLogPath(config.logPath, doc.log_path)
+      : '';
    const logFileExist = doc.log_path && (await fileExist(absolutePath));
    if (logFileExist) {
-      return await getFileContentByName(`${absolutePath}`);
+      const chunk = await readLogChunk(`${absolutePath}`, options);
+      const isRunning =
+        typeof doc.status === 'number' &&
+        [CrontabStatus.running, CrontabStatus.queued].includes(doc.status);
+      return {
+        ...chunk,
+        status: isRunning ? 'running' : 'completed',
+      };
    } else {
-      return typeof doc.status === 'number' &&
+      const status =
+        typeof doc.status === 'number' &&
        [CrontabStatus.queued, CrontabStatus.running].includes(doc.status)
-        ? '运行中...'
-        : '日志不存在...';
+          ? 'running'
+          : 'notFound';
+      return {
+        content: status === 'running' ? t('运行中...') : t('日志不存在...'),
+        status,
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
    }
  }

  public async logs(id: number) {
    const doc = await this.getDb({ id });
-    if (!doc || !doc.log_path) {
+    if (!doc || !doc.log_path || doc.log_path === '/dev/null') {
      return [];
    }

    const relativeDir = path.dirname(`${doc.log_path}`);
-    const dir = path.resolve(config.logPath, relativeDir);
+    const dir = path.dirname(resolveLogPath(config.logPath, doc.log_path));
    const dirExist = await fileExist(dir);
    if (dirExist) {
      let files = await fs.readdir(dir);
@@ -661,6 +1006,9 @@ export default class CronService {
        .replace(/;? *\n/g, ';')
        .trim()}' `;
    }
+    if (tab.work_dir) {
+      commandVariable += `work_dir='${tab.work_dir.replace(/'/g, "'\\''")}' `;
+    }

    const crontab_job_string = `${commandVariable}${command}`;
    return crontab_job_string;
@@ -690,11 +1038,13 @@ export default class CronService {

    await writeFileWithLock(config.crontabFile, crontab_string);

-    try {
-      execSync(`crontab ${config.crontabFile}`);
-    } catch (error: any) {
-      const errorMsg = error.message || String(error);
-      this.logger.error('[crontab] Failed to update system crontab:', errorMsg);
+    if (this.schedulerMode === 'system') {
+      try {
+        execSync(`crontab ${config.crontabFile}`);
+      } catch (error: any) {
+        const errorMsg = error.message || String(error);
+        this.logger.error('[crontab] Failed to update system crontab:', errorMsg);
+      }
    }

    await CrontabModel.update({ saved: true }, { where: {} });
@@ -720,7 +1070,7 @@ export default class CronService {
        if (
          command &&
          schedule &&
-          CronExpressionParser.parse(schedule).hasNext()
+          isValidCronSchedule(schedule)
        ) {
          const name = namePrefix + '_' + index;

@@ -739,29 +1089,56 @@ export default class CronService {
    });
  }

-  public async autosave_crontab() {
-    const tabs = await this.crontabs();
-    const regularCrons = tabs.data
-      .filter(
-        (x) =>
-          x.isDisabled !== 1 &&
-          this.isNodeCron(x) &&
-          !this.isSpecialSchedule(x.schedule),
-      )
-      .map((doc) => ({
-        name: doc.name || '',
-        id: String(doc.id),
-        schedule: doc.schedule!,
-        command: this.makeCommand(doc),
-        extra_schedules: doc.extra_schedules || [],
-      }));
+  public async autosave_crontab(requireScheduler = false) {
+    return withSchedulerMutation(async () => {
+      const tabs = await this.crontabs();
+      // A bad persisted rule must not block startup or all other schedules.
+      // Keep the DB row editable; omit it from both runtime scheduler snapshots.
+      tabs.data = tabs.data.filter((doc) => {
+        if (doc.isDisabled === 1) return true;
+        const invalidSchedules = getInvalidCronSchedules(doc);
+        if (invalidSchedules.length === 0) return true;
+        this.logger.warn(
+          '[crontab] Skipping task with invalid schedule (id=%s, name=%s): %s',
+          doc.id,
+          doc.name || '',
+          JSON.stringify(invalidSchedules),
+        );
+        return false;
+      });
+      const regularCrons = tabs.data
+        .filter(
+          (x) =>
+            x.isDisabled !== 1 &&
+            this.shouldUseCronClient(x),
+        )
+        .map((doc) => ({
+          name: doc.name || '',
+          id: String(doc.id),
+          schedule: doc.schedule!,
+          command: this.makeCommand(doc),
+          extra_schedules: doc.extra_schedules || [],
+        }));

-    if (isDemoEnv()) {
-      await writeFileWithLock(config.crontabFile, '');
-      return;
-    }
-    await cronClient.addCron(regularCrons);
-    this.setCrontab(tabs);
+      if (isDemoEnv()) {
+        await writeFileWithLock(config.crontabFile, '');
+        return;
+      }
+
+      // 先同步 crontab.list 与系统 crontab，确保其始终反映数据库真实状态。
+      // gRPC 调度注册为尽力而为：失败时不阻断文件同步，调度器重启后会重新注册。
+      // 这避免了因调度器短暂不可用导致 crontab.list 与数据库脱节（订阅更新误判任务已存在）。
+      await this.setCrontab(tabs);
+      try {
+        await cronClient.addCron(regularCrons, requireScheduler);
+      } catch (error: any) {
+        this.logger.warn(
+          '[crontab] Failed to register cron job in scheduler: %s',
+          error?.message || error,
+        );
+        if (requireScheduler) throw error;
+      }
+    });
  }

  public async bootTask() {
@@ -770,13 +1147,7 @@ export default class CronService {
      (x) => !x.isDisabled && this.isBootSchedule(x.schedule),
    );
    if (bootTasks.length > 0) {
-      await CrontabModel.update(
-        { status: CrontabStatus.queued },
-        { where: { id: bootTasks.map((t) => t.id!) } },
-      );
-      for (const task of bootTasks) {
-        await this.runSingle(task.id!);
-      }
+      await this.run(bootTasks.map((task) => task.id!));
    }
  }
 }
@@ -16,22 +16,27 @@ import {
  getPid,
  killTask,
  promiseExecSuccess,
+  concurrentRun,
  getInstallCommand,
  getUninstallCommand,
  getGetCommand,
 } from '../config/util';
 import dayjs from 'dayjs';
 import taskLimit from '../shared/pLimit';
+import { detectOS } from '../config/util';
+import { LINUX_DEPENDENCE_COMMAND } from '../config/const';
+import { t, tf } from '../shared/i18n';

@Service()
 export default class DependenceService {
  constructor(
    @Inject('logger') private logger: winston.Logger,
    private sockService: SockService,
-  ) { }
+  ) {}

  public async create(payloads: Dependence[]): Promise<Dependence[]> {
    const tabs = payloads.map((x) => {
+      getInstallCommand(x.type, x.name);
      const tab = new Dependence({ ...x, status: DependenceStatus.queued });
      return tab;
    });
@@ -55,6 +60,7 @@ export default class DependenceService {
      ...other,
      status: DependenceStatus.queued,
    });
+    getInstallCommand(tab.type, tab.name);
    const newDoc = await this.updateDb(tab);
    this.installDependenceOneByOne([newDoc]);
    return newDoc;
@@ -107,6 +113,11 @@ export default class DependenceService {
    query: any = {},
  ): Promise<Dependence[]> {
    let condition = query;
+    const dependenceType =
+      type && DependenceTypes[type] !== undefined
+        ? DependenceTypes[type]
+        : undefined;
+    await this.refreshInstalledStatuses(dependenceType);
    if (type && DependenceTypes[type] !== undefined) {
      condition.type = DependenceTypes[type];
    }
@@ -129,6 +140,69 @@ export default class DependenceService {
    }
  }

+  private async refreshInstalledStatuses(type?: DependenceTypes) {
+    const cacheDependenceTypes = [
+      DependenceTypes.nodejs,
+      DependenceTypes.python3,
+    ];
+    if (type !== undefined && !cacheDependenceTypes.includes(type)) {
+      return;
+    }
+
+    const docs = await DependenceModel.findAll({
+      where: {
+        status: DependenceStatus.installed,
+        type: type === undefined ? { [Op.in]: cacheDependenceTypes } : type,
+      },
+    });
+    const checks = await concurrentRun(
+      docs.map((doc) => async () => {
+        return (await this.isDependenceInstalled(doc)) ? undefined : doc.id;
+      }),
+      5,
+    );
+    const missingIds = (checks || []).filter(
+      (id): id is number => id !== undefined,
+    );
+
+    if (missingIds.length) {
+      await DependenceModel.update(
+        { status: DependenceStatus.installFailed },
+        { where: { id: missingIds } },
+      );
+    }
+  }
+
+  private async isDependenceInstalled(dependency: Dependence) {
+    let depName = dependency.name.trim();
+    const depVersionStr = versionDependenceCommandTypes[dependency.type];
+    let depVersion = '';
+    if (depName.includes(depVersionStr)) {
+      const symbolRegx = new RegExp(
+        `(.*)${depVersionStr}([0-9\\.\\-\\+a-zA-Z]*)`,
+      );
+      const [, parsedName, parsedVersion] = depName.match(symbolRegx) || [];
+      if (parsedVersion && parsedName) {
+        depName = parsedName;
+        depVersion = parsedVersion;
+      }
+    }
+
+    const depInfo = (
+      await promiseExecSuccess(getGetCommand(dependency.type, depName))
+    )
+      .replace(/\s{2,}/, ' ')
+      .replace(/\s+$/, '');
+    const nameMatches =
+      (dependency.type === DependenceTypes.nodejs &&
+        depInfo.split(' ')?.[0] === depName) ||
+      dependency.type === DependenceTypes.python3;
+
+    return Boolean(
+      depInfo && nameMatches && (!depVersion || depInfo.includes(depVersion)),
+    );
+  }
+
  public installDependenceOneByOne(
    docs: Dependence[],
    isInstall: boolean = true,
@@ -159,8 +233,19 @@ export default class DependenceService {
    const docs = await DependenceModel.findAll({ where: { id: ids } });
    for (const doc of docs) {
      taskLimit.removeQueuedDependency(doc);
-      const depInstallCommand = getInstallCommand(doc.type, doc.name);
-      const depUnInstallCommand = getUninstallCommand(doc.type, doc.name);
+      let depInstallCommand = getInstallCommand(doc.type, doc.name);
+      let depUnInstallCommand = getUninstallCommand(doc.type, doc.name);
+      const isLinuxDependence = doc.type === DependenceTypes.linux;
+
+      if (isLinuxDependence) {
+        const osType = await detectOS();
+        if (!osType) {
+          continue;
+        }
+        const linuxCommand = LINUX_DEPENDENCE_COMMAND[osType];
+        depInstallCommand = `${linuxCommand.install} ${doc.name.trim()}`;
+        depUnInstallCommand = `${linuxCommand.uninstall} ${doc.name.trim()}`;
+      }
      const pids = await Promise.all([
        getPid(depInstallCommand),
        getPid(depUnInstallCommand),
@@ -213,43 +298,102 @@ export default class DependenceService {
    force: boolean = false,
  ) {
    return taskLimit.runDependeny(dependency, () => {
+      let validatedCommand: string;
+      try {
+        validatedCommand = isInstall
+          ? getInstallCommand(dependency.type, dependency.name.trim())
+          : getUninstallCommand(dependency.type, dependency.name.trim());
+      } catch (error) {
+        taskLimit.removeQueuedDependency(dependency);
+        return DependenceModel.update(
+          {
+            status: isInstall
+              ? DependenceStatus.installFailed
+              : DependenceStatus.removeFailed,
+            log: [String(error)],
+          },
+          { where: { id: dependency.id } },
+        ).then(() => null);
+      }
      return new Promise(async (resolve) => {
        if (taskLimit.firstDependencyId !== dependency.id) {
          return resolve(null);
        }
-
-        taskLimit.removeQueuedDependency(dependency);
-
        const depIds = [dependency.id!];
+        let depName = dependency.name.trim();
+        const actionText = isInstall ? t('安装') : t('删除');
+        const socketMessageType = isInstall
+          ? 'installDependence'
+          : 'uninstallDependence';
+        const isNodeDependence = dependency.type === DependenceTypes.nodejs;
+        const isLinuxDependence = dependency.type === DependenceTypes.linux;
+        const isPythonDependence = dependency.type === DependenceTypes.python3;
+        const osType = await detectOS();
+        let linuxCommand = {} as typeof LINUX_DEPENDENCE_COMMAND.Alpine;
+        taskLimit.removeQueuedDependency(dependency);
+        if (isLinuxDependence) {
+          if (!osType) {
+            await DependenceModel.update(
+              { status: DependenceStatus.installFailed },
+              { where: { id: depIds } },
+            );
+            const startTime = dayjs();
+            const message = tf(
+              '开始%s依赖 %s，开始时间 %s\n\n当前系统不支持\n\n依赖%s失败，结束时间 %s，耗时 %s 秒',
+              actionText,
+              depName,
+              startTime.format('YYYY-MM-DD HH:mm:ss'),
+              actionText,
+              startTime.format('YYYY-MM-DD HH:mm:ss'),
+              String(startTime.diff(startTime, 'second')),
+            );
+            this.sockService.sendMessage({
+              type: socketMessageType,
+              message,
+              references: depIds,
+              status: DependenceStatus.installFailed,
+            });
+            this.updateLog(depIds, message);
+            return resolve(null);
+          }
+          linuxCommand = LINUX_DEPENDENCE_COMMAND[osType];
+        }
+
        const status = isInstall
          ? DependenceStatus.installing
          : DependenceStatus.removing;
        await DependenceModel.update({ status }, { where: { id: depIds } });

-        const socketMessageType = isInstall
-          ? 'installDependence'
-          : 'uninstallDependence';
-        let depName = dependency.name.trim();
-        const command = isInstall
-          ? getInstallCommand(dependency.type, depName)
-          : getUninstallCommand(dependency.type, depName);
-        const actionText = isInstall ? '安装' : '删除';
+        let command = validatedCommand;
+        if (isLinuxDependence) {
+          command = isInstall
+            ? `${linuxCommand.install} ${depName.trim()}`
+            : `${linuxCommand.uninstall} ${depName.trim()}`;
+        }
        const startTime = dayjs();

-        const message = `开始${actionText}依赖 ${depName}，开始时间 ${startTime.format(
-          'YYYY-MM-DD HH:mm:ss',
-        )}\n\n`;
+        const message = tf(
+          '开始%s依赖 %s，开始时间 %s\n\n',
+          actionText,
+          depName,
+          startTime.format('YYYY-MM-DD HH:mm:ss'),
+        );
        this.sockService.sendMessage({
          type: socketMessageType,
          message,
          references: depIds,
+          status,
        });
        this.updateLog(depIds, message);

        // 判断是否已经安装过依赖
        if (isInstall && !force) {
-          const getCommand = getGetCommand(dependency.type, depName);
+          let getCommand = getGetCommand(dependency.type, depName);
          const depVersionStr = versionDependenceCommandTypes[dependency.type];
+          if (isLinuxDependence) {
+            getCommand = `${linuxCommand.info} ${depName}`;
+          }
+
          let depVersion = '';
          if (depName.includes(depVersionStr)) {
            const symbolRegx = new RegExp(
@@ -261,10 +405,6 @@ export default class DependenceService {
              depVersion = _depVersion;
            }
          }
-          const isNodeDependence = dependency.type === DependenceTypes.nodejs;
-          const isLinuxDependence = dependency.type === DependenceTypes.linux;
-          const isPythonDependence =
-            dependency.type === DependenceTypes.python3;
          const depInfo = (await promiseExecSuccess(getCommand))
            .replace(/\s{2,}/, ' ')
            .replace(/\s+$/, '');
@@ -273,18 +413,24 @@ export default class DependenceService {
            depInfo &&
            ((isNodeDependence && depInfo.split(' ')?.[0] === depName) ||
              (isLinuxDependence &&
-                depInfo.toLocaleLowerCase().includes('installed')) ||
+                linuxCommand.check(depInfo.toLocaleLowerCase())) ||
              isPythonDependence) &&
            (!depVersion || depInfo.includes(depVersion))
          ) {
            const endTime = dayjs();
-            const _message = `检测到已经安装 ${depName}\n\n${depInfo}\n\n跳过安装\n\n依赖${actionText}成功，结束时间 ${endTime.format(
-              'YYYY-MM-DD HH:mm:ss',
-            )}，耗时 ${endTime.diff(startTime, 'second')} 秒`;
+            const _message = tf(
+              '检测到已经安装 %s\n\n%s\n\n跳过安装\n\n依赖%s成功，结束时间 %s，耗时 %s 秒',
+              depName,
+              depInfo,
+              actionText,
+              endTime.format('YYYY-MM-DD HH:mm:ss'),
+              String(endTime.diff(startTime, 'second')),
+            );
            this.sockService.sendMessage({
              type: socketMessageType,
              message: _message,
              references: depIds,
+              status: DependenceStatus.installed,
            });
            this.updateLog(depIds, _message);
            await DependenceModel.update(
@@ -309,6 +455,7 @@ export default class DependenceService {
            type: socketMessageType,
            message: data.toString(),
            references: depIds,
+            status,
          });
          this.updateLog(depIds, data.toString());
        });
@@ -318,6 +465,7 @@ export default class DependenceService {
            type: socketMessageType,
            message: data.toString(),
            references: depIds,
+            status,
          });
          this.updateLog(depIds, data.toString());
        });
@@ -327,6 +475,7 @@ export default class DependenceService {
            type: socketMessageType,
            message: JSON.stringify(err),
            references: depIds,
+            status,
          });
          this.updateLog(depIds, JSON.stringify(err));
        });
@@ -334,28 +483,32 @@ export default class DependenceService {
        cp.on('exit', async (code) => {
          const endTime = dayjs();
          const isSucceed = code === 0;
-          const resultText = isSucceed ? '成功' : '失败';
+          const resultText = isSucceed ? t('成功') : t('失败');

-          const message = `\n依赖${actionText}${resultText}，结束时间 ${endTime.format(
-            'YYYY-MM-DD HH:mm:ss',
-          )}，耗时 ${endTime.diff(startTime, 'second')} 秒`;
+          const message =
+            '\n' +
+            tf(
+              '依赖%s%s，结束时间 %s，耗时 %s 秒',
+              actionText,
+              resultText,
+              endTime.format('YYYY-MM-DD HH:mm:ss'),
+              String(endTime.diff(startTime, 'second')),
+            );
+          const exitStatus = isSucceed
+            ? isInstall
+              ? DependenceStatus.installed
+              : DependenceStatus.removed
+            : isInstall
+            ? DependenceStatus.installFailed
+            : DependenceStatus.removeFailed;
          this.sockService.sendMessage({
            type: socketMessageType,
            message,
            references: depIds,
+            status: exitStatus,
          });
          this.updateLog(depIds, message);

-          let status: number;
-          if (isSucceed) {
-            status = isInstall
-              ? DependenceStatus.installed
-              : DependenceStatus.removed;
-          } else {
-            status = isInstall
-              ? DependenceStatus.installFailed
-              : DependenceStatus.removeFailed;
-          }
          const docs = await DependenceModel.findAll({ where: { id: depIds } });
          const _docIds = docs
            .filter((x) => x.status !== DependenceStatus.cancelled)
@@ -363,7 +516,7 @@ export default class DependenceService {

          if (_docIds.length > 0) {
            await DependenceModel.update(
-              { status },
+              { status: exitStatus },
              { where: { id: _docIds } },
            );
          }
@@ -27,7 +27,7 @@ export default class EnvService {
      envs.length > 0 &&
      typeof envs[envs.length - 1].position === 'number'
    ) {
-      position = envs[envs.length - 1].position!;
+      position = this.getPrecisionPosition(envs[envs.length - 1].position!);
    }
    const tabs = payloads.map((x) => {
      position = position - stepPosition;
@@ -100,7 +100,7 @@ export default class EnvService {
  }

  private async checkPosition(position: number, edge: number = 0) {
-    const precisionPosition = parseFloat(position.toPrecision(16));
+    const precisionPosition = this.getPrecisionPosition(position);
    if (
      precisionPosition < minPosition ||
      precisionPosition > maxPosition ||
@@ -116,7 +116,7 @@ export default class EnvService {
  }

  private getPrecisionPosition(position: number): number {
-    return parseFloat(position.toPrecision(16));
+    return Math.trunc(parseFloat(position.toPrecision(16)));
  }

  public async envs(searchText: string = '', query: any = {}): Promise<Env[]> {
@@ -199,6 +199,44 @@ export default class EnvService {
    await EnvModel.update({ isPinned: 0 }, { where: { id: ids } });
  }

+  public async addLabels(ids: number[], labels: string[]) {
+    await sequelize.transaction(async (transaction) => {
+      const docs = await EnvModel.findAll({
+        where: { id: ids },
+        transaction,
+      });
+      for (const doc of docs) {
+        const env = doc.get({ plain: true });
+        await EnvModel.update(
+          { labels: Array.from(new Set([...(env.labels || []), ...labels])) },
+          { where: { id: env.id }, transaction },
+        );
+      }
+    });
+    return await this.find({ id: ids });
+  }
+
+  public async removeLabels(ids: number[], labels: string[]) {
+    await sequelize.transaction(async (transaction) => {
+      const docs = await EnvModel.findAll({
+        where: { id: ids },
+        transaction,
+      });
+      for (const doc of docs) {
+        const env = doc.get({ plain: true });
+        await EnvModel.update(
+          {
+            labels: (env.labels || []).filter(
+              (label: string) => !labels.includes(label),
+            ),
+          },
+          { where: { id: env.id }, transaction },
+        );
+      }
+    });
+    return await this.find({ id: ids });
+  }
+
  public async set_envs() {
    const envs = await this.envs('', {
      name: { [Op.not]: null },
@@ -224,10 +262,9 @@ export default class EnvService {
            .map((x) => x.value)
            .join('&')
            .replace(/\\/g, '\\\\')}`;
-          js_env_string += `process.env.${key}=\`${_env_value.replace(
-            /\`/g,
-            '\\`',
-          )}\`;\n`;
+          js_env_string += `process.env[${JSON.stringify(key)}]=${JSON.stringify(
+            group.map((x) => x.value).join('&'),
+          )};\n`;
          py_env_string += `os.environ['${key}']='''${_env_value.replace(
            /\'/g,
            "\\'",
@@ -4,6 +4,7 @@ import { HealthService } from '../protos/health';
 import { ApiService } from '../protos/api';
 import { addCron } from '../schedule/addCron';
 import { delCron } from '../schedule/delCron';
+import { setConcurrency } from '../schedule/setConcurrency';
 import { check } from '../schedule/health';
 import * as Api from '../schedule/api';
 import Logger from '../loaders/logger';
@@ -11,30 +12,59 @@ import { promisify } from 'util';
 import config from '../config';
 import { metricsService } from './metrics';
 import { Service } from 'typedi';
+import { initGrpcCerts } from '../config/grpcCerts';

@Service()
 export class GrpcServerService {
  private server: Server = new Server({ 'grpc.enable_http_proxy': 0 });

+  private formatGrpcAddress(host: string, port: number): string {
+    if (host === '::') {
+      return `[::]:${port}`;
+    }
+    return `${host}:${port}`;
+  }
+
  async initialize() {
    try {
      this.server.addService(HealthService, { check });
-      this.server.addService(CronService, { addCron, delCron });
+      this.server.addService(CronService, { addCron, delCron, setConcurrency });
      this.server.addService(ApiService, Api);

-      const grpcPort = config.grpcPort;
-      const bindAsync = promisify(this.server.bindAsync).bind(this.server);
-      await bindAsync(
-        `0.0.0.0:${grpcPort}`,
-        ServerCredentials.createInsecure(),
+      const tlsConfig = await initGrpcCerts();
+      const credentials = ServerCredentials.createSsl(
+        Buffer.from(tlsConfig.caCert),
+        [{ cert_chain: Buffer.from(tlsConfig.serverCert), private_key: Buffer.from(tlsConfig.serverKey) }],
+        true,
      );
-      Logger.debug(`✌️ gRPC service started successfully`);

-      metricsService.record('grpc_service_start', 1, {
-        port: grpcPort.toString(),
-      });
+      const grpcPort = config.grpcPort;
+      const hostsToTry = [
+        config.bindHostGrpc,
+        ...(config.bindHostGrpc === '::' ? ['0.0.0.0'] : [])
+      ];
+      const bindAsync = promisify(this.server.bindAsync).bind(this.server);

-      return grpcPort;
+      let lastError: Error | null = null;
+
+      for (const host of hostsToTry) {
+        try {
+          const address = this.formatGrpcAddress(host, grpcPort);
+          await bindAsync(address, credentials);
+          Logger.debug(`[boot] gRPC service started successfully on ${address}`);
+          metricsService.record('grpc_service_start', 1, {
+            port: grpcPort.toString(),
+            host
+          });
+          return grpcPort;
+        } catch (err) {
+          lastError = err as Error;
+          Logger.warn(`Failed to bind gRPC on ${host}:${grpcPort}, trying next...`, err);
+        }
+      }
+
+      Logger.error('Failed to start gRPC service on all hosts');
+      throw lastError || new Error('Failed to start gRPC service');
    } catch (err) {
      Logger.error('Failed to start gRPC service:', err);
      throw err;
@@ -1,6 +1,6 @@
 import { Service } from 'typedi';
 import Logger from '../loaders/logger';
-import { GrpcServerService } from './grpc';
+import cronClient from '../schedule/client';
 import { HttpServerService } from './http';

 interface HealthStatus {
@@ -23,7 +23,6 @@ export class HealthService {
  private startTime = Date.now();

  constructor(
-    private grpcServerService: GrpcServerService,
    private httpServerService: HttpServerService,
  ) {}

@@ -56,8 +55,7 @@ export class HealthService {
    }

    try {
-      const grpcServer = this.grpcServerService.getServer();
-      if (!grpcServer) {
+      if (!(await cronClient.readiness.check())) {
        status.services.grpc = false;
        status.status = 'error';
      }
@@ -3,31 +3,57 @@ import Logger from '../loaders/logger';
 import { metricsService } from './metrics';
 import { Service } from 'typedi';
 import { Server } from 'http';
+import config from '../config';

@Service()
 export class HttpServerService {
  private server?: Server = undefined;

  async initialize(expressApp: express.Application, port: number) {
-    try {
-      return new Promise((resolve, reject) => {
-        this.server = expressApp.listen(port, '0.0.0.0', () => {
-          Logger.debug(`✌️ HTTP service started successfully`);
-          metricsService.record('http_service_start', 1, {
-            port: port.toString(),
-          });
-          resolve(this.server);
-        });
+    const hostsToTry = [
+      config.bindHost,
+      ...(config.bindHost === '::' ? ['0.0.0.0'] : [])
+    ];

-        this.server?.on('error', (err: Error) => {
-          Logger.error('Failed to start HTTP service:', err);
-          reject(err);
+    let lastError: Error | null = null;
+
+    for (const host of hostsToTry) {
+      try {
+        const server = await this.tryListen(expressApp, port, host);
+        Logger.debug(`[boot] HTTP service started successfully on ${host}:${port}`);
+        metricsService.record('http_service_start', 1, {
+          port: port.toString(),
+          host
        });
-      });
-    } catch (err) {
-      Logger.error('Failed to start HTTP service:', err);
-      throw err;
+        this.server = server;
+        return server;
+      } catch (err) {
+        lastError = err as Error;
+        Logger.warn(`Failed to bind HTTP on ${host}:${port}, trying next...`, err);
+      }
    }
+
+    Logger.error('Failed to start HTTP service on all hosts');
+    throw lastError || new Error('Failed to start HTTP service');
+  }
+
+  private async tryListen(expressApp: express.Application, port: number, host: string): Promise<Server> {
+    return new Promise((resolve, reject) => {
+      // There is one HTTP worker; accepting here avoids primary IPC handoff
+      // for every connection. Restore shared listening for custom clusters.
+      const server = expressApp.listen({
+        port,
+        host,
+        exclusive: process.env.QL_HTTP_SHARED_LISTEN !== 'true',
+      }, () => {
+        resolve(server);
+      });
+
+      server.on('error', (err: Error) => {
+        server.close();
+        reject(err);
+      });
+    });
  }

  async shutdown() {
@@ -1,4 +1,4 @@
-import path from 'path';
+import { resolveFileAccess } from '../shared/fileAccess';
 import { Inject, Service } from 'typedi';
 import winston from 'winston';
 import config from '../config';
@@ -8,7 +8,6 @@ export default class LogService {
  constructor(@Inject('logger') private logger: winston.Logger) {}

  public checkFilePath(filePath: string, fileName: string) {
-    const finalPath = path.resolve(config.logPath, filePath, fileName);
-    return finalPath.startsWith(config.logPath) ? finalPath : '';
+    return resolveFileAccess(config.logPath, [filePath || '', fileName]);
  }
 }
@@ -8,16 +8,19 @@ interface Metric {
  tags?: Record<string, string>;
 }

-class MetricsService {
-  private metrics: Metric[] = [];
+export const MAX_METRIC_SAMPLES = 1000;
+const METRIC_RETENTION_MS = 60 * 60 * 1000;
+
+export class MetricsService {
+  private metrics: Array<Metric | undefined> = new Array(MAX_METRIC_SAMPLES);
+  private metricCount = 0;
+  private nextMetricIndex = 0;
  private static instance: MetricsService;

  private constructor() {
    // 定期清理旧数据
-    setInterval(() => {
-      const oneHourAgo = Date.now() - 3600000;
-      this.metrics = this.metrics.filter(m => m.timestamp > oneHourAgo);
-    }, 60000);
+    const cleanupTimer = setInterval(() => this.removeExpiredMetrics(), 60000);
+    cleanupTimer.unref();
  }

  static getInstance(): MetricsService {
@@ -28,12 +31,14 @@ class MetricsService {
  }

  record(name: string, value: number, tags?: Record<string, string>) {
-    this.metrics.push({
+    this.metrics[this.nextMetricIndex] = {
      name,
      value,
      timestamp: Date.now(),
      tags,
-    });
+    };
+    this.nextMetricIndex = (this.nextMetricIndex + 1) % MAX_METRIC_SAMPLES;
+    this.metricCount = Math.min(this.metricCount + 1, MAX_METRIC_SAMPLES);
  }

  measure(name: string, fn: () => void, tags?: Record<string, string>) {
@@ -46,7 +51,11 @@ class MetricsService {
    }
  }

-  async measureAsync(name: string, fn: () => Promise<void>, tags?: Record<string, string>) {
+  async measureAsync(
+    name: string,
+    fn: () => Promise<void>,
+    tags?: Record<string, string>,
+  ) {
    const start = performance.now();
    try {
      await fn();
@@ -57,28 +66,61 @@ class MetricsService {
  }

  getMetrics(name?: string, tags?: Record<string, string>) {
-    let filtered = this.metrics;
-    
+    this.removeExpiredMetrics();
+    let filtered = this.getMetricSnapshot();
+
    if (name) {
-      filtered = filtered.filter(m => m.name === name);
+      filtered = filtered.filter((m) => m.name === name);
    }
-    
+
    if (tags) {
-      filtered = filtered.filter(m => {
+      filtered = filtered.filter((m) => {
        if (!m.tags) return false;
-        return Object.entries(tags).every(([key, value]) => m.tags![key] === value);
+        return Object.entries(tags).every(
+          ([key, value]) => m.tags![key] === value,
+        );
      });
    }

+    const values = filtered.map((metric) => metric.value);
    return {
      count: filtered.length,
-      average: filtered.reduce((acc, curr) => acc + curr.value, 0) / filtered.length,
-      min: Math.min(...filtered.map(m => m.value)),
-      max: Math.max(...filtered.map(m => m.value)),
+      average: values.length
+        ? values.reduce((acc, value) => acc + value, 0) / values.length
+        : 0,
+      min: values.length ? Math.min(...values) : 0,
+      max: values.length ? Math.max(...values) : 0,
      metrics: filtered,
    };
  }

+  private getMetricSnapshot(): Metric[] {
+    if (this.metricCount < MAX_METRIC_SAMPLES) {
+      return this.metrics.slice(0, this.metricCount) as Metric[];
+    }
+    return [
+      ...this.metrics.slice(this.nextMetricIndex),
+      ...this.metrics.slice(0, this.nextMetricIndex),
+    ] as Metric[];
+  }
+
+  private removeExpiredMetrics() {
+    const oldestTimestamp = Date.now() - METRIC_RETENTION_MS;
+    const retained = this.getMetricSnapshot().filter(
+      (metric) => metric.timestamp > oldestTimestamp,
+    );
+    if (retained.length === this.metricCount) return;
+
+    this.metrics = new Array(MAX_METRIC_SAMPLES);
+    this.metricCount = 0;
+    this.nextMetricIndex = 0;
+    for (const metric of retained) {
+      this.metrics[this.nextMetricIndex] = metric;
+      this.nextMetricIndex = (this.nextMetricIndex + 1) % MAX_METRIC_SAMPLES;
+      this.metricCount++;
+    }
+  }
+
  report() {
    const report = {
      timestamp: Date.now(),
@@ -89,4 +131,4 @@ class MetricsService {
  }
 }

-export const metricsService = MetricsService.getInstance(); 
+export const metricsService = MetricsService.getInstance();
@@ -3,6 +3,7 @@ import nodemailer from 'nodemailer';
 import { Inject, Service } from 'typedi';
 import { parseBody, parseHeaders } from '../config/util';
 import { NotificationInfo } from '../data/notify';
+import { t } from '../shared/i18n';
 import UserService from './user';
 import { httpClient } from '../config/http';
 import { ProxyAgent } from 'undici';
@@ -34,6 +35,9 @@ export default class NotificationService {
    ['chronocat', this.chronocat],
    ['ntfy', this.ntfy],
    ['wxPusherBot', this.wxPusherBot],
+    ['wxPusherSpt', this.wxPusherSpt],
+    ['openiLink', this.openiLink],
+    ['wpush', this.wpush],
  ]);

  private title = '';
@@ -90,6 +94,14 @@ export default class NotificationService {
    return true;
  }

+  private parseMailRecipients(value?: string) {
+    const recipients = (value || '')
+      .split(/[;；]/)
+      .map((item) => item.trim())
+      .filter(Boolean);
+    return recipients.length > 0 ? recipients : undefined;
+  }
+
  private async gotify() {
    const { gotifyUrl, gotifyToken, gotifyPriority = 1 } = this.params;
    try {
@@ -355,7 +367,7 @@ export default class NotificationService {
          {
            title: `${this.title}`,
            thumb_media_id,
-            author: `智能助手`,
+            author: t('智能助手'),
            content_source_url: ``,
            content: `${this.content.replace(/\n/g, '<br/>')}`,
            digest: `${this.content}`,
@@ -371,7 +383,7 @@ export default class NotificationService {
            title: `${this.title}`,
            description: `${this.content}`,
            url: 'https://github.com/whyour/qinglong',
-            btntxt: '更多',
+            btntxt: t('更多'),
          },
        };
        break;
@@ -422,7 +434,7 @@ export default class NotificationService {
          roomName: `${aibotkName}`,
          message: {
            type: 1,
-            content: `【青龙快讯】\n\n${this.title}\n${this.content}`,
+            content: `【${t('青龙快讯')}】\n\n${this.title}\n${this.content}`,
          },
        };
        break;
@@ -433,7 +445,7 @@ export default class NotificationService {
          name: `${aibotkName}`,
          message: {
            type: 1,
-            content: `【青龙快讯】\n\n${this.title}\n${this.content}`,
+            content: `【${t('青龙快讯')}】\n\n${this.title}\n${this.content}`,
          },
        };
        break;
@@ -591,6 +603,7 @@ export default class NotificationService {

  private async email() {
    const { emailPass, emailService, emailUser, emailTo } = this.params;
+    const recipients = this.parseMailRecipients(emailTo) || emailUser;

    try {
      const transporter = nodemailer.createTransport({
@@ -602,8 +615,8 @@ export default class NotificationService {
      });

      const info = await transporter.sendMail({
-        from: `"青龙快讯" <${emailUser}>`,
-        to: emailTo ? emailTo.split(';') : emailUser,
+        from: `"${t('青龙快讯')}" <${emailUser}>`,
+        to: recipients,
        subject: `${this.title}`,
        html: `${this.content.replace(/\n/g, '<br/>')}`,
      });
@@ -717,7 +730,7 @@ export default class NotificationService {

    // topic_ids 和 uids 至少要有一个
    if (!topicIds.length && !uids.length) {
-      throw new Error('wxPusher 服务的 TopicIds 和 Uids 至少配置一个才行');
+      throw new Error(t('wxPusher 服务的 TopicIds 和 Uids 至少配置一个才行'));
    }

    const url = `https://wxpusher.zjiecode.com/api/send/message`;
@@ -745,6 +758,52 @@ export default class NotificationService {
    }
  }

+  private async wxPusherSpt() {
+    const { wxPusherSptList } = this.params;
+    // 处理 SPT，将逗号分隔的字符串转为数组
+    const spts = wxPusherSptList
+      ? wxPusherSptList
+          .split(',')
+          .map((spt) => spt.trim())
+          .filter((spt) => spt)
+      : [];
+
+    if (!spts.length) {
+      throw new Error(t('wxPusher SPT 不能为空'));
+    }
+    if (spts.length > 10) {
+      throw new Error(t('wxPusher SPT 最多支持 10 个'));
+    }
+
+    const url = `https://wxpusher.zjiecode.com/api/send/message/simple-push`;
+    const json: any = {
+      content: `<h1>${this.title}</h1><br/><div style='white-space: pre-wrap;'>${this.content}</div>`,
+      summary: this.title,
+      contentType: 2,
+    };
+    // 单个 SPT 用 spt，多个用 sptList
+    if (spts.length === 1) {
+      json.spt = spts[0];
+    } else {
+      json.sptList = spts;
+    }
+
+    try {
+      const res = await httpClient.post(url, {
+        ...this.gotOption,
+        json,
+      });
+
+      if (res.code === 1000) {
+        return true;
+      } else {
+        throw new Error(JSON.stringify(res));
+      }
+    } catch (error: any) {
+      throw new Error(error.response ? error.response.body : error);
+    }
+  }
+
  private async chronocat() {
    const { chronocatURL, chronocatQQ, chronocatToken } = this.params;
    try {
@@ -813,7 +872,7 @@ export default class NotificationService {
    } = this.params;

    if (!webhookUrl?.includes('$title') && !webhookBody?.includes('$title')) {
-      throw new Error('Url 或者 Body 中必须包含 $title');
+      throw new Error(t('Url 或者 Body 中必须包含 $title'));
    }

    const headers = parseHeaders(webhookHeaders);
@@ -858,4 +917,62 @@ export default class NotificationService {
    }
    return {};
  }
+
+  private async openiLink() {
+    const { openiLinkAppToken, openiLinkHubUrl, openiLinkContextToken } =
+      this.params;
+    const baseUrl = openiLinkHubUrl?.replace(/\/$/, '') || 'https://hub.openilink.com';
+    const url = `${baseUrl}/bot/v1/message/send`;
+    const body: Record<string, string> = {
+      type: 'text',
+      content: `${this.title}\n\n${this.content}`,
+    };
+    if (openiLinkContextToken) {
+      body.context_token = openiLinkContextToken;
+    }
+    try {
+      const res = await httpClient.post(url, {
+        ...this.gotOption,
+        json: body,
+        headers: {
+          'Content-Type': 'application/json',
+          Authorization: `Bearer ${openiLinkAppToken}`,
+        },
+      });
+      if (res.ok) {
+        return true;
+      } else {
+        throw new Error(JSON.stringify(res));
+      }
+    } catch (error: any) {
+      throw new Error(error.response ? error.response.body : error);
+    }
+  }
+  private async wpush() {
+    const { wpushApiKey, wpushChannel, wpushTopicCode } = this.params;
+    const url = 'https://api.wpush.cn/api/v1/send';
+    const json: Record<string, string> = {
+      apikey: `${wpushApiKey}`,
+      title: `${this.title}`,
+      content: `${this.content}`,
+      channel: `${wpushChannel || 'wechat'}`,
+    };
+    if (wpushTopicCode) {
+      json.topic_code = `${wpushTopicCode}`;
+    }
+    try {
+      const res = await httpClient.post(url, {
+        ...this.gotOption,
+        json,
+      });
+      if (res.code === 0) {
+        return true;
+      } else {
+        throw new Error(JSON.stringify(res));
+      }
+    } catch (error: any) {
+      throw new Error(error.response ? error.response.body : error);
+    }
+  }
+
 }
@@ -5,27 +5,78 @@ import { App, AppModel } from '../data/open';
 import { v4 as uuidV4 } from 'uuid';
 import sequelize, { Op } from 'sequelize';
 import { shareStore } from '../shared/store';
+import { t } from '../shared/i18n';
+import { lock } from 'proper-lockfile';
+import path from 'path';
+import os from 'os';
+import { writeFile, rename, rm, stat, chmod } from 'fs/promises';
+import config from '../config';

@Service()
 export default class OpenService {
  constructor(@Inject('logger') private logger: winston.Logger) {}

+  // Every database mutation and full authentication-cache publication shares
+  // this cross-process lock. Lock a separate marker, not token.json itself.
+  private async withAppsLock<T>(operation: () => Promise<T>): Promise<T> {
+    const target = path.join(config.configPath, '.apps-state');
+    const release = await lock(target, {
+      realpath: false,
+      stale: 10_000,
+      retries: { retries: 10, factor: 2, minTimeout: 100, maxTimeout: 3000 },
+    });
+    try {
+      return await operation();
+    } finally {
+      await release();
+    }
+  }
+
+  public async initializeSystemApp(): Promise<void> {
+    await this.withAppsLock(async () => {
+      const doc = await AppModel.findOne({ where: { name: 'system' } });
+      const app = doc?.get({ plain: true });
+      if (!app) {
+        await this.insert({
+          name: 'system',
+          scopes: ['crons', 'system', 'dashboard'],
+          client_id: createRandomString(12, 12),
+          client_secret: createRandomString(24, 24),
+        });
+      } else if (!app.scopes.includes('dashboard')) {
+        await AppModel.update(
+          { scopes: [...app.scopes, 'dashboard'] },
+          { where: { id: app.id } },
+        );
+      }
+      await shareStore.updateApps(await this.find({}));
+    });
+  }
+
+  public async refreshApps(): Promise<void> {
+    await this.withAppsLock(async () => {
+      await shareStore.updateApps(await this.find({}));
+    });
+  }
+
  public async findApps(): Promise<App[] | null> {
    const docs = await this.find({});
    return docs;
  }

  public async create(payload: App): Promise<App> {
-    const tab = { ...payload };
-    tab.client_id = createRandomString(12, 12);
-    tab.client_secret = createRandomString(24, 24);
-    const doc = await this.insert(tab);
-    const apps = await this.find({});
-    await shareStore.updateApps(apps);
-    return { ...doc, tokens: [] };
+    return this.withAppsLock(async () => {
+      const tab = { ...payload };
+      tab.client_id = createRandomString(12, 12);
+      tab.client_secret = createRandomString(24, 24);
+      const doc = await this.insert(tab);
+      const apps = await this.find({});
+      await shareStore.updateApps(apps);
+      return { ...doc, tokens: [] };
+    });
  }

-  public async insert(payload: App): Promise<App> {
+  private async insert(payload: App): Promise<App> {
    const doc = await AppModel.create(payload, { returning: true });
    return doc.get({ plain: true });
  }
@@ -40,10 +91,12 @@ export default class OpenService {
  }

  private async updateDb(payload: Partial<App>): Promise<App> {
-    await AppModel.update(payload, { where: { id: payload.id } });
-    const apps = await this.find({});
-    await shareStore.updateApps(apps);
-    return apps?.find((x) => x.id === payload.id) as App;
+    return this.withAppsLock(async () => {
+      await AppModel.update(payload, { where: { id: payload.id } });
+      const apps = await this.find({});
+      await shareStore.updateApps(apps);
+      return apps?.find((x) => x.id === payload.id) as App;
+    });
  }

  public async getDb(query: Record<string, any>): Promise<App> {
@@ -55,9 +108,11 @@ export default class OpenService {
  }

  public async remove(ids: number[]) {
-    await AppModel.destroy({ where: { id: ids } });
-    const apps = await this.find({});
-    await shareStore.updateApps(apps);
+    return this.withAppsLock(async () => {
+      await AppModel.destroy({ where: { id: ids } });
+      const apps = await this.find({});
+      await shareStore.updateApps(apps);
+    });
  }

  public async resetSecret(id: number): Promise<App> {
@@ -111,7 +166,15 @@ export default class OpenService {
    return docs.map((x) => x.get({ plain: true }));
  }

-  public async authToken({
+  public async authToken(credentials: {
+    client_id: string;
+    client_secret: string;
+  }): Promise<any> {
+    return this.withAppsLock(() => this.issueToken(credentials));
+  }
+
+  // Caller holds withAppsLock through database update and cache publication.
+  private async issueToken({
    client_id,
    client_secret,
  }: {
@@ -151,31 +214,51 @@ export default class OpenService {
        },
      };
    } else {
-      return { code: 400, message: 'client_id 或 client_seret 有误' };
+      return { code: 400, message: t('client_id 或 client_seret 有误') };
    }
  }

-  public async generateSystemToken(): Promise<{
+  public async generateSystemToken(renew = false): Promise<{
    value: string;
    expiration: number;
  }> {
-    const apps = await shareStore.getApps();
-    const systemApp = apps?.find((x) => x.name === 'system');
-    if (!systemApp) {
-      throw new Error('system app not found');
-    }
-    const now = Math.round(Date.now() / 1000);
-    const currentToken = systemApp.tokens?.find((x) => x.expiration > now);
-    if (currentToken) {
-      return currentToken;
-    }
-    const { data } = await this.authToken({
-      client_id: systemApp.client_id,
-      client_secret: systemApp.client_secret,
+    return this.withAppsLock(async () => {
+      // Read after acquiring the lock; a concurrent issuer may have renewed.
+      const systemApp = await this.getDb({ name: 'system' });
+      const now = Math.round(Date.now() / 1000);
+      let appToken = systemApp.tokens
+        ?.filter((token) => token.expiration > now)
+        .sort((left, right) => right.expiration - left.expiration)[0];
+      if (renew || !appToken) {
+        const { data } = await this.issueToken({
+          client_id: systemApp.client_id,
+          client_secret: systemApp.client_secret,
+        });
+        appToken = { value: data.token, expiration: data.expiration };
+      } else {
+        // Recover a prior database commit whose cache publication failed.
+        await shareStore.updateApps(await this.find({}));
+      }
+      const tokenFile = path.join(config.configPath, 'token.json');
+      const temporaryFile = `${tokenFile}.${process.pid}.tmp`;
+      const mode = await stat(tokenFile).then(
+        (file) => file.mode & 0o777,
+        (error: NodeJS.ErrnoException) => {
+          if (error.code === 'ENOENT') return 0o600;
+          throw error;
+        },
+      );
+      try {
+        await writeFile(temporaryFile, `${JSON.stringify(appToken)}${os.EOL}`, {
+          mode,
+        });
+        await chmod(temporaryFile, mode);
+        // Shell readers see a complete old or new document, never a partial one.
+        await rename(temporaryFile, tokenFile);
+      } finally {
+        await rm(temporaryFile, { force: true });
+      }
+      return appToken;
    });
-    return {
-      ...data,
-      value: data.token,
-    };
  }
 }
@@ -0,0 +1,155 @@
+import fs from 'fs/promises';
+import path from 'path';
+import dayjs from 'dayjs';
+import { Op } from 'sequelize';
+import { Service } from 'typedi';
+import config from '../config';
+import { CrontabStatModel } from '../data/cronStats';
+import { sequelize } from '../data';
+import { InstanceStatus, RunningInstanceModel } from '../data/runningInstance';
+import { AuthDataType, SystemModel } from '../data/system';
+import {
+  DependenceCacheType,
+  getDirectorySize,
+  isDependenceCacheType,
+  normalizeRetentionPolicy,
+  RetentionPolicy,
+} from '../shared/retention';
+
+export interface StorageCleanupRequest extends RetentionPolicy {
+  dependenceCacheTypes?: DependenceCacheType[];
+  compactDatabase?: boolean;
+}
+
+function runningInstanceWhere(days: number) {
+  const cutoff = dayjs().subtract(days, 'day').unix();
+  return {
+    status: { [Op.ne]: InstanceStatus.running },
+    [Op.or]: [
+      { finished_at: { [Op.lt]: cutoff } },
+      {
+        finished_at: { [Op.is]: null },
+        started_at: { [Op.lt]: cutoff },
+      },
+    ],
+  };
+}
+
+function cronStatWhere(days: number) {
+  return {
+    date: { [Op.lt]: dayjs().subtract(days, 'day').format('YYYY-MM-DD') },
+  };
+}
+
+@Service()
+export default class RetentionService {
+  public async updatePolicy(policy: Partial<RetentionPolicy>) {
+    const normalized = normalizeRetentionPolicy(policy);
+    const systemConfig = await SystemModel.findOne({
+      where: { type: AuthDataType.systemConfig },
+    });
+    if (!systemConfig) {
+      throw new Error('System config not found');
+    }
+    await SystemModel.update(
+      {
+        info: {
+          ...systemConfig.info,
+          ...normalized,
+        },
+      },
+      { where: { id: systemConfig.id } },
+    );
+    return normalized;
+  }
+
+  public async preview(request: StorageCleanupRequest) {
+    const policy = normalizeRetentionPolicy(request);
+    const dependenceCacheTypes = (request.dependenceCacheTypes || []).filter(
+      isDependenceCacheType,
+    );
+    const [runningInstances, cronStats, dependenceCaches] = await Promise.all([
+      policy.runningInstanceRetentionDays > 0
+        ? RunningInstanceModel.count({
+            where: runningInstanceWhere(policy.runningInstanceRetentionDays),
+          })
+        : 0,
+      policy.cronStatRetentionDays > 0
+        ? CrontabStatModel.count({
+            where: cronStatWhere(policy.cronStatRetentionDays),
+          })
+        : 0,
+      Promise.all(
+        dependenceCacheTypes.map(async (type) => ({
+          type,
+          bytes: await getDirectorySize(
+            path.join(config.dependenceCachePath, type),
+          ),
+        })),
+      ),
+    ]);
+
+    return {
+      policy,
+      runningInstances,
+      cronStats,
+      dependenceCaches,
+      dependenceCacheBytes: dependenceCaches.reduce(
+        (total, cache) => total + cache.bytes,
+        0,
+      ),
+      compactDatabase: Boolean(request.compactDatabase),
+    };
+  }
+
+  public async cleanup(request: StorageCleanupRequest) {
+    const preview = await this.preview(request);
+    const deleted = await sequelize.transaction(async (transaction) => {
+      const runningInstances =
+        preview.policy.runningInstanceRetentionDays > 0
+          ? await RunningInstanceModel.destroy({
+              where: runningInstanceWhere(
+                preview.policy.runningInstanceRetentionDays,
+              ),
+              transaction,
+            })
+          : 0;
+      const cronStats =
+        preview.policy.cronStatRetentionDays > 0
+          ? await CrontabStatModel.destroy({
+              where: cronStatWhere(preview.policy.cronStatRetentionDays),
+              transaction,
+            })
+          : 0;
+      return { runningInstances, cronStats };
+    });
+
+    const dependenceCaches = [];
+    for (const cache of preview.dependenceCaches) {
+      await fs.rm(path.join(config.dependenceCachePath, cache.type), {
+        recursive: true,
+        force: true,
+      });
+      dependenceCaches.push(cache);
+    }
+
+    if (
+      request.compactDatabase &&
+      (deleted.runningInstances || deleted.cronStats)
+    ) {
+      await sequelize.query('VACUUM');
+    }
+
+    return {
+      deleted: {
+        ...deleted,
+        dependenceCaches,
+        dependenceCacheBytes: preview.dependenceCacheBytes,
+      },
+      compactedDatabase: Boolean(
+        request.compactDatabase &&
+          (deleted.runningInstances || deleted.cronStats),
+      ),
+    };
+  }
+}
@@ -1,6 +1,6 @@
 import { Service, Inject } from 'typedi';
 import winston from 'winston';
-import nodeSchedule from 'node-schedule';
+import { createCronJob, CronJob } from '../shared/cronScheduler';
 import { ChildProcessWithoutNullStreams } from 'child_process';
 import {
  ToadScheduler,
@@ -11,6 +11,11 @@ import {
 import dayjs from 'dayjs';
 import taskLimit from '../shared/pLimit';
 import { spawn } from 'cross-spawn';
+import {
+  observeChildProcess,
+  asError,
+  ProcessResult,
+} from '../shared/childProcess';

 export interface ScheduleTaskType {
  id?: number;
@@ -27,7 +32,7 @@ export interface TaskCallbacks {
    startTime: dayjs.Dayjs,
  ) => Promise<void>;
  onEnd?: (
-    cp: ChildProcessWithoutNullStreams,
+    cp: ChildProcessWithoutNullStreams | undefined,
    endTime: dayjs.Dayjs,
    diff: number,
  ) => Promise<void>;
@@ -37,7 +42,7 @@ export interface TaskCallbacks {

@Service()
 export default class ScheduleService {
-  private scheduleStacks = new Map<string, nodeSchedule.Job>();
+  private scheduleStacks = new Map<string, CronJob>();

  private intervalSchedule = new ToadScheduler();

@@ -63,73 +68,85 @@ export default class ScheduleService {
  ) {
    const { runOrigin, ...others } = params;

-    return taskLimit[this.taskLimitMap[runOrigin]](others, () => {
-      return new Promise(async (resolve, reject) => {
-        this.logger.info(
-          `[panel][开始执行任务] 参数: ${JSON.stringify({
-            ...others,
-            command,
-          })}`,
-        );
-
-        try {
-          const startTime = dayjs();
-          await callbacks.onBefore?.(startTime);
-
-          const cp = spawn(command, { shell: '/bin/bash' });
-
-          callbacks.onStart?.(cp, startTime);
-          completionTime === 'start' && resolve(cp.pid);
-
-          cp.stdout.on('data', async (data) => {
-            await callbacks.onLog?.(data.toString());
-          });
-
-          cp.stderr.on('data', async (data) => {
-            this.logger.info(
-              '[panel][执行任务失败] 命令: %s, 错误信息: %j',
-              command,
-              data.toString(),
-            );
-            await callbacks.onError?.(data.toString());
-          });
-
-          cp.on('error', async (err) => {
-            this.logger.error(
-              '[panel][创建任务失败] 命令: %s, 错误信息: %j',
-              command,
-              err,
-            );
-            await callbacks.onError?.(JSON.stringify(err));
-          });
-
-          cp.on('exit', async (code) => {
-            this.logger.info(
-              '[panel][执行任务结束] 参数: %s, 退出码: %j',
-              JSON.stringify({
-                ...others,
-                command,
-              }),
-              code,
-            );
-            const endTime = dayjs();
-            await callbacks.onEnd?.(
-              cp,
-              endTime,
-              endTime.diff(startTime, 'seconds'),
-            );
-            resolve({ ...others, pid: cp.pid, code });
-          });
-        } catch (error) {
-          this.logger.error(
-            '[panel][执行任务失败] 命令: %s, 错误信息: %j',
-            command,
-            error,
-          );
-          await callbacks.onError?.(JSON.stringify(error));
-        }
-      });
+    let resolveStart!: (pid: number | undefined) => void;
+    let rejectStart!: (error: Error) => void;
+    const startResult = new Promise<number | undefined>((resolve, reject) => {
+      resolveStart = resolve;
+      rejectStart = reject;
    });
+    // Most scheduled callers only observe completion (or intentionally detach).
+    void startResult.catch(() => {});
+    const completion = taskLimit[this.taskLimitMap[runOrigin]](
+      others,
+      async () => {
+        const startTime = dayjs();
+        let cp: ChildProcessWithoutNullStreams | undefined;
+        let result: ProcessResult = { code: null, signal: null };
+        try {
+          this.logger.info('[panel][开始执行任务] 任务ID: %s', others.id);
+          await callbacks.onBefore?.(startTime);
+          cp = spawn(command, { shell: '/bin/bash' });
+          const child = cp;
+          const observed = observeChildProcess(child, {
+            onStart: async () => {
+              await callbacks.onStart?.(child, startTime);
+            },
+            onStdout: callbacks.onLog,
+            onStderr: callbacks.onError,
+          });
+          observed.started.then(resolveStart, rejectStart);
+          result = await observed.completed;
+        } catch (error) {
+          result.error = asError(error);
+          rejectStart(result.error);
+        }
+
+        if (result.error) {
+          this.logger.error(
+            '[panel][执行任务失败] 任务ID: %s, 错误: %s',
+            others.id,
+            result.error.message,
+          );
+          try {
+            await callbacks.onError?.(result.error.message);
+          } catch (error) {
+            this.logger.error(
+              '[panel][任务错误回调失败] %s',
+              asError(error).message,
+            );
+          }
+        }
+        // Cleanup also runs after setup/spawn failure, and only after both pipes drain.
+        const endTime = dayjs();
+        try {
+          await callbacks.onEnd?.(
+            cp,
+            endTime,
+            endTime.diff(startTime, 'seconds'),
+          );
+        } catch (error) {
+          result.error ??= asError(error);
+          this.logger.error(
+            '[panel][任务结束回调失败] %s',
+            asError(error).message,
+          );
+        }
+        this.logger.info(
+          '[panel][执行任务结束] 任务ID: %s, 退出码: %j',
+          others.id,
+          result.code,
+        );
+        return { ...others, pid: cp?.pid, ...result };
+      },
+    ).catch((error) => {
+      // Queue/setup failures must not become unhandled rejections in detached callers.
+      rejectStart(asError(error));
+      this.logger.error('[panel][任务队列失败] %s', asError(error).message);
+      return { ...others, code: null, signal: null, error: asError(error) };
+    });
+
+    // Returning a PID must not release the execution slot while the process runs.
+    return completionTime === 'start' ? startResult : completion;
  }

  async createCronTask(
@@ -146,18 +163,33 @@ export default class ScheduleService {
      command,
    );

-    this.scheduleStacks.set(
-      _id,
-      nodeSchedule.scheduleJob(_id, schedule, async () => {
-        this.runTask(command, callbacks, {
-          name,
+    if (schedule) {
+      try {
+        const job = createCronJob(
          schedule,
-          command,
-          id: _id,
-          runOrigin,
-        });
-      }),
-    );
+          () =>
+            this.runTask(command, callbacks, {
+              name,
+              schedule,
+              command,
+              id: _id,
+              runOrigin,
+            }),
+          { name: `${_id}: ${name || ''}`, logger: this.logger, start: false },
+        );
+        this.scheduleStacks.get(_id)?.cancel();
+        this.scheduleStacks.set(_id, job);
+        job.start();
+      } catch (error) {
+        // A persisted invalid subscription must not become an unhandled startup rejection.
+        this.logger.warn(
+          '[panel][跳过无效定时任务] 任务ID: %s, cron: %s, 错误: %s',
+          _id,
+          schedule,
+          asError(error).message,
+        );
+      }
+    }

    if (runImmediately) {
      this.runTask(command, callbacks, {
@@ -1,3 +1,4 @@
+import { resolveFileAccess } from '../shared/fileAccess';
 import { Service, Inject } from 'typedi';
 import winston from 'winston';
 import path, { join } from 'path';
@@ -57,16 +58,15 @@ export default class ScriptService {
      taskLimit.removeQueuedCron(relativePath.replace(/ /g, '-'));
      pid = (await getPid(`${TASK_COMMAND} ${relativePath} now`)) as number;
    }
-    try {
-      await killTask(pid);
-    } catch (error) {}
+    if (pid) {
+      await killTask(pid, true);
+    }

    return { code: 200 };
  }

  public checkFilePath(filePath: string, fileName: string) {
-    const finalPath = path.resolve(config.scriptPath, filePath, fileName);
-    return finalPath.startsWith(config.scriptPath) ? finalPath : '';
+    return resolveFileAccess(config.scriptPath, [filePath || '', fileName]);
  }

  public async getFile(filePath: string, fileName: string) {
@@ -0,0 +1,12 @@
+import path from 'path';
+import { Service } from 'typedi';
+import config from '../config';
+import { ScriptHistory } from '../shared/scriptHistory';
+
+/** Panel configuration and lifecycle; the storage core stays independently testable. */
+@Service()
+export default class ScriptHistoryService extends ScriptHistory {
+  constructor() {
+    super(config.scriptPath, path.join(config.dataPath, 'script-history'));
+  }
+}
@@ -8,10 +8,22 @@ import { formatUrl } from '../config/subscription';
 import config from '../config';
 import { fileExist, rmPath } from '../config/util';
 import { writeFileWithLock } from '../shared/utils';
+import {
+  getSubscriptionSshAlias,
+  resolveSubscriptionPath,
+} from '../shared/subscriptionPath';

@Service()
 export default class SshKeyService {
-  private homedir = os.homedir();
+  // OpenSSH uses the effective user's passwd entry, not the HOME override.
+  private get homedir(): string {
+    try {
+      return os.userInfo().homedir;
+    } catch {
+      // Custom container UIDs may have no passwd entry. Keep startup working.
+      return os.homedir();
+    }
+  }
  private sshPath = config.sshdPath;
  private sshConfigFilePath = path.resolve(this.homedir, '.ssh', 'config');
  private sshConfigHeader = `Include ${path.join(this.sshPath, '*.config')}`;
@@ -21,6 +33,10 @@ export default class SshKeyService {
  }

  private async initSshConfigFile() {
+    await fs.mkdir(path.dirname(this.sshConfigFilePath), {
+      recursive: true,
+      mode: 0o700,
+    });
    let config = '';
    const _exist = await fileExist(this.sshConfigFilePath);
    if (_exist) {
@@ -42,13 +58,11 @@ export default class SshKeyService {
    key: string,
  ): Promise<void> {
    try {
-      await writeFileWithLock(
-        path.join(this.sshPath, alias),
-        `${key}${os.EOL}`,
-        {
-          mode: '400',
-        },
-      );
+      const filePath = resolveSubscriptionPath(this.sshPath, alias);
+      try {
+        await rmPath(filePath);
+      } catch { }
+      await writeFileWithLock(filePath, `${key}${os.EOL}`, { mode: '400' });
    } catch (error) {
      this.logger.error('生成私钥文件失败', error);
    }
@@ -56,7 +70,7 @@ export default class SshKeyService {

  private async removePrivateKeyFile(alias: string): Promise<void> {
    try {
-      const filePath = path.join(this.sshPath, alias);
+      const filePath = resolveSubscriptionPath(this.sshPath, alias);
      await rmPath(filePath);
    } catch (error) {
      this.logger.error('删除私钥文件失败', error);
@@ -68,18 +82,19 @@ export default class SshKeyService {
    host: string,
    proxy?: string,
  ) {
+    if (!/^[a-zA-Z0-9.-]+$/.test(host) ||
+        (proxy && !/^[a-zA-Z0-9.:[\]-]+$/.test(proxy))) {
+      throw Object.assign(new Error('Invalid SSH host or proxy'), { status: 400 });
+    }
    if (host === 'github.com') {
      host = `ssh.github.com\n    Port 443\n    HostkeyAlgorithms +ssh-rsa`;
    }
    const proxyStr = proxy
      ? `    ProxyCommand nc -v -x ${proxy} %h %p 2>/dev/null\n`
      : '';
-    const config = `Host ${alias}\n    Hostname ${host}\n    IdentityFile ${path.join(
-      this.sshPath,
-      alias,
-    )}\n    StrictHostKeyChecking no\n${proxyStr}`;
+    const config = `Host ${alias}\n    Hostname ${host}\n    IdentityFile ${resolveSubscriptionPath(this.sshPath, alias)}\n    StrictHostKeyChecking no\n${proxyStr}`;
    await writeFileWithLock(
-      `${path.join(this.sshPath, `${alias}.config`)}`,
+      `${resolveSubscriptionPath(this.sshPath, alias, '.config')}`,
      config,
      {
        encoding: 'utf8',
@@ -90,7 +105,7 @@ export default class SshKeyService {

  private async removeSshConfig(alias: string) {
    try {
-      const filePath = path.join(this.sshPath, `${alias}.config`);
+      const filePath = resolveSubscriptionPath(this.sshPath, alias, '.config');
      await rmPath(filePath);
    } catch (error) {
      this.logger.error(`删除ssh配置文件${alias}失败`, error);
@@ -103,6 +118,7 @@ export default class SshKeyService {
    host: string,
    proxy?: string,
  ): Promise<void> {
+    alias = getSubscriptionSshAlias(alias);
    await this.generatePrivateKeyFile(alias, key);
    await this.generateSingleSshConfig(alias, host, proxy);
  }
@@ -112,6 +128,7 @@ export default class SshKeyService {
    host: string,
    proxy?: string,
  ): Promise<void> {
+    alias = getSubscriptionSshAlias(alias);
    await this.removePrivateKeyFile(alias);
    await this.removeSshConfig(alias);
  }
@@ -119,15 +136,19 @@ export default class SshKeyService {
  public async setSshConfig(docs: Subscription[]) {
    for (const doc of docs) {
      if (doc.type === 'private-repo' && doc.pull_type === 'ssh-key') {
-        const { alias, proxy } = doc;
-        const { host } = formatUrl(doc);
-        await this.removePrivateKeyFile(alias);
-        await this.removeSshConfig(alias);
-        await this.generatePrivateKeyFile(
-          alias,
-          (doc.pull_option as any).private_key,
-        );
-        await this.generateSingleSshConfig(alias, host, proxy);
+        try {
+          const alias = getSubscriptionSshAlias(doc.alias);
+          const { host } = formatUrl(doc);
+          await this.removePrivateKeyFile(alias);
+          await this.removeSshConfig(alias);
+          await this.generatePrivateKeyFile(
+            alias,
+            (doc.pull_option as any).private_key,
+          );
+          await this.generateSingleSshConfig(alias, host, doc.proxy);
+        } catch (error) {
+          this.logger.warn('跳过订阅 %s 的无效 SSH 配置: %s', doc.id, error);
+        }
      }
    }
  }
@@ -146,12 +167,9 @@ export default class SshKeyService {

  private async generateGlobalSshConfig(alias: string) {
    // Create a config that matches all hosts, making this key globally available
-    const config = `Host *\n    IdentityFile ${path.join(
-      this.sshPath,
-      alias,
-    )}\n    StrictHostKeyChecking no\n`;
+    const config = `Host *\n    IdentityFile ${resolveSubscriptionPath(this.sshPath, alias)}\n    StrictHostKeyChecking no\n`;
    await writeFileWithLock(
-      `${path.join(this.sshPath, `${alias}.config`)}`,
+      `${resolveSubscriptionPath(this.sshPath, alias, '.config')}`,
      config,
      {
        encoding: 'utf8',
@@ -20,18 +20,28 @@ import {
 } from '../config/util';
 import fs from 'fs/promises';
 import { FindOptions, Op } from 'sequelize';
-import path, { join } from 'path';
+import path from 'path';
 import ScheduleService, { TaskCallbacks } from './schedule';
 import { SimpleIntervalSchedule } from 'toad-scheduler';
 import SockService from './sock';
+import { t, tf } from '../shared/i18n';
 import SshKeyService from './sshKey';
 import dayjs from 'dayjs';
 import { LOG_END_SYMBOL } from '../config/const';
 import { formatCommand, formatUrl } from '../config/subscription';
 import { CrontabModel } from '../data/cron';
+import { assertSubscriptionAlias } from '../shared/subscriptionPath';
+import { resolveLogPath } from '../shared/logPath';
+import {
+  getSubscriptionStorageCandidates,
+  isSubscriptionStorageName,
+  resolveSubscriptionStoragePath,
+} from '../shared/subscriptionStorage';
+import { withSubscriptionMutation } from '../shared/subscriptionMutationLock';
 import CrontabService from './cron';
 import taskLimit from '../shared/pLimit';
 import { logStreamManager } from '../shared/logStreamManager';
+import { LogReadOptions, readLogChunk } from '../shared/logReader';

@Service()
 export default class SubscriptionService {
@@ -86,9 +96,11 @@ export default class SubscriptionService {
    needCreate = true,
    runImmediately = false,
  ) {
-    const { url } = formatUrl(doc);
-
-    doc.command = formatCommand(doc, url as string);
+    // Cancelling by ID must also work for invalid historical subscriptions.
+    if (needCreate) {
+      const { url } = formatUrl(doc);
+      doc.command = formatCommand(doc, url as string);
+    }

    if (doc.schedule_type === 'crontab') {
      this.scheduleService.cancelCronTask(doc as any);
@@ -130,14 +142,14 @@ export default class SubscriptionService {
        );
        const absolutePath = await handleLogPath(
          logPath as string,
-          `## 开始执行... ${startTime.format('YYYY-MM-DD HH:mm:ss')}\n`,
+          tf('## 开始执行... %s\n', startTime.format('YYYY-MM-DD HH:mm:ss')),
        );

        // 执行sub_before
        let beforeStr = '';
        try {
          if (doc.sub_before) {
-            await logStreamManager.write(absolutePath, `\n## 执行before命令...\n\n`);
+            await logStreamManager.write(absolutePath, `\n## ${t('执行before命令...')}\n\n`);
            beforeStr = await promiseExec(doc.sub_before);
          }
        } catch (error: any) {
@@ -157,44 +169,54 @@ export default class SubscriptionService {
        );
      },
      onEnd: async (cp, endTime, diff) => {
-        const sub = await this.getDb({ id: doc.id });
-        const absolutePath = await handleLogPath(sub.log_path as string);
-
-        // 执行 sub_after
-        let afterStr = '';
+        let absolutePath: string | undefined;
        try {
-          if (sub.sub_after) {
-            await logStreamManager.write(absolutePath, `\n\n## 执行after命令...\n\n`);
-            afterStr = await promiseExec(sub.sub_after);
+          const sub = await this.getDb({ id: doc.id });
+          absolutePath = await handleLogPath(sub.log_path as string);
+
+          // 执行 sub_after
+          let afterStr = '';
+          try {
+            if (sub.sub_after) {
+              await logStreamManager.write(
+                absolutePath,
+                `\n\n## ${t('执行after命令...')}\n\n`,
+              );
+              afterStr = await promiseExec(sub.sub_after);
+            }
+          } catch (error: any) {
+            afterStr =
+              (error.stderr && error.stderr.toString()) || JSON.stringify(error);
+          }
+          if (afterStr) {
+            await logStreamManager.write(absolutePath, `${afterStr}\n`);
+          }
+
+          await logStreamManager.write(
+            absolutePath,
+            '\n' +
+              tf(
+                '## 执行结束... %s  耗时 %s 秒',
+                endTime.format('YYYY-MM-DD HH:mm:ss'),
+                String(diff),
+              ) +
+              LOG_END_SYMBOL,
+          );
+        } finally {
+          try {
+            if (absolutePath) await logStreamManager.closeStream(absolutePath);
+          } finally {
+            await SubscriptionModel.update(
+              { status: SubscriptionStatus.idle, pid: null } as any,
+              { where: { id: doc.id } },
+            );
+            this.sockService.sendMessage({
+              type: 'runSubscriptionEnd',
+              message: t('订阅执行完成'),
+              references: [doc.id as number],
+            });
          }
-        } catch (error: any) {
-          afterStr =
-            (error.stderr && error.stderr.toString()) || JSON.stringify(error);
        }
-        if (afterStr) {
-          await logStreamManager.write(absolutePath, `${afterStr}\n`);
-        }
-
-        await logStreamManager.write(
-          absolutePath,
-          `\n## 执行结束... ${endTime.format(
-            'YYYY-MM-DD HH:mm:ss',
-          )}  耗时 ${diff} 秒${LOG_END_SYMBOL}`,
-        );
-
-        // Close the stream after task completion
-        await logStreamManager.closeStream(absolutePath);
-
-        await SubscriptionModel.update(
-          { status: SubscriptionStatus.idle, pid: undefined },
-          { where: { id: sub.id } },
-        );
-
-        this.sockService.sendMessage({
-          type: 'runSubscriptionEnd',
-          message: '订阅执行完成',
-          references: [doc.id as number],
-        });
      },
      onError: async (message: string) => {
        const sub = await this.getDb({ id: doc.id });
@@ -210,11 +232,14 @@ export default class SubscriptionService {
  }

  public async create(payload: Subscription): Promise<Subscription> {
-    const tab = new Subscription(payload);
-    const doc = await this.insert(tab);
-    await this.handleTask(doc.get({ plain: true }));
-    await this.setSshConfig();
-    return doc;
+    assertSubscriptionAlias(payload.alias);
+    return withSubscriptionMutation(async () => {
+      const tab = new Subscription(payload);
+      const doc = await this.insert(tab);
+      await this.handleTask(doc.get({ plain: true }));
+      await this.setSshConfig();
+      return doc;
+    });
  }

  public async insert(payload: Subscription): Promise<SubscriptionInstance> {
@@ -222,12 +247,27 @@ export default class SubscriptionService {
  }

  public async update(payload: Subscription): Promise<Subscription> {
-    const doc = await this.getDb({ id: payload.id });
-    const tab = new Subscription({ ...doc, ...payload });
-    const newDoc = await this.updateDb(tab);
-    await this.handleTask(newDoc, !newDoc.is_disabled);
-    await this.setSshConfig();
-    return newDoc;
+    assertSubscriptionAlias(payload.alias);
+    return withSubscriptionMutation(async () => {
+      const doc = await this.getDb({ id: payload.id });
+      const tab = new Subscription({ ...doc, ...payload });
+      const newDoc = await this.updateDb(tab);
+      await this.handleTask(newDoc, !newDoc.is_disabled);
+      await this.removeSshConfigForSubscription(doc);
+      await this.setSshConfig();
+      return newDoc;
+    });
+  }
+
+  private async removeSshConfigForSubscription(doc: Subscription) {
+    if (doc.type !== 'private-repo' || doc.pull_type !== 'ssh-key') return;
+    // Invalid historical rows must remain deletable without deriving unsafe paths.
+    try {
+      assertSubscriptionAlias(doc.alias);
+    } catch {
+      return;
+    }
+    await this.sshKeyService.removeSSHKey(doc.alias, '');
  }

  public async updateDb(payload: Subscription): Promise<Subscription> {
@@ -250,6 +290,7 @@ export default class SubscriptionService {
    last_running_time: number;
    last_execution_time: number;
  }) {
+    if (log_path) resolveLogPath(config.logPath, log_path);
    const options: any = {
      status,
      pid,
@@ -267,25 +308,71 @@ export default class SubscriptionService {
  }

  public async remove(ids: number[], query: { force?: boolean }) {
-    const docs = await SubscriptionModel.findAll({ where: { id: ids } });
-    for (const doc of docs) {
-      await this.handleTask(doc.get({ plain: true }), false);
-    }
-    await SubscriptionModel.destroy({ where: { id: ids } });
-    await this.setSshConfig();
-
-    if (query?.force === true) {
-      const crons = await CrontabModel.findAll({ where: { sub_id: ids } });
-      if (crons?.length) {
-        await this.crontabService.remove(crons.map((x) => x.id!));
+    return withSubscriptionMutation(async () => {
+      const docs = await SubscriptionModel.findAll({ where: { id: ids } });
+      const storagePaths: string[] = [];
+      const retainedPaths: string[] = [];
+      const collectPaths = (
+        doc: Subscription,
+        targets: string[],
+        deleting = false,
+      ) => {
+        for (const names of getSubscriptionStorageCandidates(doc)) {
+          // A URL accepted by one engine can produce an invalid name in another.
+          // Validate each candidate independently, retaining the safe candidate.
+          if (!Object.values(names).every(isSubscriptionStorageName)) continue;
+          const resolve = deleting ? resolveSubscriptionStoragePath : path.resolve;
+          const scriptPath = resolve(config.scriptPath, names.script);
+          if (scriptPath) targets.push(scriptPath);
+          if (names.repo) {
+            const repoPath = resolve(config.repoPath, names.repo);
+            if (repoPath) targets.push(repoPath);
+          }
+          if (names.raw) {
+            const rawPath = resolve(path.join(config.dataPath, 'raw'), names.raw);
+            if (rawPath) targets.push(rawPath);
+          }
+        }
+      };
+      // Validate all deletion targets before changing rows or removing any files.
+      if (query?.force === true) {
+        for (const doc of docs) {
+          assertSubscriptionAlias(doc.alias);
+          resolveSubscriptionStoragePath(config.scriptPath, doc.alias);
+          resolveSubscriptionStoragePath(config.repoPath, doc.alias);
+          collectPaths(doc, storagePaths, true);
+        }
+        const remaining = await SubscriptionModel.findAll({ where: {} });
+        for (const doc of remaining) {
+          if (!ids.includes(doc.id!)) collectPaths(doc, retainedPaths);
+        }
      }
      for (const doc of docs) {
-        const filePath = join(config.scriptPath, doc.alias);
-        const repoPath = join(config.repoPath, doc.alias);
-        await rmPath(filePath);
-        await rmPath(repoPath);
+        await this.handleTask(doc.get({ plain: true }), false);
      }
-    }
+      await SubscriptionModel.destroy({ where: { id: ids } });
+      for (const doc of docs) {
+        await this.removeSshConfigForSubscription(doc);
+      }
+      await this.setSshConfig();
+
+      if (query?.force === true) {
+        const crons = await CrontabModel.findAll({ where: { sub_id: ids } });
+        if (crons?.length) {
+          await this.crontabService.remove(crons.map((x) => x.id!));
+        }
+        for (const storagePath of new Set(storagePaths)) {
+          // rm is recursive: an ancestor must also be retained if a surviving
+          // subscription uses a nested branch, and vice versa.
+          const shared = retainedPaths.some((retained) =>
+            retained === storagePath ||
+            retained.startsWith(`${storagePath}${path.sep}`) ||
+            storagePath.startsWith(`${retained}${path.sep}`),
+          );
+          if (!shared) await rmPath(storagePath);
+        }
+      }
+    });
  }

  public async getDb(
@@ -310,20 +397,22 @@ export default class SubscriptionService {

  public async stop(ids: number[]) {
    const docs = await SubscriptionModel.findAll({ where: { id: ids } });
+    let failure: unknown;
    for (const doc of docs) {
-      if (doc.pid) {
-        try {
-          await killTask(doc.pid);
-        } catch (error) {
-          this.logger.error(error);
+      try {
+        if (doc.pid) {
+          await killTask(doc.pid, true);
        }
+        await SubscriptionModel.update(
+          { status: SubscriptionStatus.idle, pid: null } as any,
+          { where: { id: doc.id, pid: doc.pid ?? null } },
+        );
+      } catch (error) {
+        this.logger.error(error);
+        failure ??= error;
      }
    }
-
-    await SubscriptionModel.update(
-      { status: SubscriptionStatus.idle, pid: undefined },
-      { where: { id: ids } },
-    );
+    if (failure) throw failure;
  }

  private async runSingle(subscriptionId: number) {
@@ -361,14 +450,20 @@ export default class SubscriptionService {
    }
  }

-  public async log(id: number) {
+  public async log(id: number, options: LogReadOptions = {}) {
    const doc = await this.getDb({ id });
    if (!doc || !doc.log_path) {
-      return '';
+      return {
+        content: '',
+        offset: 0,
+        nextOffset: 0,
+        total: 0,
+        truncated: false,
+      };
    }

    const absolutePath = await handleLogPath(doc.log_path as string);
-    return await getFileContentByName(absolutePath);
+    return await readLogChunk(absolutePath, options);
  }

  public async logs(id: number) {
@@ -379,7 +474,7 @@ export default class SubscriptionService {

    if (doc.log_path) {
      const relativeDir = path.dirname(`${doc.log_path}`);
-      const dir = path.resolve(config.logPath, relativeDir);
+      const dir = path.dirname(resolveLogPath(config.logPath, doc.log_path));
      const _exist = await fileExist(dir);
      if (_exist) {
        let files = await fs.readdir(dir);
@@ -1,8 +1,10 @@
 import { spawn } from 'cross-spawn';
+import { execFile } from 'child_process';
+import { promisify } from 'util';
+import { shellQuote } from '../shared/shellQuote';
 import { Response } from 'express';
 import fs from 'fs';
 import { Agent, request } from 'undici';
-import sum from 'lodash/sum';
 import path from 'path';
 import { Inject, Service } from 'typedi';
 import winston from 'winston';
@@ -32,11 +34,15 @@ import {
  SystemModelInfo,
 } from '../data/system';
 import taskLimit from '../shared/pLimit';
+import cronClient from '../schedule/client';
+import { withSchedulerMutation } from '../shared/schedulerMutationLock';
 import NotificationService from './notify';
 import ScheduleService, { TaskCallbacks } from './schedule';
 import SockService from './sock';
 import os from 'os';
 import dayjs from 'dayjs';
+import { t, setLang } from '../shared/i18n';
+import { updateLinuxMirrorFile } from '../config/util';

@Service()
 export default class SystemService {
@@ -76,8 +82,8 @@ export default class SystemService {
    const code = Math.random().toString().slice(-6);
    const isSuccess = await this.notificationService.testNotify(
      notificationInfo,
-      '青龙',
-      `【蛟龙】测试通知 https://t.me/jiao_long`,
+      t('青龙'),
+      t('【蛟龙】测试通知 https://t.me/jiao_long'),
    );
    if (isSuccess) {
      const result = await this.updateAuthDb({
@@ -86,7 +92,7 @@ export default class SystemService {
      });
      return { code: 200, data: { ...result, code } };
    } else {
-      return { code: 400, message: '通知发送失败，请检查参数' };
+      return { code: 400, message: t('通知发送失败，请检查参数') };
    }
  }

@@ -98,7 +104,7 @@ export default class SystemService {
    });
    const cron = {
      id: result.id as number,
-      name: '删除日志',
+      name: t('删除日志'),
      command: `ql rmlog ${info.logRemoveFrequency}`,
      runOrigin: 'system' as const,
    };
@@ -118,15 +124,29 @@ export default class SystemService {
  }

  public async updateCronConcurrency(info: SystemModelInfo) {
-    const oDoc = await this.getSystemConfig();
-    await this.updateAuthDb({
-      ...oDoc,
-      info: { ...oDoc.info, ...info },
-    });
-    if (info.cronConcurrency) {
-      await taskLimit.setCustomLimit(info.cronConcurrency);
+    const limit = info.cronConcurrency ?? 0;
+    if (!Number.isInteger(limit) || limit < 0 || limit > 2147483647) {
+      throw Object.assign(new Error(t('参数错误')), { status: 400 });
    }
-    return { code: 200, data: info };
+    return withSchedulerMutation(async () => {
+      const oDoc = await this.getSystemConfig();
+      await this.updateAuthDb({
+        ...oDoc,
+        info: { ...oDoc.info, ...info, cronConcurrency: limit },
+      });
+      await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
+      await cronClient.setConcurrency(limit);
+      return { code: 200, data: info };
+    });
+  }
+
+  public async restoreCronConcurrency() {
+    await withSchedulerMutation(async () => {
+      const doc = await this.getSystemConfig();
+      const limit = doc.info?.cronConcurrency ?? 0;
+      await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
+      await cronClient.setConcurrency(limit);
+    });
  }

  public async updateDependenceProxy(info: SystemModelInfo) {
@@ -138,7 +158,7 @@ export default class SystemService {
    if (info.dependenceProxy) {
      await fs.promises.writeFile(
        config.dependenceProxyFile,
-        `export http_proxy="${info.dependenceProxy}"\nexport https_proxy="${info.dependenceProxy}"`,
+        `export http_proxy=${shellQuote(info.dependenceProxy)}\nexport https_proxy=${shellQuote(info.dependenceProxy)}`,
      );
    } else {
      await fs.promises.rm(config.dependenceProxyFile);
@@ -154,7 +174,7 @@ export default class SystemService {
    });
    let cmd = 'pnpm config delete registry';
    if (info.nodeMirror) {
-      cmd = `pnpm config set registry ${info.nodeMirror}`;
+      cmd = `pnpm config set registry ${shellQuote(info.nodeMirror)}`;
    }
    let command = `cd && ${cmd}`;
    const docs = await DependenceModel.findAll({
@@ -177,6 +197,7 @@ export default class SystemService {
          this.sockService.sendMessage({
            type: 'updateNodeMirror',
            message: 'update node mirror end',
+            status: 'completed',
          });
        },
        onError: async (message: string) => {
@@ -202,7 +223,7 @@ export default class SystemService {
    });
    let cmd = 'pip config unset global.index-url';
    if (info.pythonMirror) {
-      cmd = `pip3 config set global.index-url ${info.pythonMirror}`;
+      cmd = `pip3 config set global.index-url ${shellQuote(info.pythonMirror)}`;
    }
    await promiseExec(cmd);
    return { code: 200, data: info };
@@ -214,33 +235,10 @@ export default class SystemService {
    onEnd?: () => void,
  ) {
    const oDoc = await this.getSystemConfig();
-    await this.updateAuthDb({
-      ...oDoc,
-      info: { ...oDoc.info, ...info },
-    });
-    let defaultDomain = 'https://dl-cdn.alpinelinux.org';
-    let targetDomain = 'https://dl-cdn.alpinelinux.org';
    if (os.platform() !== 'linux') {
      return;
    }
-    const content = await fs.promises.readFile('/etc/apk/repositories', {
-      encoding: 'utf-8',
-    });
-    const domainMatch = content.match(/(http.*)\/alpine\/.*/);
-    if (domainMatch) {
-      defaultDomain = domainMatch[1];
-    }
-    if (info.linuxMirror) {
-      targetDomain = info.linuxMirror;
-    }
-    const command = `sed -i 's/${defaultDomain.replace(
-      /\//g,
-      '\\/',
-    )}/${targetDomain.replace(
-      /\//g,
-      '\\/',
-    )}/g' /etc/apk/repositories && apk update -f`;
-
+    const command = await updateLinuxMirrorFile(info.linuxMirror || '');
    this.scheduleService.runTask(
      command,
      {
@@ -248,12 +246,20 @@ export default class SystemService {
          res?.setHeader('QL-Task-Pid', `${cp.pid}`);
          res?.end();
        },
-        onEnd: async () => {
+        onEnd: async (cp) => {
          this.sockService.sendMessage({
            type: 'updateLinuxMirror',
            message: 'update linux mirror end',
+            status: 'completed',
          });
          onEnd?.();
+          // Tools such as sudo and apt may write warnings to stderr on success.
+          if (cp?.exitCode === 0 && cp.signalCode === null) {
+            await this.updateAuthDb({
+              ...oDoc,
+              info: { ...oDoc.info, ...info },
+            });
+          }
        },
        onError: async (message: string) => {
          this.sockService.sendMessage({ type: 'updateLinuxMirror', message });
@@ -353,6 +359,15 @@ export default class SystemService {
      this.sockService.sendMessage({
        type: 'updateSystemVersion',
        message: JSON.stringify(err),
+        status: 'failed',
+      });
+    });
+
+    cp.on('exit', (code) => {
+      this.sockService.sendMessage({
+        type: 'updateSystemVersion',
+        message: '',
+        status: code === 0 ? 'success' : 'failed',
      });
    });

@@ -360,7 +375,10 @@ export default class SystemService {
  }

  public async reloadSystem(target?: 'system' | 'data') {
-    const cmd = `real_time=true ql reload ${target || ''}`;
+    if (target && !['system', 'data'].includes(target)) {
+      return { code: 400, message: t('参数错误') };
+    }
+    const cmd = `real_time=true ql reload ${shellQuote(target || '')}`;
    const cp = spawn(cmd, {
      shell: '/bin/bash',
      detached: true,
@@ -395,9 +413,9 @@ export default class SystemService {
      notificationInfo,
    );
    if (isSuccess) {
-      return { code: 200, message: '通知发送成功' };
+      return { code: 200, message: t('通知发送成功') };
    } else {
-      return { code: 400, message: '通知发送失败，请检查系统设置/通知配置' };
+      return { code: 400, message: t('通知发送失败，请检查系统设置/通知配置') };
    }
  }

@@ -415,7 +433,7 @@ export default class SystemService {

  public async stop({ command, pid }: { command: string; pid: number }) {
    if (!pid && !command) {
-      return { code: 400, message: '参数错误' };
+      return { code: 400, message: t('参数错误') };
    }

    if (pid) {
@@ -431,7 +449,7 @@ export default class SystemService {
      await killTask(_pid);
      return { code: 200 };
    } else {
-      return { code: 400, message: '任务未找到' };
+      return { code: 400, message: t('任务未找到') };
    }
  }

@@ -441,11 +459,20 @@ export default class SystemService {
      if (type && type.length) {
        dataDirs = dataDirs.concat(type.filter((x) => x !== 'base'));
      }
-      const dataPaths = dataDirs.map((dir) => `data/${dir}`);
-      await promiseExec(
-        `cd ${config.dataPath} && cd ../ && tar -zcvf ${config.dataTgzFile
-        } ${dataPaths.join(' ')}`,
-      );
+      const allowed = new Set(['db', 'upload', 'config', 'scripts', 'log', 'deps',
+        'syslog', 'dep_cache', 'raw', 'repo', 'ssh.d', 'script-history']);
+      if (dataDirs.some((dir) => !allowed.has(dir))) {
+        return res.status(400).send({ code: 400, message: t('参数错误') });
+      }
+      if (dataDirs.includes('scripts')) dataDirs.push('script-history');
+      // History is created lazily; old/fresh installations must still export.
+      if (!fs.existsSync(path.join(config.dataPath, 'script-history'))) {
+        dataDirs = dataDirs.filter((dir) => dir !== 'script-history');
+      }
+      const dataPaths = [...new Set(dataDirs)].map((dir) => `data/${dir}`);
+      await promisify(execFile)('tar', ['-zcf', config.dataTgzFile, '--', ...dataPaths], {
+        cwd: path.dirname(config.dataPath),
+      });
      res.download(config.dataTgzFile);
    } catch (error: any) {
      return res.send({ code: 400, message: error.message });
@@ -469,6 +496,7 @@ export default class SystemService {
    query: {
      startTime?: string;
      endTime?: string;
+      limit?: number;
    },
  ) {
    const startTime = dayjs(query.startTime || undefined)
@@ -483,8 +511,30 @@ export default class SystemService {
      .filter((x) => x.title.endsWith('.log'))
      .filter((x) => x.createTime >= startTime && x.createTime <= endTime);

+    const limit = Math.min(
+      Math.max(Number(query.limit) || 1024 * 1024, 1),
+      1024 * 1024,
+    );
+    const total = logs.reduce((size, log) => size + (log.size || 0), 0);
+    let remaining = limit;
+    const selected: Array<
+      (typeof logs)[number] & { start: number; length: number }
+    > = [];
+    for (let index = logs.length - 1; index >= 0 && remaining > 0; index--) {
+      const log = logs[index];
+      const size = log.size || 0;
+      const length = Math.min(size, remaining);
+      if (length > 0) {
+        selected.unshift({ ...log, start: size - length, length });
+        remaining -= length;
+      }
+    }
+
+    const contentLength = selected.reduce((size, log) => size + log.length, 0);
    res.set({
-      'Content-Length': sum(logs.map((x) => x.size)),
+      'Content-Length': contentLength,
+      'X-QL-Log-Total': total,
+      'X-QL-Log-Truncated': total > contentLength ? 'true' : 'false',
    });
    (function sendFiles(res, fileNames) {
      if (fileNames.length === 0) {
@@ -496,13 +546,17 @@ export default class SystemService {
      if (currentLog) {
        const currentFileStream = fs.createReadStream(
          path.join(config.systemLogPath, currentLog.title),
+          {
+            start: currentLog.start,
+            end: currentLog.start + currentLog.length - 1,
+          },
        );
        currentFileStream.on('end', () => {
          sendFiles(res, fileNames);
        });
        currentFileStream.pipe(res, { end: false });
      }
-    })(res, logs);
+    })(res, selected);
  }

  public async deleteSystemLog() {
@@ -526,39 +580,77 @@ export default class SystemService {
    if (success) {
      return { code: 200, data: info };
    } else {
-      return { code: 400, message: '设置时区失败' };
+      return { code: 400, message: t('设置时区失败') };
    }
  }

+  public async updateLanguage(info: SystemModelInfo) {
+    const oDoc = await this.getSystemConfig();
+    const lang = info.lang || 'zh';
+    await this.updateAuthDb({
+      ...oDoc,
+      info: { ...oDoc.info, lang },
+    });
+    // Write to standalone lang_env.sh, sourced by shell scripts
+    try {
+      await fs.promises.writeFile(
+        config.langEnvFile,
+        `export QL_LANG=${shellQuote(lang)}\n`,
+      );
+    } catch (error) {
+      this.logger.error(`Failed to write lang_env.sh: ${error}`);
+    }
+    setLang(lang);
+    return { code: 200, data: { lang } };
+  }
+
+  public async updatePanelTitle(info: SystemModelInfo) {
+    const oDoc = await this.getSystemConfig();
+    const panelTitle = info.panelTitle?.trim() || '';
+    await this.updateAuthDb({
+      ...oDoc,
+      info: { ...oDoc.info, panelTitle },
+    });
+    return { code: 200, data: { panelTitle } };
+  }
+
  public async updateGlobalSshKey(info: SystemModelInfo) {
    const oDoc = await this.getSystemConfig();
    const result = await this.updateAuthDb({
      ...oDoc,
      info: { ...oDoc.info, ...info },
    });
-    
+
    // Apply the global SSH key
    const SshKeyService = require('./sshKey').default;
    const Container = require('typedi').Container;
    const sshKeyService = Container.get(SshKeyService);
-    
+
    if (info.globalSshKey) {
      await sshKeyService.addGlobalSSHKey(info.globalSshKey, 'global');
    } else {
      await sshKeyService.removeGlobalSSHKey('global');
    }
-    
+
    return { code: 200, data: result };
  }

  public async cleanDependence(type: 'node' | 'python3') {
    if (!type || !['node', 'python3'].includes(type)) {
-      return { code: 400, message: '参数错误' };
+      return { code: 400, message: t('参数错误') };
    }
-    try {
-      const finalPath = path.join(config.dependenceCachePath, type);
-      await fs.promises.rm(finalPath, { recursive: true });
-    } catch (error) { }
+    const finalPath = path.join(config.dependenceCachePath, type);
+    await fs.promises.rm(finalPath, { recursive: true, force: true });
+    await DependenceModel.update(
+      { status: DependenceStatus.installFailed },
+      {
+        where: {
+          type:
+            type === 'node' ? DependenceTypes.nodejs : DependenceTypes.python3,
+          status: DependenceStatus.installed,
+        },
+      },
+    );
    return { code: 200 };
  }
 }
@@ -20,14 +20,24 @@ import ScheduleService from './schedule';
 import SockService from './sock';
 import dayjs from 'dayjs';
 import IP2Region from 'ip2region';
-import requestIp from 'request-ip';
 import uniq from 'lodash/uniq';
 import pickBy from 'lodash/pickBy';
 import isNil from 'lodash/isNil';
 import { shareStore } from '../shared/store';
+import { t, tf } from '../shared/i18n';
+import { getClientIp, normalizeClientIp } from '../shared/clientIp';
+import { isDefaultAuthInfo } from '../shared/auth';
+import {
+  hashPassword,
+  isPasswordHash,
+  verifyPassword,
+} from '../shared/password';
+import { serializeAuthMutation } from '../shared/authMutation';
+import { LoginThrottle } from '../shared/loginThrottle';

@Service()
 export default class UserService {
+  private loginThrottle = new LoginThrottle();
  @Inject((type) => NotificationService)
  private notificationService!: NotificationService;

@@ -37,7 +47,15 @@ export default class UserService {
    private sockService: SockService,
  ) {}

+  @serializeAuthMutation
  public async login(
+    payloads: { username: string; password: string },
+    req: Request,
+  ): Promise<any> {
+    return this.authenticate(payloads, req);
+  }
+
+  private async authenticate(
    payloads: {
      username: string;
      password: string;
@@ -47,7 +65,18 @@ export default class UserService {
  ): Promise<any> {
    let { username, password } = payloads;
    const content = await this.getAuthInfo();
+    if (isDefaultAuthInfo(content)) {
+      return { code: 450, message: t('请先初始化') };
+    }
    const timestamp = Date.now();
+    const ip = getClientIp(req);
+    const query = new IP2Region();
+    const ipAddress = query.search(ip);
+    let address = '';
+    if (ipAddress) {
+      const { country, province, city, isp } = ipAddress;
+      address = uniq([country, province, city, isp]).filter(Boolean).join(' ');
+    }
    let {
      username: cUsername,
      password: cPassword,
@@ -58,27 +87,43 @@ export default class UserService {
      twoFactorActivated,
      tokens = {},
      platform,
+      blockedIps = [],
    } = content;
-    const retriesTime = Math.pow(3, retries) * 1000;
-    if (retries > 2 && timestamp - lastlogon < retriesTime) {
-      const waitTime = Math.ceil(
-        (retriesTime - (timestamp - lastlogon)) / 1000,
-      );
+
+    if (
+      ip &&
+      blockedIps.some((blockedIp) => normalizeClientIp(blockedIp) === ip)
+    ) {
+      await this.insertDb({
+        type: AuthDataType.loginLog,
+        info: {
+          timestamp,
+          address,
+          ip,
+          platform: req.platform,
+          status: LoginStatus.fail,
+        },
+      });
+      this.getLoginLog();
+      return { code: 403, message: t('该 IP 已被列入黑名单') };
+    }
+
+    const waitTime = this.loginThrottle.retryAfter(ip, timestamp);
+    if (waitTime > 0) {
      return {
        code: 410,
-        message: `失败次数过多，请${waitTime}秒后重试`,
+        message: tf('失败次数过多，请%s秒后重试', waitTime),
        data: waitTime,
      };
    }

-    if (
-      username === cUsername &&
-      password === cPassword &&
-      twoFactorActivated &&
-      needTwoFactor
-    ) {
+    const passwordMatches =
+      username === cUsername && (await verifyPassword(password, cPassword));
+
+    if (passwordMatches && twoFactorActivated && needTwoFactor) {
      await this.updateAuthInfo(content, {
        isTwoFactorChecking: true,
+        twoFactorExpiresAt: timestamp + 5 * 60 * 1000,
      });
      return {
        code: 420,
@@ -86,15 +131,8 @@ export default class UserService {
      };
    }

-    const ip = requestIp.getClientIp(req) || '';
-    const query = new IP2Region();
-    const ipAddress = query.search(ip);
-    let address = '';
-    if (ipAddress) {
-      const { country, province, city, isp } = ipAddress;
-      address = uniq([country, province, city, isp]).filter(Boolean).join(' ');
-    }
-    if (username === cUsername && password === cPassword) {
+    if (passwordMatches) {
+      this.loginThrottle.reset(ip);
      const data = createRandomString(50, 100);
      const expiration = twoFactorActivated ? '60d' : '20d';
      let token = jwt.sign({ data }, config.jwt.secret, {
@@ -117,6 +155,9 @@ export default class UserService {
      );

      await this.updateAuthInfo(content, {
+        password: isPasswordHash(cPassword)
+          ? cPassword
+          : await hashPassword(password),
        token,
        tokens: updatedTokens,
        lastlogon: timestamp,
@@ -125,12 +166,22 @@ export default class UserService {
        lastaddr: address,
        platform: req.platform,
        isTwoFactorChecking: false,
+        twoFactorExpiresAt: 0,
      });
      this.notificationService.notify(
-        '登录通知',
-        `你于${dayjs(timestamp).format('YYYY-MM-DD HH:mm:ss')}在 ${address} ${
-          req.platform
-        }端 登录成功，ip地址 ${ip}`,
+        t('登录通知'),
+        t('你于') +
+          dayjs(timestamp).format('YYYY-MM-DD HH:mm:ss') +
+          t('在') +
+          address +
+          ' ' +
+          req.platform +
+          t('端') +
+          ' ' +
+          t('登录成功') +
+          t('，ip地址') +
+          ' ' +
+          ip,
      );
      await this.insertDb({
        type: AuthDataType.loginLog,
@@ -155,18 +206,28 @@ export default class UserService {
        },
      };
    } else {
+      const failures = this.loginThrottle.fail(ip, timestamp);
      await this.updateAuthInfo(content, {
-        retries: retries + 1,
+        retries: failures,
        lastlogon: timestamp,
        lastip: ip,
        lastaddr: address,
        platform: req.platform,
      });
      this.notificationService.notify(
-        '登录通知',
-        `你于${dayjs(timestamp).format('YYYY-MM-DD HH:mm:ss')}在 ${address} ${
-          req.platform
-        }端 登录失败，ip地址 ${ip}`,
+        t('登录通知'),
+        t('你于') +
+          dayjs(timestamp).format('YYYY-MM-DD HH:mm:ss') +
+          t('在') +
+          address +
+          ' ' +
+          req.platform +
+          t('端') +
+          ' ' +
+          t('登录失败') +
+          t('，ip地址') +
+          ' ' +
+          ip,
      );
      await this.insertDb({
        type: AuthDataType.loginLog,
@@ -179,19 +240,20 @@ export default class UserService {
        },
      });
      this.getLoginLog();
-      if (retries > 2) {
-        const waitTime = Math.round(Math.pow(3, retries + 1));
+      if (failures > 3) {
+        const waitTime = this.loginThrottle.retryAfter(ip, timestamp);
        return {
          code: 410,
-          message: `失败次数过多，请${waitTime}秒后重试`,
+          message: tf('失败次数过多，请%s秒后重试', waitTime),
          data: waitTime,
        };
      } else {
-        return { code: 400, message: config.authError };
+        return { code: 400, message: t('错误的用户名密码，请重试') };
      }
    }
  }

+  @serializeAuthMutation
  public async logout(platform: string, tokenValue: string): Promise<any> {
    if (!platform || !tokenValue) {
      this.logger.warn('Invalid logout parameters - empty platform or token');
@@ -245,11 +307,67 @@ export default class UserService {
    return [];
  }

+  public async getIpBlacklist(): Promise<string[]> {
+    const authInfo = await this.getAuthInfo();
+    return uniq((authInfo.blockedIps || []).map(normalizeClientIp)).filter(
+      Boolean,
+    );
+  }
+
+  @serializeAuthMutation
+  public async blockIp(ip: string): Promise<string[]> {
+    const authInfo = await this.getAuthInfo();
+    const blockedIps = uniq([
+      ...(authInfo.blockedIps || []).map(normalizeClientIp),
+      normalizeClientIp(ip),
+    ]).filter(Boolean);
+    await this.updateAuthInfo(authInfo, { blockedIps });
+    return blockedIps;
+  }
+
+  @serializeAuthMutation
+  public async unblockIp(ip: string): Promise<string[]> {
+    const authInfo = await this.getAuthInfo();
+    const normalizedIp = normalizeClientIp(ip);
+    const blockedIps = (authInfo.blockedIps || [])
+      .map(normalizeClientIp)
+      .filter((blockedIp) => blockedIp && blockedIp !== normalizedIp);
+    await this.updateAuthInfo(authInfo, { blockedIps });
+    return blockedIps;
+  }
+
  private async insertDb(payload: SystemInfo): Promise<SystemInfo> {
    const doc = await SystemModel.create({ ...payload }, { returning: true });
    return doc;
  }

+  @serializeAuthMutation
+  public async initializeUser({
+    username,
+    password,
+  }: {
+    username: string;
+    password: string;
+  }) {
+    const authInfo = await this.getAuthInfo();
+    if (!isDefaultAuthInfo(authInfo)) {
+      return { code: 450, message: t('未知错误') };
+    }
+    if (password === 'admin') {
+      return { code: 400, message: t('密码不能设置为admin') };
+    }
+    await this.updateAuthInfo(authInfo, {
+      username,
+      password: await hashPassword(password),
+      token: '',
+      tokens: {},
+      isTwoFactorChecking: false,
+      twoFactorExpiresAt: 0,
+    });
+    return { code: 200, message: t('更新成功') };
+  }
+
+  @serializeAuthMutation
  public async updateUsernameAndPassword({
    username,
    password,
@@ -258,27 +376,40 @@ export default class UserService {
    password: string;
  }) {
    if (password === 'admin') {
-      return { code: 400, message: '密码不能设置为admin' };
+      return { code: 400, message: t('密码不能设置为admin') };
    }
    const authInfo = await this.getAuthInfo();
-    await this.updateAuthInfo(authInfo, { username, password });
-    return { code: 200, message: '更新成功' };
+    await this.updateAuthInfo(authInfo, {
+      username,
+      password: await hashPassword(password),
+      token: '',
+      tokens: {},
+      isTwoFactorChecking: false,
+      twoFactorExpiresAt: 0,
+    });
+    return { code: 200, message: t('更新成功') };
  }

+  @serializeAuthMutation
  public async updateAvatar(avatar: string) {
    const authInfo = await this.getAuthInfo();
    await this.updateAuthInfo(authInfo, { avatar });
-    return { code: 200, data: avatar, message: '更新成功' };
+    return { code: 200, data: avatar, message: t('更新成功') };
  }

+  @serializeAuthMutation
  public async initTwoFactor() {
    const secret = authenticator.generateSecret();
    const authInfo = await this.getAuthInfo();
+    if (authInfo.twoFactorActivated) {
+      throw new Error(t('请先关闭两步验证'));
+    }
    const otpauth = authenticator.keyuri(authInfo.username, 'qinglong', secret);
    await this.updateAuthInfo(authInfo, { twoFactorSecret: secret });
    return { secret, url: otpauth };
  }

+  @serializeAuthMutation
  public async activeTwoFactor(code: string) {
    const authInfo = await this.getAuthInfo();
    const isValid = authenticator.verify({
@@ -286,11 +417,18 @@ export default class UserService {
      secret: authInfo.twoFactorSecret,
    });
    if (isValid) {
-      await this.updateAuthInfo(authInfo, { twoFactorActivated: true });
+      await this.updateAuthInfo(authInfo, {
+        twoFactorActivated: true,
+        token: '',
+        tokens: {},
+        isTwoFactorChecking: false,
+        twoFactorExpiresAt: 0,
+      });
    }
    return isValid;
  }

+  @serializeAuthMutation
  public async twoFactorLogin(
    {
      username,
@@ -301,17 +439,30 @@ export default class UserService {
  ) {
    const authInfo = await this.getAuthInfo();
    const { isTwoFactorChecking, twoFactorSecret } = authInfo;
-    if (!isTwoFactorChecking) {
-      return { code: 450, message: '未知错误' };
+    const now = Date.now();
+    const source = getClientIp(req);
+    if (this.loginThrottle.retryAfter(source, now) > 0) {
+      return { code: 410, message: t('失败次数过多，请稍后重试') };
    }
-    const isValid = authenticator.verify({
-      token: code,
-      secret: twoFactorSecret,
-    });
+    if (
+      !isTwoFactorChecking ||
+      !authInfo.twoFactorActivated ||
+      !authInfo.twoFactorExpiresAt ||
+      authInfo.twoFactorExpiresAt <= now
+    ) {
+      return { code: 450, message: t('未知错误') };
+    }
+    const step = Math.floor(now / 30000);
+    const isValid =
+      username === authInfo.username &&
+      (await verifyPassword(password, authInfo.password)) &&
+      authInfo.lastTwoFactorStep !== step &&
+      authenticator.verify({ token: code, secret: twoFactorSecret });
    if (isValid) {
-      return this.login({ username, password }, req, false);
+      await this.updateAuthInfo(authInfo, { lastTwoFactorStep: step });
+      return this.authenticate({ username, password }, req, false);
    } else {
-      const ip = requestIp.getClientIp(req) || '';
+      const ip = getClientIp(req);
      const query = new IP2Region();
      const ipAddress = query.search(ip);
      let address = '';
@@ -322,19 +473,26 @@ export default class UserService {
          .join(' ');
      }
      await this.updateAuthInfo(authInfo, {
+        retries: this.loginThrottle.fail(source, now),
+        lastlogon: now,
        lastip: ip,
        lastaddr: address,
        platform: req.platform,
      });
-      return { code: 430, message: '验证失败' };
+      return { code: 430, message: t('验证失败') };
    }
  }

-  public async deactiveTwoFactor() {
+  @serializeAuthMutation
+  public async deactivateTwoFactor() {
    const authInfo = await this.getAuthInfo();
    await this.updateAuthInfo(authInfo, {
      twoFactorActivated: false,
      twoFactorSecret: '',
+      token: '',
+      tokens: {},
+      isTwoFactorChecking: false,
+      twoFactorExpiresAt: 0,
    });
    return true;
  }
@@ -355,6 +513,9 @@ export default class UserService {
      type: AuthDataType.authConfig,
      info: result,
    });
+    if (info.tokens && Object.keys(info.tokens).length === 0) {
+      this.sockService.getClients().forEach((conn) => conn.close('401'));
+    }
  }

  public async getNotificationMode(): Promise<NotificationInfo> {
@@ -388,8 +549,8 @@ export default class UserService {
    const code = Math.random().toString().slice(-6);
    const isSuccess = await this.notificationService.testNotify(
      notificationInfo,
-      '青龙',
-      `【蛟龙】测试通知 https://t.me/jiao_long`,
+      t('青龙'),
+      t('【蛟龙】测试通知 https://t.me/jiao_long'),
    );
    if (isSuccess) {
      const result = await this.updateAuthDb({
@@ -398,7 +559,7 @@ export default class UserService {
      });
      return { code: 200, data: { ...result, code } };
    } else {
-      return { code: 400, message: '通知发送失败，请检查参数' };
+      return { code: 400, message: t('通知发送失败，请检查参数') };
    }
  }

@@ -496,8 +657,12 @@ export default class UserService {
    return undefined;
  }

+  @serializeAuthMutation
  public async resetAuthInfo(info: Partial<AuthInfo>) {
    const { retries, twoFactorActivated, password, username } = info;
+    if (password === 'admin') {
+      return { code: 400, message: t('密码不能设置为admin') };
+    }
    const authInfo = await this.getAuthInfo();
    const payload = pickBy(
      {
@@ -509,6 +674,24 @@ export default class UserService {
      (x) => !isNil(x),
    );

+    if (password !== undefined) {
+      payload.password = await hashPassword(password);
+    }
+    if (
+      password !== undefined ||
+      username !== undefined ||
+      twoFactorActivated !== undefined
+    ) {
+      Object.assign(payload, {
+        token: '',
+        tokens: {},
+        isTwoFactorChecking: false,
+        twoFactorExpiresAt: 0,
+      });
+    }
    await this.updateAuthInfo(authInfo, payload);
+    if (retries === 0 || password !== undefined || username !== undefined) {
+      this.loginThrottle.clear();
+    }
  }
 }