feat(storage): pause Drive uploads on quota cooldown

This commit is contained in:
hicocos committed 2026-07-06 17:26:04 +08:00
1 parent 7df6f9111a
commit 0cc0543e6e
14 files changed
+808 -25

No files matched your search

+297 -12
View File
@@ -97,6 +97,18 @@ async function initializeDatabase() {
await pool.query(`ALTER TABLE telegram_channel_subscriptions ADD COLUMN IF NOT EXISTS source_type TEXT DEFAULT 'public'`);
await pool.query(`ALTER TABLE telegram_channel_subscriptions ADD COLUMN IF NOT EXISTS disabled_reason TEXT`);
await pool.query(`ALTER TABLE telegram_channel_subscriptions ADD COLUMN IF NOT EXISTS disabled_at TIMESTAMPTZ`);
await pool.query(`CREATE TABLE IF NOT EXISTS storage_account_cooldowns (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
storage_account_id UUID REFERENCES storage_accounts(id) ON DELETE CASCADE,
provider VARCHAR(50) NOT NULL,
reason VARCHAR(100) NOT NULL,
cooldown_until TIMESTAMPTZ NOT NULL,
last_error TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(storage_account_id, provider, reason)
)`);
await pool.query(`CREATE INDEX IF NOT EXISTS idx_storage_account_cooldowns_until ON storage_account_cooldowns(cooldown_until)`);
await pool.query(`CREATE UNIQUE INDEX IF NOT EXISTS idx_api_keys_key_hash ON api_keys(key_hash) WHERE key_hash IS NOT NULL`);
console.log("\u2705 \u6570\u636E\u5E93\u8868\u7ED3\u6784\u521D\u59CB\u5316\u5B8C\u6210");
} catch (err) {
@@ -347,6 +359,72 @@ var init_localPath = __esm({
}
});
// src/services/storageCooldown.ts
async function ensureStorageCooldownSchema() {
await query(`
CREATE TABLE IF NOT EXISTS storage_account_cooldowns (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
storage_account_id UUID REFERENCES storage_accounts(id) ON DELETE CASCADE,
provider VARCHAR(50) NOT NULL,
reason VARCHAR(100) NOT NULL,
cooldown_until TIMESTAMPTZ NOT NULL,
last_error TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(storage_account_id, provider, reason)
)
`);
await query(`CREATE INDEX IF NOT EXISTS idx_storage_account_cooldowns_until ON storage_account_cooldowns(cooldown_until)`);
}
async function markStorageAccountCooldown(storageAccountId, provider, reason, cooldownUntil, error) {
if (!storageAccountId) return;
await ensureStorageCooldownSchema();
await query(
`INSERT INTO storage_account_cooldowns (storage_account_id, provider, reason, cooldown_until, last_error, updated_at)
VALUES ($1, $2, $3, $4, $5, NOW())
ON CONFLICT (storage_account_id, provider, reason)
DO UPDATE SET cooldown_until = EXCLUDED.cooldown_until, last_error = EXCLUDED.last_error, updated_at = NOW()`,
[storageAccountId, provider, reason, cooldownUntil, error || null]
);
}
async function getStorageAccountCooldown(storageAccountId, provider, reason = STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT) {
if (!storageAccountId) return null;
await ensureStorageCooldownSchema();
const result = await query(
`SELECT storage_account_id, provider, reason, cooldown_until, last_error
FROM storage_account_cooldowns
WHERE storage_account_id = $1
AND provider = $2
AND reason = $3
AND cooldown_until > NOW()
LIMIT 1`,
[storageAccountId, provider, reason]
);
const row = result.rows[0];
if (!row) return null;
return {
storageAccountId: row.storage_account_id,
provider: row.provider,
reason: row.reason,
cooldownUntil: new Date(row.cooldown_until),
lastError: row.last_error
};
}
async function clearExpiredStorageCooldowns() {
await ensureStorageCooldownSchema();
const result = await query(`DELETE FROM storage_account_cooldowns WHERE cooldown_until <= NOW()`);
return result.rowCount || 0;
}
var STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT, DEFAULT_STORAGE_COOLDOWN_MS;
var init_storageCooldown = __esm({
"src/services/storageCooldown.ts"() {
"use strict";
init_db();
STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT = "daily_upload_limit";
DEFAULT_STORAGE_COOLDOWN_MS = 24 * 60 * 60 * 1e3;
}
});
// src/services/storage.ts
var storage_exports = {};
__export(storage_exports, {
@@ -356,7 +434,9 @@ __export(storage_exports, {
OneDriveStorageProvider: () => OneDriveStorageProvider,
S3StorageProvider: () => S3StorageProvider,
StorageManager: () => StorageManager,
StorageQuotaCooldownError: () => StorageQuotaCooldownError,
WebDAVStorageProvider: () => WebDAVStorageProvider,
isStorageQuotaCooldownError: () => isStorageQuotaCooldownError,
storageManager: () => storageManager
});
import fs4 from "fs";
@@ -367,13 +447,48 @@ import { S3Client, PutObjectCommand, GetObjectCommand, DeleteObjectCommand, Head
import { getSignedUrl as getS3SignedUrl } from "@aws-sdk/s3-request-presigner";
import { createClient } from "webdav";
import { google } from "googleapis";
var LocalStorageProvider, AliyunOSSStorageProvider, S3StorageProvider, WebDAVStorageProvider, OneDriveStorageProvider, GoogleDriveStorageProvider, StorageManager, storageManager;
function isStorageQuotaCooldownError(error) {
return error instanceof StorageQuotaCooldownError || error?.name === "StorageQuotaCooldownError";
}
function googleDriveErrorText(error) {
const pieces = [
error?.message,
error?.errorMessage,
error?.response?.data ? JSON.stringify(error.response.data) : void 0,
error?.errors ? JSON.stringify(error.errors) : void 0
].filter(Boolean);
return pieces.join(" ");
}
function isGoogleDriveDailyUploadLimitError(error) {
const status = Number(error?.code || error?.response?.status || error?.status || 0);
const text = googleDriveErrorText(error);
if (status !== 403) return false;
return /Drive upload limit|exceeded their Drive upload limit|exceeded.*upload limit|upload limit exceeded|upload limit/i.test(text);
}
var StorageQuotaCooldownError, LocalStorageProvider, AliyunOSSStorageProvider, S3StorageProvider, WebDAVStorageProvider, OneDriveStorageProvider, GoogleDriveStorageProvider, StorageManager, storageManager;
var init_storage = __esm({
"src/services/storage.ts"() {
"use strict";
init_db();
init_localPath();
init_credentialCrypto();
init_storageCooldown();
StorageQuotaCooldownError = class extends Error {
provider;
reason;
storageAccountId;
cooldownUntil;
originalError;
constructor(message, options) {
super(message);
this.name = "StorageQuotaCooldownError";
this.provider = options.provider;
this.reason = options.reason;
this.storageAccountId = options.storageAccountId;
this.cooldownUntil = options.cooldownUntil || new Date(Date.now() + DEFAULT_STORAGE_COOLDOWN_MS);
this.originalError = options.originalError;
}
};
LocalStorageProvider = class {
name = "local";
uploadDir;
@@ -1222,6 +1337,14 @@ var init_storage = __esm({
return file.data.id;
} catch (error) {
console.error("[GoogleDrive] Upload failed:", error.message);
if (isGoogleDriveDailyUploadLimitError(error)) {
throw new StorageQuotaCooldownError("Google Drive \u4ECA\u65E5\u4E0A\u4F20\u989D\u5EA6\u5DF2\u8FBE\u4E0A\u9650\uFF0C\u4EFB\u52A1\u5C06\u81EA\u52A8\u6682\u505C 24 \u5C0F\u65F6\u540E\u7EE7\u7EED\u3002", {
provider: "google_drive",
reason: STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT,
storageAccountId: this.id,
originalError: error
});
}
throw new Error(`Google Drive upload failed: ${error.message}`);
}
}
@@ -2850,6 +2973,58 @@ async function getImageDimensions(filePath, mimeType) {
// src/services/telegramUpload.ts
init_storage();
// src/services/storageCooldownGuard.ts
init_storage();
init_storageCooldown();
function formatStorageCooldownNotice(cooldownUntil) {
return [
"\u23F8\uFE0F Google Drive \u4ECA\u65E5\u4E0A\u4F20\u989D\u5EA6\u5DF2\u8FBE\u4E0A\u9650",
"",
"\u5F53\u524D\u4EFB\u52A1\u5DF2\u81EA\u52A8\u6682\u505C\uFF0C\u9884\u8BA1 24 \u5C0F\u65F6\u540E\u7EE7\u7EED\u3002",
"\u5269\u4F59\u6587\u4EF6\u4E0D\u4F1A\u4E22\u5931\uFF0C\u6062\u590D\u540E\u4F1A\u4ECE\u672A\u5B8C\u6210\u90E8\u5206\u7EE7\u7EED\u5904\u7406\u3002",
"",
`\u6062\u590D\u65F6\u95F4\uFF1A${cooldownUntil.toISOString()}`
].join("\n");
}
function buildStorageCooldownHttpError(error) {
return {
status: 429,
body: {
error: error.message || "Google Drive \u4ECA\u65E5\u4E0A\u4F20\u989D\u5EA6\u5DF2\u8FBE\u4E0A\u9650\uFF0C\u8BF7\u7A0D\u540E\u91CD\u8BD5\u3002",
code: "storage_account_cooling",
provider: error.provider,
reason: error.reason,
retryAt: error.cooldownUntil.toISOString()
}
};
}
function sendStorageCooldownHttpError(res, error) {
const payload = buildStorageCooldownHttpError(error);
res.status(payload.status).json(payload.body);
}
async function getActiveStorageCooldown() {
const provider = storageManager.getProvider();
const activeAccountId = storageManager.getActiveAccountId();
if (provider.name !== "google_drive" || !activeAccountId) return null;
return getStorageAccountCooldown(activeAccountId, provider.name, STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT);
}
async function assertActiveStorageWritable() {
const cooldown = await getActiveStorageCooldown();
if (!cooldown) return;
throw new StorageQuotaCooldownError("Google Drive \u4ECA\u65E5\u4E0A\u4F20\u989D\u5EA6\u5DF2\u8FBE\u4E0A\u9650\uFF0C\u8BF7\u7B49\u5F85\u81EA\u52A8\u6062\u590D\u540E\u518D\u4E0A\u4F20\uFF0C\u6216\u4E34\u65F6\u5207\u6362\u5176\u5B83\u5B58\u50A8\u6E90\u3002", {
provider: cooldown.provider,
reason: cooldown.reason,
storageAccountId: cooldown.storageAccountId,
cooldownUntil: cooldown.cooldownUntil
});
}
function isStorageCooldownError(error) {
return isStorageQuotaCooldownError(error);
}
// src/services/telegramUpload.ts
init_storageCooldown();
// src/services/telegramUserClient.ts
import fs6 from "fs";
import path7 from "path";
@@ -4430,6 +4605,7 @@ async function processFileUpload(client2, file, queue) {
let storedName;
try {
const activeAccountId = storageManager.getActiveAccountId();
await assertActiveStorageWritable();
const chatName = await getTelegramChatName(file.message);
const batchFolder = null;
const storageRules = await getStoragePathRules();
@@ -4511,7 +4687,13 @@ async function processFileUpload(client2, file, queue) {
});
} catch (error) {
console.error("\u{1F916} \u6587\u4EF6\u4E0A\u4F20\u5931\u8D25:", error);
file.error = error.message;
if (isStorageQuotaCooldownError(error)) {
await markStorageAccountCooldown(error.storageAccountId || storageManager.getActiveAccountId(), error.provider, error.reason, error.cooldownUntil, error.message);
file.storageCooldownUntil = error.cooldownUntil;
file.error = formatStorageCooldownNotice(error.cooldownUntil);
} else {
file.error = error.message;
}
if (localFilePath && fs7.existsSync(localFilePath)) {
try {
fs7.unlinkSync(localFilePath);
@@ -4526,7 +4708,7 @@ async function processFileUpload(client2, file, queue) {
const queueTask = async (signal) => {
file.status = "uploading";
const firstAttemptSuccess = await attemptUpload(signal);
if (!firstAttemptSuccess && !signal.aborted && !file.retried) {
if (!firstAttemptSuccess && !signal.aborted && !file.retried && !file.storageCooldownUntil) {
file.retried = true;
file.status = "uploading";
file.error = void 0;
@@ -4902,6 +5084,13 @@ async function downloadTelegramChannelRange(botClient, requestMessage, source, s
successful += 1;
successfulMessageIds.push(item.id);
await onItemSettled?.(item, "success");
} else if (uploadItem.storageCooldownUntil) {
throw new StorageQuotaCooldownError(uploadItem.error || "Google Drive \u4ECA\u65E5\u4E0A\u4F20\u989D\u5EA6\u5DF2\u8FBE\u4E0A\u9650\uFF0C\u4EFB\u52A1\u5C06\u81EA\u52A8\u6682\u505C 24 \u5C0F\u65F6\u540E\u7EE7\u7EED\u3002", {
provider: "google_drive",
reason: "daily_upload_limit",
storageAccountId: storageManager.getActiveAccountId() || void 0,
cooldownUntil: uploadItem.storageCooldownUntil
});
} else {
failed += 1;
failedMessageIds.push(item.id);
@@ -5299,8 +5488,10 @@ init_storage();
// src/services/telegramChannelJobs.ts
init_db();
init_storage();
import { Api as Api5 } from "telegram";
import { getPeerId } from "telegram/Utils.js";
init_storageCooldown();
var SUBSCRIPTION_INTERVAL_MS = Math.max(6e4, parseInt(process.env.TELEGRAM_SUBSCRIPTION_INTERVAL_MS || "300000", 10) || 3e5);
var SUBSCRIPTION_SCAN_LIMIT = Math.max(1, parseInt(process.env.TELEGRAM_SUBSCRIPTION_SCAN_LIMIT || "100", 10) || 100);
var TG_JOB_RECOVERY_DELAY_MS = Math.max(1e3, parseInt(process.env.TG_JOB_RECOVERY_DELAY_MS || "10000", 10) || 1e4);
@@ -5802,12 +5993,62 @@ function isFloodWait(error) {
if (seconds > 0 || /FLOOD|Too many requests/i.test(text)) return { seconds: Math.max(30, seconds || 60) };
return null;
}
async function putJobIntoStorageCooldown(jobId, cooldownUntil, reasonText) {
await updateJob(jobId, {
status: "cooling",
download_status: "cooling",
cooldown_until: cooldownUntil,
error: reasonText
});
await query(
`UPDATE telegram_download_items
SET status = 'pending', locked_at = NULL, last_error = $2, updated_at = NOW()
WHERE job_id = $1
AND status = 'downloading'`,
[jobId, reasonText]
);
}
async function applyStorageCooldownIfNeeded(jobId) {
const provider = storageManager.getProvider();
const activeAccountId = storageManager.getActiveAccountId();
if (provider.name !== "google_drive" || !activeAccountId) return null;
const cooldown = await getStorageAccountCooldown(activeAccountId, provider.name, STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT);
if (!cooldown) return null;
await putJobIntoStorageCooldown(jobId, cooldown.cooldownUntil, `Google Drive \u4ECA\u65E5\u4E0A\u4F20\u989D\u5EA6\u5DF2\u8FBE\u4E0A\u9650\uFF0C\u81EA\u52A8\u6682\u505C\u5230 ${cooldown.cooldownUntil.toISOString()}`);
return cooldown;
}
async function notifyStorageCooldownOnce(botClient, job, cooldownUntil) {
const params = job.params || {};
if (params.storageQuotaNoticeSentAt) return;
const nextParams = { ...params, storageQuotaNoticeSentAt: (/* @__PURE__ */ new Date()).toISOString(), storageQuotaCooldownUntil: cooldownUntil.toISOString() };
await updateJob(job.id, { params: JSON.stringify(nextParams) });
const targetChat = job.chat_id || job.user_id;
await botClient.sendMessage(targetChat, {
message: [
formatStorageCooldownNotice(cooldownUntil),
"",
`\u4EFB\u52A1\uFF1A${String(job.id).slice(0, 8)}`
].join("\n")
}).catch(() => void 0);
}
async function handleStorageQuotaCooldownError(botClient, jobId, error) {
if (!isStorageQuotaCooldownError(error)) return false;
const cooldownUntil = error.cooldownUntil;
await markStorageAccountCooldown(error.storageAccountId || storageManager.getActiveAccountId(), error.provider, error.reason, cooldownUntil, error.message);
await putJobIntoStorageCooldown(jobId, cooldownUntil, error.message);
const job = await getJob(jobId);
if (job) await notifyStorageCooldownOnce(botClient, job, cooldownUntil);
return true;
}
async function ensureJobCanRun(jobId) {
const job = await getJob(jobId);
if (!job) return "cancelled";
if (job.cancelled_at || job.status === "cancelled") return "cancelled";
if (job.paused_at || job.status === "paused") return "paused";
if (job.cooldown_until && new Date(job.cooldown_until).getTime() > Date.now()) return "cooldown";
if (job.status === "cooling" && (!job.cooldown_until || new Date(job.cooldown_until).getTime() <= Date.now())) return "run";
const storageCooldown = await applyStorageCooldownIfNeeded(jobId);
if (storageCooldown) return "cooldown";
return "run";
}
async function waitUntilRunnable(jobId, options) {
@@ -5862,6 +6103,9 @@ async function downloadClaimedRefs(botClient, requestMessage, jobId, source, ref
for (const ref of refs) await markDownloadRefStatus(jobId, ref, "failed", `FloodWait ${flood.seconds}s`);
return { found: 0, skipped: 0, failed: refs.length, successful: 0 };
}
if (await handleStorageQuotaCooldownError(botClient, jobId, error)) {
return { found: 0, skipped: 0, failed: 0, successful: 0 };
}
for (const ref of refs) await markDownloadRefStatus(jobId, ref, "failed", error instanceof Error ? error.message : String(error));
throw error;
}
@@ -6078,7 +6322,7 @@ async function listTelegramActiveTaskQueues(userId, limit = 10) {
)
)
OR (
j.status = 'paused'
j.status IN ('paused', 'cooling')
AND (COALESCE(s.pending_count, 0) > 0 OR COALESCE(s.downloading_count, 0) > 0 OR j.scan_status = 'scanning')
)
)
@@ -6187,6 +6431,11 @@ async function runSubscriptionScan(botClient) {
const subscriptionRefs = candidateMessages.map((message) => toChannelDownloadRef(row.source, message)).filter((ref) => Boolean(ref));
await markDownloadRefsDownloading(jobId, subscriptionRefs);
const downloadResult = await downloadTelegramChannelRange(botClient, requestMessage, row.source, 0, ids.length, "newer", ids, row.folder_override || null, subscriptionRefs, (ref, status, error) => markDownloadRefStatus(jobId, ref, status, error));
const cooledJob = await getJob(jobId);
if (cooledJob?.status === "cooling") {
await notifyStorageCooldownOnce(botClient, cooledJob, new Date(cooledJob.cooldown_until));
continue;
}
await updateJob(jobId, {
status: downloadResult.failed > 0 ? "completed_with_errors" : "completed",
enqueued_count: downloadResult.found,
@@ -6261,6 +6510,11 @@ async function recoverTelegramJob(botClient, job) {
refs,
(ref, status, error) => markDownloadRefStatus(job.id, ref, status, error)
);
const cooledJob = await getJob(job.id);
if (cooledJob?.status === "cooling") {
await notifyStorageCooldownOnce(botClient, cooledJob, new Date(cooledJob.cooldown_until));
return;
}
await updateJob(job.id, {
status: result.failed > 0 ? "completed_with_errors" : "completed",
enqueued_count: result.found,
@@ -6280,6 +6534,16 @@ async function recoverInterruptedTelegramJobs(botClient) {
if (recoveryRunning) return;
recoveryRunning = true;
try {
await clearExpiredStorageCooldowns();
await query(
`UPDATE telegram_background_jobs
SET status = 'running', download_status = 'active', cooldown_until = NULL, error = NULL, updated_at = NOW()
WHERE status = 'cooling'
AND cooldown_until IS NOT NULL
AND cooldown_until <= NOW()
AND cancelled_at IS NULL
AND finished_at IS NULL`
);
await query(
`UPDATE telegram_download_items
SET status = 'pending', locked_at = NULL, updated_at = NOW()
@@ -6293,7 +6557,7 @@ async function recoverInterruptedTelegramJobs(botClient) {
WHERE j.kind IN ('date_range', 'tag_download')
AND j.finished_at IS NULL
AND j.cancelled_at IS NULL
AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors')
AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors', 'cooling')
AND i.status = 'pending'
ORDER BY j.created_at ASC
LIMIT 5`
@@ -7710,6 +7974,7 @@ import fs10 from "fs";
import path14 from "path";
import { spawn } from "child_process";
import os2 from "os";
init_storageCooldown();
var YtDlpQueue = class {
constructor(maxConcurrent) {
this.maxConcurrent = maxConcurrent;
@@ -7803,6 +8068,7 @@ async function runYtDlpDownload(url, taskDir) {
async function uploadDownloadedFile(localFilePath, originalFileName) {
const provider = storageManager.getProvider();
const activeAccountId = storageManager.getActiveAccountId();
await assertActiveStorageWritable();
const safeName = sanitizeFilename(originalFileName);
const ext = path14.extname(safeName) || path14.extname(localFilePath) || "";
const mimeType = getMimeTypeFromFilename(safeName);
@@ -7876,12 +8142,23 @@ Task: ${task.id}` });
task.status = "failed";
task.finishedAt = Date.now();
task.error = e instanceof Error ? e.message : String(e);
const errText = (task.error || "\u672A\u77E5\u9519\u8BEF").toString().trim();
const trimmed = errText.length > 1500 ? errText.slice(0, 1500) + "..." : errText;
try {
await message.reply({ message: `\u274C \u4E0B\u8F7D/\u4E0A\u4F20\u5931\u8D25
let replyText;
if (isStorageQuotaCooldownError(e)) {
await markStorageAccountCooldown(e.storageAccountId || storageManager.getActiveAccountId(), e.provider, e.reason, e.cooldownUntil, e.message);
replyText = [
formatStorageCooldownNotice(e.cooldownUntil),
"",
"yt-dlp \u4EFB\u52A1\u6CA1\u6709\u6301\u4E45\u5316\u6062\u590D\u961F\u5217\uFF1B\u8BF7\u5728\u6062\u590D\u65F6\u95F4\u540E\u91CD\u65B0\u53D1\u9001\u8BE5\u94FE\u63A5\uFF0C\u6216\u5148\u5207\u6362\u5176\u5B83\u5B58\u50A8\u6E90\u3002"
].join("\n");
} else {
const errText = (task.error || "\u672A\u77E5\u9519\u8BEF").toString().trim();
const trimmed = errText.length > 1500 ? errText.slice(0, 1500) + "..." : errText;
replyText = `\u274C \u4E0B\u8F7D/\u4E0A\u4F20\u5931\u8D25
\u539F\u56E0: ${trimmed}` });
\u539F\u56E0: ${trimmed}`;
}
try {
await message.reply({ message: replyText });
} catch {
}
} finally {
@@ -8148,7 +8425,7 @@ function isDateOnly(text) {
async function updateJobProgressMessage(statusMessage, summary) {
const totalDone = summary.completed + summary.failed + summary.skipped;
const lines = [
summary.status === "paused" ? `\u23F8\uFE0F **\u9891\u9053\u4E0B\u8F7D\u5DF2\u6682\u505C**` : summary.status === "cancelled" ? `\u{1F6D1} **\u9891\u9053\u4E0B\u8F7D\u5DF2\u53D6\u6D88**` : totalDone >= summary.totalMediaFound && summary.scanStatus === "done" ? `\u2705 **\u9891\u9053\u4EFB\u52A1\u5B8C\u6210**` : `\u{1F50E} **\u9891\u9053\u4EFB\u52A1\u8FD0\u884C\u4E2D**`,
summary.status === "paused" ? `\u23F8\uFE0F **\u9891\u9053\u4E0B\u8F7D\u5DF2\u6682\u505C**` : summary.status === "cooling" ? `\u23F8\uFE0F **Google Drive \u9650\u989D\u51B7\u5374\u4E2D**` : summary.status === "cancelled" ? `\u{1F6D1} **\u9891\u9053\u4E0B\u8F7D\u5DF2\u53D6\u6D88**` : totalDone >= summary.totalMediaFound && summary.scanStatus === "done" ? `\u2705 **\u9891\u9053\u4EFB\u52A1\u5B8C\u6210**` : `\u{1F50E} **\u9891\u9053\u4EFB\u52A1\u8FD0\u884C\u4E2D**`,
`\u{1F194} job: ${summary.jobId.slice(0, 8)}`,
`\u{1F4CD} \u9891\u9053\uFF1A${summary.source}`,
``,
@@ -8158,7 +8435,7 @@ async function updateJobProgressMessage(statusMessage, summary) {
``,
`\u2B07\uFE0F \u4E0B\u8F7D\uFF1A${summary.downloadStatus}`,
`\u2705 \u6210\u529F ${summary.completed}\u3000\u23F3 \u5F85\u4E0B\u8F7D ${summary.pending}\u3000\u{1F504} \u4E0B\u8F7D\u4E2D ${summary.downloading}\u3000\u274C \u5931\u8D25 ${summary.failed}\u3000\u23ED \u8DF3\u8FC7 ${summary.skipped}`,
summary.cooldownUntil ? `\u23F3 FloodWait \u51B7\u5374\u5230\uFF1A${summary.cooldownUntil}` : "",
summary.cooldownUntil ? summary.status === "cooling" ? `\u23F8\uFE0F Google Drive \u9650\u989D\u51B7\u5374\u5230\uFF1A${summary.cooldownUntil}` : `\u23F3 FloodWait \u51B7\u5374\u5230\uFF1A${summary.cooldownUntil}` : "",
``,
`\u63A7\u5236\uFF1A/task_pause ${summary.jobId.slice(0, 8)} \xB7 /task_resume ${summary.jobId.slice(0, 8)} \xB7 /task_cancel ${summary.jobId.slice(0, 8)}`
].filter(Boolean);
@@ -10682,6 +10959,7 @@ var handleUpload = async (req, res, source = "web") => {
console.log(`[Upload] \u{1F3E0} Local temp path: ${tempPath}`);
try {
const provider = storageManager.getProvider();
await assertActiveStorageWritable();
console.log(`[Upload] \u{1F6E0}\uFE0F Current storage provider: ${provider.name}, activeAccountId: ${activeAccountId || "none (local)"}`);
let thumbnailPath = null;
let previewPath = null;
@@ -10784,6 +11062,9 @@ var handleUpload = async (req, res, source = "web") => {
} catch (error) {
console.error("\u4E0A\u4F20\u5904\u7406\u5931\u8D25:", error);
if (fs13.existsSync(tempPath)) fs13.unlinkSync(tempPath);
if (isStorageCooldownError(error)) {
return sendStorageCooldownHttpError(res, error);
}
res.status(500).json({ error: "\u6587\u4EF6\u4E0A\u4F20\u5931\u8D25" });
}
};
@@ -11511,6 +11792,7 @@ router6.post("/complete", async (req, res) => {
let width = null;
let height = null;
const provider = storageManager.getProvider();
await assertActiveStorageWritable();
if (provider.name === "local" && (session.mimeType.startsWith("image/") || session.mimeType.startsWith("video/"))) {
try {
console.log(`[ChunkedComplete] \u{1F5BC}\uFE0F MIME: ${session.mimeType}, starting generation...`);
@@ -11585,6 +11867,9 @@ router6.post("/complete", async (req, res) => {
});
} catch (error) {
console.error("\u5B8C\u6210\u4E0A\u4F20\u5931\u8D25:", error);
if (isStorageCooldownError(error)) {
return sendStorageCooldownHttpError(res, error);
}
res.status(500).json({ error: "\u5B8C\u6210\u4E0A\u4F20\u5931\u8D25" });
}
});
+12
View File
@@ -107,6 +107,18 @@ async function initializeDatabase() {
await pool.query(`ALTER TABLE telegram_channel_subscriptions ADD COLUMN IF NOT EXISTS source_type TEXT DEFAULT 'public'`);
await pool.query(`ALTER TABLE telegram_channel_subscriptions ADD COLUMN IF NOT EXISTS disabled_reason TEXT`);
await pool.query(`ALTER TABLE telegram_channel_subscriptions ADD COLUMN IF NOT EXISTS disabled_at TIMESTAMPTZ`);
await pool.query(`CREATE TABLE IF NOT EXISTS storage_account_cooldowns (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
storage_account_id UUID REFERENCES storage_accounts(id) ON DELETE CASCADE,
provider VARCHAR(50) NOT NULL,
reason VARCHAR(100) NOT NULL,
cooldown_until TIMESTAMPTZ NOT NULL,
last_error TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(storage_account_id, provider, reason)
)`);
await pool.query(`CREATE INDEX IF NOT EXISTS idx_storage_account_cooldowns_until ON storage_account_cooldowns(cooldown_until)`);
await pool.query(`CREATE UNIQUE INDEX IF NOT EXISTS idx_api_keys_key_hash ON api_keys(key_hash) WHERE key_hash IS NOT NULL`);
console.log('✅ 数据库表结构初始化完成');
+20
View File
@@ -28,6 +28,26 @@ CREATE OR REPLACE TRIGGER storage_accounts_updated_at
FOR EACH ROW
EXECUTE FUNCTION update_updated_at();
-- 存储账户冷却表(如 Google Drive 每日上传限额触发后暂停 24 小时)
CREATE TABLE IF NOT EXISTS storage_account_cooldowns (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
storage_account_id UUID REFERENCES storage_accounts(id) ON DELETE CASCADE,
provider VARCHAR(50) NOT NULL,
reason VARCHAR(100) NOT NULL,
cooldown_until TIMESTAMPTZ NOT NULL,
last_error TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(storage_account_id, provider, reason)
);
CREATE INDEX IF NOT EXISTS idx_storage_account_cooldowns_until ON storage_account_cooldowns(cooldown_until);
CREATE OR REPLACE TRIGGER storage_account_cooldowns_updated_at
BEFORE UPDATE ON storage_account_cooldowns
FOR EACH ROW
EXECUTE FUNCTION update_updated_at();
-- 文件表
CREATE TABLE IF NOT EXISTS files (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+5
View File
@@ -6,6 +6,7 @@ import { pipeline } from 'stream/promises';
import { query } from '../db/index.js';
import { generateThumbnail, getImageDimensions, generateMediaPreview } from '../utils/thumbnail.js';
import { storageManager } from '../services/storage.js';
import { assertActiveStorageWritable, isStorageCooldownError, sendStorageCooldownHttpError } from '../services/storageCooldownGuard.js';
import { getSignedUrl } from '../middleware/signedUrl.js';
import { getUniqueStoredName } from '../utils/fileUtils.js';
import { buildStorageFolderWithRules, getStoragePathRules } from '../utils/storagePath.js';
@@ -321,6 +322,7 @@ router.post('/complete', async (req: Request, res: Response) => {
let width = null;
let height = null;
const provider = storageManager.getProvider();
await assertActiveStorageWritable();
if (provider.name === 'local' && (session.mimeType.startsWith('image/') || session.mimeType.startsWith('video/'))) {
try {
@@ -412,6 +414,9 @@ router.post('/complete', async (req: Request, res: Response) => {
});
} catch (error) {
console.error('完成上传失败:', error);
if (isStorageCooldownError(error)) {
return sendStorageCooldownHttpError(res, error);
}
res.status(500).json({ error: '完成上传失败' });
}
});
+5
View File
@@ -8,6 +8,7 @@ import { query } from '../db/index.js';
import { validateApiKey } from '../middleware/apiKey.js';
import { generateThumbnail, getImageDimensions, generateMediaPreview } from '../utils/thumbnail.js';
import { storageManager } from '../services/storage.js';
import { assertActiveStorageWritable, isStorageCooldownError, sendStorageCooldownHttpError } from '../services/storageCooldownGuard.js';
import { getSignedUrl } from '../middleware/signedUrl.js';
import { getUniqueStoredName } from '../utils/fileUtils.js';
import { buildStorageFolderWithRules, getStoragePathRules } from '../utils/storagePath.js';
@@ -97,6 +98,7 @@ const handleUpload = async (req: Request, res: Response, source: string = 'web')
try {
// 1. 获取当前存储提供商
const provider = storageManager.getProvider();
await assertActiveStorageWritable();
console.log(`[Upload] 🛠️ Current storage provider: ${provider.name}, activeAccountId: ${activeAccountId || 'none (local)'}`);
// 2. 在保存到永久存储前生成缩略图和获取尺寸
@@ -219,6 +221,9 @@ const handleUpload = async (req: Request, res: Response, source: string = 'web')
} catch (error) {
console.error('上传处理失败:', error);
if (fs.existsSync(tempPath)) fs.unlinkSync(tempPath);
if (isStorageCooldownError(error)) {
return sendStorageCooldownHttpError(res, error);
}
res.status(500).json({ error: '文件上传失败' });
}
};
@@ -0,0 +1,90 @@
import assert from 'node:assert/strict';
import { mkdtemp, writeFile, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { GoogleDriveStorageProvider, StorageQuotaCooldownError } from './storage.js';
async function main() {
const tempDir = await mkdtemp(join(tmpdir(), 'tg-vault-gdrive-quota-test-'));
const tempFile = join(tempDir, 'upload.txt');
await writeFile(tempFile, 'demo');
const provider = new GoogleDriveStorageProvider(
'account-id',
'client-id',
'client-secret',
'refresh-token',
'http://localhost/callback'
);
(provider as any).oauth2Client = {
getAccessToken: async () => ({ token: 'access-token', res: { data: { expiry_date: Date.now() + 3600_000 } } })
};
(provider as any).drive = {
files: {
list: async () => ({ data: { files: [{ id: 'tg-vault-folder-id' }] } }),
create: async () => {
const error: any = new Error('The user has exceeded their Drive upload limit. Please try again later.');
error.code = 403;
error.response = {
status: 403,
data: {
error: {
status: 'PERMISSION_DENIED',
errors: [{ reason: 'dailyLimitExceeded', message: 'The user has exceeded their Drive upload limit' }]
}
}
};
throw error;
},
},
};
try {
await assert.rejects(
() => provider.saveFile(tempFile, 'demo.txt', 'text/plain', null),
(error: unknown) => {
assert.ok(error instanceof StorageQuotaCooldownError, 'daily upload quota errors should use StorageQuotaCooldownError');
assert.equal((error as StorageQuotaCooldownError).provider, 'google_drive');
assert.equal((error as StorageQuotaCooldownError).reason, 'daily_upload_limit');
assert.equal((error as StorageQuotaCooldownError).storageAccountId, 'account-id');
assert.ok((error as StorageQuotaCooldownError).cooldownUntil.getTime() > Date.now() + 23 * 60 * 60 * 1000);
return true;
}
);
(provider as any).folderIdCache.clear();
(provider as any).drive.files.create = async () => {
const error: any = new Error('User rate limit exceeded');
error.code = 403;
error.response = {
status: 403,
data: {
error: {
status: 'RESOURCE_EXHAUSTED',
errors: [{ reason: 'userRateLimitExceeded', message: 'User rate limit exceeded' }]
}
}
};
throw error;
};
await assert.rejects(
() => provider.saveFile(tempFile, 'demo.txt', 'text/plain', null),
(error: unknown) => {
assert.ok(!(error instanceof StorageQuotaCooldownError), 'generic API rate limits must not pause uploads for 24 hours');
assert.match((error as Error).message, /Google Drive upload failed/);
return true;
}
);
} finally {
await rm(tempDir, { recursive: true, force: true });
}
console.log('google drive quota cooldown error ok');
}
main().catch(error => {
console.error(error);
process.exit(1);
});
+51
View File
@@ -18,6 +18,49 @@ import {
isSensitiveSettingKey,
storageConfigNeedsEncryption,
} from '../utils/credentialCrypto.js';
import { DEFAULT_STORAGE_COOLDOWN_MS, STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT } from './storageCooldown.js';
export class StorageQuotaCooldownError extends Error {
provider: string;
reason: string;
storageAccountId?: string;
cooldownUntil: Date;
originalError?: unknown;
constructor(message: string, options: { provider: string; reason: string; storageAccountId?: string; cooldownUntil?: Date; originalError?: unknown }) {
super(message);
this.name = 'StorageQuotaCooldownError';
this.provider = options.provider;
this.reason = options.reason;
this.storageAccountId = options.storageAccountId;
this.cooldownUntil = options.cooldownUntil || new Date(Date.now() + DEFAULT_STORAGE_COOLDOWN_MS);
this.originalError = options.originalError;
}
}
export function isStorageQuotaCooldownError(error: unknown): error is StorageQuotaCooldownError {
return error instanceof StorageQuotaCooldownError || (error as any)?.name === 'StorageQuotaCooldownError';
}
function googleDriveErrorText(error: any): string {
const pieces = [
error?.message,
error?.errorMessage,
error?.response?.data ? JSON.stringify(error.response.data) : undefined,
error?.errors ? JSON.stringify(error.errors) : undefined,
].filter(Boolean);
return pieces.join(' ');
}
function isGoogleDriveDailyUploadLimitError(error: any): boolean {
const status = Number(error?.code || error?.response?.status || error?.status || 0);
const text = googleDriveErrorText(error);
if (status !== 403) return false;
// Treat only explicit Drive/upload-limit responses as the 24h daily upload cap.
// Generic userRateLimitExceeded/rateLimitExceeded/429 errors are usually short-lived API throttles
// and must not pause background transfers for a full day.
return /Drive upload limit|exceeded their Drive upload limit|exceeded.*upload limit|upload limit exceeded|upload limit/i.test(text);
}
// 接口定义
export interface IStorageProvider {
@@ -1050,6 +1093,14 @@ export class GoogleDriveStorageProvider implements IStorageProvider {
return file.data.id;
} catch (error: any) {
console.error('[GoogleDrive] Upload failed:', error.message);
if (isGoogleDriveDailyUploadLimitError(error)) {
throw new StorageQuotaCooldownError('Google Drive 今日上传额度已达上限,任务将自动暂停 24 小时后继续。', {
provider: 'google_drive',
reason: STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT,
storageAccountId: this.id,
originalError: error,
});
}
throw new Error(`Google Drive upload failed: ${error.message}`);
}
}
+85
View File
@@ -0,0 +1,85 @@
import { query } from '../db/index.js';
export const STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT = 'daily_upload_limit';
export const DEFAULT_STORAGE_COOLDOWN_MS = 24 * 60 * 60 * 1000;
export interface StorageAccountCooldown {
storageAccountId: string;
provider: string;
reason: string;
cooldownUntil: Date;
lastError?: string | null;
}
export async function ensureStorageCooldownSchema(): Promise<void> {
await query(`
CREATE TABLE IF NOT EXISTS storage_account_cooldowns (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
storage_account_id UUID REFERENCES storage_accounts(id) ON DELETE CASCADE,
provider VARCHAR(50) NOT NULL,
reason VARCHAR(100) NOT NULL,
cooldown_until TIMESTAMPTZ NOT NULL,
last_error TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
updated_at TIMESTAMPTZ DEFAULT NOW(),
UNIQUE(storage_account_id, provider, reason)
)
`);
await query(`CREATE INDEX IF NOT EXISTS idx_storage_account_cooldowns_until ON storage_account_cooldowns(cooldown_until)`);
}
export async function markStorageAccountCooldown(
storageAccountId: string | null | undefined,
provider: string,
reason: string,
cooldownUntil: Date,
error?: string | null,
): Promise<void> {
if (!storageAccountId) return;
await ensureStorageCooldownSchema();
await query(
`INSERT INTO storage_account_cooldowns (storage_account_id, provider, reason, cooldown_until, last_error, updated_at)
VALUES ($1, $2, $3, $4, $5, NOW())
ON CONFLICT (storage_account_id, provider, reason)
DO UPDATE SET cooldown_until = EXCLUDED.cooldown_until, last_error = EXCLUDED.last_error, updated_at = NOW()`,
[storageAccountId, provider, reason, cooldownUntil, error || null],
);
}
export async function getStorageAccountCooldown(
storageAccountId: string | null | undefined,
provider: string,
reason = STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT,
): Promise<StorageAccountCooldown | null> {
if (!storageAccountId) return null;
await ensureStorageCooldownSchema();
const result = await query(
`SELECT storage_account_id, provider, reason, cooldown_until, last_error
FROM storage_account_cooldowns
WHERE storage_account_id = $1
AND provider = $2
AND reason = $3
AND cooldown_until > NOW()
LIMIT 1`,
[storageAccountId, provider, reason],
);
const row = result.rows[0];
if (!row) return null;
return {
storageAccountId: row.storage_account_id,
provider: row.provider,
reason: row.reason,
cooldownUntil: new Date(row.cooldown_until),
lastError: row.last_error,
};
}
export async function isStorageAccountCooling(storageAccountId: string | null | undefined, provider: string): Promise<boolean> {
return Boolean(await getStorageAccountCooldown(storageAccountId, provider));
}
export async function clearExpiredStorageCooldowns(): Promise<number> {
await ensureStorageCooldownSchema();
const result = await query(`DELETE FROM storage_account_cooldowns WHERE cooldown_until <= NOW()`);
return result.rowCount || 0;
}
@@ -0,0 +1,34 @@
import assert from 'node:assert/strict';
import { StorageQuotaCooldownError } from './storage.js';
import { buildStorageCooldownHttpError, formatStorageCooldownNotice } from './storageCooldownGuard.js';
async function main() {
const cooldownUntil = new Date('2030-01-02T03:04:05.000Z');
const error = new StorageQuotaCooldownError('Google Drive 今日上传额度已达上限,任务将自动暂停 24 小时后继续。', {
provider: 'google_drive',
reason: 'daily_upload_limit',
storageAccountId: 'account-id',
cooldownUntil,
});
const payload = buildStorageCooldownHttpError(error);
assert.equal(payload.status, 429);
assert.equal(payload.body.code, 'storage_account_cooling');
assert.equal(payload.body.provider, 'google_drive');
assert.equal(payload.body.reason, 'daily_upload_limit');
assert.equal(payload.body.retryAt, cooldownUntil.toISOString());
assert.match(payload.body.error, /Google Drive/);
assert.match(payload.body.error, /24 小时/);
const notice = formatStorageCooldownNotice(cooldownUntil);
assert.match(notice, /Google Drive 今日上传额度已达上限/);
assert.match(notice, /2030-01-02T03:04:05.000Z/);
assert.match(notice, /剩余文件不会丢失/);
console.log('storage cooldown guard formatting ok');
}
main().catch(error => {
console.error(error);
process.exit(1);
});
@@ -0,0 +1,66 @@
import { Response } from 'express';
import { storageManager, StorageQuotaCooldownError, isStorageQuotaCooldownError } from './storage.js';
import {
getStorageAccountCooldown,
STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT,
type StorageAccountCooldown,
} from './storageCooldown.js';
export interface StorageCooldownHttpErrorBody {
error: string;
code: 'storage_account_cooling';
provider: string;
reason: string;
retryAt: string;
}
export function formatStorageCooldownNotice(cooldownUntil: Date): string {
return [
'⏸️ Google Drive 今日上传额度已达上限',
'',
'当前任务已自动暂停,预计 24 小时后继续。',
'剩余文件不会丢失,恢复后会从未完成部分继续处理。',
'',
`恢复时间:${cooldownUntil.toISOString()}`,
].join('\n');
}
export function buildStorageCooldownHttpError(error: StorageQuotaCooldownError): { status: number; body: StorageCooldownHttpErrorBody } {
return {
status: 429,
body: {
error: error.message || 'Google Drive 今日上传额度已达上限,请稍后重试。',
code: 'storage_account_cooling',
provider: error.provider,
reason: error.reason,
retryAt: error.cooldownUntil.toISOString(),
},
};
}
export function sendStorageCooldownHttpError(res: Response, error: StorageQuotaCooldownError): void {
const payload = buildStorageCooldownHttpError(error);
res.status(payload.status).json(payload.body);
}
export async function getActiveStorageCooldown(): Promise<StorageAccountCooldown | null> {
const provider = storageManager.getProvider();
const activeAccountId = storageManager.getActiveAccountId();
if (provider.name !== 'google_drive' || !activeAccountId) return null;
return getStorageAccountCooldown(activeAccountId, provider.name, STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT);
}
export async function assertActiveStorageWritable(): Promise<void> {
const cooldown = await getActiveStorageCooldown();
if (!cooldown) return;
throw new StorageQuotaCooldownError('Google Drive 今日上传额度已达上限,请等待自动恢复后再上传,或临时切换其它存储源。', {
provider: cooldown.provider,
reason: cooldown.reason,
storageAccountId: cooldown.storageAccountId,
cooldownUntil: cooldown.cooldownUntil,
});
}
export function isStorageCooldownError(error: unknown): error is StorageQuotaCooldownError {
return isStorageQuotaCooldownError(error);
}
+14 -2
View File
@@ -302,7 +302,15 @@ interface TelegramDownloadScanSummary {
async function updateJobProgressMessage(statusMessage: Api.Message, summary: TelegramJobProgressSummary): Promise<void> {
const totalDone = summary.completed + summary.failed + summary.skipped;
const lines = [
summary.status === 'paused' ? `⏸️ **频道下载已暂停**` : summary.status === 'cancelled' ? `🛑 **频道下载已取消**` : totalDone >= summary.totalMediaFound && summary.scanStatus === 'done' ? `✅ **频道任务完成**` : `🔎 **频道任务运行中**`,
summary.status === 'paused'
? `⏸️ **频道下载已暂停**`
: summary.status === 'cooling'
? `⏸️ **Google Drive 限额冷却中**`
: summary.status === 'cancelled'
? `🛑 **频道下载已取消**`
: totalDone >= summary.totalMediaFound && summary.scanStatus === 'done'
? `✅ **频道任务完成**`
: `🔎 **频道任务运行中**`,
`🆔 job: ${summary.jobId.slice(0, 8)}`,
`📍 频道:${summary.source}`,
``,
@@ -312,7 +320,11 @@ async function updateJobProgressMessage(statusMessage: Api.Message, summary: Tel
``,
`⬇️ 下载:${summary.downloadStatus}`,
`✅ 成功 ${summary.completed} ⏳ 待下载 ${summary.pending} 🔄 下载中 ${summary.downloading} ❌ 失败 ${summary.failed} ⏭ 跳过 ${summary.skipped}`,
summary.cooldownUntil ? `⏳ FloodWait 冷却到:${summary.cooldownUntil}` : '',
summary.cooldownUntil
? (summary.status === 'cooling'
? `⏸️ Google Drive 限额冷却到:${summary.cooldownUntil}`
: `⏳ FloodWait 冷却到:${summary.cooldownUntil}`)
: '',
``,
`控制:/task_pause ${summary.jobId.slice(0, 8)} · /task_resume ${summary.jobId.slice(0, 8)} · /task_cancel ${summary.jobId.slice(0, 8)}`,
].filter(Boolean);
+91 -4
View File
@@ -1,7 +1,15 @@
import { Api, TelegramClient } from 'telegram';
import { getPeerId } from 'telegram/Utils.js';
import { query } from '../db/index.js';
import { storageManager } from './storage.js';
import { storageManager, isStorageQuotaCooldownError } from './storage.js';
import { formatStorageCooldownNotice } from './storageCooldownGuard.js';
import {
clearExpiredStorageCooldowns,
getStorageAccountCooldown,
markStorageAccountCooldown,
STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT,
type StorageAccountCooldown,
} from './storageCooldown.js';
import { getTelegramUserClient, isTelegramUserClientReady } from './telegramUserClient.js';
import { downloadTelegramChannelRange, getTelegramDownloadPreview, type TelegramDownloadMessageRef } from './telegramUpload.js';
import { getSetting } from '../utils/settings.js';
@@ -685,12 +693,68 @@ function isFloodWait(error: unknown): { seconds: number } | null {
return null;
}
async function ensureJobCanRun(jobId: string): Promise<'run' | 'paused' | 'cancelled' | 'cooldown'> {
type TelegramJobControlState = 'run' | 'paused' | 'cancelled' | 'cooldown';
async function putJobIntoStorageCooldown(jobId: string, cooldownUntil: Date, reasonText: string): Promise<void> {
await updateJob(jobId, {
status: 'cooling',
download_status: 'cooling',
cooldown_until: cooldownUntil,
error: reasonText,
});
await query(
`UPDATE telegram_download_items
SET status = 'pending', locked_at = NULL, last_error = $2, updated_at = NOW()
WHERE job_id = $1
AND status = 'downloading'`,
[jobId, reasonText],
);
}
async function applyStorageCooldownIfNeeded(jobId: string): Promise<StorageAccountCooldown | null> {
const provider = storageManager.getProvider();
const activeAccountId = storageManager.getActiveAccountId();
if (provider.name !== 'google_drive' || !activeAccountId) return null;
const cooldown = await getStorageAccountCooldown(activeAccountId, provider.name, STORAGE_COOLDOWN_REASON_DAILY_UPLOAD_LIMIT);
if (!cooldown) return null;
await putJobIntoStorageCooldown(jobId, cooldown.cooldownUntil, `Google Drive 今日上传额度已达上限,自动暂停到 ${cooldown.cooldownUntil.toISOString()}`);
return cooldown;
}
async function notifyStorageCooldownOnce(botClient: TelegramClient, job: any, cooldownUntil: Date): Promise<void> {
const params = job.params || {};
if (params.storageQuotaNoticeSentAt) return;
const nextParams = { ...params, storageQuotaNoticeSentAt: new Date().toISOString(), storageQuotaCooldownUntil: cooldownUntil.toISOString() };
await updateJob(job.id, { params: JSON.stringify(nextParams) });
const targetChat = job.chat_id || job.user_id;
await botClient.sendMessage(targetChat, {
message: [
formatStorageCooldownNotice(cooldownUntil),
'',
`任务:${String(job.id).slice(0, 8)}`,
].join('\n'),
}).catch(() => undefined);
}
async function handleStorageQuotaCooldownError(botClient: TelegramClient, jobId: string, error: unknown): Promise<boolean> {
if (!isStorageQuotaCooldownError(error)) return false;
const cooldownUntil = error.cooldownUntil;
await markStorageAccountCooldown(error.storageAccountId || storageManager.getActiveAccountId(), error.provider, error.reason, cooldownUntil, error.message);
await putJobIntoStorageCooldown(jobId, cooldownUntil, error.message);
const job = await getJob(jobId);
if (job) await notifyStorageCooldownOnce(botClient, job, cooldownUntil);
return true;
}
async function ensureJobCanRun(jobId: string): Promise<TelegramJobControlState> {
const job = await getJob(jobId);
if (!job) return 'cancelled';
if (job.cancelled_at || job.status === 'cancelled') return 'cancelled';
if (job.paused_at || job.status === 'paused') return 'paused';
if (job.cooldown_until && new Date(job.cooldown_until).getTime() > Date.now()) return 'cooldown';
if (job.status === 'cooling' && (!job.cooldown_until || new Date(job.cooldown_until).getTime() <= Date.now())) return 'run';
const storageCooldown = await applyStorageCooldownIfNeeded(jobId);
if (storageCooldown) return 'cooldown';
return 'run';
}
@@ -750,6 +814,9 @@ async function downloadClaimedRefs(botClient: TelegramClient, requestMessage: Ap
for (const ref of refs) await markDownloadRefStatus(jobId, ref, 'failed', `FloodWait ${flood.seconds}s`);
return { found: 0, skipped: 0, failed: refs.length, successful: 0 };
}
if (await handleStorageQuotaCooldownError(botClient, jobId, error)) {
return { found: 0, skipped: 0, failed: 0, successful: 0 };
}
for (const ref of refs) await markDownloadRefStatus(jobId, ref, 'failed', error instanceof Error ? error.message : String(error));
throw error;
}
@@ -998,7 +1065,7 @@ export async function listTelegramActiveTaskQueues(userId: number, limit = 10) {
)
)
OR (
j.status = 'paused'
j.status IN ('paused', 'cooling')
AND (COALESCE(s.pending_count, 0) > 0 OR COALESCE(s.downloading_count, 0) > 0 OR j.scan_status = 'scanning')
)
)
@@ -1122,6 +1189,11 @@ async function runSubscriptionScan(botClient: TelegramClient) {
.filter((ref): ref is TelegramDownloadMessageRef => Boolean(ref));
await markDownloadRefsDownloading(jobId, subscriptionRefs);
const downloadResult = await downloadTelegramChannelRange(botClient, requestMessage, row.source, 0, ids.length, 'newer', ids, row.folder_override || null, subscriptionRefs, (ref, status, error) => markDownloadRefStatus(jobId, ref, status, error));
const cooledJob = await getJob(jobId);
if (cooledJob?.status === 'cooling') {
await notifyStorageCooldownOnce(botClient, cooledJob, new Date(cooledJob.cooldown_until));
continue;
}
await updateJob(jobId, {
status: downloadResult.failed > 0 ? 'completed_with_errors' : 'completed',
enqueued_count: downloadResult.found,
@@ -1201,6 +1273,11 @@ async function recoverTelegramJob(botClient: TelegramClient, job: any): Promise<
refs,
(ref, status, error) => markDownloadRefStatus(job.id, ref, status, error),
);
const cooledJob = await getJob(job.id);
if (cooledJob?.status === 'cooling') {
await notifyStorageCooldownOnce(botClient, cooledJob, new Date(cooledJob.cooldown_until));
return;
}
await updateJob(job.id, {
status: result.failed > 0 ? 'completed_with_errors' : 'completed',
enqueued_count: result.found,
@@ -1221,6 +1298,16 @@ export async function recoverInterruptedTelegramJobs(botClient: TelegramClient):
if (recoveryRunning) return;
recoveryRunning = true;
try {
await clearExpiredStorageCooldowns();
await query(
`UPDATE telegram_background_jobs
SET status = 'running', download_status = 'active', cooldown_until = NULL, error = NULL, updated_at = NOW()
WHERE status = 'cooling'
AND cooldown_until IS NOT NULL
AND cooldown_until <= NOW()
AND cancelled_at IS NULL
AND finished_at IS NULL`
);
await query(
`UPDATE telegram_download_items
SET status = 'pending', locked_at = NULL, updated_at = NOW()
@@ -1234,7 +1321,7 @@ export async function recoverInterruptedTelegramJobs(botClient: TelegramClient):
WHERE j.kind IN ('date_range', 'tag_download')
AND j.finished_at IS NULL
AND j.cancelled_at IS NULL
AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors')
AND j.status IN ('pending', 'running', 'failed', 'completed_with_errors', 'cooling')
AND i.status = 'pending'
ORDER BY j.created_at ASC
LIMIT 5`
+20 -3
View File
@@ -6,7 +6,9 @@ import crypto from 'crypto';
import bigInt from 'big-integer';
import { query } from '../db/index.js';
import { generateThumbnail, getImageDimensions } from '../utils/thumbnail.js';
import { storageManager } from '../services/storage.js';
import { storageManager, isStorageQuotaCooldownError, StorageQuotaCooldownError } from '../services/storage.js';
import { assertActiveStorageWritable, formatStorageCooldownNotice } from './storageCooldownGuard.js';
import { markStorageAccountCooldown } from './storageCooldown.js';
import { getTelegramUserClient, isTelegramUserClientReady } from './telegramUserClient.js';
import { getSetting } from '../utils/settings.js';
import { isAuthenticatedAsync } from './telegramState.js';
@@ -1241,6 +1243,7 @@ interface FileUploadItem {
size?: number;
fileType?: string;
error?: string;
storageCooldownUntil?: Date;
retried?: boolean; // 是否已重试过
targetDir?: string; // 批量上传时文件的目标目录
folderOverride?: string | null; // 任务级保存目录覆盖,不影响全局路径规则
@@ -1415,6 +1418,7 @@ async function processFileUpload(client: TelegramClient, file: FileUploadItem, q
try {
const activeAccountId = storageManager.getActiveAccountId();
await assertActiveStorageWritable();
const chatName = await getTelegramChatName(file.message);
const batchFolder = null;
const storageRules = await getStoragePathRules();
@@ -1510,7 +1514,13 @@ async function processFileUpload(client: TelegramClient, file: FileUploadItem, q
} catch (error) {
console.error('🤖 文件上传失败:', error);
file.error = (error as Error).message;
if (isStorageQuotaCooldownError(error)) {
await markStorageAccountCooldown(error.storageAccountId || storageManager.getActiveAccountId(), error.provider, error.reason, error.cooldownUntil, error.message);
file.storageCooldownUntil = error.cooldownUntil;
file.error = formatStorageCooldownNotice(error.cooldownUntil);
} else {
file.error = (error as Error).message;
}
// 立即清理本地临时文件
if (localFilePath && fs.existsSync(localFilePath)) {
try {
@@ -1531,7 +1541,7 @@ async function processFileUpload(client: TelegramClient, file: FileUploadItem, q
const firstAttemptSuccess = await attemptUpload(signal);
if (!firstAttemptSuccess && !signal.aborted && !file.retried) {
if (!firstAttemptSuccess && !signal.aborted && !file.retried && !file.storageCooldownUntil) {
file.retried = true;
file.status = 'uploading'; // 保持 uploading 状态供外部显示
file.error = undefined;
@@ -2037,6 +2047,13 @@ export async function downloadTelegramChannelRange(
successful += 1;
successfulMessageIds.push(item.id);
await onItemSettled?.(item, 'success');
} else if (uploadItem.storageCooldownUntil) {
throw new StorageQuotaCooldownError(uploadItem.error || 'Google Drive 今日上传额度已达上限,任务将自动暂停 24 小时后继续。', {
provider: 'google_drive',
reason: 'daily_upload_limit',
storageAccountId: storageManager.getActiveAccountId() || undefined,
cooldownUntil: uploadItem.storageCooldownUntil,
});
} else {
failed += 1;
failedMessageIds.push(item.id);
+18 -4
View File
@@ -4,7 +4,9 @@ import path from 'path';
import { spawn } from 'child_process';
import os from 'os';
import { query } from '../db/index.js';
import { storageManager } from './storage.js';
import { storageManager, isStorageQuotaCooldownError } from './storage.js';
import { assertActiveStorageWritable, formatStorageCooldownNotice } from './storageCooldownGuard.js';
import { markStorageAccountCooldown } from './storageCooldown.js';
import { formatBytes, getFileType, getMimeTypeFromFilename, sanitizeFilename } from '../utils/telegramUtils.js';
import { generateThumbnail, getImageDimensions } from '../utils/thumbnail.js';
import { getUniqueStoredName } from '../utils/fileUtils.js';
@@ -136,6 +138,7 @@ async function runYtDlpDownload(url: string, taskDir: string): Promise<void> {
async function uploadDownloadedFile(localFilePath: string, originalFileName: string): Promise<{ finalPath: string; providerName: string; size: number; storedName: string; folder: string }> {
const provider = storageManager.getProvider();
const activeAccountId = storageManager.getActiveAccountId();
await assertActiveStorageWritable();
const safeName = sanitizeFilename(originalFileName);
const ext = path.extname(safeName) || path.extname(localFilePath) || '';
@@ -226,11 +229,22 @@ export async function handleYtDlpCommand(message: Api.Message, url: string): Pro
task.finishedAt = Date.now();
task.error = (e instanceof Error) ? e.message : String(e);
const errText = (task.error || '未知错误').toString().trim();
const trimmed = errText.length > 1500 ? errText.slice(0, 1500) + '...' : errText;
let replyText: string;
if (isStorageQuotaCooldownError(e)) {
await markStorageAccountCooldown(e.storageAccountId || storageManager.getActiveAccountId(), e.provider, e.reason, e.cooldownUntil, e.message);
replyText = [
formatStorageCooldownNotice(e.cooldownUntil),
'',
'yt-dlp 任务没有持久化恢复队列;请在恢复时间后重新发送该链接,或先切换其它存储源。',
].join('\n');
} else {
const errText = (task.error || '未知错误').toString().trim();
const trimmed = errText.length > 1500 ? errText.slice(0, 1500) + '...' : errText;
replyText = `❌ 下载/上传失败\n\n原因: ${trimmed}`;
}
try {
await message.reply({ message: `❌ 下载/上传失败\n\n原因: ${trimmed}` });
await message.reply({ message: replyText });
} catch {
}
} finally {