From fa51a3455f6175d2400ef40ddce842ae514dcbc1 Mon Sep 17 00:00:00 2001 From: liangxu Date: Sat, 30 May 2026 19:52:17 +0800 Subject: [PATCH] fix(publish): prevent stuck publish queue and duplicate posting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The desktop publish queue could stall for a long time and end users assumed the software was broken. Root cause: a hung adapter held its execution slot with no wall-clock timeout while auto-renewing its lease forever, so the server never reclaimed it and every queued task behind it stayed 等待发布. Auto-recovery also risked silently re-posting a non-idempotent article. Client (Electron): - per-task wall-clock deadline + abort; progress-gated lease renewal that stops and aborts a stalled task instead of renewing it forever - decouple the concurrency cap from CDP-induced CPU/memory pressure (admission gate instead of self-throttling collapse); 15s watchdog pump - all adapter network I/O now has fetch timeouts and honors context.signal; bounded image-upload concurrency with per-image timeout - surface live adapter progress, elapsed time, queue position and a working-vs-queued distinction in the publish view Server (tenant-api): - publish lease-recovery worker (every 3m) + supporting index: reclaim expired in_progress publish leases, requeue (<3 attempts) or terminal-fail - 3-minute lease TTL with client-presence-gated extension; max 3 attempts Idempotency (production-grade core): - durable publish_submit_started_at marker, set before the irreversible platform submit POST; recovery and abort route a maybe-submitted task to unknown (manual reconcile, kept in the dedup set) instead of re-posting - desktop UI requires explicit confirmation before retrying a possibly- already-published task Verified: go build/vet/test (incl. resolvePublishRecoveryOutcome), vue-tsc, vitest 141/141, gofmt all clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 1 + .../src/main/adapters/baijiahao.ts | 37 +- .../src/main/adapters/bilibili.ts | 24 +- .../src/main/adapters/common.ts | 240 +++++++++- .../src/main/adapters/dongchedi.ts | 83 +++- .../src/main/adapters/jianshu.ts | 9 +- .../src/main/adapters/juejin.ts | 104 +++-- .../src/main/adapters/media-image.ts | 15 +- .../src/main/adapters/qiehao.ts | 72 +-- .../desktop-client/src/main/adapters/smzdm.ts | 75 ++- .../src/main/adapters/sohuhao.ts | 56 ++- .../src/main/adapters/toutiao.test.ts | 4 +- .../src/main/adapters/toutiao.ts | 15 +- .../src/main/adapters/wangyihao.ts | 123 +++-- .../src/main/adapters/weixin-gzh.ts | 128 ++++-- .../desktop-client/src/main/adapters/zhihu.ts | 8 +- apps/desktop-client/src/main/adapters/zol.ts | 23 +- apps/desktop-client/src/main/bootstrap.ts | 4 + .../src/main/runtime-controller.ts | 332 +++++++++++--- .../desktop-client/src/main/runtime-events.ts | 2 +- .../src/main/transport/api-client.ts | 10 + apps/desktop-client/src/preload/bridge.ts | 7 +- apps/desktop-client/src/renderer/env.d.ts | 3 +- apps/desktop-client/src/renderer/types.ts | 1 + .../renderer/views/PublishManagementView.vue | 231 +++++++++- server/cmd/scheduler/main.go | 10 + .../publish_lease_recovery_worker.go | 114 +++++ .../tenant/app/desktop_task_service.go | 432 +++++++++++++++++- .../tenant/app/desktop_task_service_test.go | 73 +++ .../repository/generated/desktop_task.sql.go | 18 +- .../generated/desktop_task_sql_test.go | 25 + .../tenant/repository/generated/models.go | 126 +++++ .../repository/queries/desktop_task.sql | 18 +- .../tenant/transport/desktop_task_handler.go | 20 + server/internal/tenant/transport/router.go | 1 + ...publish_task_lease_recovery_index.down.sql | 1 + ...d_publish_task_lease_recovery_index.up.sql | 5 + ...esktop_task_publish_submit_marker.down.sql | 2 + ..._desktop_task_publish_submit_marker.up.sql | 6 + ...lish_lease_recovery_scheduler_job.down.sql | 8 + ...ublish_lease_recovery_scheduler_job.up.sql | 16 + 41 files changed, 2124 insertions(+), 358 deletions(-) create mode 100644 server/internal/scheduler/publish_lease_recovery_worker.go create mode 100644 server/migrations/20260530100000_add_publish_task_lease_recovery_index.down.sql create mode 100644 server/migrations/20260530100000_add_publish_task_lease_recovery_index.up.sql create mode 100644 server/migrations/20260530101000_add_desktop_task_publish_submit_marker.down.sql create mode 100644 server/migrations/20260530101000_add_desktop_task_publish_submit_marker.up.sql create mode 100644 server/migrations_ops/20260530100000_add_publish_lease_recovery_scheduler_job.down.sql create mode 100644 server/migrations_ops/20260530100000_add_publish_lease_recovery_scheduler_job.up.sql diff --git a/.gitignore b/.gitignore index 98d4d3e..54c633b 100644 --- a/.gitignore +++ b/.gitignore @@ -29,3 +29,4 @@ apps/*/stats.html ops-api tmp/* *.pem +outputs/* diff --git a/apps/desktop-client/src/main/adapters/baijiahao.ts b/apps/desktop-client/src/main/adapters/baijiahao.ts index 33ec1f2..7fb2da2 100644 --- a/apps/desktop-client/src/main/adapters/baijiahao.ts +++ b/apps/desktop-client/src/main/adapters/baijiahao.ts @@ -11,6 +11,7 @@ import { PUBLISH_IMAGE_FETCH_FAILED_MESSAGE } from '../../shared/publisher-error import { resolveDesktopApiURL } from '../transport/api-client' import type { PublishAdapter, PublishAdapterContext } from './base' import { + adapterFetch, normalizeArticleHtml, sessionFetchJson, sessionFetchText, @@ -149,19 +150,25 @@ async function fetchAppInfo(context: PublishAdapterContext): Promise null) return response?.data?.user ?? null } async function getPublishToken(context: PublishAdapterContext): Promise { - const html = await sessionFetchText(context.session, BAIJIAHAO_EDIT_URL, { - credentials: 'include', - headers: baijiahaoHeaders({ - accept: - 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8', - }), - }) + const html = await sessionFetchText( + context.session, + BAIJIAHAO_EDIT_URL, + { + credentials: 'include', + headers: baijiahaoHeaders({ + accept: + 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8', + }), + }, + { signal: context.signal }, + ) return /window\.__BJH__INIT__AUTH__\s*=\s*(["'])(.*?)\1/.exec(html)?.[2] ?? '' } @@ -172,7 +179,7 @@ async function uploadImage( appId: string, cropCover: boolean, ): Promise { - const image = await fetchBaijiahaoImageBlob(sourceUrl) + const image = await fetchBaijiahaoImageBlob(sourceUrl, context.signal) if (!image) { throw new Error(cropCover ? 'baijiahao_cover_fetch_failed' : 'baijiahao_image_fetch_failed') } @@ -196,6 +203,7 @@ async function uploadImage( headers: baijiahaoHeaders(), body: form, }, + { signal: context.signal }, ).catch((error) => { throw new Error( error instanceof Error @@ -232,6 +240,7 @@ async function uploadImage( cutting_type: 'cover_image', }), }, + { signal: context.signal }, ).catch((error) => { throw new Error( error instanceof Error @@ -279,9 +288,14 @@ function buildImageURLCandidates(sourceUrl: string): string[] { return candidates } -async function fetchBaijiahaoImageBlob(sourceUrl: string): Promise { +async function fetchBaijiahaoImageBlob( + sourceUrl: string, + signal?: AbortSignal, +): Promise { for (const candidate of buildImageURLCandidates(sourceUrl)) { - const response = await fetch(candidate).catch(() => null) + const response = await adapterFetch(candidate, {}, { signal, timeoutMs: 10_000 }).catch( + () => null, + ) if (!response?.ok) { continue } @@ -332,7 +346,7 @@ async function processContentImages( const processed = await uploadHtmlImages(html, async (sourceUrl) => { await waitForHumanPace(context.signal) return await uploadImage(context, sourceUrl, appId, false) - }) + }, { signal: context.signal }) return processed.html } @@ -570,6 +584,7 @@ export const baijiahaoAdapter: PublishAdapter = { activity_list: buildActivityList(), }), }, + { signal: context.signal }, ).catch( (requestError): BaijiahaoPublishResponse => ({ errno: -1, diff --git a/apps/desktop-client/src/main/adapters/bilibili.ts b/apps/desktop-client/src/main/adapters/bilibili.ts index b2e8936..89edc83 100644 --- a/apps/desktop-client/src/main/adapters/bilibili.ts +++ b/apps/desktop-client/src/main/adapters/bilibili.ts @@ -7,6 +7,7 @@ import { PUBLISH_IMAGE_FETCH_FAILED_MESSAGE } from '../../shared/publisher-error import { resolveDesktopApiURL } from '../transport/api-client' import type { PublishAdapter, PublishAdapterContext } from './base' import { + adapterFetch, extractImageSources, fetchImageBlob, normalizeArticleHtml, @@ -117,11 +118,16 @@ async function fetchNav(context: PublishAdapterContext): Promise null) } -async function mainFetchJson(input: string, init?: RequestInit): Promise { - const response = await fetch(input, init) +async function mainFetchJson( + input: string, + init?: RequestInit, + options: { signal?: AbortSignal } = {}, +): Promise { + const response = await adapterFetch(input, init, { signal: options.signal }) const text = await response.text() if (!response.ok) { throw new Error(text || `request_failed_${response.status}`) @@ -230,9 +236,9 @@ function buildAssetURLCandidates(sourceUrl: string): string[] { return candidates } -async function fetchAssetBlob(sourceUrl: string): Promise { +async function fetchAssetBlob(sourceUrl: string, signal?: AbortSignal): Promise { for (const candidate of buildAssetURLCandidates(sourceUrl)) { - const blob = await fetchImageBlob(candidate) + const blob = await fetchImageBlob(candidate, { signal }) if (blob) { return blob } @@ -244,7 +250,7 @@ async function uploadImage( context: PublishAdapterContext, sourceUrl: string, ): Promise { - const blob = await fetchAssetBlob(sourceUrl) + const blob = await fetchAssetBlob(sourceUrl, context.signal) if (!blob) { throw new Error('bilibili_image_fetch_failed') } @@ -268,6 +274,7 @@ async function uploadImage( headers: await bilibiliHeaders(context), body: form, }, + { signal: context.signal }, ).catch( (error): BilibiliUploadResponse => ({ code: -1, @@ -284,8 +291,10 @@ async function uploadImage( } async function processContentImages(context: PublishAdapterContext, html: string): Promise { - const processed = await uploadHtmlImages(html, async (sourceUrl) => - uploadImage(context, sourceUrl), + const processed = await uploadHtmlImages( + html, + async (sourceUrl) => uploadImage(context, sourceUrl), + { signal: context.signal }, ) return processed.html } @@ -547,6 +556,7 @@ async function findPublishedArticle( credentials: 'include', headers: await bilibiliHeaders(context), }, + { signal: context.signal }, ).catch(() => null) const items = response?.data?.items ?? [] diff --git a/apps/desktop-client/src/main/adapters/common.ts b/apps/desktop-client/src/main/adapters/common.ts index 90f6d3c..e34a48d 100644 --- a/apps/desktop-client/src/main/adapters/common.ts +++ b/apps/desktop-client/src/main/adapters/common.ts @@ -4,6 +4,26 @@ import { marked } from 'marked' import { STANDARD_ACCEPT_LANGUAGES, STANDARD_USER_AGENT } from '../user-agent' +const defaultFetchTimeoutMs = 15_000 +const defaultImageFetchTimeoutMs = 10_000 +const defaultHtmlImageUploadTimeoutMs = 30_000 +const defaultHtmlImageUploadConcurrency = 3 + +export interface AdapterFetchOptions { + signal?: AbortSignal + timeoutMs?: number +} + +export interface UploadHtmlImagesOptions extends AdapterFetchOptions { + imageTimeoutMs?: number + concurrency?: number +} + +export interface AbortablePromiseOptions extends AdapterFetchOptions { + abortErrorMessage?: string + timeoutErrorMessage?: string +} + export function normalizeText(value: unknown): string | null { if (typeof value !== 'string') { return null @@ -76,13 +96,133 @@ export function extractImageSources(html: string): string[] { return [...sources] } -export async function fetchImageBlob(sourceUrl: string): Promise { +export function mergeAbortSignals( + signals: Array, +): AbortSignal | undefined { + const activeSignals = signals.filter((signal): signal is AbortSignal => Boolean(signal)) + if (!activeSignals.length) { + return undefined + } + const anyAbortSignal = (AbortSignal as typeof AbortSignal & { + any?: (signals: AbortSignal[]) => AbortSignal + }).any + if (typeof anyAbortSignal === 'function') { + return anyAbortSignal(activeSignals) + } + + const controller = new AbortController() + const abort = (signal: AbortSignal) => { + if (!controller.signal.aborted) { + controller.abort(signal.reason) + } + } + for (const signal of activeSignals) { + if (signal.aborted) { + abort(signal) + break + } + signal.addEventListener('abort', () => abort(signal), { once: true }) + } + return controller.signal +} + +export function timeoutSignal(timeoutMs = defaultFetchTimeoutMs): AbortSignal { + const timeoutAbortSignal = (AbortSignal as typeof AbortSignal & { + timeout?: (timeoutMs: number) => AbortSignal + }).timeout + if (typeof timeoutAbortSignal === 'function') { + return timeoutAbortSignal(timeoutMs) + } + + const controller = new AbortController() + setTimeout(() => controller.abort(new Error('adapter_fetch_timeout')), timeoutMs) + return controller.signal +} + +export function buildAdapterAbortSignal(options: AdapterFetchOptions = {}): AbortSignal { + return mergeAbortSignals([ + options.signal, + timeoutSignal(options.timeoutMs ?? defaultFetchTimeoutMs), + ]) as AbortSignal +} + +export function withRequestTimeout(init: RequestInit = {}, options: AdapterFetchOptions = {}): RequestInit { + return { + ...init, + signal: mergeAbortSignals([ + init.signal, + options.signal, + timeoutSignal(options.timeoutMs ?? defaultFetchTimeoutMs), + ]), + } +} + +export async function adapterFetch( + input: string | URL | Request, + init: RequestInit = {}, + options: AdapterFetchOptions = {}, +): Promise { + return fetch(input, withRequestTimeout(init, options)) +} + +export async function raceWithAbort( + promise: Promise, + options: AbortablePromiseOptions = {}, +): Promise { + const timeoutMs = options.timeoutMs ?? defaultFetchTimeoutMs + const abortErrorMessage = options.abortErrorMessage ?? 'adapter_aborted' + const timeoutErrorMessage = options.timeoutErrorMessage ?? 'adapter_fetch_timeout' + + if (options.signal?.aborted) { + throw new Error(abortErrorMessage) + } + + let timeoutHandle: ReturnType | null = null + return await new Promise((resolve, reject) => { + const cleanup = () => { + if (timeoutHandle) { + clearTimeout(timeoutHandle) + timeoutHandle = null + } + options.signal?.removeEventListener('abort', onAbort) + } + const onAbort = () => { + cleanup() + reject(new Error(abortErrorMessage)) + } + + timeoutHandle = setTimeout(() => { + cleanup() + reject(new Error(timeoutErrorMessage)) + }, timeoutMs) + options.signal?.addEventListener('abort', onAbort, { once: true }) + + promise.then( + (value) => { + cleanup() + resolve(value) + }, + (error) => { + cleanup() + reject(error) + }, + ) + }) +} + +export async function fetchImageBlob( + sourceUrl: string, + options: AdapterFetchOptions = {}, +): Promise { const normalizedUrl = normalizeRemoteUrl(sourceUrl) if (!normalizedUrl) { return null } - const response = await fetch(normalizedUrl).catch(() => null) + const response = await adapterFetch(normalizedUrl, {}, { + timeoutMs: options.timeoutMs ?? defaultImageFetchTimeoutMs, + signal: options.signal, + }).catch(() => null) if (!response?.ok) { return null } @@ -98,6 +238,7 @@ export async function fetchImageBlob(sourceUrl: string): Promise { export async function uploadHtmlImages( html: string, uploader: (sourceUrl: string) => Promise, + options: UploadHtmlImagesOptions = {}, ): Promise<{ html: string; uploaded: Map }> { const sources = extractImageSources(html) if (!sources.length) { @@ -108,13 +249,68 @@ export async function uploadHtmlImages( } const uploaded = new Map() - for (const source of sources) { - const target = await uploader(source) + const overallSignal = buildAdapterAbortSignal({ + signal: options.signal, + timeoutMs: options.timeoutMs ?? defaultHtmlImageUploadTimeoutMs, + }) + const concurrency = Math.max( + 1, + Math.min(options.concurrency ?? defaultHtmlImageUploadConcurrency, sources.length), + ) + let cursor = 0 + + const uploadOne = async (source: string) => { + if (overallSignal.aborted) { + throw new Error('adapter_aborted') + } + + const target = await Promise.race([ + uploader(source), + new Promise((resolve) => { + const signal = timeoutSignal(options.imageTimeoutMs ?? defaultImageFetchTimeoutMs) + if (signal.aborted) { + resolve(null) + return + } + signal.addEventListener('abort', () => resolve(null), { once: true }) + }), + ]).catch(() => null) if (target) { uploaded.set(source, target) } } + const workers = Array.from({ length: concurrency }, async () => { + // Stop pulling new images once the overall signal aborts so abandoned workers + // don't keep firing fresh network requests after the caller has given up. + while (cursor < sources.length && !overallSignal.aborted) { + const index = cursor + cursor += 1 + await uploadOne(sources[index]) + } + }) + + try { + await Promise.race([ + Promise.all(workers), + new Promise((_, reject) => { + if (overallSignal.aborted) { + reject(new Error('adapter_aborted')) + return + } + overallSignal.addEventListener('abort', () => reject(new Error('adapter_aborted')), { + once: true, + }) + }), + ]) + } catch { + if (options.signal?.aborted) { + throw new Error('adapter_aborted') + } + // Best-effort image replacement: keep successfully uploaded images and leave + // slow or broken image URLs untouched instead of blocking the whole article. + } + let next = html for (const [from, to] of uploaded.entries()) { next = next.split(from).join(to) @@ -127,6 +323,7 @@ export async function sessionFetchText( session: Session, input: string, init?: RequestInit, + options: AdapterFetchOptions = {}, ): Promise { const headers = new Headers(init?.headers) if (!headers.has('user-agent')) { @@ -139,6 +336,11 @@ export async function sessionFetchText( const response = await session.fetch(input, { ...init, headers, + signal: mergeAbortSignals([ + init?.signal, + options.signal, + timeoutSignal(options.timeoutMs ?? defaultFetchTimeoutMs), + ]), }) const text = await response.text() if (!response.ok) { @@ -151,8 +353,9 @@ export async function sessionFetchJson( session: Session, input: string, init?: RequestInit, + options: AdapterFetchOptions = {}, ): Promise { - const text = await sessionFetchText(session, input, init) + const text = await sessionFetchText(session, input, init, options) if (!text) { return {} as T } @@ -169,7 +372,8 @@ export interface PageFetchInit { export async function sessionCookieFetchJson( session: Session, url: string, - init?: { method?: string; headers?: Record; body?: string }, + init?: { method?: string; headers?: Record; body?: string; signal?: AbortSignal }, + options: AdapterFetchOptions = {}, ): Promise { let parsedURL: URL try { @@ -208,6 +412,11 @@ export async function sessionCookieFetchJson( method: init?.method ?? 'GET', headers, body: init?.body, + signal: mergeAbortSignals([ + init?.signal, + options.signal, + timeoutSignal(options.timeoutMs ?? defaultFetchTimeoutMs), + ]), }) } catch { return null @@ -238,6 +447,7 @@ export async function pageFetchJson( webContents: WebContents, url: string, init?: PageFetchInit, + options: AdapterFetchOptions = {}, ): Promise { if (!webContents || webContents.isDestroyed()) { return null @@ -249,8 +459,11 @@ export async function pageFetchJson( } const script = `(async () => { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), ${JSON.stringify(options.timeoutMs ?? defaultFetchTimeoutMs)}); try { - const response = await fetch(${JSON.stringify(url)}, ${JSON.stringify(effectiveInit)}); + const init = ${JSON.stringify(effectiveInit)}; + const response = await fetch(${JSON.stringify(url)}, { ...init, signal: controller.signal }); if (!response.ok) { return JSON.stringify({ __ok: false, status: response.status }); } @@ -261,12 +474,17 @@ export async function pageFetchJson( return JSON.stringify({ __ok: true, body: text }); } catch (error) { return JSON.stringify({ __ok: false, error: String(error && error.message || error) }); + } finally { + clearTimeout(timeout); } })()` let serialized: string try { - serialized = (await webContents.executeJavaScript(script, true)) as string + serialized = (await raceWithAbort(webContents.executeJavaScript(script, true) as Promise, { + signal: options.signal, + timeoutMs: options.timeoutMs, + })) as string } catch { return null } @@ -331,7 +549,11 @@ export async function ensureViewLoaded( return } - await view.webContents.loadURL(url) + await raceWithAbort(view.webContents.loadURL(url), { + signal, + timeoutMs: 45_000, + timeoutErrorMessage: 'adapter_view_load_timeout', + }) await waitForLoadStop(view, signal) } diff --git a/apps/desktop-client/src/main/adapters/dongchedi.ts b/apps/desktop-client/src/main/adapters/dongchedi.ts index 0947d97..d515624 100644 --- a/apps/desktop-client/src/main/adapters/dongchedi.ts +++ b/apps/desktop-client/src/main/adapters/dongchedi.ts @@ -11,7 +11,9 @@ import { ensureViewLoaded, extractImageSources, normalizeArticleHtml, + raceWithAbort, sessionFetchJson, + sessionFetchText, } from './common' import { countDongchediContentWordCount, prepareDongchediArticleHtml } from './dongchedi-content' import { cropImageAssetBlob, fetchImageAssetBlob, type ImageAssetBlob } from './media-image' @@ -276,6 +278,9 @@ async function dongchediPageFetchJson( }, ): Promise { await ensureDongchediEditorContext(context) + if (context.signal.aborted) { + throw new Error('adapter_aborted') + } const effectiveInit = { credentials: 'include' as const, @@ -284,20 +289,30 @@ async function dongchediPageFetchJson( const page = context.playwright?.page if (page && !page.isClosed()) { - const envelope = await page.evaluate( + const envelope = await raceWithAbort( + page.evaluate( async ({ fetchInput, fetchInit }) => { + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 30_000) try { - const response = await fetch(fetchInput, fetchInit) + const response = await fetch(fetchInput, { + ...fetchInit, + signal: controller.signal, + }) const text = await response.text().catch(() => '') return { ok: response.ok, status: response.status, body: text } } catch (error) { return { ok: false, error: String((error && (error as Error).message) || error) } + } finally { + clearTimeout(timeout) } }, { fetchInput: input, fetchInit: effectiveInit, }, + ), + { signal: context.signal, timeoutMs: 35_000 }, ) return parseDongchediFetchEnvelope(envelope) } @@ -308,17 +323,24 @@ async function dongchediPageFetchJson( } const script = `(async () => { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 30000); try { const init = ${JSON.stringify(effectiveInit)}; - const response = await fetch(${JSON.stringify(input)}, init); + const response = await fetch(${JSON.stringify(input)}, { ...init, signal: controller.signal }); const text = await response.text().catch(() => ""); return JSON.stringify({ ok: response.ok, status: response.status, body: text }); } catch (error) { return JSON.stringify({ ok: false, error: String(error && error.message || error) }); + } finally { + clearTimeout(timeout); } })()` - const serialized = await webContents.executeJavaScript(script, true) + const serialized = await raceWithAbort(webContents.executeJavaScript(script, true) as Promise, { + signal: context.signal, + timeoutMs: 35_000, + }) if (typeof serialized !== 'string' || !serialized) { throw new Error('dongchedi_page_fetch_failed') } @@ -378,13 +400,20 @@ function parseDongchediFetchEnvelope(envelope: { } async function fetchDongchediCsrfToken(context: PublishAdapterContext): Promise { + if (context.signal.aborted) { + throw new Error('adapter_aborted') + } + const page = context.playwright?.page if (page && !page.isClosed()) { - const token = await page.evaluate(async (url) => { + const token = await raceWithAbort(page.evaluate(async (url) => { + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 15_000) try { const response = await fetch(url, { method: 'HEAD', credentials: 'include', + signal: controller.signal, headers: { 'x-secsdk-csrf-request': '1', 'x-secsdk-csrf-version': '1.2.10', @@ -393,8 +422,13 @@ async function fetchDongchediCsrfToken(context: PublishAdapterContext): Promise< return response.headers.get('x-ware-csrf-token') || '' } catch { return '' + } finally { + clearTimeout(timeout) } - }, `${DONGCHEDI_ORIGIN}/motor/mp_index/api/user/info?`) + }, `${DONGCHEDI_ORIGIN}/motor/mp_index/api/user/info?`), { + signal: context.signal, + timeoutMs: 18_000, + }) const parts = token.split(',') return parts.length >= 2 && parts[1] ? parts[1] : null } @@ -405,10 +439,13 @@ async function fetchDongchediCsrfToken(context: PublishAdapterContext): Promise< } const script = `(async () => { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 15000); try { const response = await fetch(${JSON.stringify(`${DONGCHEDI_ORIGIN}/motor/mp_index/api/user/info?`)}, { method: "HEAD", credentials: "include", + signal: controller.signal, headers: { "x-secsdk-csrf-request": "1", "x-secsdk-csrf-version": "1.2.10" @@ -417,10 +454,15 @@ async function fetchDongchediCsrfToken(context: PublishAdapterContext): Promise< return JSON.stringify({ token: response.headers.get("x-ware-csrf-token") || "" }); } catch { return JSON.stringify({ token: "" }); + } finally { + clearTimeout(timeout); } })()` - const serialized = await webContents.executeJavaScript(script, true).catch(() => '') + const serialized = await raceWithAbort( + webContents.executeJavaScript(script, true) as Promise, + { signal: context.signal, timeoutMs: 18_000 }, + ).catch(() => '') if (typeof serialized !== 'string' || !serialized) { return null } @@ -440,12 +482,7 @@ async function imagexFetchText( input: string, init?: RequestInit, ): Promise { - const response = await context.session.fetch(input, init) - const text = await response.text() - if (!response.ok) { - throw new Error(text || `request_failed_${response.status}`) - } - return text + return await sessionFetchText(context.session, input, init, { signal: context.signal }) } async function imagexFetchJson( @@ -471,6 +508,7 @@ async function fetchAccount(context: PublishAdapterContext): Promise null) const user = response?.data @@ -595,6 +633,7 @@ async function fetchUploadToken( { credentials: 'include', }, + { signal: context.signal }, ).catch(() => null) const token = response?.data?.token @@ -625,7 +664,7 @@ async function uploadImage( sourceUrl: string, ratio?: number, ): Promise { - const source = await fetchImageAssetBlob(sourceUrl) + const source = await fetchImageAssetBlob(sourceUrl, { signal: context.signal }) if (!source) { throw new Error(ratio ? 'dongchedi_cover_fetch_failed' : 'dongchedi_image_fetch_failed') } @@ -680,11 +719,15 @@ async function uploadImage( 'content-crc32': crc32(new Uint8Array(buffer)), authorization: storeInfo.Auth, }) - const putResponse = await context.session.fetch(`https://${uploadHost}/${storeUri}`, { - method: 'PUT', - headers: putHeaders, - body: image.blob, - }) + const putResponse = await raceWithAbort( + context.session.fetch(`https://${uploadHost}/${storeUri}`, { + method: 'PUT', + headers: putHeaders, + body: image.blob, + signal: context.signal, + }), + { signal: context.signal, timeoutMs: 30_000 }, + ) if (!putResponse.ok) { const text = await putResponse.text().catch(() => '') throw new Error(text || `dongchedi_image_tos_upload_failed_${putResponse.status}`) @@ -729,6 +772,7 @@ async function uploadImage( credentials: 'include', body: `img_uris=${storeUri}&img_url_type=2&img_param=noop&img_format=image`, }, + { signal: context.signal }, ).catch(() => null) const imageUrl = imageUrlResponse?.data?.img_url_map?.[storeUri]?.main_url?.trim() || '' @@ -765,6 +809,7 @@ async function spiceImage( imageUrl, }), }, + { signal: context.signal }, ).catch(() => null) return response?.data ?? null diff --git a/apps/desktop-client/src/main/adapters/jianshu.ts b/apps/desktop-client/src/main/adapters/jianshu.ts index a965ff5..132ec5d 100644 --- a/apps/desktop-client/src/main/adapters/jianshu.ts +++ b/apps/desktop-client/src/main/adapters/jianshu.ts @@ -41,9 +41,9 @@ function buildAssetURLCandidates(sourceUrl: string): string[] { return candidates } -async function fetchAssetBlob(sourceUrl: string): Promise { +async function fetchAssetBlob(sourceUrl: string, signal?: AbortSignal): Promise { for (const candidate of buildAssetURLCandidates(sourceUrl)) { - const blob = await fetchImageBlob(candidate) + const blob = await fetchImageBlob(candidate, { signal }) if (blob) { return blob } @@ -83,8 +83,9 @@ export const jianshuAdapter: PublishAdapter = { coverAssetUrl: context.article.cover_asset_url, }, { - fetchJson: (input, init) => sessionFetchJson(context.session, input, init), - fetchImageBlob: fetchAssetBlob, + fetchJson: (input, init) => + sessionFetchJson(context.session, input, init, { signal: context.signal }), + fetchImageBlob: (sourceUrl) => fetchAssetBlob(sourceUrl, context.signal), reportProgress(stage) { context.reportProgress(`jianshu.${stage}`) }, diff --git a/apps/desktop-client/src/main/adapters/juejin.ts b/apps/desktop-client/src/main/adapters/juejin.ts index f268c66..28b029e 100644 --- a/apps/desktop-client/src/main/adapters/juejin.ts +++ b/apps/desktop-client/src/main/adapters/juejin.ts @@ -5,6 +5,7 @@ import type { JsonValue } from '@geo/shared-types' import { resolveDesktopApiURL } from '../transport/api-client' import type { PublishAdapter, PublishAdapterContext } from './base' import { + adapterFetch, fetchImageBlob, normalizeArticleHtml, sessionCookieHeader, @@ -180,6 +181,7 @@ async function fetchProfile(context: PublishAdapterContext): Promise null) return response?.data?.bui_user ?? null @@ -233,6 +235,7 @@ function shouldSkipImageSource(source: string): boolean { async function uploadMarkdownImages( markdown: string, uploader: (sourceUrl: string) => Promise, + signal?: AbortSignal, ): Promise { const sources = extractMarkdownImageSources(markdown) if (!sources.length) { @@ -241,10 +244,27 @@ async function uploadMarkdownImages( const uploaded = new Map() for (const source of sources) { + if (signal?.aborted) { + throw new Error('adapter_aborted') + } if (shouldSkipImageSource(source)) { continue } - const target = await uploader(source) + let imageTimeout: ReturnType | null = null + const target = await Promise.race([ + uploader(source), + new Promise((resolve) => { + imageTimeout = setTimeout(() => resolve(null), 10_000) + signal?.addEventListener('abort', () => resolve(null), { once: true }) + }), + ]) + .catch(() => null) + .finally(() => { + if (imageTimeout) { + clearTimeout(imageTimeout) + imageTimeout = null + } + }) if (target) { uploaded.set(source, target) } @@ -386,9 +406,9 @@ function buildAssetURLCandidates(sourceUrl: string): string[] { return candidates } -async function fetchAssetBlob(sourceUrl: string): Promise { +async function fetchAssetBlob(sourceUrl: string, signal?: AbortSignal): Promise { for (const candidate of buildAssetURLCandidates(sourceUrl)) { - const blob = await fetchImageBlob(candidate) + const blob = await fetchImageBlob(candidate, { signal }) if (blob) { return blob } @@ -504,8 +524,12 @@ function crc32Table(): Uint32Array { return table } -async function mainFetchJson(input: string, init?: RequestInit): Promise { - const response = await fetch(input, init) +async function mainFetchJson( + input: string, + init?: RequestInit, + options: { signal?: AbortSignal } = {}, +): Promise { + const response = await adapterFetch(input, init, { signal: options.signal }) const text = await response.text() if (!response.ok) { throw new Error(text || `request_failed_${response.status}`) @@ -532,6 +556,7 @@ async function fetchImageXToken( credentials: 'include', headers: await juejinHeaders(context), }, + { signal: context.signal }, ) if (response.err_no && response.err_no !== 0) { @@ -550,7 +575,7 @@ async function fetchImageXToken( } } -async function applyImageUpload(token: JuejinImageXToken) { +async function applyImageUpload(token: JuejinImageXToken, signal?: AbortSignal) { const url = `${JUEJIN_IMAGEX_ORIGIN}/?Action=ApplyImageUpload&Version=2018-08-01&ServiceId=${JUEJIN_IMAGEX_SERVICE_ID}` const signedHeaders = signAWS4({ method: 'GET', @@ -559,10 +584,14 @@ async function applyImageUpload(token: JuejinImageXToken) { secretAccessKey: token.secretAccessKey, securityToken: token.sessionToken, }) - const response = await mainFetchJson(url, { - method: 'GET', - headers: signedHeaders, - }) + const response = await mainFetchJson( + url, + { + method: 'GET', + headers: signedHeaders, + }, + { signal }, + ) const uploadAddress = response.Result?.UploadAddress const storeInfo = uploadAddress?.StoreInfos?.[0] const uploadHost = uploadAddress?.UploadHosts?.[0] @@ -580,24 +609,33 @@ async function applyImageUpload(token: JuejinImageXToken) { async function uploadToTOS( upload: Awaited>, blob: Blob, + signal?: AbortSignal, ): Promise { const buffer = Buffer.from(await blob.arrayBuffer()) - const response = await fetch(`https://${upload.uploadHost}/${upload.storeUri}`, { - method: 'PUT', - headers: { - authorization: upload.auth, - 'content-type': blob.type || 'application/octet-stream', - 'content-crc32': crc32(new Uint8Array(buffer)), + const response = await adapterFetch( + `https://${upload.uploadHost}/${upload.storeUri}`, + { + method: 'PUT', + headers: { + authorization: upload.auth, + 'content-type': blob.type || 'application/octet-stream', + 'content-crc32': crc32(new Uint8Array(buffer)), + }, + body: blob, }, - body: blob, - }) + { signal }, + ) if (!response.ok) { const text = await response.text().catch(() => '') throw new Error(text || `juejin_image_tos_upload_failed_${response.status}`) } } -async function commitImageUpload(token: JuejinImageXToken, sessionKey: string): Promise { +async function commitImageUpload( + token: JuejinImageXToken, + sessionKey: string, + signal?: AbortSignal, +): Promise { const url = `${JUEJIN_IMAGEX_ORIGIN}/?Action=CommitImageUpload&Version=2018-08-01&SessionKey=${encodeURIComponent(sessionKey)}&ServiceId=${JUEJIN_IMAGEX_SERVICE_ID}` const signedHeaders = signAWS4({ method: 'POST', @@ -606,13 +644,17 @@ async function commitImageUpload(token: JuejinImageXToken, sessionKey: string): secretAccessKey: token.secretAccessKey, securityToken: token.sessionToken, }) - const response = await mainFetchJson(url, { - method: 'POST', - headers: { - ...signedHeaders, - 'content-length': '0', + const response = await mainFetchJson( + url, + { + method: 'POST', + headers: { + ...signedHeaders, + 'content-length': '0', + }, }, - }) + { signal }, + ) if (!response.Result) { throw new Error('juejin_image_commit_upload_failed') } @@ -631,6 +673,7 @@ async function resolveImageURL( credentials: 'include', headers: await juejinHeaders(context), }, + { signal: context.signal }, ) if (response.err_no && response.err_no !== 0) { throw new Error(response.err_msg || 'juejin_image_url_failed') @@ -647,15 +690,15 @@ async function uploadImage( uuid: string, sourceUrl: string, ): Promise { - const blob = await fetchAssetBlob(sourceUrl) + const blob = await fetchAssetBlob(sourceUrl, context.signal) if (!blob || !blob.type.startsWith('image/')) { return null } const token = await fetchImageXToken(context, uuid) - const upload = await applyImageUpload(token) - await uploadToTOS(upload, blob) - await commitImageUpload(token, upload.sessionKey) + const upload = await applyImageUpload(token, context.signal) + await uploadToTOS(upload, blob, context.signal) + await commitImageUpload(token, upload.sessionKey, context.signal) return await resolveImageURL(context, uuid, upload.storeUri) } @@ -697,6 +740,7 @@ async function createDraft( pics: [], }), }, + { signal: context.signal }, ) if (response.err_no && response.err_no !== 0) { @@ -892,6 +936,7 @@ async function findPublishedArticle( page_no: 1, }), }, + { signal: context.signal }, ).catch(() => null) const article = extractArticleFromListResponse(response, draftId) @@ -1012,6 +1057,7 @@ export const juejinAdapter: PublishAdapter = { markdown = await uploadMarkdownImages( markdown, async (sourceUrl) => await uploadImage(context, uuid, sourceUrl).catch(() => null), + context.signal, ) context.reportProgress('juejin.create_draft') diff --git a/apps/desktop-client/src/main/adapters/media-image.ts b/apps/desktop-client/src/main/adapters/media-image.ts index 58afc13..8ea676f 100644 --- a/apps/desktop-client/src/main/adapters/media-image.ts +++ b/apps/desktop-client/src/main/adapters/media-image.ts @@ -3,6 +3,7 @@ import { Buffer } from 'node:buffer' import { nativeImage } from 'electron' import { fetchDesktopApiURL, resolveDesktopApiURL } from '../transport/api-client' +import { adapterFetch, mergeAbortSignals, timeoutSignal, type AdapterFetchOptions } from './common' export interface ImageAssetBlob { blob: Blob @@ -97,10 +98,20 @@ function fileNameForImageType(sourceType: string): string { return 'image.png' } -export async function fetchImageAssetBlob(sourceUrl: string): Promise { +export async function fetchImageAssetBlob( + sourceUrl: string, + options: AdapterFetchOptions = {}, +): Promise { for (const candidate of buildAssetURLCandidates(sourceUrl)) { const response = await ( - shouldUseDesktopAssetFetch(candidate) ? fetchDesktopApiURL(candidate) : fetch(candidate) + shouldUseDesktopAssetFetch(candidate) + ? fetchDesktopApiURL(candidate, { + signal: mergeAbortSignals([ + options.signal, + timeoutSignal(options.timeoutMs ?? 10_000), + ]), + }) + : adapterFetch(candidate, {}, { timeoutMs: options.timeoutMs ?? 10_000, signal: options.signal }) ).catch(() => null) if (!response?.ok) { continue diff --git a/apps/desktop-client/src/main/adapters/qiehao.ts b/apps/desktop-client/src/main/adapters/qiehao.ts index 3234738..2c43fc6 100644 --- a/apps/desktop-client/src/main/adapters/qiehao.ts +++ b/apps/desktop-client/src/main/adapters/qiehao.ts @@ -9,6 +9,7 @@ import { import { resolveDesktopApiURL } from '../transport/api-client' import type { PublishAdapter, PublishAdapterContext } from './base' import { + adapterFetch, fetchImageBlob, normalizeArticleHtml, sessionCookieHeader, @@ -171,13 +172,18 @@ async function fetchBasicInfo(context: PublishAdapterContext): Promise null) return response?.data?.cpInfo ?? null } -async function mainFetchJson(input: string, init?: RequestInit): Promise { - const response = await fetch(input, init) +async function mainFetchJson( + input: string, + init?: RequestInit, + options: { signal?: AbortSignal } = {}, +): Promise { + const response = await adapterFetch(input, init, { signal: options.signal }) const text = await response.text() if (!response.ok) { throw new Error(text || `request_failed_${response.status}`) @@ -224,9 +230,9 @@ function buildAssetURLCandidates(sourceUrl: string): string[] { return candidates } -async function fetchAssetBlob(sourceUrl: string): Promise { +async function fetchAssetBlob(sourceUrl: string, signal?: AbortSignal): Promise { for (const candidate of buildAssetURLCandidates(sourceUrl)) { - const blob = await fetchImageBlob(candidate) + const blob = await fetchImageBlob(candidate, { signal }) if (blob) { return blob } @@ -239,7 +245,7 @@ async function uploadImage( sourceUrl: string, includeCoverMeta: boolean, ): Promise { - const blob = await fetchAssetBlob(sourceUrl) + const blob = await fetchAssetBlob(sourceUrl, context.signal) if (!blob) { throw new Error(includeCoverMeta ? 'qiehao_cover_fetch_failed' : 'qiehao_image_fetch_failed') } @@ -259,6 +265,7 @@ async function uploadImage( headers: await qiehaoHeaders(context, QIEHAO_ORIGIN, {}, 'image.om.qq.com'), body: form, }, + { signal: context.signal }, ).catch( (error): QiehaoUploadStepOneResponse => ({ msg: error instanceof Error ? error.message : 'qiehao_image_upload_failed', @@ -306,6 +313,7 @@ async function uploadImage( relogin: 1, }), }, + { signal: context.signal }, ).catch( (error): QiehaoUploadStepTwoResponse => ({ msg: error instanceof Error ? error.message : 'qiehao_cover_upload_failed', @@ -341,6 +349,7 @@ async function findArticleUrl( credentials: 'include', headers: await qiehaoHeaders(context, QIEHAO_ORIGIN), }, + { signal: context.signal }, ).catch(() => null) const article = list?.data?.articles?.find( @@ -502,8 +511,10 @@ export const qiehaoAdapter: PublishAdapter = { } context.reportProgress('qiehao.upload_content_images') - const processed = await uploadHtmlImages(html, async (sourceUrl) => - uploadImage(context, sourceUrl, false), + const processed = await uploadHtmlImages( + html, + async (sourceUrl) => uploadImage(context, sourceUrl, false), + { signal: context.signal }, ) const coverImages: string[] = [] @@ -524,27 +535,32 @@ export const qiehaoAdapter: PublishAdapter = { : `${QIEHAO_ORIGIN}/marticlepublish/omSave` context.reportProgress('qiehao.submit') - const response = await sessionFetchJson(context.session, endpoint, { - method: 'POST', - credentials: 'include', - headers: await qiehaoHeaders(context, QIEHAO_ORIGIN, { - 'content-type': 'application/json', - }), - body: JSON.stringify({ - title: context.article.title?.trim() || '未命名文章', - content: processed.html, - imgurl_ext: JSON.stringify(coverImages), - cover_type: '1', - imgurlsrc: 'custom', - orignal: 0, - user_original: 0, - mediaId, - needpub: 1, - type: 0, - relogin: 1, - resource_aigc_mark_info: JSON.stringify(resourceMarkInfo), - }), - }).catch( + const response = await sessionFetchJson( + context.session, + endpoint, + { + method: 'POST', + credentials: 'include', + headers: await qiehaoHeaders(context, QIEHAO_ORIGIN, { + 'content-type': 'application/json', + }), + body: JSON.stringify({ + title: context.article.title?.trim() || '未命名文章', + content: processed.html, + imgurl_ext: JSON.stringify(coverImages), + cover_type: '1', + imgurlsrc: 'custom', + orignal: 0, + user_original: 0, + mediaId, + needpub: 1, + type: 0, + relogin: 1, + resource_aigc_mark_info: JSON.stringify(resourceMarkInfo), + }), + }, + { signal: context.signal }, + ).catch( (requestError): QiehaoPublishResponse => ({ code: -1, msg: requestError instanceof Error ? requestError.message : 'qiehao_publish_failed', diff --git a/apps/desktop-client/src/main/adapters/smzdm.ts b/apps/desktop-client/src/main/adapters/smzdm.ts index 2d0780a..9ca7126 100644 --- a/apps/desktop-client/src/main/adapters/smzdm.ts +++ b/apps/desktop-client/src/main/adapters/smzdm.ts @@ -8,8 +8,10 @@ import { PUBLISH_IMAGE_FETCH_FAILED_MESSAGE } from '../../shared/publisher-error import { resolveDesktopApiURL } from '../transport/api-client' import type { PublishAdapter, PublishAdapterContext } from './base' import { + adapterFetch, extractImageSources, normalizeArticleHtml, + raceWithAbort, sessionCookieHeader, sessionCookieValue, sessionFetchJson, @@ -203,6 +205,7 @@ async function fetchCurrentAccount(context: PublishAdapterContext): Promise null) const user = response?.data ?? response @@ -257,10 +260,13 @@ async function getArticleId(context: PublishAdapterContext): Promise { throw new Error('smzdm_playwright_page_missing') } - await page.goto(SMZDM_TOU_GAO_URL, { - waitUntil: 'domcontentloaded', - timeout: 45_000, - }) + await raceWithAbort( + page.goto(SMZDM_TOU_GAO_URL, { + waitUntil: 'domcontentloaded', + timeout: 45_000, + }), + { signal: context.signal, timeoutMs: 45_000 }, + ) await page.waitForLoadState('networkidle', { timeout: 5_000 }).catch(() => undefined) const href = await page @@ -320,9 +326,14 @@ function buildAssetURLCandidates(sourceUrl: string): string[] { return candidates } -async function fetchSmzdmImageBlob(sourceUrl: string): Promise { +async function fetchSmzdmImageBlob( + sourceUrl: string, + signal?: AbortSignal, +): Promise { for (const candidate of buildAssetURLCandidates(sourceUrl)) { - const response = await fetch(candidate).catch(() => null) + const response = await adapterFetch(candidate, {}, { signal, timeoutMs: 10_000 }).catch( + () => null, + ) if (!response?.ok) { continue } @@ -386,7 +397,7 @@ async function uploadImage( articleId: string, cropCover: boolean, ): Promise { - const image = await fetchSmzdmImageBlob(sourceUrl) + const image = await fetchSmzdmImageBlob(sourceUrl, context.signal) if (!image) { throw new Error(cropCover ? 'smzdm_cover_fetch_failed' : 'smzdm_image_fetch_failed') } @@ -406,6 +417,7 @@ async function uploadImage( headers: await smzdmHeaders(context), body: form, }, + { signal: context.signal }, ).catch( (error): SmzdmImageUploadResponse => ({ error_code: -1, @@ -425,17 +437,22 @@ async function uploadImage( const uploadedId = stringId(uploaded.data.id) if (uploadedId) { - void sessionFetchJson(context.session, `${SMZDM_ORIGIN}/api/editor/image_add_time/record`, { - method: 'POST', - credentials: 'include', - headers: await smzdmHeaders(context, { - 'content-type': 'application/x-www-form-urlencoded', - }), - body: new URLSearchParams({ - article_id: articleId, - ids: uploadedId, - }), - }).catch(() => null) + void sessionFetchJson( + context.session, + `${SMZDM_ORIGIN}/api/editor/image_add_time/record`, + { + method: 'POST', + credentials: 'include', + headers: await smzdmHeaders(context, { + 'content-type': 'application/x-www-form-urlencoded', + }), + body: new URLSearchParams({ + article_id: articleId, + ids: uploadedId, + }), + }, + { signal: context.signal }, + ).catch(() => null) } if (!cropCover) { @@ -460,6 +477,7 @@ async function uploadImage( pic_url: uploaded.data.url, }), }, + { signal: context.signal }, ).catch(() => null) const crop = getSmzdmCoverCropRect(image.width, image.height) @@ -497,6 +515,7 @@ async function uploadImage( headers: await smzdmHeaders(context), body: cropForm, }, + { signal: context.signal }, ).catch( (error): SmzdmCropResponse => ({ error_code: -1, @@ -570,14 +589,17 @@ async function calculateEditorSignature( throw new Error('smzdm_playwright_page_missing') } - await page.goto(`${SMZDM_ORIGIN}/edit/${encodeURIComponent(articleId)}`, { - waitUntil: 'domcontentloaded', - timeout: 45_000, - }) + await raceWithAbort( + page.goto(`${SMZDM_ORIGIN}/edit/${encodeURIComponent(articleId)}`, { + waitUntil: 'domcontentloaded', + timeout: 45_000, + }), + { signal: context.signal, timeoutMs: 45_000 }, + ) await page.waitForLoadState('networkidle', { timeout: 8_000 }).catch(() => undefined) - const textCount = await page - .evaluate((content) => { + const textCount = await raceWithAbort( + page.evaluate((content) => { const editorNode = document.querySelector('.ProseMirror[contenteditable]') as | (HTMLElement & { editor?: { @@ -600,7 +622,9 @@ async function calculateEditorSignature( }, }) return editor.commands.getTextCount() - }, html) + }, html), + { signal: context.signal, timeoutMs: 20_000 }, + ) .catch(() => null) const normalizedCount = @@ -904,6 +928,7 @@ export const smzdmAdapter: PublishAdapter = { }), ), }, + { signal: context.signal }, ).catch( (requestError): SmzdmSubmitResponse => ({ error_code: -1, diff --git a/apps/desktop-client/src/main/adapters/sohuhao.ts b/apps/desktop-client/src/main/adapters/sohuhao.ts index 01dc60a..9f22a4e 100644 --- a/apps/desktop-client/src/main/adapters/sohuhao.ts +++ b/apps/desktop-client/src/main/adapters/sohuhao.ts @@ -141,6 +141,7 @@ async function fetchAccount(context: PublishAdapterContext): Promise null) const account = accountFromRaw(registerInfo?.data?.account) @@ -157,6 +158,7 @@ async function fetchAccount(context: PublishAdapterContext): Promise null) for (const group of list?.data?.data ?? []) { @@ -176,7 +178,7 @@ async function uploadImage( accountId: string, sourceUrl: string, ): Promise { - const image = await fetchImageAssetBlob(sourceUrl) + const image = await fetchImageAssetBlob(sourceUrl, { signal: context.signal }) if (!image) { return null } @@ -188,14 +190,19 @@ async function uploadImage( const url = new URL(`${SOHU_ORIGIN}/commons/front/outerUpload/image/file`) url.searchParams.set('accountId', accountId) - const response = await sessionFetchJson(context.session, url.toString(), { - method: 'POST', - credentials: 'include', - headers: await sohuHeaders(context, { - 'sp-cm': await sohuSpCm(context), - }), - body: form, - }).catch(() => null) + const response = await sessionFetchJson( + context.session, + url.toString(), + { + method: 'POST', + credentials: 'include', + headers: await sohuHeaders(context, { + 'sp-cm': await sohuSpCm(context), + }), + body: form, + }, + { signal: context.signal }, + ).catch(() => null) return response?.url?.trim() || null } @@ -225,16 +232,21 @@ async function submitArticle( infoResource: 0, } - const response = await sessionFetchJson(context.session, url.toString(), { - method: 'POST', - credentials: 'include', - headers: await sohuHeaders(context, { - 'content-type': 'application/json', - 'dv-id': generateDeviceId(), - 'sp-cm': await sohuSpCm(context), - }), - body: JSON.stringify(body), - }).catch( + const response = await sessionFetchJson( + context.session, + url.toString(), + { + method: 'POST', + credentials: 'include', + headers: await sohuHeaders(context, { + 'content-type': 'application/json', + 'dv-id': generateDeviceId(), + 'sp-cm': await sohuSpCm(context), + }), + body: JSON.stringify(body), + }, + { signal: context.signal }, + ).catch( (error): SohuSubmitResponse => ({ code: -1, msg: error instanceof Error ? error.message : 'sohuhao_submit_failed', @@ -355,8 +367,10 @@ export const sohuhaoAdapter: PublishAdapter = { } context.reportProgress('sohuhao.upload_content_images') - const processed = await uploadHtmlImages(html, async (sourceUrl) => - uploadImage(context, account.id, sourceUrl), + const processed = await uploadHtmlImages( + html, + async (sourceUrl) => uploadImage(context, account.id, sourceUrl), + { signal: context.signal }, ) const publishType = resolvePublishType(payload) diff --git a/apps/desktop-client/src/main/adapters/toutiao.test.ts b/apps/desktop-client/src/main/adapters/toutiao.test.ts index 1abc7ec..615daed 100644 --- a/apps/desktop-client/src/main/adapters/toutiao.test.ts +++ b/apps/desktop-client/src/main/adapters/toutiao.test.ts @@ -60,7 +60,7 @@ describe('toutiao adapter', () => { expect(bodyBlob?.type).toBe('image/png') expect(coverBlob?.type).toBe('image/png') - expect(fetchImageAssetBlob).toHaveBeenCalledWith('/api/public/assets/body-token') - expect(fetchImageAssetBlob).toHaveBeenCalledWith('/api/public/assets/cover-token') + expect(fetchImageAssetBlob).toHaveBeenCalledWith('/api/public/assets/body-token', expect.anything()) + expect(fetchImageAssetBlob).toHaveBeenCalledWith('/api/public/assets/cover-token', expect.anything()) }) }) diff --git a/apps/desktop-client/src/main/adapters/toutiao.ts b/apps/desktop-client/src/main/adapters/toutiao.ts index 025522d..3980e41 100644 --- a/apps/desktop-client/src/main/adapters/toutiao.ts +++ b/apps/desktop-client/src/main/adapters/toutiao.ts @@ -5,8 +5,11 @@ import type { PublishAdapter } from './base' import { normalizeArticleHtml, sessionFetchJson, uploadHtmlImages } from './common' import { fetchImageAssetBlob } from './media-image' -async function fetchToutiaoImageBlob(sourceUrl: string): Promise { - return (await fetchImageAssetBlob(sourceUrl))?.blob ?? null +async function fetchToutiaoImageBlob( + sourceUrl: string, + signal?: AbortSignal, +): Promise { + return (await fetchImageAssetBlob(sourceUrl, { signal }))?.blob ?? null } function buildResultPayload( @@ -40,9 +43,11 @@ export const toutiaoAdapter: PublishAdapter = { publishType: 'publish', }, { - fetchJson: (input, init) => sessionFetchJson(context.session, input, init), - fetchImageBlob: fetchToutiaoImageBlob, - uploadHtmlImages, + fetchJson: (input, init) => + sessionFetchJson(context.session, input, init, { signal: context.signal }), + fetchImageBlob: (sourceUrl) => fetchToutiaoImageBlob(sourceUrl, context.signal), + uploadHtmlImages: (sourceHtml, uploader) => + uploadHtmlImages(sourceHtml, uploader, { signal: context.signal }), reportProgress(stage) { context.reportProgress(`toutiao.${stage}`) }, diff --git a/apps/desktop-client/src/main/adapters/wangyihao.ts b/apps/desktop-client/src/main/adapters/wangyihao.ts index 8b54461..bd953ca 100644 --- a/apps/desktop-client/src/main/adapters/wangyihao.ts +++ b/apps/desktop-client/src/main/adapters/wangyihao.ts @@ -9,6 +9,7 @@ import { import type { PublishAdapter, PublishAdapterContext } from './base' import { extractImageSources, + raceWithAbort, sessionCookieHeader, sessionFetchJson, uploadHtmlImages, @@ -180,6 +181,7 @@ async function fetchAccount(context: PublishAdapterContext): Promise null) const media = response?.data?.mediaInfo @@ -213,6 +215,7 @@ async function fetchPublishDetail( credentials: 'include', headers: await wangyiHeaders(context), }, + { signal: context.signal }, ).catch(() => null) if (String(response?.code) !== '1' || !response?.data) { @@ -235,12 +238,13 @@ async function getPublishToken(context: PublishAdapterContext): Promise } for (let attempt = 0; attempt < 2; attempt += 1) { - await page - .goto(WANGYI_HOME_URL, { + await raceWithAbort( + page.goto(WANGYI_HOME_URL, { waitUntil: attempt === 0 ? 'domcontentloaded' : 'load', timeout: 20_000, - }) - .catch(() => null) + }), + { signal: context.signal, timeoutMs: 20_000 }, + ).catch(() => null) await page.waitForLoadState('networkidle', { timeout: 3_000 }).catch(() => null) if (attempt > 0) { await sleep(800, context.signal) @@ -445,7 +449,7 @@ async function uploadImage( account: WangyiAccount, sourceUrl: string, ): Promise { - const image = await fetchImageAssetBlob(sourceUrl) + const image = await fetchImageAssetBlob(sourceUrl, { signal: context.signal }) if (!image) { return null } @@ -460,15 +464,20 @@ async function uploadImage( url.searchParams.set('wemediaId', account.wemediaId) url.searchParams.set('realUserId', account.realUserId) - const response = await sessionFetchJson(context.session, url.toString(), { - method: 'POST', - credentials: 'include', - headers: await wangyiHeaders(context, { - 'x-b3-sampled': '1', - 'x-b3-spanid': '0', - }), - body: form, - }).catch(() => null) + const response = await sessionFetchJson( + context.session, + url.toString(), + { + method: 'POST', + credentials: 'include', + headers: await wangyiHeaders(context, { + 'x-b3-sampled': '1', + 'x-b3-spanid': '0', + }), + body: form, + }, + { signal: context.signal }, + ).catch(() => null) return response?.data?.url?.trim() || null } @@ -544,16 +553,21 @@ async function saveDraftWithSession( url: string, body: URLSearchParams, ): Promise { - return await sessionFetchJson(context.session, url, { - method: 'POST', - credentials: 'include', - headers: await wangyiHeaders(context, { - 'content-type': 'application/x-www-form-urlencoded', - 'x-b3-sampled': '1', - 'x-b3-spanid': '0', - }), - body, - }).catch( + return await sessionFetchJson( + context.session, + url, + { + method: 'POST', + credentials: 'include', + headers: await wangyiHeaders(context, { + 'content-type': 'application/x-www-form-urlencoded', + 'x-b3-sampled': '1', + 'x-b3-spanid': '0', + }), + body, + }, + { signal: context.signal }, + ).catch( (error): WangyiDraftResponse => ({ code: -1, message: error instanceof Error ? error.message : 'wangyihao_draft_save_failed', @@ -571,24 +585,34 @@ async function saveDraftWithPage( return null } - const serialized = await page.evaluate( - async ({ requestUrl, requestBody }) => { - const response = await fetch(requestUrl, { - method: 'POST', - credentials: 'include', - headers: { - 'content-type': 'application/x-www-form-urlencoded', - 'x-b3-sampled': '1', - 'x-b3-spanid': '0', - }, - body: requestBody, - }) - return await response.text() - }, - { - requestUrl: url, - requestBody: body.toString(), - }, + const serialized = await raceWithAbort( + page.evaluate( + async ({ requestUrl, requestBody }) => { + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 30_000) + try { + const response = await fetch(requestUrl, { + method: 'POST', + credentials: 'include', + signal: controller.signal, + headers: { + 'content-type': 'application/x-www-form-urlencoded', + 'x-b3-sampled': '1', + 'x-b3-spanid': '0', + }, + body: requestBody, + }) + return await response.text() + } finally { + clearTimeout(timeout) + } + }, + { + requestUrl: url, + requestBody: body.toString(), + }, + ), + { signal: context.signal, timeoutMs: 35_000 }, ) if (!serialized) { @@ -835,9 +859,12 @@ async function publishDraft(context: PublishAdapterContext, draftId: string): Pr } const draftUrl = `${WANGYI_CONSOLE_ORIGIN}/subscribe_v4/index.html#/article-publish/${encodeURIComponent(draftId)}?option=editDraft` - await page.goto(draftUrl, { - waitUntil: 'domcontentloaded', - }) + await raceWithAbort( + page.goto(draftUrl, { + waitUntil: 'domcontentloaded', + }), + { signal: context.signal, timeoutMs: 45_000 }, + ) await page.waitForLoadState('networkidle', { timeout: 8_000 }).catch(() => null) await waitForWangyiEditorReady(page) @@ -997,8 +1024,10 @@ export const wangyihaoAdapter: PublishAdapter = { } context.reportProgress('wangyihao.upload_content_images') - const processed = await uploadHtmlImages(html, async (sourceUrl) => - uploadImage(context, account, sourceUrl), + const processed = await uploadHtmlImages( + html, + async (sourceUrl) => uploadImage(context, account, sourceUrl), + { signal: context.signal }, ) context.reportProgress('wangyihao.save_draft') diff --git a/apps/desktop-client/src/main/adapters/weixin-gzh.ts b/apps/desktop-client/src/main/adapters/weixin-gzh.ts index c18eed6..e2b869a 100644 --- a/apps/desktop-client/src/main/adapters/weixin-gzh.ts +++ b/apps/desktop-client/src/main/adapters/weixin-gzh.ts @@ -9,6 +9,7 @@ import type { PublishAdapter, PublishAdapterContext } from './base' import { normalizeArticleHtml, normalizeRemoteUrl, + raceWithAbort, sessionFetchJson, sessionFetchText, uploadHtmlImages, @@ -198,12 +199,17 @@ function parseMeta(html: string): WeixinMeta | null { } async function fetchMeta(context: PublishAdapterContext): Promise { - const html = await sessionFetchText(context.session, WEIXIN_HOME_URL, { - credentials: 'include', - headers: await weixinHeaders(context, { - accept: 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', - }), - }).catch(() => '') + const html = await sessionFetchText( + context.session, + WEIXIN_HOME_URL, + { + credentials: 'include', + headers: await weixinHeaders(context, { + accept: 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', + }), + }, + { signal: context.signal }, + ).catch(() => '') return html ? parseMeta(html) : null } @@ -268,7 +274,7 @@ async function uploadImage( return normalized } - const image = await fetchImageAssetBlob(sourceUrl) + const image = await fetchImageAssetBlob(sourceUrl, { signal: context.signal }) if (!image) { return null } @@ -297,12 +303,17 @@ async function uploadImage( url.searchParams.set('seq', String(Date.now())) url.searchParams.set('t', String(Math.random())) - const response = await sessionFetchJson(context.session, url.toString(), { - method: 'POST', - credentials: 'include', - headers: await weixinHeaders(context), - body: form, - }).catch(() => null) + const response = await sessionFetchJson( + context.session, + url.toString(), + { + method: 'POST', + credentials: 'include', + headers: await weixinHeaders(context), + body: form, + }, + { signal: context.signal }, + ).catch(() => null) if (response?.base_resp?.err_msg && response.base_resp.err_msg !== 'ok') { const message = response.base_resp.err_msg @@ -402,7 +413,7 @@ async function uploadCover( meta: WeixinMeta, sourceUrl: string, ): Promise { - const source = await fetchImageAssetBlob(sourceUrl) + const source = await fetchImageAssetBlob(sourceUrl, { signal: context.signal }) if (!source) { return null } @@ -412,7 +423,7 @@ async function uploadCover( return null } - const cdnImage = await fetchImageAssetBlob(cdnUrl).catch(() => null) + const cdnImage = await fetchImageAssetBlob(cdnUrl, { signal: context.signal }).catch(() => null) const width = cdnImage?.width || source.width const height = cdnImage?.height || source.height const rect235 = cropPercent(width, height, 2.35) @@ -426,14 +437,19 @@ async function uploadCover( const url = new URL(`${WEIXIN_ORIGIN}/cgi-bin/cropimage`) url.searchParams.set('action', 'crop_multi') - const response = await sessionFetchJson(context.session, url.toString(), { - method: 'POST', - credentials: 'include', - headers: await weixinHeaders(context, { - 'content-type': 'application/x-www-form-urlencoded', - }), - body, - }).catch(() => null) + const response = await sessionFetchJson( + context.session, + url.toString(), + { + method: 'POST', + credentials: 'include', + headers: await weixinHeaders(context, { + 'content-type': 'application/x-www-form-urlencoded', + }), + body, + }, + { signal: context.signal }, + ).catch(() => null) return response ? coverFromCropResponse(cdnUrl, body, response) : null } @@ -1014,10 +1030,12 @@ function isWeixinPublishNetworkResponse(response: PlaywrightResponse): boolean { async function waitForWeixinPublishNetworkResponse( page: PlaywrightPage, timeoutMs = WEIXIN_PUBLISH_RESPONSE_TIMEOUT_MS, + signal?: AbortSignal, ): Promise { - const response = await page - .waitForResponse(isWeixinPublishNetworkResponse, { timeout: timeoutMs }) - .catch(() => null) + const response = await raceWithAbort( + page.waitForResponse(isWeixinPublishNetworkResponse, { timeout: timeoutMs }), + { signal, timeoutMs: timeoutMs + 500 }, + ).catch(() => null) if (!response) { return null } @@ -1028,21 +1046,23 @@ async function waitForWeixinPublishNetworkResponse( return parseWeixinOperateResponseText(text) } -async function waitForWeixinEditorReady(page: PlaywrightPage): Promise { - await page - .waitForLoadState('domcontentloaded', { timeout: WEIXIN_EDITOR_READY_TIMEOUT_MS }) - .catch(() => null) +async function waitForWeixinEditorReady(page: PlaywrightPage, signal?: AbortSignal): Promise { + await raceWithAbort( + page.waitForLoadState('domcontentloaded', { timeout: WEIXIN_EDITOR_READY_TIMEOUT_MS }), + { signal, timeoutMs: WEIXIN_EDITOR_READY_TIMEOUT_MS + 500 }, + ).catch(() => null) await page.waitForLoadState('networkidle', { timeout: 10_000 }).catch(() => null) - await page - .waitForFunction( + await raceWithAbort( + page.waitForFunction( () => { const text = document.body?.innerText || '' return /发表|发布|草稿|图文|登录/.test(text) }, undefined, { timeout: WEIXIN_EDITOR_READY_TIMEOUT_MS }, - ) - .catch(() => null) + ), + { signal, timeoutMs: WEIXIN_EDITOR_READY_TIMEOUT_MS + 500 }, + ).catch(() => null) } async function readWeixinPageMessage( @@ -1606,7 +1626,11 @@ async function completeWeixinPublishConfirmation( } await waitForWeixinDelay(context.signal, 300) - const responsePromise = waitForWeixinPublishNetworkResponse(page) + const responsePromise = waitForWeixinPublishNetworkResponse( + page, + WEIXIN_PUBLISH_RESPONSE_TIMEOUT_MS, + context.signal, + ) const confirm = await clickWeixinVisibleAction(page, ['发表', '确认发表', '发布', '确认发布'], { scope: 'dialog', timeoutMs: 8_000, @@ -1831,10 +1855,13 @@ async function publishDraftFromDraftManagePage( throw new Error('weixin_gzh_playwright_page_missing') } - await page.goto(buildDraftManageUrl(meta.token), { - waitUntil: 'domcontentloaded', - }) - await waitForWeixinEditorReady(page) + await raceWithAbort( + page.goto(buildDraftManageUrl(meta.token), { + waitUntil: 'domcontentloaded', + }), + { signal: context.signal, timeoutMs: WEIXIN_EDITOR_READY_TIMEOUT_MS }, + ) + await waitForWeixinEditorReady(page, context.signal) const initialMessage = await readWeixinPageMessage(page) if (/loginpage|\/cgi-bin\/login/i.test(page.url()) || /登录超时|重新登录/.test(initialMessage)) { @@ -1844,6 +1871,7 @@ async function publishDraftFromDraftManagePage( const directResponsePromise = waitForWeixinPublishNetworkResponse( page, WEIXIN_PUBLISH_DIALOG_TIMEOUT_MS + 4_000, + context.signal, ) const popupPromise = page.waitForEvent('popup', { timeout: 8_000 }).catch(() => null) const openDialog = await clickWeixinDraftCardPublish(page, articleId, title, context.signal) @@ -1858,7 +1886,7 @@ async function publishDraftFromDraftManagePage( if (popup && !popup.isClosed()) { popup.setDefaultTimeout(WEIXIN_EDITOR_READY_TIMEOUT_MS) popup.setDefaultNavigationTimeout(WEIXIN_EDITOR_READY_TIMEOUT_MS) - await waitForWeixinEditorReady(popup) + await waitForWeixinEditorReady(popup, context.signal) const popupMessage = await readWeixinPageMessage(popup) if (/loginpage|\/cgi-bin\/login/i.test(popup.url()) || /登录超时|重新登录/.test(popupMessage)) { throw new Error('weixin_gzh_not_logged_in') @@ -1866,7 +1894,11 @@ async function publishDraftFromDraftManagePage( const popupResponse = await completeWeixinPublishConfirmation( context, popup, - waitForWeixinPublishNetworkResponse(popup, WEIXIN_PUBLISH_DIALOG_TIMEOUT_MS + 4_000), + waitForWeixinPublishNetworkResponse( + popup, + WEIXIN_PUBLISH_DIALOG_TIMEOUT_MS + 4_000, + context.signal, + ), ) await popup.close().catch(() => undefined) return popupResponse @@ -1885,10 +1917,13 @@ async function publishDraftFromEditorPage( throw new Error('weixin_gzh_playwright_page_missing') } - await page.goto(buildDraftEditUrl(articleId, meta.token), { - waitUntil: 'domcontentloaded', - }) - await waitForWeixinEditorReady(page) + await raceWithAbort( + page.goto(buildDraftEditUrl(articleId, meta.token), { + waitUntil: 'domcontentloaded', + }), + { signal: context.signal, timeoutMs: WEIXIN_EDITOR_READY_TIMEOUT_MS }, + ) + await waitForWeixinEditorReady(page, context.signal) const initialMessage = await readWeixinPageMessage(page) if (/loginpage|\/cgi-bin\/login/i.test(page.url()) || /登录超时|重新登录/.test(initialMessage)) { @@ -1898,6 +1933,7 @@ async function publishDraftFromEditorPage( const directResponsePromise = waitForWeixinPublishNetworkResponse( page, WEIXIN_PUBLISH_DIALOG_TIMEOUT_MS + 4_000, + context.signal, ) const openDialog = await clickWeixinVisibleAction(page, ['发表', '发布'], { timeoutMs: 12_000, @@ -2091,6 +2127,7 @@ export const weixinGzhAdapter: PublishAdapter = { context.reportProgress('weixin_gzh.upload_content_images') const processed = await uploadHtmlImages(html, async (sourceUrl) => uploadImage(context, meta, sourceUrl), + { signal: context.signal }, ) const content = wrapWeixinContent(processed.html) @@ -2112,6 +2149,7 @@ export const weixinGzhAdapter: PublishAdapter = { }), body: form, }, + { signal: context.signal }, ).catch( (error): WeixinOperateResponse => ({ ret: -1, diff --git a/apps/desktop-client/src/main/adapters/zhihu.ts b/apps/desktop-client/src/main/adapters/zhihu.ts index d8748e6..136c5d6 100644 --- a/apps/desktop-client/src/main/adapters/zhihu.ts +++ b/apps/desktop-client/src/main/adapters/zhihu.ts @@ -6,6 +6,7 @@ import { marked } from 'marked' import type { PublishAdapter } from './base' import { + adapterFetch, extractImageSources, fetchImageBlob, normalizeRemoteUrl, @@ -56,7 +57,7 @@ async function zhihuFetch( return await sessionFetchJson(context.session, input, { ...init, headers: buildHeaders(init?.headers), - }) + }, { signal: context.signal }) } function markdownToHtml(markdown: string): string { @@ -134,7 +135,7 @@ async function uploadBinaryImage( context: Parameters>[0], sourceUrl: string, ): Promise { - const blob = await fetchImageBlob(sourceUrl) + const blob = await fetchImageBlob(sourceUrl, { signal: context.signal }) if (!blob || !blob.type.startsWith('image/')) { return null } @@ -159,7 +160,7 @@ async function uploadBinaryImage( const stringToSign = `PUT\n\n${blob.type}\n${ossDate}\nx-oss-date:${ossDate}\nx-oss-security-token:${token.upload_token.access_token}\nx-oss-user-agent:${ossUserAgent}\n/zhihu-pics/v2-${imageHash}` const signature = hmacSha1Base64(token.upload_token.access_key || '', stringToSign) - const uploadResponse = await fetch( + const uploadResponse = await adapterFetch( `https://zhihu-pics-upload.zhimg.com/${token.upload_file.object_key}`, { method: 'PUT', @@ -173,6 +174,7 @@ async function uploadBinaryImage( }, body: blob, }, + { signal: context.signal, timeoutMs: 30_000 }, ) if (!uploadResponse.ok) { throw new Error(`zhihu_oss_upload_failed_${uploadResponse.status}`) diff --git a/apps/desktop-client/src/main/adapters/zol.ts b/apps/desktop-client/src/main/adapters/zol.ts index 180bbfa..180f062 100644 --- a/apps/desktop-client/src/main/adapters/zol.ts +++ b/apps/desktop-client/src/main/adapters/zol.ts @@ -168,6 +168,7 @@ async function fetchAccount(context: PublishAdapterContext): Promise null) const user = response?.data @@ -217,6 +218,7 @@ async function createDraftId(context: PublishAdapterContext, form: FormData): Pr headers: await zolHeaders(context), body: form, }, + { signal: context.signal }, ).catch( (error): ZolDraftResponse => ({ errcode: -1, @@ -254,6 +256,7 @@ async function uploadCoverVariant( headers: await zolHeaders(context), body: form, }, + { signal: context.signal }, ).catch( (error): ZolImageUploadResponse => ({ errcode: -1, @@ -281,7 +284,7 @@ async function uploadCover( context: PublishAdapterContext, sourceUrl: string, ): Promise { - const source = await fetchImageAssetBlob(sourceUrl) + const source = await fetchImageAssetBlob(sourceUrl, { signal: context.signal }) if (!source) { throw new Error('zol_cover_fetch_failed') } @@ -327,6 +330,7 @@ async function uploadContentImageByNetworkPic( headers: await zolHeaders(context), body: form, }, + { signal: context.signal }, ).catch(() => null) const uploaded = zolUploadedImageUrl(response) @@ -345,7 +349,7 @@ async function uploadContentImageByFile( context: PublishAdapterContext, sourceUrl: string, ): Promise { - const image = await fetchImageAssetBlob(sourceUrl) + const image = await fetchImageAssetBlob(sourceUrl, { signal: context.signal }) if (!image) { throw new Error( `zol_content_image_upload_failed:${sourceUrl.slice(0, 160)}:source_fetch_failed`, @@ -367,6 +371,7 @@ async function uploadContentImageByFile( headers: await zolHeaders(context), body: form, }, + { signal: context.signal }, ).catch( (error): ZolImageUploadResponse => ({ errcode: -1, @@ -421,10 +426,15 @@ async function applyRecommendSubjects( url.searchParams.set('keyword', '') url.searchParams.set('page', '1') - const response = await sessionFetchJson(context.session, url.toString(), { - credentials: 'include', - headers: await zolHeaders(context), - }).catch(() => null) + const response = await sessionFetchJson( + context.session, + url.toString(), + { + credentials: 'include', + headers: await zolHeaders(context), + }, + { signal: context.signal }, + ).catch(() => null) const subjects = (response?.data?.list ?? []) .filter((item) => item.subjectId != null && item.subjectName?.trim()) @@ -598,6 +608,7 @@ export const zolAdapter: PublishAdapter = { headers: await zolHeaders(context), body: form, }, + { signal: context.signal }, ).catch( (error): ZolPublishResponse => ({ errcode: -1, diff --git a/apps/desktop-client/src/main/bootstrap.ts b/apps/desktop-client/src/main/bootstrap.ts index b245cfa..6c6723e 100644 --- a/apps/desktop-client/src/main/bootstrap.ts +++ b/apps/desktop-client/src/main/bootstrap.ts @@ -66,6 +66,7 @@ import { initScheduler } from './scheduler' import { initSessionRegistry } from './session-registry' import { initSingleInstance } from './single-instance' import { + cancelDesktopTask, checkDesktopClientRelease, initTransport, listDesktopPublishTasks, @@ -1013,6 +1014,9 @@ function registerBridgeHandlers(): void { safeHandle('desktop:retry-publish-task', async (_event, taskId: string) => { return retryDesktopPublishTask(taskId) }) + safeHandle('desktop:cancel-publish-task', async (_event, taskId: string) => { + return cancelDesktopTask(taskId, { reason: 'user_cancelled_publish' }) + }) safeHandle('desktop:open-external-url', async (_event, url: string) => { if (!isSafeExternalUrl(url)) { throw new Error('unsupported_external_url') diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index 9bfa3f4..d7610f8 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -118,6 +118,7 @@ import { leaseDesktopTask, leaseMonitoringTasks, listDesktopAccounts, + markDesktopTaskPublishSubmitStarted, noteTransportHeartbeat, noteTransportPull, offlineDesktopClient, @@ -148,7 +149,11 @@ type MonitorExecutionPhase = const heartbeatIntervalMs = 15_000 const pullIntervalMs = 60_000 +const pumpWatchdogIntervalMs = 15_000 const leaseExtendIntervalMs = 60_000 +const defaultPublishTaskTimeoutMs = 5 * 60_000 +const publishTaskStaleProgressMs = 2 * 60_000 +const publishLongRunningNotificationMs = 15 * 60_000 const maxActivityItems = 60 const monitorBusinessTimeZone = 'Asia/Shanghai' const authRequiredMonitorPlatforms = new Set(aiPlatformCatalog.map((platform) => platform.id)) @@ -171,6 +176,7 @@ interface RuntimeTaskRecord { status: RuntimeTaskStatus routing: RuntimeTaskRouting leaseExpiresAt: number | null + startedAt: number | null updatedAt: number summary: string payload: Record @@ -226,6 +232,16 @@ interface ActiveRuntimeExecution { interruptGeneration: number interruptReason: string | null lastSafePointAt: number | null + startedAt: number + lastProgressAt: number + lastProgressSummary: string | null + deadlineAt: number | null + deadlineHandle: ReturnType | null + longRunningNotifiedAt: number | null + // Set the moment a publish adapter enters its irreversible submit/save_draft stage. Once set, + // an interrupted publish is completed as 'unknown' (manual reconcile) instead of 'failed', so a + // possibly-already-published article is never silently re-posted by a retry. + submitStartedAt: number | null } interface RuntimeState { @@ -241,6 +257,7 @@ interface RuntimeState { publishFallbackBackoffUntil: number heartbeatTimer: ReturnType | null pullTimer: ReturnType | null + pumpWatchdogTimer: ReturnType | null accountHealthReportTimer: ReturnType | null accountHealthReportQueue: Map accountHealthReportBadAccounts: Set @@ -271,6 +288,7 @@ const state: RuntimeState = { publishFallbackBackoffUntil: 0, heartbeatTimer: null, pullTimer: null, + pumpWatchdogTimer: null, accountHealthReportTimer: null, accountHealthReportQueue: new Map(), accountHealthReportBadAccounts: new Set(), @@ -659,7 +677,10 @@ export function getRuntimeControllerSnapshot(): RuntimeControllerSnapshot { ), tasks: [...state.tasks.values()] .sort((left, right) => right.updatedAt - left.updatedAt) - .map(({ leaseToken: _leaseToken, ...task }) => ({ ...task })), + .map(({ leaseToken: _leaseToken, ...task }) => ({ + ...task, + startedAt: state.activeExecutions.get(task.id)?.startedAt ?? task.startedAt, + })), activity: [...state.activity], lastHeartbeatAt: state.lastHeartbeatAt, lastHeartbeatStatus: state.lastHeartbeatStatus, @@ -690,6 +711,7 @@ function startRuntime(): void { connectDispatchWs() scheduleHeartbeatLoop() schedulePullLoop() + schedulePumpWatchdogLoop() void sendHeartbeat('startup') void syncAccounts('startup') @@ -714,6 +736,10 @@ function stopRuntime(): void { state.running = false state.leaseInFlightKinds.clear() for (const execution of state.activeExecutions.values()) { + if (execution.deadlineHandle) { + clearTimeout(execution.deadlineHandle) + execution.deadlineHandle = null + } execution.abortController.abort() } state.activeExecutions.clear() @@ -726,6 +752,10 @@ function stopRuntime(): void { clearInterval(state.pullTimer) state.pullTimer = null } + if (state.pumpWatchdogTimer) { + clearInterval(state.pumpWatchdogTimer) + state.pumpWatchdogTimer = null + } if (state.accountHealthReportTimer) { clearTimeout(state.accountHealthReportTimer) state.accountHealthReportTimer = null @@ -793,6 +823,17 @@ function schedulePullLoop(): void { }, pullIntervalMs) } +function schedulePumpWatchdogLoop(): void { + if (state.pumpWatchdogTimer) { + clearInterval(state.pumpWatchdogTimer) + } + + state.pumpWatchdogTimer = setInterval(() => { + notifyLongRunningPublishTasks() + pumpExecutionLoop() + }, pumpWatchdogIntervalMs) +} + function connectDispatchWs(): void { if (!canRunLive(state.session)) { return @@ -909,6 +950,7 @@ function handleTaskEvent(event: DesktopTaskEventMessage): void { status: event.status, routing: existing?.routing ?? 'rabbitmq-primary', leaseExpiresAt: event.status === 'in_progress' ? (existing?.leaseExpiresAt ?? null) : null, + startedAt: event.status === 'in_progress' ? (existing?.startedAt ?? eventUpdatedAt) : null, updatedAt: eventUpdatedAt, summary: summaryFromEventType(event.type, event.status), payload: existing?.payload ?? {}, @@ -1577,6 +1619,114 @@ function enqueuePublishTask(taskId: string, platform: string, routing: RuntimeTa syncSchedulerSurface() } +function createActiveExecution( + taskRecord: RuntimeTaskRecord, + abortController: AbortController, +): ActiveRuntimeExecution { + const now = Date.now() + return { + taskId: taskRecord.id, + kind: taskRecord.kind, + platform: taskRecord.platform, + accountId: taskRecord.accountId, + abortController, + executionPhase: taskRecord.kind === 'monitor' ? 'PREPARING' : null, + interruptRequested: + taskRecord.kind === 'monitor' && Boolean(taskRecord.payload.interrupt_requested), + interruptGeneration: + taskRecord.kind === 'monitor' + ? toInterruptGeneration(taskRecord.payload.interrupt_generation) + : 0, + interruptReason: taskRecord.kind === 'monitor' ? interruptReasonFromTask(taskRecord) : null, + lastSafePointAt: taskRecord.kind === 'monitor' ? now : null, + startedAt: now, + lastProgressAt: now, + lastProgressSummary: taskRecord.summary, + deadlineAt: null, + deadlineHandle: null, + longRunningNotifiedAt: null, + submitStartedAt: null, + } +} + +function armPublishTaskDeadline( + taskRecord: RuntimeTaskRecord, + active: ActiveRuntimeExecution, +): void { + if (taskRecord.kind !== 'publish') { + return + } + + const taskDeadlineMs = resolvePublishTaskTimeoutMs(taskRecord.platform) + active.deadlineAt = active.startedAt + taskDeadlineMs + active.deadlineHandle = setTimeout(() => { + if (active.abortController.signal.aborted) { + return + } + + const timeoutError = new Error('publish_task_timeout') + recordActivity( + 'warn', + '任务超时', + `${taskRecord.title} 执行超过 ${Math.round(taskDeadlineMs / 1000)}s，已中止并释放。`, + ) + updateTaskProgress(taskRecord.id, `${taskRecord.title} 执行超时，正在释放任务。`) + active.abortController.abort(timeoutError) + }, taskDeadlineMs) +} + +function resolvePublishTaskTimeoutMs(platform: string): number { + const envKey = `GEO_DESKTOP_PUBLISH_TIMEOUT_${platform.toUpperCase().replace(/[^A-Z0-9]/g, '_')}_MS` + return resolvePositiveEnvInteger(envKey) ?? resolvePositiveEnvInteger('GEO_DESKTOP_PUBLISH_TIMEOUT_MS') ?? defaultPublishTaskTimeoutMs +} + +function resolvePositiveEnvInteger(key: string): number | null { + const raw = process.env[key]?.trim() + if (!raw) { + return null + } + const parsed = Number.parseInt(raw, 10) + return Number.isFinite(parsed) && parsed > 0 ? parsed : null +} + +// Stages emitted by an adapter right before it performs an irreversible platform write (creating +// the article/draft). After this point the article may exist on the platform regardless of whether +// the local task records it, so an interrupted publish must not be auto-retried. +function isIrreversiblePublishStage(stage: string): boolean { + const normalized = stage.trim().toLowerCase() + return ( + normalized.endsWith('.submit') || + normalized.endsWith('.save_draft') || + normalized === 'submit' || + normalized === 'save_draft' + ) +} + +// Persist a durable "submit started" marker (server-side, before/at the irreversible POST) and a +// local flag the abort path reads. Fires at most once per execution and never blocks publishing. +function markPublishSubmitStartedIfNeeded(taskId: string, stage: string): void { + if (!isIrreversiblePublishStage(stage)) { + return + } + const active = state.activeExecutions.get(taskId) + if (!active || active.kind !== 'publish' || active.submitStartedAt !== null) { + return + } + active.submitStartedAt = Date.now() + state.activeExecutions.set(taskId, active) + + const leaseToken = state.tasks.get(taskId)?.leaseToken + if (!leaseToken) { + return + } + void markDesktopTaskPublishSubmitStarted(taskId, leaseToken).catch((error) => { + console.warn('[desktop-runtime] mark publish submit started failed', { + taskId, + message: error instanceof Error ? error.message : String(error), + }) + }) +} + function pumpExecutionLoop(): void { if (!state.running) { return @@ -1837,6 +1987,7 @@ function enqueueLeasedMonitoringTasks( status: 'queued', routing, leaseExpiresAt: parseTimestamp(task.lease_expires_at), + startedAt: null, updatedAt: now, summary: `已领取监控租约，等待本地调度执行（${describeRouting(routing)}）。`, payload: { @@ -1920,18 +2071,7 @@ async function executeLeasedMonitoringTask( recordActivity('info', '监控任务开始执行', `${taskRecord.title} 已进入本地执行队列。`) const abortController = new AbortController() - state.activeExecutions.set(taskRecord.id, { - taskId: taskRecord.id, - kind: 'monitor', - platform: taskRecord.platform, - accountId: taskRecord.accountId, - abortController, - executionPhase: 'PREPARING', - interruptRequested: Boolean(taskRecord.payload.interrupt_requested), - interruptGeneration: toInterruptGeneration(taskRecord.payload.interrupt_generation), - interruptReason: interruptReasonFromTask(taskRecord), - lastSafePointAt: Date.now(), - }) + state.activeExecutions.set(taskRecord.id, createActiveExecution(taskRecord, abortController)) noteMonitorTaskActivated(taskId) syncSchedulerSurface() queueMicrotask(() => pumpExecutionLoop()) @@ -2382,6 +2522,7 @@ function noteMonitorExecutionSafePoint(taskId: string, phase: MonitorExecutionPh active.executionPhase = phase } active.lastSafePointAt = Date.now() + active.lastProgressAt = Date.now() state.activeExecutions.set(taskId, active) const taskRecord = state.tasks.get(taskId) @@ -2406,6 +2547,7 @@ function requestMonitorStaleBusinessDateDrop(taskId: string, staleBusinessDate: active.interruptReason = 'stale_business_date' active.lastSafePointAt = Date.now() + active.lastProgressAt = Date.now() state.activeExecutions.set(taskId, active) const existing = state.tasks.get(taskId) @@ -2523,22 +2665,9 @@ async function executeLeasedTask( ) const abortController = new AbortController() - state.activeExecutions.set(taskRecord.id, { - taskId: taskRecord.id, - kind: taskRecord.kind, - platform: taskRecord.platform, - accountId: taskRecord.accountId, - abortController, - executionPhase: taskRecord.kind === 'monitor' ? 'PREPARING' : null, - interruptRequested: - taskRecord.kind === 'monitor' && Boolean(taskRecord.payload.interrupt_requested), - interruptGeneration: - taskRecord.kind === 'monitor' - ? toInterruptGeneration(taskRecord.payload.interrupt_generation) - : 0, - interruptReason: taskRecord.kind === 'monitor' ? interruptReasonFromTask(taskRecord) : null, - lastSafePointAt: taskRecord.kind === 'monitor' ? Date.now() : null, - }) + const activeExecution = createActiveExecution(taskRecord, abortController) + armPublishTaskDeadline(taskRecord, activeExecution) + state.activeExecutions.set(taskRecord.id, activeExecution) if (task.kind === 'monitor') { noteMonitorTaskLeased(task, routing) @@ -2598,7 +2727,23 @@ async function executeLeasedTask( const structuredError = toStructuredError(error) const accountIdentity = accountIdentityFromTask(taskRecord) - if (accountIdentity) { + // If a publish task was interrupted AFTER it may have entered the irreversible submit phase, + // the article might already be live on the platform. Completing as a clean 'failed' would drop + // it from the server dedup set and let a retry double-post. Complete as 'unknown' instead so it + // stays for manual reconcile. (A pre-submit interruption is a true failure and stays 'failed'.) + const submitMaybeStarted = + taskRecord.kind === 'publish' && activeExecution.submitStartedAt !== null + const terminalStatus: 'failed' | 'unknown' = submitMaybeStarted ? 'unknown' : 'failed' + const interruptedSummary = '发布可能已提交到平台，结果待人工确认，已避免自动重复发布。' + // Tag the persisted error so the desktop UI can require explicit confirmation before a manual + // retry of a possibly-already-published task (mirrors the server recovery payload marker). + const terminalError = submitMaybeStarted + ? { ...structuredError, publish_submit_uncertain: true } + : structuredError + + // A submit-phase interruption is typically a network/timeout stall, not an auth problem — don't + // wrongly degrade account health for it. + if (accountIdentity && !submitMaybeStarted) { await reportAccountFailure(accountIdentity, { summary: failureMessage, error: structuredError, @@ -2615,22 +2760,24 @@ async function executeLeasedTask( try { const completed = await completeDesktopTask(taskRecord.id, { lease_token: leased.lease_token, - status: 'failed', - error: structuredError, + status: terminalStatus, + error: terminalError, }) upsertTaskFromInfo(completed, { routing, - summary: failureMessage || '桌面任务执行失败。', + summary: submitMaybeStarted ? interruptedSummary : failureMessage || '桌面任务执行失败。', attemptId: null, leaseToken: null, }) } catch (completeError) { const existing = state.tasks.get(taskRecord.id) if (existing) { - existing.status = 'failed' - existing.error = structuredError - existing.summary = `${failureMessage || '桌面任务执行失败。'}（result 回写失败：${errorMessage(completeError)}）` + existing.status = terminalStatus + existing.error = terminalError + existing.summary = submitMaybeStarted + ? `${interruptedSummary}（result 回写失败：${errorMessage(completeError)}）` + : `${failureMessage || '桌面任务执行失败。'}（result 回写失败：${errorMessage(completeError)}）` existing.updatedAt = Date.now() state.tasks.set(taskRecord.id, existing) } @@ -2638,12 +2785,18 @@ async function executeLeasedTask( noteLeaseReleased('failed', taskRecord.id) recordActivity( - 'danger', - '任务执行失败', - `${taskRecord.title} 执行失败：${failureMessage || '未知错误'}`, + submitMaybeStarted ? 'warn' : 'danger', + submitMaybeStarted ? '发布结果待确认' : '任务执行失败', + submitMaybeStarted + ? `${taskRecord.title} 进入提交阶段后中断，可能已发布，已转人工确认以避免重复发文。` + : `${taskRecord.title} 执行失败：${failureMessage || '未知错误'}`, ) } finally { clearInterval(extendHandle) + if (activeExecution.deadlineHandle) { + clearTimeout(activeExecution.deadlineHandle) + activeExecution.deadlineHandle = null + } if (taskRecord.kind === 'publish') { notePublishTaskCompleted(taskRecord.id, taskRecord.platform) } @@ -2656,6 +2809,7 @@ async function executeLeasedTask( async function extendActiveLease(taskId: string, leaseToken: string): Promise { const existing = state.tasks.get(taskId) + const active = state.activeExecutions.get(taskId) if (existing?.kind === 'monitor') { const staleBusinessDate = resolveStaleMonitoringBusinessDate(existing.payload) if (staleBusinessDate) { @@ -2663,6 +2817,28 @@ async function extendActiveLease(taskId: string, leaseToken: string): Promise= publishTaskStaleProgressMs) { + const detail = active.lastProgressSummary ? `，最后进度：${active.lastProgressSummary}` : '' + existing.summary = `发布任务超过 ${Math.round(idleMs / 1000)} 秒没有新进度，已中止并释放，准备由服务端重排${detail}。` + existing.updatedAt = Date.now() + state.tasks.set(taskId, existing) + recordActivity( + 'warn', + '发布任务疑似卡住', + `${existing.title} ${existing.summary}`, + ) + // Abort the local execution instead of merely letting the lease lapse: this + // releases the slot immediately and, crucially, closes the window where the + // server could re-queue the task while a stalled adapter is still mid-publish + // (which would double-post a non-idempotent article). + if (!active.abortController.signal.aborted) { + active.abortController.abort(new Error('publish_stale_progress')) + } + return + } + } try { const task = await extendDesktopTask(taskId, { lease_token: leaseToken }) @@ -2673,7 +2849,10 @@ async function extendActiveLease(taskId: string, leaseToken: string): Promise sum + item.percentCPUUsage, 0) @@ -3371,19 +3606,10 @@ function resolveRuntimeHealthTotalConcurrency(hardwareCap: number): number { mainPrivateBytesKB >= 1_000_000 || processCount >= 40 ) { - return minimumForegroundTotalConcurrency + return activeAdmissionCount() < minimumForegroundTotalConcurrency } - if ( - totalCPUUsage >= 120 || - totalPrivateBytesKB >= 1_800_000 || - mainPrivateBytesKB >= 750_000 || - processCount >= 32 - ) { - return Math.min(hardwareCap, 3) - } - - return hardwareCap + return true } function resolveFeatureFlagTotalConcurrency(): number | null { diff --git a/apps/desktop-client/src/main/runtime-events.ts b/apps/desktop-client/src/main/runtime-events.ts index 75c81b9..6feb4f8 100644 --- a/apps/desktop-client/src/main/runtime-events.ts +++ b/apps/desktop-client/src/main/runtime-events.ts @@ -1,4 +1,4 @@ -type RuntimeInvalidationReason = 'account-health' | 'publish-task-lease' +type RuntimeInvalidationReason = 'account-health' | 'publish-task-lease' | 'publish-task-progress' type RuntimeInvalidationListener = (event: { reason: RuntimeInvalidationReason diff --git a/apps/desktop-client/src/main/transport/api-client.ts b/apps/desktop-client/src/main/transport/api-client.ts index c88e0dd..6807d39 100644 --- a/apps/desktop-client/src/main/transport/api-client.ts +++ b/apps/desktop-client/src/main/transport/api-client.ts @@ -690,6 +690,16 @@ export async function completeDesktopTask( ) } +export async function markDesktopTaskPublishSubmitStarted( + taskId: string, + leaseToken: string, +): Promise { + await desktopPost<{ ok: boolean }, { lease_token: string }>( + `/api/desktop/tasks/${taskId}/publish-submit-marker`, + { lease_token: leaseToken }, + ) +} + export async function cancelDesktopTask( taskId: string, payload: CancelDesktopTaskRequest, diff --git a/apps/desktop-client/src/preload/bridge.ts b/apps/desktop-client/src/preload/bridge.ts index 90d4f5d..7b20d65 100644 --- a/apps/desktop-client/src/preload/bridge.ts +++ b/apps/desktop-client/src/preload/bridge.ts @@ -7,6 +7,7 @@ import type { DesktopClientUpdateResult, DesktopPublishTaskListResponse, DesktopRuntimeSessionSyncRequest, + DesktopTaskInfo, ListDesktopPublishTasksParams, } from '@geo/shared-types' import type { IpcRendererEvent } from 'electron' @@ -167,6 +168,8 @@ const desktopBridge = { ) as Promise, retryPublishTask: (taskId: string) => ipcRenderer.invoke('desktop:retry-publish-task', taskId) as Promise, + cancelPublishTask: (taskId: string) => + ipcRenderer.invoke('desktop:cancel-publish-task', taskId) as Promise, openExternalUrl: (url: string) => ipcRenderer.invoke('desktop:open-external-url', url) as Promise, openSettingsWindow: () => ipcRenderer.invoke('desktop:open-settings-window') as Promise, @@ -197,7 +200,7 @@ const desktopBridge = { ipcRenderer.invoke('desktop:set-window-mode', mode, request) as Promise, onRuntimeInvalidated: ( listener: (event: { - reason: 'account-health' | 'publish-task-lease' + reason: 'account-health' | 'publish-task-lease' | 'publish-task-progress' at: number accountId?: string taskId?: string @@ -206,7 +209,7 @@ const desktopBridge = { const wrapped = ( _event: IpcRendererEvent, payload: { - reason: 'account-health' | 'publish-task-lease' + reason: 'account-health' | 'publish-task-lease' | 'publish-task-progress' at: number accountId?: string taskId?: string diff --git a/apps/desktop-client/src/renderer/env.d.ts b/apps/desktop-client/src/renderer/env.d.ts index fcf6d33..e0a9674 100644 --- a/apps/desktop-client/src/renderer/env.d.ts +++ b/apps/desktop-client/src/renderer/env.d.ts @@ -77,6 +77,7 @@ declare global { params?: ListDesktopPublishTasksParams, ): Promise retryPublishTask(taskId: string): Promise + cancelPublishTask(taskId: string): Promise openExternalUrl(url: string): Promise openSettingsWindow(): Promise getAppSettings(): Promise @@ -96,7 +97,7 @@ declare global { ): Promise onRuntimeInvalidated( listener: (event: { - reason: 'account-health' | 'publish-task-lease' + reason: 'account-health' | 'publish-task-lease' | 'publish-task-progress' at: number accountId?: string taskId?: string diff --git a/apps/desktop-client/src/renderer/types.ts b/apps/desktop-client/src/renderer/types.ts index 819d8c1..c9a6d21 100644 --- a/apps/desktop-client/src/renderer/types.ts +++ b/apps/desktop-client/src/renderer/types.ts @@ -59,6 +59,7 @@ export interface RuntimeTask { status: 'queued' | 'in_progress' | 'succeeded' | 'failed' | 'unknown' | 'aborted' routing: 'rabbitmq-primary' | 'db-recovery' leaseExpiresAt: number | null + startedAt: number | null updatedAt: number summary: string } diff --git a/apps/desktop-client/src/renderer/views/PublishManagementView.vue b/apps/desktop-client/src/renderer/views/PublishManagementView.vue index a874121..96e316e 100644 --- a/apps/desktop-client/src/renderer/views/PublishManagementView.vue +++ b/apps/desktop-client/src/renderer/views/PublishManagementView.vue @@ -7,9 +7,10 @@ import { ReloadOutlined, RetweetOutlined, SearchOutlined, + StopOutlined, } from '@ant-design/icons-vue' import type { DesktopPublishTaskListResponse, DesktopTaskInfo, JsonValue } from '@geo/shared-types' -import { notification } from 'ant-design-vue' +import { Modal, notification } from 'ant-design-vue' import { computed, onMounted, onUnmounted, ref, watch } from 'vue' import { normalizePublisherErrorMessage } from '../../shared/publisher-errors' @@ -27,6 +28,8 @@ const PAGE_SIZE = 10 const INLINE_ERROR_HEAD_CHARS = 48 const INLINE_ERROR_TAIL_CHARS = 10 const ACTIVE_TASK_POLL_INTERVAL_MS = 5_000 +const DEFAULT_ESTIMATED_TASK_MS = 3 * 60_000 +const PROGRESS_REFRESH_THROTTLE_MS = 3_000 const publishPlatformIndex: Record = Object.fromEntries(desktopPublishMediaCatalog.map((item) => [item.id, item])) @@ -40,6 +43,8 @@ interface PublishTaskItem { accountName: string status: DesktopTaskInfo['status'] summary: string + queueSummary: string | null + running: boolean attempts: number leaseExpiresAt: number | null createdAt: number @@ -48,11 +53,13 @@ interface PublishTaskItem { publishType: string | null fallbackReason: string | null errorMessage: string | null + submitUncertain: boolean complianceBlockedRecordId: number | null complianceBlockedAt: number | null } let refreshTimer: ReturnType | null = null +let lastProgressRefreshAt = 0 let runtimeInvalidationUnsubscribe: (() => void) | null = null function platformMeta(platform: string) { @@ -114,8 +121,9 @@ function hasActivePublishingTask(page: DesktopPublishTaskListResponse | null): b function statusTone(status: DesktopTaskInfo['status']): 'info' | 'warn' | 'success' | 'danger' { switch (status) { - case 'queued': case 'in_progress': + return 'info' + case 'queued': return 'warn' case 'succeeded': return 'success' @@ -311,6 +319,58 @@ function summaryForTask( } } +function elapsedTimeLabel(since: number, now = Date.now()): string { + const elapsedMs = Math.max(0, now - since) + if (elapsedMs < 60_000) { + return '不到 1 分钟' + } + if (elapsedMs < 60 * 60_000) { + return `${Math.max(1, Math.floor(elapsedMs / 60_000))} 分钟` + } + const hours = Math.max(1, Math.floor(elapsedMs / 3_600_000)) + const minutes = Math.floor((elapsedMs % 3_600_000) / 60_000) + return minutes > 0 ? `${hours} 小时 ${minutes} 分钟` : `${hours} 小时` +} + +function stageSummaryFromRuntime(raw: string | null | undefined): string | null { + const summary = raw?.trim() + if (!summary) { + return null + } + const stage = summary.match(/publish adapter progress:\s*([a-z0-9_.-]+)/i)?.[1] ?? summary + const stageKey = stage.trim().toLowerCase() + const stageMap: Record = { + detect_login: '正在校验登录态', + normalize_html: '正在整理正文', + upload_content_images: '正在上传正文图片', + upload_cover: '正在上传封面图', + submit: '正在提交发布', + save_draft: '正在保存草稿', + create_draft: '正在创建草稿', + create_article_id: '正在打开投稿入口', + calculate_signature: '正在计算页面签名', + get_token: '正在获取发布凭证', + publish: '正在确认发表', + publish_draft: '正在发表草稿', + resolve_public_url: '正在确认公开链接', + resolve_article_url: '正在确认文章链接', + apply_subjects: '正在匹配推荐话题', + disable_mass_notification: '正在关闭群发通知', + continue_publish_without_notification: '正在继续确认发表', + } + const suffix = stageKey.split('.').pop() ?? stageKey + if (stageMap[stageKey]) { + return stageMap[stageKey] + } + if (stageMap[suffix]) { + return stageMap[suffix] + } + if (summary.startsWith('publish adapter progress:')) { + return `正在执行 ${stage}` + } + return summary +} + function parseTimestamp(value: string | null): number { if (!value) { return Date.now() @@ -330,6 +390,7 @@ const error = ref(null) const consolePendingTaskId = ref(null) const copiedErrorTaskId = ref(null) const retryingTaskId = ref(null) +const cancelingTaskId = ref(null) function notifySuccess(message: string) { notification.success({ @@ -428,11 +489,22 @@ function bindPublishLeaseListener() { } runtimeInvalidationUnsubscribe = window.desktopBridge.app.onRuntimeInvalidated((event) => { - if (event.reason !== 'publish-task-lease') { + if (event.reason !== 'publish-task-lease' && event.reason !== 'publish-task-progress') { return } - void refreshTasks(1) + if (event.reason === 'publish-task-lease') { + void refreshTasks(1) + return + } + + // publish-task-progress fires once per adapter stage; throttle the server-list + // refresh so the status badge / queue position stay fresh without hammering the API. + const now = Date.now() + if (now - lastProgressRefreshAt >= PROGRESS_REFRESH_THROTTLE_MS) { + lastProgressRefreshAt = now + void refreshTasks() + } }) } @@ -480,7 +552,25 @@ function canRetryTask(record: PublishTaskItem): boolean { return record.status === 'failed' || record.status === 'aborted' || record.status === 'unknown' } +function confirmUncertainRetry(record: PublishTaskItem): Promise { + return new Promise((resolve) => { + Modal.confirm({ + title: '该文章可能已发布', + content: `「${record.title}」上次执行已进入提交阶段后中断，可能已经发布到平台。重新提交可能导致重复发文。请先到平台核实，确认尚未发布后再继续。`, + okText: '已核实，仍要重新提交', + okType: 'danger', + cancelText: '取消', + onOk: () => resolve(true), + onCancel: () => resolve(false), + }) + }) +} + async function retryTask(record: PublishTaskItem) { + if (record.submitUncertain && !(await confirmUncertainRetry(record))) { + return + } + retryingTaskId.value = record.id try { @@ -512,6 +602,27 @@ async function retryTask(record: PublishTaskItem) { } } +function canCancelTask(record: PublishTaskItem): boolean { + return record.status === 'queued' +} + +async function cancelTask(record: PublishTaskItem) { + cancelingTaskId.value = record.id + + try { + await window.desktopBridge.app.cancelPublishTask(record.id) + notifySuccess('已取消排队中的发布任务') + await refreshTasks(currentPage.value) + } catch (err) { + notifyActionError( + normalizeTaskErrorMessage(unwrapClientErrorMessage(err, '取消发布任务失败')) ?? + '取消发布任务失败', + ) + } finally { + cancelingTaskId.value = null + } +} + async function copyTextToClipboard(text: string): Promise { if (navigator.clipboard?.writeText) { await navigator.clipboard.writeText(text) @@ -592,9 +703,14 @@ const accountMap = computed(() => { return new Map((snapshot.value?.accounts ?? []).map((item) => [item.id, item])) }) +const runtimeTaskMap = computed(() => { + return new Map((snapshot.value?.tasks ?? []).map((item) => [item.id, item])) +}) + const publishTasks = computed(() => (taskPage.value?.items ?? []).map((task) => { const normalizedStatus = normalizePublishTaskStatus(task.status) + const runtimeTask = runtimeTaskMap.value.get(task.id) const payload = task.payload ?? {} const result = task.result ?? {} const taskError = task.error ?? {} @@ -619,6 +735,44 @@ const publishTasks = computed(() => const articleId = extractNumber(payload.article_id) ?? extractNumber(nestedContentRef?.article_id ?? nestedContentRef?.id) + const pendingTasks = + taskPage.value?.items.filter((item) => + isPendingStatus(normalizePublishTaskStatus(item.status)), + ) ?? [] + const queuedTasks = pendingTasks.filter( + (item) => normalizePublishTaskStatus(item.status) === 'queued', + ) + const queuedIndex = queuedTasks.findIndex((item) => item.id === task.id) + const runningCount = pendingTasks.filter( + (item) => normalizePublishTaskStatus(item.status) === 'in_progress', + ).length + // Queue position / ahead-count are only globally accurate on page 1, where all + // pending tasks (which the server always sorts first) are present. On later pages + // the per-page index would restart from 1 and mislead the user, so we drop the + // position there and keep only the always-correct queued elapsed time. + const positionIsAccurate = currentPage.value === 1 + const queuePosition = queuedIndex >= 0 && positionIsAccurate ? queuedIndex + 1 : null + const aheadCount = queuePosition === null ? null : runningCount + Math.max(0, queuePosition - 1) + const estimatedMs = + aheadCount === null ? null : Math.max(0, aheadCount) * DEFAULT_ESTIMATED_TASK_MS + const queuedSummary = + normalizedStatus === 'queued' + ? `${ + queuePosition !== null ? `第 ${queuePosition} 个，前面还有 ${aheadCount} 个 / ` : '' + }已排队 ${elapsedTimeLabel(parseTimestamp(task.created_at))}${ + estimatedMs && estimatedMs > 0 + ? ` / 预计 ${Math.max(1, Math.ceil(estimatedMs / 60_000))} 分钟后执行` + : '' + }` + : null + const runningSummary = + normalizedStatus === 'in_progress' + ? `已执行 ${elapsedTimeLabel( + runtimeTask?.startedAt ?? runtimeTask?.updatedAt ?? parseTimestamp(task.updated_at), + )}` + : null + const runtimeSummary = + normalizedStatus === 'in_progress' ? stageSummaryFromRuntime(runtimeTask?.summary) : null return { id: task.id, @@ -628,7 +782,9 @@ const publishTasks = computed(() => accountId: task.target_account_id, accountName: accountNameMap.value.get(task.target_account_id) ?? '待同步账号', status: normalizedStatus, - summary: summaryForTask(normalizedStatus, publishType, fallbackReason), + summary: runtimeSummary ?? summaryForTask(normalizedStatus, publishType, fallbackReason), + queueSummary: queuedSummary ?? runningSummary, + running: normalizedStatus === 'in_progress', attempts: task.attempts, leaseExpiresAt: task.lease_expires_at ? parseTimestamp(task.lease_expires_at) : null, createdAt: parseTimestamp(task.created_at), @@ -648,6 +804,7 @@ const publishTasks = computed(() => extractString(taskError.code), task.platform, ), + submitUncertain: task.status === 'unknown' && taskError.publish_submit_uncertain === true, complianceBlockedRecordId, complianceBlockedAt: task.compliance_blocked_at ? parseTimestamp(task.compliance_blocked_at) : null, } @@ -791,7 +948,13 @@ const tableScroll = { x: 1080 } as const