diff --git a/backend/dist/index.js b/backend/dist/index.js index 5fa2e9c..1b637b4 100644 --- a/backend/dist/index.js +++ b/backend/dist/index.js @@ -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" }); } }); diff --git a/backend/src/db/index.ts b/backend/src/db/index.ts index ac0c11c..05b38b9 100644 --- a/backend/src/db/index.ts +++ b/backend/src/db/index.ts @@ -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('✅ 数据库表结构初始化完成'); diff --git a/backend/src/db/schema.sql b/backend/src/db/schema.sql index 948f6bd..c5dd059 100644 --- a/backend/src/db/schema.sql +++ b/backend/src/db/schema.sql @@ -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(), diff --git a/backend/src/routes/chunkedUpload.ts b/backend/src/routes/chunkedUpload.ts index 8b26903..087ecd3 100644 --- a/backend/src/routes/chunkedUpload.ts +++ b/backend/src/routes/chunkedUpload.ts @@ -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: '完成上传失败' }); } }); diff --git a/backend/src/routes/upload.ts b/backend/src/routes/upload.ts index c568c4a..6b19b72 100644 --- a/backend/src/routes/upload.ts +++ b/backend/src/routes/upload.ts @@ -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: '文件上传失败' }); } }; diff --git a/backend/src/services/googleDriveQuotaCooldown.test.ts b/backend/src/services/googleDriveQuotaCooldown.test.ts new file mode 100644 index 0000000..66de4ca --- /dev/null +++ b/backend/src/services/googleDriveQuotaCooldown.test.ts @@ -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); +}); diff --git a/backend/src/services/storage.ts b/backend/src/services/storage.ts index 86dbda2..3775323 100644 --- a/backend/src/services/storage.ts +++ b/backend/src/services/storage.ts @@ -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}`); } } diff --git a/backend/src/services/storageCooldown.ts b/backend/src/services/storageCooldown.ts new file mode 100644 index 0000000..2bfc804 --- /dev/null +++ b/backend/src/services/storageCooldown.ts @@ -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 { + 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 { + 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 { + 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 { + return Boolean(await getStorageAccountCooldown(storageAccountId, provider)); +} + +export async function clearExpiredStorageCooldowns(): Promise { + await ensureStorageCooldownSchema(); + const result = await query(`DELETE FROM storage_account_cooldowns WHERE cooldown_until <= NOW()`); + return result.rowCount || 0; +} diff --git a/backend/src/services/storageCooldownGuard.test.ts b/backend/src/services/storageCooldownGuard.test.ts new file mode 100644 index 0000000..9d3a751 --- /dev/null +++ b/backend/src/services/storageCooldownGuard.test.ts @@ -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); +}); diff --git a/backend/src/services/storageCooldownGuard.ts b/backend/src/services/storageCooldownGuard.ts new file mode 100644 index 0000000..e85850b --- /dev/null +++ b/backend/src/services/storageCooldownGuard.ts @@ -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 { + 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 { + 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); +} diff --git a/backend/src/services/telegramBot.ts b/backend/src/services/telegramBot.ts index 2e30f09..2045849 100644 --- a/backend/src/services/telegramBot.ts +++ b/backend/src/services/telegramBot.ts @@ -302,7 +302,15 @@ interface TelegramDownloadScanSummary { async function updateJobProgressMessage(statusMessage: Api.Message, summary: TelegramJobProgressSummary): Promise { 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); diff --git a/backend/src/services/telegramChannelJobs.ts b/backend/src/services/telegramChannelJobs.ts index d5d4f3f..0293f92 100644 --- a/backend/src/services/telegramChannelJobs.ts +++ b/backend/src/services/telegramChannelJobs.ts @@ -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 { + 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 { + 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 { + 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 { + 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 { 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` diff --git a/backend/src/services/telegramUpload.ts b/backend/src/services/telegramUpload.ts index f22daf3..304ba5f 100644 --- a/backend/src/services/telegramUpload.ts +++ b/backend/src/services/telegramUpload.ts @@ -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); diff --git a/backend/src/services/ytDlpDownload.ts b/backend/src/services/ytDlpDownload.ts index edde847..a8bdca7 100644 --- a/backend/src/services/ytDlpDownload.ts +++ b/backend/src/services/ytDlpDownload.ts @@ -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 { 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 {