Compare commits

...

2 Commits

Author SHA1 Message Date
root ed48674ab5 fix: allow retry after definitive publish failures
Frontend CI / Frontend (push) Successful in 3m18s
Backend CI / Backend (push) Successful in 16m16s
Desktop Client Build / Publish Client Artifacts to NAS (push) Successful in 26s
Desktop Client Build / Resolve Build Metadata (push) Successful in 26s
Desktop Client Build / Build Desktop Client (push) Successful in 25m45s
2026-06-25 00:22:59 +08:00
root 2ec266201c fix desktop smzdm csrf publishing 2026-06-25 00:18:24 +08:00
17 changed files with 761 additions and 269 deletions
@@ -21,6 +21,8 @@ import {
encodeSmzdmForm, encodeSmzdmForm,
extractSmzdmArticleIdFromHref, extractSmzdmArticleIdFromHref,
getSmzdmCoverCropRect, getSmzdmCoverCropRect,
isSmzdmCsrfMissingMessage,
smzdmCsrfHeaders,
smzdmArticleUrl, smzdmArticleUrl,
} from './smzdm' } from './smzdm'
@@ -54,6 +56,18 @@ describe('smzdm adapter helpers', () => {
expect(buildSmzdmAwne('4242025233', 123)).toBe('m2At8ovjSz5kmnLNUFEW9g==') expect(buildSmzdmAwne('4242025233', 123)).toBe('m2At8ovjSz5kmnLNUFEW9g==')
}) })
it('builds SMZDM CSRF headers like the reference extension submit request', () => {
expect(smzdmCsrfHeaders('token-1')).toMatchObject({
_csrf_token: 'token-1',
'x-csrf-token': 'token-1',
})
})
it('recognizes SMZDM missing csrf responses', () => {
expect(isSmzdmCsrfMissingMessage('CSRF token缺失')).toBe(true)
expect(isSmzdmCsrfMissingMessage('csrf token missing')).toBe(true)
})
it('centers the cover crop at the required SMZDM ratio', () => { it('centers the cover crop at the required SMZDM ratio', () => {
const crop = getSmzdmCoverCropRect(2000, 1000) const crop = getSmzdmCoverCropRect(2000, 1000)
+234 -105
View File
@@ -2,19 +2,18 @@ import { Buffer } from 'node:buffer'
import { createCipheriv, createHash } from 'node:crypto' import { createCipheriv, createHash } from 'node:crypto'
import type { JsonValue } from '@geo/shared-types' import type { JsonValue } from '@geo/shared-types'
import { nativeImage } from 'electron'
import { PUBLISH_IMAGE_FETCH_FAILED_MESSAGE } from '../../shared/publisher-errors' import { PUBLISH_IMAGE_FETCH_FAILED_MESSAGE } from '../../shared/publisher-errors'
import { resolveDesktopApiURL } from '../transport/api-client'
import type { PublishAdapter, PublishAdapterContext } from './base' import type { PublishAdapter, PublishAdapterContext } from './base'
import { import {
adapterFetch,
extractImageSources, extractImageSources,
normalizeArticleHtml, normalizeArticleHtml,
raceWithAbort, raceWithAbort,
sessionCookieHeader, sessionCookieHeader,
sessionCookieValue,
sessionFetchJson, sessionFetchJson,
} from './common' } from './common'
import { fetchImageAssetBlob } from './media-image'
type SmzdmPublishType = 'publish' type SmzdmPublishType = 'publish'
@@ -64,6 +63,14 @@ type SmzdmCropResponse = {
}> }>
} }
type SmzdmEditorTokenResponse = {
error_code?: number | string
error_msg?: string
data?: {
token?: string
}
}
type SmzdmSubmitResponse = { type SmzdmSubmitResponse = {
error_code?: number | string error_code?: number | string
error_msg?: string error_msg?: string
@@ -75,12 +82,6 @@ type SmzdmSubmitResponse = {
} }
} }
type SmzdmImageBlob = {
blob: Blob
width: number
height: number
}
type SmzdmUploadedImage = { type SmzdmUploadedImage = {
id: string id: string
origin: string origin: string
@@ -151,6 +152,29 @@ function isSmzdmChallengeMessage(message: string): boolean {
) )
} }
export function isSmzdmCsrfMissingMessage(message: string): boolean {
return /csrf/i.test(message) && /(缺失|missing|required|invalid|校验|验证|403)/i.test(message)
}
export function smzdmCsrfHeaders(token: string): Record<string, string> {
return {
_csrf_token: token,
'x-csrf-token': token,
'x-xsrf-token': token,
'csrf-token': token,
}
}
function appendSmzdmCsrfFields(target: FormData | URLSearchParams, token: string): void {
if (!token) {
return
}
target.append('_csrf', token)
target.append('csrf_token', token)
target.append('csrfToken', token)
target.append('csrf', token)
}
async function waitForHumanPace(signal?: AbortSignal): Promise<void> { async function waitForHumanPace(signal?: AbortSignal): Promise<void> {
if (signal?.aborted) { if (signal?.aborted) {
throw new Error('adapter_aborted') throw new Error('adapter_aborted')
@@ -178,16 +202,152 @@ async function smzdmCookieHeader(context: PublishAdapterContext): Promise<string
return await sessionCookieHeader(context.session, 'smzdm.com').catch(() => '') return await sessionCookieHeader(context.session, 'smzdm.com').catch(() => '')
} }
async function smzdmCookieCsrfToken(context: PublishAdapterContext): Promise<string> {
const names = [
'csrf_token',
'csrfToken',
'_csrf',
'csrf',
'XSRF-TOKEN',
'xsrf-token',
'csrftoken',
'CSRF-TOKEN',
'SMZDM_CSRF_TOKEN',
]
for (const name of names) {
const value = await sessionCookieValue(context.session, 'smzdm.com', name).catch(() => '')
const decoded = decodeCookieText(value).trim()
if (decoded) {
return decoded
}
}
const cookies = await context.session.cookies.get({ domain: 'smzdm.com' }).catch(() => [])
const matched = cookies.find((cookie) => /(csrf|xsrf)/i.test(cookie.name) && cookie.value.trim())
return matched ? decodeCookieText(matched.value).trim() : ''
}
async function smzdmPageCsrfToken(context: PublishAdapterContext): Promise<string> {
const page = context.playwright?.page
if (!page) {
return ''
}
return await page
.evaluate(() => {
const candidates: string[] = []
const push = (value: unknown) => {
if (typeof value !== 'string') {
return
}
const trimmed = value.trim()
if (trimmed && !candidates.includes(trimmed)) {
candidates.push(trimmed)
}
}
const selectors = [
'meta[name="csrf-token"]',
'meta[name="csrf_token"]',
'meta[name="_csrf"]',
'meta[name="csrf"]',
'meta[name="x-csrf-token"]',
'meta[name="xsrf-token"]',
]
for (const selector of selectors) {
push(document.querySelector<HTMLMetaElement>(selector)?.content)
}
const inputNames = ['_csrf', 'csrf', 'csrf_token', 'csrfToken', 'x-csrf-token']
for (const name of inputNames) {
push(document.querySelector<HTMLInputElement>(`input[name="${name}"]`)?.value)
}
const globals = window as unknown as Window &
Record<string, unknown> & {
csrfToken?: string
csrf_token?: string
_csrf?: string
CSRF_TOKEN?: string
SMZDM_CSRF_TOKEN?: string
PAGE_CONFIG?: Record<string, unknown>
__INITIAL_STATE__?: Record<string, unknown>
}
push(globals.csrfToken)
push(globals.csrf_token)
push(globals._csrf)
push(globals.CSRF_TOKEN)
push(globals.SMZDM_CSRF_TOKEN)
push(globals.PAGE_CONFIG?.csrfToken)
push(globals.PAGE_CONFIG?.csrf_token)
push(globals.__INITIAL_STATE__?.csrfToken)
push(globals.__INITIAL_STATE__?.csrf_token)
const readStorage = (storage: Storage) => {
for (let index = 0; index < storage.length; index += 1) {
const key = storage.key(index) || ''
if (/(csrf|xsrf)/i.test(key)) {
push(storage.getItem(key))
}
}
}
try {
readStorage(window.localStorage)
} catch {
// Ignore inaccessible storage in restricted platform pages.
}
try {
readStorage(window.sessionStorage)
} catch {
// Ignore inaccessible storage in restricted platform pages.
}
for (const item of document.cookie.split(';')) {
const [rawName, ...rawValue] = item.split('=')
if (/(csrf|xsrf)/i.test(rawName || '')) {
push(decodeURIComponent(rawValue.join('=')))
}
}
return candidates[0] || ''
})
.catch(() => '')
}
async function smzdmCsrfToken(context: PublishAdapterContext): Promise<string> {
return (await smzdmCookieCsrfToken(context)) || (await smzdmPageCsrfToken(context))
}
async function fetchSmzdmEditorToken(context: PublishAdapterContext): Promise<string> {
const response = await sessionFetchJson<SmzdmEditorTokenResponse>(
context.session,
`${SMZDM_ORIGIN}/api/editor/get_token`,
{
method: 'GET',
credentials: 'include',
headers: await smzdmHeaders(context, {}, ''),
},
{ signal: context.signal },
).catch(() => null)
return stringId(response?.data?.token) || (await smzdmCsrfToken(context))
}
async function smzdmHeaders( async function smzdmHeaders(
context: PublishAdapterContext, context: PublishAdapterContext,
extra: Record<string, string> = {}, extra: Record<string, string> = {},
csrfToken?: string,
): Promise<Record<string, string>> { ): Promise<Record<string, string>> {
const cookie = await smzdmCookieHeader(context) const cookie = await smzdmCookieHeader(context)
const token = csrfToken ?? (await smzdmCsrfToken(context))
return { return {
accept: 'application/json, text/plain, */*', accept: 'application/json, text/plain, */*',
origin: SMZDM_ORIGIN, origin: SMZDM_ORIGIN,
referer: SMZDM_TOU_GAO_URL, referer: SMZDM_TOU_GAO_URL,
'x-requested-with': 'XMLHttpRequest',
...(cookie ? { cookie } : {}), ...(cookie ? { cookie } : {}),
...(token ? smzdmCsrfHeaders(token) : {}),
...extra, ...extra,
} }
} }
@@ -289,81 +449,6 @@ async function getArticleId(context: PublishAdapterContext): Promise<string> {
return articleId return articleId
} }
function buildAssetURLCandidates(sourceUrl: string): string[] {
const trimmed = sourceUrl.trim()
if (!trimmed) {
return []
}
if (/^(data|blob):/i.test(trimmed)) {
return [trimmed]
}
const candidates: string[] = []
const pushCandidate = (value: string) => {
if (!candidates.includes(value)) {
candidates.push(value)
}
}
try {
const parsed = new URL(trimmed, resolveDesktopApiURL('/'))
if (parsed.pathname.startsWith('/api/')) {
const apiAssetUrl = new URL(resolveDesktopApiURL(`${parsed.pathname}${parsed.search}`))
if (apiAssetUrl.pathname.startsWith('/api/public/assets/')) {
apiAssetUrl.searchParams.set('format', 'png')
}
pushCandidate(apiAssetUrl.toString())
}
pushCandidate(parsed.toString())
} catch {
pushCandidate(trimmed)
}
return candidates
}
async function fetchSmzdmImageBlob(
sourceUrl: string,
signal?: AbortSignal,
): Promise<SmzdmImageBlob | null> {
for (const candidate of buildAssetURLCandidates(sourceUrl)) {
const response = await adapterFetch(candidate, {}, { signal, timeoutMs: 10_000 }).catch(
() => null,
)
if (!response?.ok) {
continue
}
const sourceBlob = await response.blob().catch(() => null)
if (!sourceBlob || !sourceBlob.type.startsWith('image/')) {
continue
}
const buffer = Buffer.from(await sourceBlob.arrayBuffer())
const image = nativeImage.createFromBuffer(buffer)
if (image.isEmpty()) {
continue
}
const size = image.getSize()
const normalizedBlob =
sourceBlob.type.toLowerCase() === 'image/webp'
? new Blob([new Uint8Array(image.toPNG())], { type: 'image/png' })
: sourceBlob
return {
blob: normalizedBlob,
width: size.width,
height: size.height,
}
}
return null
}
export function getSmzdmCoverCropRect( export function getSmzdmCoverCropRect(
width: number, width: number,
height: number, height: number,
@@ -396,16 +481,18 @@ async function uploadImage(
articleId: string, articleId: string,
cropCover: boolean, cropCover: boolean,
): Promise<SmzdmUploadedImage | null> { ): Promise<SmzdmUploadedImage | null> {
const image = await fetchSmzdmImageBlob(sourceUrl, context.signal) const image = await fetchImageAssetBlob(sourceUrl, { signal: context.signal })
if (!image) { if (!image) {
throw new Error(cropCover ? 'smzdm_cover_fetch_failed' : 'smzdm_image_fetch_failed') throw new Error(cropCover ? 'smzdm_cover_fetch_failed' : 'smzdm_image_fetch_failed')
} }
const csrfToken = await smzdmCsrfToken(context)
const form = new FormData() const form = new FormData()
form.append('imgFile', image.blob, 'image.png') form.append('imgFile', image.blob, image.fileName)
form.append('id', 'WU_FILE_0') form.append('id', 'WU_FILE_0')
form.append('type', image.blob.type || 'image/png') form.append('type', image.blob.type || 'image/png')
form.append('article_id', articleId) form.append('article_id', articleId)
appendSmzdmCsrfFields(form, csrfToken)
const uploaded = await sessionFetchJson<SmzdmImageUploadResponse>( const uploaded = await sessionFetchJson<SmzdmImageUploadResponse>(
context.session, context.session,
@@ -413,7 +500,7 @@ async function uploadImage(
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context), headers: await smzdmHeaders(context, {}, csrfToken),
body: form, body: form,
}, },
{ signal: context.signal }, { signal: context.signal },
@@ -426,6 +513,9 @@ async function uploadImage(
if (!isSuccessfulCode(uploaded.error_code) || !uploaded.data?.url) { if (!isSuccessfulCode(uploaded.error_code) || !uploaded.data?.url) {
const message = smzdmResponseMessage(uploaded) const message = smzdmResponseMessage(uploaded)
if (isSmzdmCsrfMissingMessage(message)) {
throw new Error(`smzdm_csrf_missing:${message}`)
}
if (isSmzdmChallengeMessage(message)) { if (isSmzdmChallengeMessage(message)) {
throw new Error(`smzdm_challenge_required:${message}`) throw new Error(`smzdm_challenge_required:${message}`)
} }
@@ -436,19 +526,26 @@ async function uploadImage(
const uploadedId = stringId(uploaded.data.id) const uploadedId = stringId(uploaded.data.id)
if (uploadedId) { if (uploadedId) {
const recordBody = new URLSearchParams({
article_id: articleId,
ids: uploadedId,
})
appendSmzdmCsrfFields(recordBody, csrfToken)
void sessionFetchJson( void sessionFetchJson(
context.session, context.session,
`${SMZDM_ORIGIN}/api/editor/image_add_time/record`, `${SMZDM_ORIGIN}/api/editor/image_add_time/record`,
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context, { headers: await smzdmHeaders(
'content-type': 'application/x-www-form-urlencoded', context,
}), {
body: new URLSearchParams({ 'content-type': 'application/x-www-form-urlencoded',
article_id: articleId, },
ids: uploadedId, csrfToken,
}), ),
body: recordBody,
}, },
{ signal: context.signal }, { signal: context.signal },
).catch(() => null) ).catch(() => null)
@@ -468,13 +565,21 @@ async function uploadImage(
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context, { headers: await smzdmHeaders(
'content-type': 'application/x-www-form-urlencoded', context,
}), {
body: new URLSearchParams({ 'content-type': 'application/x-www-form-urlencoded',
article_id: articleId, },
pic_url: uploaded.data.url, csrfToken,
}), ),
body: (() => {
const body = new URLSearchParams({
article_id: articleId,
pic_url: uploaded.data.url,
})
appendSmzdmCsrfFields(body, csrfToken)
return body
})(),
}, },
{ signal: context.signal }, { signal: context.signal },
).catch(() => null) ).catch(() => null)
@@ -504,6 +609,7 @@ async function uploadImage(
cropForm.append('cut_pic_list[0][original_pic_width]', String(image.width)) cropForm.append('cut_pic_list[0][original_pic_width]', String(image.width))
cropForm.append('cut_pic_list[0][cutUrl]', original?.data?.original_url || '') cropForm.append('cut_pic_list[0][cutUrl]', original?.data?.original_url || '')
cropForm.append('cut_pic_list[0][is_head]', '1') cropForm.append('cut_pic_list[0][is_head]', '1')
appendSmzdmCsrfFields(cropForm, csrfToken)
const cropped = await sessionFetchJson<SmzdmCropResponse>( const cropped = await sessionFetchJson<SmzdmCropResponse>(
context.session, context.session,
@@ -511,7 +617,7 @@ async function uploadImage(
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context), headers: await smzdmHeaders(context, {}, csrfToken),
body: cropForm, body: cropForm,
}, },
{ signal: context.signal }, { signal: context.signal },
@@ -796,6 +902,17 @@ function failureResult(
} }
} }
if (message.startsWith('smzdm_csrf_missing') || isSmzdmCsrfMissingMessage(message)) {
return {
status: 'failed',
summary: '什么值得买 CSRF token 缺失,请重新登录或重新绑定账号后再试。',
error: {
code: 'smzdm_csrf_missing',
message,
},
}
}
if (message.startsWith('smzdm_cover_fetch_failed')) { if (message.startsWith('smzdm_cover_fetch_failed')) {
return { return {
status: 'failed', status: 'failed',
@@ -905,15 +1022,24 @@ export const smzdmAdapter: PublishAdapter = {
await context.reportProgress('smzdm.submit') await context.reportProgress('smzdm.submit')
await waitForHumanPace(context.signal) await waitForHumanPace(context.signal)
const csrfToken = await fetchSmzdmEditorToken(context)
if (!csrfToken) {
throw new Error('smzdm_csrf_missing:editor_token_missing')
}
const response = await sessionFetchJson<SmzdmSubmitResponse>( const response = await sessionFetchJson<SmzdmSubmitResponse>(
context.session, context.session,
`${SMZDM_ORIGIN}/api/editor/article/submit`, `${SMZDM_ORIGIN}/api/editor/article/submit`,
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context, { headers: await smzdmHeaders(
'content-type': 'application/x-www-form-urlencoded', context,
}), {
'content-type': 'application/x-www-form-urlencoded',
},
csrfToken,
),
body: encodeSmzdmForm( body: encodeSmzdmForm(
buildSubmitForm({ buildSubmitForm({
articleId, articleId,
@@ -937,6 +1063,9 @@ export const smzdmAdapter: PublishAdapter = {
const topicCode = response.data?.update_topic?.error_code const topicCode = response.data?.update_topic?.error_code
if (!isSuccessfulCode(response.error_code) || !isSuccessfulCode(topicCode)) { if (!isSuccessfulCode(response.error_code) || !isSuccessfulCode(topicCode)) {
const message = response.data?.update_topic?.error_msg || smzdmResponseMessage(response) const message = response.data?.update_topic?.error_msg || smzdmResponseMessage(response)
if (isSmzdmCsrfMissingMessage(message)) {
throw new Error(`smzdm_csrf_missing:${message}`)
}
if (isSmzdmChallengeMessage(message)) { if (isSmzdmChallengeMessage(message)) {
throw new Error(`smzdm_challenge_required:${message}`) throw new Error(`smzdm_challenge_required:${message}`)
} }
@@ -94,6 +94,18 @@ describe('platform auth adapters', () => {
).toBe('auth_failure') ).toBe('auth_failure')
}) })
it('classifies SMZDM missing csrf as auth failure', () => {
const adapter = getPlatformAdapter('smzdm')
expect(
adapter.classifyFailure({
error: {
code: 'smzdm_csrf_missing',
},
}),
).toBe('auth_failure')
})
it('classifies Juejin missing login as auth failure', () => { it('classifies Juejin missing login as auth failure', () => {
const adapter = getPlatformAdapter('juejin') const adapter = getPlatformAdapter('juejin')
@@ -234,6 +234,20 @@ function classifyDongchediFailure(input: PlatformFailureInput): PlatformFailureC
return classifyFailure(input) return classifyFailure(input)
} }
function classifySmzdmFailure(input: PlatformFailureInput): PlatformFailureClassification {
const code = typeof input.error?.code === 'string' ? input.error.code : ''
if (code === 'smzdm_not_logged_in' || code === 'smzdm_csrf_missing') {
return 'auth_failure'
}
if (code === 'smzdm_challenge_required') {
return 'challenge'
}
if (code.startsWith('smzdm_')) {
return 'ok'
}
return classifyFailure(input)
}
const aiAdapter: PlatformAdapter = { const aiAdapter: PlatformAdapter = {
async probe(account, partition) { async probe(account, partition) {
return await probeAIAccountSession(account, partition) return await probeAIAccountSession(account, partition)
@@ -364,6 +378,16 @@ const dongchediAdapter: PlatformAdapter = {
classifyFailure: classifyDongchediFailure, classifyFailure: classifyDongchediFailure,
} }
const smzdmAdapter: PlatformAdapter = {
async probe(account, partition) {
return await probePublishAccountSession(account, partition)
},
async silentRefresh(account, partition) {
return await silentRefreshPublishAccountSession(account, partition)
},
classifyFailure: classifySmzdmFailure,
}
const adapterRegistry = new Map<string, PlatformAdapter>( const adapterRegistry = new Map<string, PlatformAdapter>(
[ [
...aiPlatformCatalog.map((platform) => platform.id), ...aiPlatformCatalog.map((platform) => platform.id),
@@ -402,8 +426,10 @@ const adapterRegistry = new Map<string, PlatformAdapter>(
? weixinGzhAdapter ? weixinGzhAdapter
: platform === 'zol' : platform === 'zol'
? zolAdapter ? zolAdapter
: platform === 'dongchedi' : platform === 'dongchedi'
? dongchediAdapter ? dongchediAdapter
: platform === 'smzdm'
? smzdmAdapter
: aiPlatformCatalog.some((item) => item.id === platform) : aiPlatformCatalog.some((item) => item.id === platform)
? aiAdapter ? aiAdapter
: genericAdapter, : genericAdapter,
@@ -0,0 +1,72 @@
import type { JsonValue } from '@geo/shared-types'
import type { AdapterExecutionResult } from './adapters/base'
export function markAuthorizationFailureNonRetryable(
error: Record<string, JsonValue>,
category: 'ai_platform_authorization' | 'media_account_authorization',
): Record<string, JsonValue> {
return {
...error,
retryable: false,
non_retryable: true,
failure_category: category,
}
}
export function isAuthorizationFailureCode(code: string | null | undefined): boolean {
if (!code) {
return false
}
const normalized = code.trim().toLowerCase()
return (
normalized === 'desktop_account_auth_expired' ||
normalized === 'desktop_account_challenge_required' ||
normalized === 'desktop_account_risk_control' ||
normalized.endsWith('_not_logged_in') ||
normalized.endsWith('_login_required') ||
normalized.endsWith('_login_expired') ||
normalized.endsWith('_challenge_required')
)
}
export function isDefinitivePublishFailure(result: AdapterExecutionResult): boolean {
if (result.status !== 'failed') {
return false
}
const code = typeof result.error?.code === 'string' ? result.error.code.toLowerCase() : ''
const message =
typeof result.error?.message === 'string' ? result.error.message.toLowerCase() : ''
const summary = result.summary.toLowerCase()
const combined = `${code} ${message} ${summary}`
if (
result.error?.non_retryable === true ||
result.error?.retryable === false ||
isAuthorizationFailureCode(code)
) {
return true
}
return /(?:csrf|forbidden|\b403\b|not_logged_in|login_required|token_missing|cookie_missing|challenge_required|risk_control|captcha|验证码|滑块|人机验证|安全验证|风控|风险|参数错误|内容为空|image_fetch_failed|cover_fetch_failed|upload_failed|signature_failed)/i.test(
combined,
)
}
export function asSubmitUncertainExecution(result: AdapterExecutionResult): AdapterExecutionResult {
if (result.status !== 'failed' || isDefinitivePublishFailure(result)) {
return result
}
return {
...result,
status: 'unknown',
summary: '发布可能已提交到平台,结果待人工确认,已避免自动重复发布。',
error: {
...(result.error ?? {}),
publish_submit_uncertain: true,
original_status: 'failed',
original_summary: result.summary,
},
}
}
@@ -0,0 +1,38 @@
import { describe, expect, it } from 'vitest'
import type { AdapterExecutionResult } from './adapters/base'
import {
asSubmitUncertainExecution,
isDefinitivePublishFailure,
} from './publish-result-classification'
describe('publish retry result classification', () => {
it('keeps CSRF 403 publish failures definitive after submit started', () => {
const result: AdapterExecutionResult = {
status: 'failed',
summary: '什么值得买发布失败。',
error: {
code: 'smzdm_publish_failed',
message: '{"error_code":403,"error_msg":"CSRF token invalid"}',
},
}
expect(isDefinitivePublishFailure(result)).toBe(true)
expect(asSubmitUncertainExecution(result).status).toBe('failed')
})
it('marks ambiguous submit-phase publish failures as unknown', () => {
const result: AdapterExecutionResult = {
status: 'failed',
summary: '发布失败。',
error: {
code: 'smzdm_publish_failed',
message: 'network interrupted after submit',
},
}
const normalized = asSubmitUncertainExecution(result)
expect(normalized.status).toBe('unknown')
expect(normalized.error?.publish_submit_uncertain).toBe(true)
})
})
@@ -80,6 +80,11 @@ import {
startHiddenPlaywrightReaper, startHiddenPlaywrightReaper,
} from './playwright-cdp' } from './playwright-cdp'
import { getProcessMetricsSnapshot } from './process-metrics' import { getProcessMetricsSnapshot } from './process-metrics'
import {
asSubmitUncertainExecution,
isAuthorizationFailureCode,
markAuthorizationFailureNonRetryable,
} from './publish-result-classification'
import { import {
enqueuePublishTask as enqueuePublishSchedulerTask, enqueuePublishTask as enqueuePublishSchedulerTask,
getPublishSchedulerSnapshot, getPublishSchedulerSnapshot,
@@ -500,34 +505,6 @@ function eligibleMonitorPlatformIds(): string[] {
.filter((platform) => !blocked.has(platform)) .filter((platform) => !blocked.has(platform))
} }
function markAuthorizationFailureNonRetryable(
error: Record<string, JsonValue>,
category: 'ai_platform_authorization' | 'media_account_authorization',
): Record<string, JsonValue> {
return {
...error,
retryable: false,
non_retryable: true,
failure_category: category,
}
}
function isAuthorizationFailureCode(code: string | null | undefined): boolean {
if (!code) {
return false
}
const normalized = code.trim().toLowerCase()
return (
normalized === 'desktop_account_auth_expired' ||
normalized === 'desktop_account_challenge_required' ||
normalized === 'desktop_account_risk_control' ||
normalized.endsWith('_not_logged_in') ||
normalized.endsWith('_login_required') ||
normalized.endsWith('_login_expired') ||
normalized.endsWith('_challenge_required')
)
}
function isAuthorizationFailureMessage(message: string | null | undefined): boolean { function isAuthorizationFailureMessage(message: string | null | undefined): boolean {
if (!message) { if (!message) {
return false return false
@@ -782,7 +759,8 @@ function startRuntime(): void {
return return
} }
if (playwrightCDPFatal) { if (playwrightCDPFatal) {
const message = '浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。' const message =
'浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。'
state.lastError = message state.lastError = message
setSchedulerPhase('paused') setSchedulerPhase('paused')
setSchedulerError(message) setSchedulerError(message)
@@ -1922,23 +1900,6 @@ async function markPublishSubmitStartedIfNeeded(taskId: string, stage: string):
state.activeExecutions.set(taskId, active) state.activeExecutions.set(taskId, active)
} }
function asSubmitUncertainExecution(result: AdapterExecutionResult): AdapterExecutionResult {
if (result.status !== 'failed') {
return result
}
return {
...result,
status: 'unknown',
summary: '发布可能已提交到平台,结果待人工确认,已避免自动重复发布。',
error: {
...(result.error ?? {}),
publish_submit_uncertain: true,
original_status: 'failed',
original_summary: result.summary,
},
}
}
async function retainPlaywrightPageOrThrow( async function retainPlaywrightPageOrThrow(
task: RuntimeTaskRecord, task: RuntimeTaskRecord,
payload: Record<string, JsonValue>, payload: Record<string, JsonValue>,
@@ -2183,7 +2144,12 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise<void> {
async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise<void> { async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise<void> {
const leaseLimit = resolveLegacyMonitoringLeaseLimit() const leaseLimit = resolveLegacyMonitoringLeaseLimit()
if (!canRunLive(state.session) || isLeaseInFlight('monitor') || leaseLimit <= 0 || playwrightCDPFatal) { if (
!canRunLive(state.session) ||
isLeaseInFlight('monitor') ||
leaseLimit <= 0 ||
playwrightCDPFatal
) {
return return
} }
@@ -2205,7 +2171,8 @@ async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise<void> {
noteSchedulerPull() noteSchedulerPull()
enqueueLeasedMonitoringTasks(leased.tasks, routing) enqueueLeasedMonitoringTasks(leased.tasks, routing)
state.monitorFallbackBackoffUntil = leased.tasks.length > 0 ? 0 : Date.now() + monitorPullFallbackIntervalMs state.monitorFallbackBackoffUntil =
leased.tasks.length > 0 ? 0 : Date.now() + monitorPullFallbackIntervalMs
} catch (error) { } catch (error) {
state.lastPullAt = Date.now() state.lastPullAt = Date.now()
state.lastPullStatus = 'failed' state.lastPullStatus = 'failed'
@@ -3720,7 +3687,9 @@ async function ensurePlaywrightCDPAdmissionForRequest(
return ready return ready
} }
async function ensurePlaywrightCDPAdmissionForFallback(kind: 'publish' | 'monitor'): Promise<boolean> { async function ensurePlaywrightCDPAdmissionForFallback(
kind: 'publish' | 'monitor',
): Promise<boolean> {
if (kind === 'publish') { if (kind === 'publish') {
const queuedPlaywrightPublish = getPublishSchedulerSnapshot().tasks.some( const queuedPlaywrightPublish = getPublishSchedulerSnapshot().tasks.some(
(task) => task.state === 'queued' && adapterRequiresPlaywright('publish', task.platform), (task) => task.state === 'queued' && adapterRequiresPlaywright('publish', task.platform),
@@ -3759,7 +3728,9 @@ async function ensurePlaywrightCDPAdmission(): Promise<boolean> {
} }
} }
function isPlaywrightCDPInfrastructureError(error: unknown): error is PlaywrightCDPInfrastructureError { function isPlaywrightCDPInfrastructureError(
error: unknown,
): error is PlaywrightCDPInfrastructureError {
return error instanceof PlaywrightCDPInfrastructureError || isPlaywrightCDPUnavailableError(error) return error instanceof PlaywrightCDPInfrastructureError || isPlaywrightCDPUnavailableError(error)
} }
@@ -3779,7 +3750,8 @@ function deferPendingTaskRequest(
} }
function adapterRequiresPlaywright(kind: 'publish' | 'monitor', platform: string): boolean { function adapterRequiresPlaywright(kind: 'publish' | 'monitor', platform: string): boolean {
const adapter = kind === 'publish' ? selectPublishAdapter(platform) : selectMonitorAdapter(platform) const adapter =
kind === 'publish' ? selectPublishAdapter(platform) : selectMonitorAdapter(platform)
return adapter?.executionMode === 'playwright' return adapter?.executionMode === 'playwright'
} }
@@ -3820,8 +3792,7 @@ function markPlaywrightCDPFatal(message: string, taskId?: string): void {
playwrightCDPFatal = true playwrightCDPFatal = true
const fatalMessage = const fatalMessage =
message || message || '浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。'
'浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。'
state.lastError = fatalMessage state.lastError = fatalMessage
setSchedulerPhase('paused') setSchedulerPhase('paused')
setSchedulerError(fatalMessage) setSchedulerError(fatalMessage)
@@ -201,7 +201,7 @@ function buildWangyihaoArticleUrl(
} }
function normalizePublishTaskStatus(status: DesktopTaskInfo['status']): DesktopTaskInfo['status'] { function normalizePublishTaskStatus(status: DesktopTaskInfo['status']): DesktopTaskInfo['status'] {
return status === 'unknown' ? 'failed' : status return status
} }
function normalizeTaskErrorMessage( function normalizeTaskErrorMessage(
@@ -810,7 +810,9 @@ const publishTasks = computed<PublishTaskItem[]>(() =>
extractString(taskError.code), extractString(taskError.code),
task.platform, task.platform,
), ),
submitUncertain: task.status === 'unknown' && taskError.publish_submit_uncertain === true, submitUncertain:
task.status === 'unknown' &&
(taskError.publish_submit_uncertain === true || Boolean(task.publish_submit_started_at)),
complianceBlockedRecordId, complianceBlockedRecordId,
complianceBlockedAt: task.compliance_blocked_at complianceBlockedAt: task.compliance_blocked_at
? parseTimestamp(task.compliance_blocked_at) ? parseTimestamp(task.compliance_blocked_at)
+1
View File
@@ -317,6 +317,7 @@ export interface DesktopTaskInfo {
dedup_key: string | null dedup_key: string | null
active_attempt_id: string | null active_attempt_id: string | null
lease_expires_at: string | null lease_expires_at: string | null
publish_submit_started_at?: string | null
attempts: number attempts: number
result: Record<string, JsonValue> | null result: Record<string, JsonValue> | null
error: Record<string, JsonValue> | null error: Record<string, JsonValue> | null
@@ -95,6 +95,7 @@ type DesktopTaskView struct {
DedupKey *string `json:"dedup_key"` DedupKey *string `json:"dedup_key"`
ActiveAttemptID *string `json:"active_attempt_id"` ActiveAttemptID *string `json:"active_attempt_id"`
LeaseExpiresAt *time.Time `json:"lease_expires_at"` LeaseExpiresAt *time.Time `json:"lease_expires_at"`
PublishSubmitStartedAt *time.Time `json:"publish_submit_started_at,omitempty"`
Attempts int `json:"attempts"` Attempts int `json:"attempts"`
Result json.RawMessage `json:"result"` Result json.RawMessage `json:"result"`
Error json.RawMessage `json:"error"` Error json.RawMessage `json:"error"`
@@ -1563,6 +1564,7 @@ func buildListPublishTasksByStatusesQuery(
t.error, t.error,
t.created_at, t.created_at,
t.updated_at, t.updated_at,
t.publish_submit_started_at,
j.status, j.status,
j.compliance_blocked_record_id, j.compliance_blocked_record_id,
j.compliance_blocked_at, j.compliance_blocked_at,
@@ -1638,28 +1640,29 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
items := make([]DesktopTaskView, 0) items := make([]DesktopTaskView, 0)
for rows.Next() { for rows.Next() {
var ( var (
desktopID uuid.UUID desktopID uuid.UUID
jobID uuid.UUID jobID uuid.UUID
tenantID int64 tenantID int64
workspaceID int64 workspaceID int64
targetAccountID uuid.UUID targetAccountID uuid.UUID
targetClientID uuid.UUID targetClientID uuid.UUID
platform string platform string
kind string kind string
payload []byte payload []byte
status string status string
dedupKey pgtype.Text dedupKey pgtype.Text
activeAttemptID pgtype.UUID activeAttemptID pgtype.UUID
leaseExpiresAt pgtype.Timestamptz leaseExpiresAt pgtype.Timestamptz
attempts int32 attempts int32
resultJSON []byte resultJSON []byte
errorJSON []byte errorJSON []byte
createdAt time.Time createdAt time.Time
updatedAt time.Time updatedAt time.Time
jobStatus string publishSubmitStartedAt pgtype.Timestamptz
blockedRecordID pgtype.Int8 jobStatus string
blockedAt pgtype.Timestamptz blockedRecordID pgtype.Int8
blockedReason pgtype.Text blockedAt pgtype.Timestamptz
blockedReason pgtype.Text
) )
if err := rows.Scan( if err := rows.Scan(
&desktopID, &desktopID,
@@ -1680,6 +1683,7 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
&errorJSON, &errorJSON,
&createdAt, &createdAt,
&updatedAt, &updatedAt,
&publishSubmitStartedAt,
&jobStatus, &jobStatus,
&blockedRecordID, &blockedRecordID,
&blockedAt, &blockedAt,
@@ -1708,6 +1712,11 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
value := leaseExpiresAt.Time value := leaseExpiresAt.Time
leaseExpiresAtValue = &value leaseExpiresAtValue = &value
} }
var publishSubmitStartedAtValue *time.Time
if publishSubmitStartedAt.Valid {
value := publishSubmitStartedAt.Time
publishSubmitStartedAtValue = &value
}
publishJobStatusText := strings.TrimSpace(jobStatus) publishJobStatusText := strings.TrimSpace(jobStatus)
var publishJobStatus *string var publishJobStatus *string
@@ -1751,6 +1760,7 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
DedupKey: dedupKeyText, DedupKey: dedupKeyText,
ActiveAttemptID: activeAttemptIDText, ActiveAttemptID: activeAttemptIDText,
LeaseExpiresAt: leaseExpiresAtValue, LeaseExpiresAt: leaseExpiresAtValue,
PublishSubmitStartedAt: publishSubmitStartedAtValue,
Attempts: int(attempts), Attempts: int(attempts),
Result: json.RawMessage(resultJSON), Result: json.RawMessage(resultJSON),
Error: json.RawMessage(errorJSON), Error: json.RawMessage(errorJSON),
@@ -1833,24 +1843,25 @@ func buildDesktopTaskView(task *repository.DesktopTask) DesktopTaskView {
} }
return DesktopTaskView{ return DesktopTaskView{
ID: task.DesktopID.String(), ID: task.DesktopID.String(),
JobID: task.JobID.String(), JobID: task.JobID.String(),
TenantID: task.TenantID, TenantID: task.TenantID,
WorkspaceID: task.WorkspaceID, WorkspaceID: task.WorkspaceID,
TargetAccountID: task.TargetAccountID.String(), TargetAccountID: task.TargetAccountID.String(),
TargetClientID: task.TargetClientID.String(), TargetClientID: task.TargetClientID.String(),
Platform: task.Platform, Platform: task.Platform,
Kind: task.Kind, Kind: task.Kind,
Payload: json.RawMessage(task.Payload), Payload: json.RawMessage(task.Payload),
Status: normalizeDesktopTaskViewStatus(task.Kind, task.Status), Status: normalizeDesktopTaskViewStatus(task.Kind, task.Status),
DedupKey: task.DedupKey, DedupKey: task.DedupKey,
ActiveAttemptID: activeAttemptID, ActiveAttemptID: activeAttemptID,
LeaseExpiresAt: task.LeaseExpiresAt, LeaseExpiresAt: task.LeaseExpiresAt,
Attempts: task.Attempts, PublishSubmitStartedAt: task.PublishSubmitStartedAt,
Result: json.RawMessage(task.Result), Attempts: task.Attempts,
Error: json.RawMessage(task.Error), Result: json.RawMessage(task.Result),
CreatedAt: task.CreatedAt, Error: json.RawMessage(task.Error),
UpdatedAt: task.UpdatedAt, CreatedAt: task.CreatedAt,
UpdatedAt: task.UpdatedAt,
} }
} }
@@ -1899,8 +1910,10 @@ func normalizeDesktopTaskCompletionStatusForKind(kind string, status string) str
func normalizeDesktopTaskViewStatus(kind string, status string) string { func normalizeDesktopTaskViewStatus(kind string, status string) string {
switch strings.TrimSpace(kind) { switch strings.TrimSpace(kind) {
case "monitor", "publish": case "monitor":
return normalizeDesktopTaskTerminalStatus(status) return normalizeDesktopTaskTerminalStatus(status)
case "publish":
return strings.TrimSpace(status)
default: default:
return strings.TrimSpace(status) return strings.TrimSpace(status)
} }
@@ -257,14 +257,14 @@ func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) {
} }
} }
func TestNormalizeDesktopTaskViewStatusMapsMonitorAndPublishUnknownToFailed(t *testing.T) { func TestNormalizeDesktopTaskViewStatusMapsMonitorUnknownToFailedButKeepsPublishUnknown(t *testing.T) {
t.Parallel() t.Parallel()
if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" { if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" {
t.Fatalf("monitor unknown view status = %q, want failed", got) t.Fatalf("monitor unknown view status = %q, want failed", got)
} }
if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "failed" { if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "unknown" {
t.Fatalf("publish unknown view status = %q, want failed", got) t.Fatalf("publish unknown view status = %q, want unknown", got)
} }
} }
@@ -4,6 +4,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"regexp"
"sort" "sort"
"strconv" "strconv"
"strings" "strings"
@@ -218,6 +219,14 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
if err := lockDesktopPublishDedupKeys(ctx, tx, dedupKeys); err != nil { if err := lockDesktopPublishDedupKeys(ctx, tx, dedupKeys); err != nil {
return nil, err return nil, err
} }
releasedPublishOutcomes := make([]*desktopPublishSyncOutcome, 0)
if len(dedupKeys) > 0 {
outcomes, releaseErr := releaseDefinitiveFailedUnknownPublishTasks(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, dedupKeys)
if releaseErr != nil {
return nil, releaseErr
}
releasedPublishOutcomes = append(releasedPublishOutcomes, outcomes...)
}
existingTasks, err := loadExistingPublishRecordTargets(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, articleID, targets) existingTasks, err := loadExistingPublishRecordTargets(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, articleID, targets)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -338,6 +347,11 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
return nil, response.ErrInternal(50100, "desktop_publish_job_commit_failed", "failed to commit desktop publish job") return nil, response.ErrInternal(50100, "desktop_publish_job_commit_failed", "failed to commit desktop publish job")
} }
invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID) invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID)
for _, outcome := range releasedPublishOutcomes {
if outcome != nil {
invalidateArticleCaches(ctx, s.cache, outcome.TenantID, &outcome.ArticleID)
}
}
for _, task := range createdTasks { for _, task := range createdTasks {
dispatchEvent := DesktopDispatchEventFromTask(task, "task_available") dispatchEvent := DesktopDispatchEventFromTask(task, "task_available")
@@ -514,6 +528,155 @@ func lockDesktopPublishDedupKeys(ctx context.Context, tx pgx.Tx, keys []string)
return nil return nil
} }
func releaseDefinitiveFailedUnknownPublishTasks(
ctx context.Context,
tx pgx.Tx,
tenantID int64,
workspaceID int64,
dedupKeys []string,
) ([]*desktopPublishSyncOutcome, error) {
if len(dedupKeys) == 0 {
return nil, nil
}
rows, err := tx.Query(ctx, `
SELECT desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error
FROM desktop_tasks
WHERE tenant_id = $1
AND workspace_id = $2
AND kind = 'publish'
AND status = 'unknown'
AND dedup_key = ANY($3::text[])
FOR UPDATE
`, tenantID, workspaceID, dedupKeys)
if err != nil {
return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to inspect stale publish deduplication tasks")
}
defer rows.Close()
tasks := make([]repository.DesktopTask, 0)
for rows.Next() {
var task repository.DesktopTask
if scanErr := rows.Scan(
&task.DesktopID,
&task.TenantID,
&task.WorkspaceID,
&task.Platform,
&task.Kind,
&task.Payload,
&task.Status,
&task.Result,
&task.Error,
); scanErr != nil {
return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to parse stale publish deduplication task")
}
tasks = append(tasks, task)
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to iterate stale publish deduplication tasks")
}
outcomes := make([]*desktopPublishSyncOutcome, 0)
for index := range tasks {
task := tasks[index]
if !publishUnknownTaskHasDefinitiveFailure(&task) {
continue
}
updated, updateErr := markUnknownPublishTaskDefinitiveFailed(ctx, tx, &task)
if updateErr != nil {
return nil, updateErr
}
outcome, syncErr := syncDesktopPublishTaskState(ctx, tx, updated)
if syncErr != nil {
return nil, syncErr
}
if outcome != nil {
outcomes = append(outcomes, outcome)
}
}
return outcomes, nil
}
func markUnknownPublishTaskDefinitiveFailed(
ctx context.Context,
tx pgx.Tx,
task *repository.DesktopTask,
) (*repository.DesktopTask, error) {
if task == nil {
return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
}
row := tx.QueryRow(ctx, `
UPDATE desktop_tasks
SET status = 'failed',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
publish_submit_started_at = NULL,
updated_at = NOW()
WHERE desktop_id = $1
AND workspace_id = $2
AND status = 'unknown'
RETURNING desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error
`, task.DesktopID, task.WorkspaceID)
var updated repository.DesktopTask
if err := row.Scan(
&updated.DesktopID,
&updated.TenantID,
&updated.WorkspaceID,
&updated.Platform,
&updated.Kind,
&updated.Payload,
&updated.Status,
&updated.Result,
&updated.Error,
); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrConflict(40985, "desktop_task_reconcile_conflict", "desktop task is not waiting for reconcile")
}
return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to release stale publish deduplication task")
}
return &updated, nil
}
func publishUnknownTaskHasDefinitiveFailure(task *repository.DesktopTask) bool {
if task == nil || task.Kind != "publish" || task.Status != "unknown" {
return false
}
errorPayload := unmarshalJSONObject(task.Error)
if boolValueFromMap(errorPayload, "publish_submit_uncertain") {
return false
}
combined := strings.ToLower(strings.Join([]string{
stringPointerValue(extractStringPointer(errorPayload, "code", "error_code")),
stringPointerValue(extractStringPointer(errorPayload, "message", "error_msg", "detail")),
}, " "))
if combined == "" {
return false
}
return definitivePublishFailurePattern.MatchString(combined)
}
func boolValueFromMap(payload map[string]any, key string) bool {
if payload == nil {
return false
}
value, ok := payload[key]
if !ok {
return false
}
switch typed := value.(type) {
case bool:
return typed
case string:
normalized := strings.TrimSpace(strings.ToLower(typed))
return normalized == "true" || normalized == "1" || normalized == "yes"
default:
return false
}
}
var definitivePublishFailurePattern = regexp.MustCompile(`csrf|forbidden|\b403\b|not_logged_in|login_required|token_missing|cookie_missing|challenge_required|risk_control|captcha|验证码|滑块|人机验证|安全验证|风控|风险|参数错误|内容为空|image_fetch_failed|cover_fetch_failed|upload_failed|signature_failed`)
func loadExistingPublishRecordTargets( func loadExistingPublishRecordTargets(
ctx context.Context, ctx context.Context,
tx pgx.Tx, tx pgx.Tx,
@@ -567,7 +730,7 @@ func loadExistingPublishRecordTargets(
AND pr.article_id = $3 AND pr.article_id = $3
AND pr.platform_account_id = ANY($4::bigint[]) AND pr.platform_account_id = ANY($4::bigint[])
AND pr.deleted_at IS NULL AND pr.deleted_at IS NULL
AND pr.status IN ('queued', 'publishing', 'success') AND (pr.status IN ('queued', 'publishing', 'success') OR dt.status = 'unknown')
AND dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown') AND dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown')
ORDER BY pr.platform_account_id, ORDER BY pr.platform_account_id,
CASE CASE
@@ -785,10 +948,18 @@ func (s *PublishJobService) requeueUnknownPublishTask(
return nil, response.ErrInternal(50094, "desktop_task_reconcile_failed", "failed to reconcile desktop task") return nil, response.ErrInternal(50094, "desktop_task_reconcile_failed", "failed to reconcile desktop task")
} }
publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, requeued)
if err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil { if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50106, "desktop_task_reconcile_commit_failed", "failed to commit desktop task reconcile") return nil, response.ErrInternal(50106, "desktop_task_reconcile_commit_failed", "failed to commit desktop task reconcile")
} }
if publishOutcome != nil {
invalidateArticleCaches(ctx, s.cache, publishOutcome.TenantID, &publishOutcome.ArticleID)
}
publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, requeued, "task_available") publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, requeued, "task_available")
return createPublishJobResponseForRequeuedTask(requeued), nil return createPublishJobResponseForRequeuedTask(requeued), nil
} }
@@ -203,6 +203,32 @@ func TestCreatePublishJobResponseForRequeuedTaskReportsRequeuedTaskID(t *testing
} }
} }
func TestPublishUnknownTaskHasDefinitiveFailureReleasesCSRF403(t *testing.T) {
t.Parallel()
task := &repository.DesktopTask{
Kind: "publish",
Status: "unknown",
Error: []byte(`{"error_code":403,"error_msg":"CSRF token invalid"}`),
}
if !publishUnknownTaskHasDefinitiveFailure(task) {
t.Fatal("unknown publish task with CSRF 403 should be treated as definitive failed")
}
}
func TestPublishUnknownTaskHasDefinitiveFailureKeepsSubmitUncertain(t *testing.T) {
t.Parallel()
task := &repository.DesktopTask{
Kind: "publish",
Status: "unknown",
Error: []byte(`{"publish_submit_uncertain":true,"message":"network interrupted after submit"}`),
}
if publishUnknownTaskHasDefinitiveFailure(task) {
t.Fatal("submit-uncertain unknown publish task must stay in dedup set")
}
}
func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) { func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) {
t.Parallel() t.Parallel()
@@ -220,6 +246,9 @@ func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) {
if !strings.Contains(normalizedSQL, "dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown')") { if !strings.Contains(normalizedSQL, "dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown')") {
t.Fatalf("publish dedup lookup should include unknown task status; query:\n%s", tx.sql) t.Fatalf("publish dedup lookup should include unknown task status; query:\n%s", tx.sql)
} }
if !strings.Contains(normalizedSQL, "OR dt.status = 'unknown'") {
t.Fatalf("publish dedup lookup should include unknown tasks even when publish record is failed; query:\n%s", tx.sql)
}
if !strings.Contains(normalizedSQL, "AND dt.tenant_id = $1") { if !strings.Contains(normalizedSQL, "AND dt.tenant_id = $1") {
t.Fatalf("publish dedup lookup should constrain desktop_tasks by tenant for the payload index; query:\n%s", tx.sql) t.Fatalf("publish dedup lookup should constrain desktop_tasks by tenant for the payload index; query:\n%s", tx.sql)
} }
@@ -37,37 +37,38 @@ type CreateDesktopPublishJobParams struct {
} }
type DesktopTask struct { type DesktopTask struct {
DesktopID uuid.UUID DesktopID uuid.UUID
JobID uuid.UUID JobID uuid.UUID
TenantID int64 TenantID int64
WorkspaceID int64 WorkspaceID int64
TargetAccountID uuid.UUID TargetAccountID uuid.UUID
TargetClientID uuid.UUID TargetClientID uuid.UUID
Platform string Platform string
Kind string Kind string
Payload []byte Payload []byte
Status string Status string
Priority int Priority int
Lane string Lane string
LaneWeight int LaneWeight int
Source string Source string
SchedulerGroupKey *string SchedulerGroupKey *string
MonitorTaskID *int64 MonitorTaskID *int64
SupersedesTaskID *uuid.UUID SupersedesTaskID *uuid.UUID
ControlFlags []byte ControlFlags []byte
InterruptGeneration int InterruptGeneration int
DedupKey *string DedupKey *string
ActiveAttemptID *uuid.UUID ActiveAttemptID *uuid.UUID
LeaseExpiresAt *time.Time LeaseExpiresAt *time.Time
Attempts int Attempts int
Result []byte Result []byte
Error []byte Error []byte
StartedAt *time.Time StartedAt *time.Time
InterruptedAt *time.Time InterruptedAt *time.Time
InterruptReason *string InterruptReason *string
EnqueuedAt time.Time EnqueuedAt time.Time
CreatedAt time.Time PublishSubmitStartedAt *time.Time
UpdatedAt time.Time CreatedAt time.Time
UpdatedAt time.Time
} }
type CreateDesktopTaskParams struct { type CreateDesktopTaskParams struct {
@@ -355,37 +356,38 @@ func desktopPublishJobFromGenerated(row generated.DesktopPublishJob) *DesktopPub
func desktopTaskFromGenerated(row generated.DesktopTask) *DesktopTask { func desktopTaskFromGenerated(row generated.DesktopTask) *DesktopTask {
return &DesktopTask{ return &DesktopTask{
DesktopID: uuidFromPG(row.DesktopID), DesktopID: uuidFromPG(row.DesktopID),
JobID: uuidFromPG(row.JobID), JobID: uuidFromPG(row.JobID),
TenantID: row.TenantID, TenantID: row.TenantID,
WorkspaceID: row.WorkspaceID, WorkspaceID: row.WorkspaceID,
TargetAccountID: uuidFromPG(row.TargetAccountID), TargetAccountID: uuidFromPG(row.TargetAccountID),
TargetClientID: uuidFromPG(row.TargetClientID), TargetClientID: uuidFromPG(row.TargetClientID),
Platform: row.PlatformID, Platform: row.PlatformID,
Kind: row.Kind, Kind: row.Kind,
Payload: row.Payload, Payload: row.Payload,
Status: row.Status, Status: row.Status,
Priority: int(row.Priority), Priority: int(row.Priority),
Lane: row.Lane, Lane: row.Lane,
LaneWeight: int(row.LaneWeight), LaneWeight: int(row.LaneWeight),
Source: row.Source, Source: row.Source,
SchedulerGroupKey: nullableText(row.SchedulerGroupKey), SchedulerGroupKey: nullableText(row.SchedulerGroupKey),
MonitorTaskID: nullableInt64(row.MonitorTaskID), MonitorTaskID: nullableInt64(row.MonitorTaskID),
SupersedesTaskID: nullableUUID(row.SupersedesTaskID), SupersedesTaskID: nullableUUID(row.SupersedesTaskID),
ControlFlags: row.ControlFlags, ControlFlags: row.ControlFlags,
InterruptGeneration: int(row.InterruptGeneration), InterruptGeneration: int(row.InterruptGeneration),
DedupKey: nullableText(row.DedupKey), DedupKey: nullableText(row.DedupKey),
ActiveAttemptID: nullableUUID(row.ActiveAttemptID), ActiveAttemptID: nullableUUID(row.ActiveAttemptID),
LeaseExpiresAt: optionalTime(row.LeaseExpiresAt), LeaseExpiresAt: optionalTime(row.LeaseExpiresAt),
Attempts: int(row.Attempts), Attempts: int(row.Attempts),
Result: row.Result, Result: row.Result,
Error: row.Error, Error: row.Error,
StartedAt: optionalTime(row.StartedAt), StartedAt: optionalTime(row.StartedAt),
InterruptedAt: optionalTime(row.InterruptedAt), InterruptedAt: optionalTime(row.InterruptedAt),
InterruptReason: nullableText(row.InterruptReason), InterruptReason: nullableText(row.InterruptReason),
EnqueuedAt: timeFromTimestamp(row.EnqueuedAt), EnqueuedAt: timeFromTimestamp(row.EnqueuedAt),
CreatedAt: timeFromTimestamp(row.CreatedAt), PublishSubmitStartedAt: optionalTime(row.PublishSubmitStartedAt),
UpdatedAt: timeFromTimestamp(row.UpdatedAt), CreatedAt: timeFromTimestamp(row.CreatedAt),
UpdatedAt: timeFromTimestamp(row.UpdatedAt),
} }
} }
@@ -849,6 +849,10 @@ SET status = CASE
active_attempt_id = NULL, active_attempt_id = NULL,
lease_token_hash = NULL, lease_token_hash = NULL,
lease_expires_at = NULL, lease_expires_at = NULL,
publish_submit_started_at = CASE
WHEN $1::text = 'retry' THEN NULL
ELSE publish_submit_started_at
END,
attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END, attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END,
updated_at = now() updated_at = now()
WHERE desktop_id = $4 WHERE desktop_id = $4
@@ -39,6 +39,10 @@ func TestReconcileDesktopTaskRetryResetsAttempts(t *testing.T) {
if !strings.Contains(query, "attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END") { if !strings.Contains(query, "attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END") {
t.Fatalf("retry reconcile must reset attempts so manual retry can be leased; query:\n%s", query) t.Fatalf("retry reconcile must reset attempts so manual retry can be leased; query:\n%s", query)
} }
if !strings.Contains(query, "publish_submit_started_at = CASE") ||
!strings.Contains(query, "WHEN $1::text = 'retry' THEN NULL") {
t.Fatalf("retry reconcile must clear stale publish submit marker; query:\n%s", query)
}
if strings.Contains(query, "attempts + CASE") { if strings.Contains(query, "attempts + CASE") {
t.Fatalf("retry reconcile must not increment attempts; query:\n%s", query) t.Fatalf("retry reconcile must not increment attempts; query:\n%s", query)
} }
@@ -225,6 +225,10 @@ SET status = CASE
active_attempt_id = NULL, active_attempt_id = NULL,
lease_token_hash = NULL, lease_token_hash = NULL,
lease_expires_at = NULL, lease_expires_at = NULL,
publish_submit_started_at = CASE
WHEN sqlc.arg(status)::text = 'retry' THEN NULL
ELSE publish_submit_started_at
END,
attempts = CASE WHEN sqlc.arg(status)::text = 'retry' THEN 0 ELSE attempts END, attempts = CASE WHEN sqlc.arg(status)::text = 'retry' THEN 0 ELSE attempts END,
updated_at = now() updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id) WHERE desktop_id = sqlc.arg(desktop_id)