root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: root/geo:f1334aac80e5e9b752a7d016ab598601ab7fef23
Branches Tags
root/geo:main
...
compare: root/geo:5d29703265144ce09c55f0558c21dd74ba97d8d9
Branches Tags
root/geo:main

3 Commits
f1334aac80 ... 5d29703265

Author SHA1 Message Date
root 5d29703265 perf(admin-web): reduce redundant polling and cover active publish status
Frontend CI / Frontend (push) Successful in 3m35s
Details
Backend CI / Backend (push) Failing after 9m46s
Details 
Unify list/detail poll intervals to 5s and skip refetch while a query is
already in flight, replacing broad invalidateQueries cascades with targeted
refetches. Poll now also triggers on active publish status via the new
hasActivePublishStatus helper. Dedupe concurrent refreshBrands calls with a
shared promise and skip the redundant AppShell refresh when already
initialized. Gate publish-record loading on popover open.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
 2026-07-08 23:16:47 +08:00
root 2c11910b6c chore(admin-web): add favicon 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
 2026-07-08 22:56:45 +08:00
root 832d7f507d feat(publish): cancel stale queued publish tasks after queue timeout 
Add CancelStaleQueuedPublishTasks to abort publish tasks that sit in the
queue past a configurable timeout (default 3 days) without being claimed by
a desktop client, so they no longer linger indefinitely as 'queued'. The
lease recovery worker runs the cleanup each cycle and reports a
cancelled_queued count; the timeout is configurable via job run config.
Aborted tasks carry a publish_queue_timeout error payload with a readable
Chinese duration, and the admin-web publish summary surfaces it as an
auto-cancelled queue-timeout message.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
 2026-07-08 22:56:37 +08:00
19 changed files with 530 additions and 74 deletions
Expand all files Collapse all files
apps/admin-web/index.html
+5 -1
View File
@@ -6,6 +6,11 @@
name="viewport" name="viewport"
content="width=device-width, initial-scale=1.0" content="width=device-width, initial-scale=1.0"
/> />
<link
rel="icon"
href="/favicon.ico"
sizes="any"
/>
<title>省心推</title> <title>省心推</title>
</head> </head>
<body> <body>
@@ -16,4 +21,3 @@
></script> ></script>
</body> </body>
</html> </html>
apps/admin-web/public/favicon.ico
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 15 KiB

apps/admin-web/src/components/ArticleDetailDrawer.vue
+25 -18
View File
@@ -1,7 +1,7 @@
<script setup lang="ts"> <script setup lang="ts">
import { CopyOutlined, LinkOutlined } from '@ant-design/icons-vue' import { CopyOutlined, LinkOutlined } from '@ant-design/icons-vue'
import type { ArticleVersion, PublishRecord } from '@geo/shared-types' import type { ArticleVersion, PublishRecord } from '@geo/shared-types'
import { useQuery, useQueryClient } from '@tanstack/vue-query' import { useQuery } from '@tanstack/vue-query'
import type { TableColumnsType } from 'ant-design-vue' import type { TableColumnsType } from 'ant-design-vue'
import { message } from 'ant-design-vue' import { message } from 'ant-design-vue'
import { computed, ref, watch } from 'vue' import { computed, ref, watch } from 'vue'
@@ -22,6 +22,7 @@ import {
getPublishStatusMeta, getPublishStatusMeta,
getSourceTypeLabel, getSourceTypeLabel,
hasActiveGenerationStatus, hasActiveGenerationStatus,
hasActivePublishStatus,
} from '@/lib/display' } from '@/lib/display'
import { useCompanyStore } from '@/stores/company' import { useCompanyStore } from '@/stores/company'
@@ -34,7 +35,6 @@ const emit = defineEmits<{
close: [] close: []
}>() }>()
const queryClient = useQueryClient()
const { t } = useI18n() const { t } = useI18n()
const companyStore = useCompanyStore() const companyStore = useCompanyStore()
const activeTab = ref('content') const activeTab = ref('content')
@@ -43,6 +43,7 @@ const streamTitle = ref('')
const streamMarkdown = ref('') const streamMarkdown = ref('')
const streamStatus = ref<string | null>(null) const streamStatus = ref<string | null>(null)
const streamFeatureEnabled = isGenerationStreamEnabled() const streamFeatureEnabled = isGenerationStreamEnabled()
const ARTICLE_DETAIL_POLL_INTERVAL_MS = 5000
const detailQuery = useQuery({ const detailQuery = useQuery({
queryKey: computed(() => ['articles', 'detail', companyStore.currentBrandId, props.articleId]), queryKey: computed(() => ['articles', 'detail', companyStore.currentBrandId, props.articleId]),
@@ -178,18 +179,30 @@ watch(
) )
watch( watch(
[() => props.open, () => props.articleId, () => detail.value?.generate_status], [
([open, articleId, generateStatus], _prev, onCleanup) => { () => props.open,
if (!open || !articleId || !hasActiveGenerationStatus(generateStatus)) { () => props.articleId,
() => detail.value?.generate_status,
() => detail.value?.publish_status,
],
([open, articleId, generateStatus, publishStatus], _prev, onCleanup) => {
const shouldPoll =
hasActiveGenerationStatus(generateStatus) || hasActivePublishStatus(publishStatus)
if (!open || !articleId || !shouldPoll) {
return return
} }
const timer = window.setInterval( const timer = window.setInterval(() => {
() => { if (
void refreshArticleData() detailQuery.isFetching.value ||
}, versionsQuery.isFetching.value ||
streamFeatureEnabled ? 5000 : 2000, publishRecordsQuery.isFetching.value
) ) {
return
}
void refreshArticleData()
}, ARTICLE_DETAIL_POLL_INTERVAL_MS)
onCleanup(() => { onCleanup(() => {
window.clearInterval(timer) window.clearInterval(timer)
@@ -234,13 +247,7 @@ function resetStreamState(): void {
} }
async function refreshArticleData(): Promise<void> { async function refreshArticleData(): Promise<void> {
await Promise.all([ await Promise.all([detailQuery.refetch(), versionsQuery.refetch(), publishRecordsQuery.refetch()])
detailQuery.refetch(),
versionsQuery.refetch(),
publishRecordsQuery.refetch(),
queryClient.invalidateQueries({ queryKey: ['articles'] }),
queryClient.invalidateQueries({ queryKey: ['workspace'] }),
])
} }
function resolvePublishLink(record: PublishRecord): string { function resolvePublishLink(record: PublishRecord): string {
apps/admin-web/src/components/ArticlePublishStatus.vue
+3 -1
View File
@@ -63,7 +63,9 @@ const publishRecordsQuery = useQuery({
companyStore.currentBrandId, companyStore.currentBrandId,
props.articleId, props.articleId,
]), ]),
enabled: computed(() => shouldLoadRecords.value && Boolean(companyStore.currentBrandId)), enabled: computed(
() => popoverOpen.value && shouldLoadRecords.value && Boolean(companyStore.currentBrandId),
),
queryFn: () => articlesApi.publishRecords(props.articleId as number), queryFn: () => articlesApi.publishRecords(props.articleId as number),
staleTime: 30 * 1000, staleTime: 30 * 1000,
}) })
apps/admin-web/src/components/CustomArticleTab.vue
+14 -4
View File
@@ -17,7 +17,12 @@ import { articlesApi, promptRulesApi } from '@/lib/api'
import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions' import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions'
import { useArticleRegenerateAction } from '@/lib/article-regenerate-action' import { useArticleRegenerateAction } from '@/lib/article-regenerate-action'
import { copyTextToClipboard } from '@/lib/clipboard' import { copyTextToClipboard } from '@/lib/clipboard'
import { getGenerateStatusMeta, getPublishStatusMeta } from '@/lib/display' import {
getGenerateStatusMeta,
getPublishStatusMeta,
hasActiveGenerationStatus,
hasActivePublishStatus,
} from '@/lib/display'
import { formatError } from '@/lib/errors' import { formatError } from '@/lib/errors'
import { formatPublishPlatformList } from '@/lib/publish-platforms' import { formatPublishPlatformList } from '@/lib/publish-platforms'
import { useCompanyStore } from '@/stores/company' import { useCompanyStore } from '@/stores/company'
@@ -109,6 +114,7 @@ const listQuery = useQuery({
}) })
let pollingTimer: number | null = null let pollingTimer: number | null = null
const ARTICLE_LIST_POLL_INTERVAL_MS = 5000
const deleteMutation = useMutation({ const deleteMutation = useMutation({
mutationFn: (id: number) => articlesApi.remove(id), mutationFn: (id: number) => articlesApi.remove(id),
@@ -205,8 +211,10 @@ function handleTableChange(nextPage: number, nextPageSize: number): void {
function startPolling(): void { function startPolling(): void {
if (pollingTimer !== null) return if (pollingTimer !== null) return
pollingTimer = window.setInterval(() => { pollingTimer = window.setInterval(() => {
void listQuery.refetch() if (!listQuery.isFetching.value) {
}, 3000) void listQuery.refetch()
}
}, ARTICLE_LIST_POLL_INTERVAL_MS)
} }
function stopPolling(): void { function stopPolling(): void {
@@ -236,7 +244,9 @@ watch(
() => listQuery.data.value?.items ?? [], () => listQuery.data.value?.items ?? [],
(items) => { (items) => {
const hasActive = items.some( const hasActive = items.some(
(item) => item.generate_status === 'generating' || item.generate_status === 'running', (item) =>
hasActiveGenerationStatus(item.generate_status) ||
hasActivePublishStatus(item.publish_status),
) )
if (hasActive) { if (hasActive) {
startPolling() startPolling()
apps/admin-web/src/components/InstantTaskTab.vue
+5 -2
View File
@@ -117,6 +117,7 @@ const columns = computed<TableColumnsType<InstantTaskItem>>(() => [
]) ])
let pollingTimer: number | null = null let pollingTimer: number | null = null
const TASK_LIST_POLL_INTERVAL_MS = 5000
function applyFilters(): void { function applyFilters(): void {
page.value = 1 page.value = 1
@@ -153,8 +154,10 @@ function openTaskArticles(task: InstantTaskItem): void {
function startPolling(): void { function startPolling(): void {
if (pollingTimer !== null) return if (pollingTimer !== null) return
pollingTimer = window.setInterval(() => { pollingTimer = window.setInterval(() => {
void listQuery.refetch() if (!listQuery.isFetching.value) {
}, 3000) void listQuery.refetch()
}
}, TASK_LIST_POLL_INTERVAL_MS)
} }
function stopPolling(): void { function stopPolling(): void {
apps/admin-web/src/components/ScheduleTaskTab.vue
+5 -2
View File
@@ -235,12 +235,15 @@ function openTaskArticles(task: ScheduleTask): void {
} }
let pollingTimer: number | null = null let pollingTimer: number | null = null
const TASK_LIST_POLL_INTERVAL_MS = 5000
function startPolling(): void { function startPolling(): void {
if (pollingTimer !== null) return if (pollingTimer !== null) return
pollingTimer = window.setInterval(() => { pollingTimer = window.setInterval(() => {
void listQuery.refetch() if (!listQuery.isFetching.value) {
}, 3000) void listQuery.refetch()
}
}, TASK_LIST_POLL_INTERVAL_MS)
} }
function stopPolling(): void { function stopPolling(): void {
apps/admin-web/src/layouts/AppShell.vue
+1 -1
View File
@@ -518,7 +518,7 @@ onMounted(() => {
syncMobileState(mobileMediaQuery) syncMobileState(mobileMediaQuery)
mobileMediaQuery.addEventListener('change', syncMobileState) mobileMediaQuery.addEventListener('change', syncMobileState)
if (!isMembershipBlocked.value) { if (!isMembershipBlocked.value && !companyStore.initialized) {
void companyStore.refreshBrands() void companyStore.refreshBrands()
} }
}) })
apps/admin-web/src/lib/display.ts
+12
View File
@@ -89,6 +89,18 @@ export function hasActiveGenerationStatus(status?: string | null): boolean {
) )
} }
export function hasActivePublishStatus(status?: string | null): boolean {
const normalized = String(status ?? '')
.trim()
.toLowerCase()
return (
normalized === 'queued' ||
normalized === 'pending' ||
normalized === 'publishing' ||
normalized === 'running'
)
}
export function getPublishStatusMeta(status?: string | null): { label: string; color: string } { export function getPublishStatusMeta(status?: string | null): { label: string; color: string } {
if (!status) { if (!status) {
return { label: '--', color: 'default' } return { label: '--', color: 'default' }
apps/admin-web/src/stores/company.ts
+20 -8
View File
@@ -15,6 +15,7 @@ export const useCompanyStore = defineStore('company', () => {
const currentBrandId = ref<number | null>(readStoredCurrentBrandId()) const currentBrandId = ref<number | null>(readStoredCurrentBrandId())
const loading = ref(false) const loading = ref(false)
const initialized = ref(false) const initialized = ref(false)
let refreshPromise: Promise<Brand[]> | null = null
const currentBrand = computed( const currentBrand = computed(
() => brands.value.find((brand) => brand.id === currentBrandId.value) ?? null, () => brands.value.find((brand) => brand.id === currentBrandId.value) ?? null,
@@ -56,15 +57,25 @@ export const useCompanyStore = defineStore('company', () => {
} }
async function refreshBrands(): Promise<Brand[]> { async function refreshBrands(): Promise<Brand[]> {
loading.value = true if (refreshPromise) {
try { return refreshPromise
const nextBrands = normalizeBrandList(await brandsApi.list())
setBrands(nextBrands)
initialized.value = true
return nextBrands
} finally {
loading.value = false
} }
loading.value = true
refreshPromise = brandsApi
.list()
.then((value) => {
const nextBrands = normalizeBrandList(value)
setBrands(nextBrands)
initialized.value = true
return nextBrands
})
.finally(() => {
loading.value = false
refreshPromise = null
})
return refreshPromise
} }
function setBrands(value: Brand[] | null | undefined): void { function setBrands(value: Brand[] | null | undefined): void {
@@ -78,6 +89,7 @@ export const useCompanyStore = defineStore('company', () => {
currentBrandId.value = null currentBrandId.value = null
initialized.value = false initialized.value = false
loading.value = false loading.value = false
refreshPromise = null
clearStoredCurrentBrandId() clearStoredCurrentBrandId()
} }
apps/admin-web/src/views/FreeCreateView.vue
+44 -2
View File
@@ -5,7 +5,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/vue-query'
import type { TableColumnsType } from 'ant-design-vue' import type { TableColumnsType } from 'ant-design-vue'
import { message } from 'ant-design-vue' import { message } from 'ant-design-vue'
import type { Dayjs } from 'dayjs' import type { Dayjs } from 'dayjs'
import { computed, reactive, ref } from 'vue' import { computed, onBeforeUnmount, reactive, ref, watch } from 'vue'
import { useI18n } from 'vue-i18n' import { useI18n } from 'vue-i18n'
import { useRouter } from 'vue-router' import { useRouter } from 'vue-router'
@@ -18,7 +18,7 @@ import { articlesApi } from '@/lib/api'
import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions' import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions'
import { useArticleRegenerateAction } from '@/lib/article-regenerate-action' import { useArticleRegenerateAction } from '@/lib/article-regenerate-action'
import { copyTextToClipboard } from '@/lib/clipboard' import { copyTextToClipboard } from '@/lib/clipboard'
import { getPublishStatusMeta } from '@/lib/display' import { getPublishStatusMeta, hasActivePublishStatus } from '@/lib/display'
import { formatError } from '@/lib/errors' import { formatError } from '@/lib/errors'
import { useCompanyStore } from '@/stores/company' import { useCompanyStore } from '@/stores/company'
@@ -34,6 +34,8 @@ const selectedArticleId = ref<number | null>(null)
const page = ref(1) const page = ref(1)
const pageSize = ref(10) const pageSize = ref(10)
const createdRange = ref<[Dayjs, Dayjs] | null>(null) const createdRange = ref<[Dayjs, Dayjs] | null>(null)
let articlePollingTimer: number | null = null
const ARTICLE_LIST_POLL_INTERVAL_MS = 5000
const draftFilters = reactive<{ const draftFilters = reactive<{
publish_status?: string publish_status?: string
@@ -172,11 +174,51 @@ function openPreview(article: ArticleListItem): void {
articleDrawerOpen.value = true articleDrawerOpen.value = true
} }
watch(
() => articleListQuery.data.value?.items || [],
(items: ArticleListItem[]) => {
const hasPublishingArticles = items.some((item) => hasActivePublishStatus(item.publish_status))
if (hasPublishingArticles) {
startArticlePolling()
return
}
stopArticlePolling()
},
{ immediate: true },
)
onBeforeUnmount(() => {
stopArticlePolling()
})
function handleTableChange(nextPage: number, nextPageSize: number): void { function handleTableChange(nextPage: number, nextPageSize: number): void {
page.value = nextPage page.value = nextPage
pageSize.value = nextPageSize pageSize.value = nextPageSize
} }
function startArticlePolling(): void {
if (articlePollingTimer !== null) {
return
}
articlePollingTimer = window.setInterval(() => {
if (!articleListQuery.isFetching.value) {
void articleListQuery.refetch()
}
}, ARTICLE_LIST_POLL_INTERVAL_MS)
}
function stopArticlePolling(): void {
if (articlePollingTimer === null) {
return
}
window.clearInterval(articlePollingTimer)
articlePollingTimer = null
}
async function handleDelete(articleId: number): Promise<void> { async function handleDelete(articleId: number): Promise<void> {
await deleteMutation.mutateAsync(articleId) await deleteMutation.mutateAsync(articleId)
} }
apps/admin-web/src/views/ImitationView.vue
+8 -4
View File
@@ -23,6 +23,7 @@ import {
getGenerateStatusMeta, getGenerateStatusMeta,
getPublishStatusMeta, getPublishStatusMeta,
hasActiveGenerationStatus, hasActiveGenerationStatus,
hasActivePublishStatus,
} from '@/lib/display' } from '@/lib/display'
import { formatError } from '@/lib/errors' import { formatError } from '@/lib/errors'
import { useCompanyStore } from '@/stores/company' import { useCompanyStore } from '@/stores/company'
@@ -93,6 +94,7 @@ const articleListQuery = useQuery({
}) })
let articlePollingTimer: number | null = null let articlePollingTimer: number | null = null
const ARTICLE_LIST_POLL_INTERVAL_MS = 5000
const deleteMutation = useMutation({ const deleteMutation = useMutation({
mutationFn: (articleId: number) => articlesApi.remove(articleId), mutationFn: (articleId: number) => articlesApi.remove(articleId),
@@ -180,8 +182,9 @@ watch(
const hasGeneratingArticles = items.some((item) => const hasGeneratingArticles = items.some((item) =>
hasActiveGenerationStatus(item.generate_status), hasActiveGenerationStatus(item.generate_status),
) )
const hasPublishingArticles = items.some((item) => hasActivePublishStatus(item.publish_status))
if (hasGeneratingArticles) { if (hasGeneratingArticles || hasPublishingArticles) {
startArticlePolling() startArticlePolling()
return return
} }
@@ -245,9 +248,10 @@ function startArticlePolling(): void {
} }
articlePollingTimer = window.setInterval(() => { articlePollingTimer = window.setInterval(() => {
void queryClient.invalidateQueries({ queryKey: ['articles'] }) if (!articleListQuery.isFetching.value) {
void articleListQuery.refetch() void articleListQuery.refetch()
}, 3000) }
}, ARTICLE_LIST_POLL_INTERVAL_MS)
} }
function stopArticlePolling(): void { function stopArticlePolling(): void {
apps/admin-web/src/views/PublishManagementView.vue
+4
View File
@@ -118,6 +118,7 @@ function inlineErrorPreview(messageText: string): string {
function summaryForRecord(record: PublishRecord, status: string): string { function summaryForRecord(record: PublishRecord, status: string): string {
const targetLabel = record.target_type === 'enterprise_site' ? '企业站点' : '目标平台' const targetLabel = record.target_type === 'enterprise_site' ? '企业站点' : '目标平台'
const errorMessage = normalizeRecordErrorMessage(record.error_message, record.platform_id)
switch (status) { switch (status) {
case 'queued': case 'queued':
return record.target_type === 'enterprise_site' return record.target_type === 'enterprise_site'
@@ -134,6 +135,9 @@ function summaryForRecord(record: PublishRecord, status: string): string {
? '本次企业站发布失败，请检查站点连接或发布错误。' ? '本次企业站发布失败，请检查站点连接或发布错误。'
: '本次发送失败，请检查账号状态或平台提示后重新提交。' : '本次发送失败，请检查账号状态或平台提示后重新提交。'
case 'cancelled': case 'cancelled':
if (errorMessage && /等待通道|publish_queue_timeout/.test(errorMessage)) {
return '等待通道超时，系统已自动取消。'
}
return '本次发布已取消。' return '本次发布已取消。'
case 'unknown': case 'unknown':
default: default:
apps/admin-web/src/views/TemplatesView.vue
+17 -5
View File
@@ -42,6 +42,7 @@ import {
getPublishStatusMeta, getPublishStatusMeta,
getTemplateMeta, getTemplateMeta,
hasActiveGenerationStatus, hasActiveGenerationStatus,
hasActivePublishStatus,
} from '@/lib/display' } from '@/lib/display'
import { formatError } from '@/lib/errors' import { formatError } from '@/lib/errors'
import { useCompanyStore } from '@/stores/company' import { useCompanyStore } from '@/stores/company'
@@ -207,6 +208,7 @@ const displayedArticleTotal = computed(() =>
) )
let articlePollingTimer: number | null = null let articlePollingTimer: number | null = null
const ARTICLE_LIST_POLL_INTERVAL_MS = 5000
const deleteMutation = useMutation({ const deleteMutation = useMutation({
mutationFn: (articleId: number) => articlesApi.remove(articleId), mutationFn: (articleId: number) => articlesApi.remove(articleId),
@@ -418,8 +420,11 @@ watch(
const hasGeneratingArticles = items.some((item: ArticleListItem) => const hasGeneratingArticles = items.some((item: ArticleListItem) =>
hasActiveGenerationStatus(item.generate_status), hasActiveGenerationStatus(item.generate_status),
) )
const hasPublishingArticles = items.some((item: ArticleListItem) =>
hasActivePublishStatus(item.publish_status),
)
if (hasGeneratingArticles) { if (hasGeneratingArticles || hasPublishingArticles) {
startArticlePolling() startArticlePolling()
return return
} }
@@ -444,10 +449,17 @@ function startArticlePolling(): void {
} }
articlePollingTimer = window.setInterval(() => { articlePollingTimer = window.setInterval(() => {
void queryClient.invalidateQueries({ queryKey: ['articles'] }) if (shouldUseRecentFallback.value) {
void queryClient.invalidateQueries({ queryKey: ['workspace', 'recent-articles'] }) if (!recentArticlesQuery.isFetching.value) {
void Promise.all([articleListQuery.refetch(), recentArticlesQuery.refetch()]) void recentArticlesQuery.refetch()
}, 3000) }
return
}
if (!articleListQuery.isFetching.value) {
void articleListQuery.refetch()
}
}, ARTICLE_LIST_POLL_INTERVAL_MS)
} }
function stopArticlePolling(): void { function stopArticlePolling(): void {
apps/admin-web/src/views/WorkspaceView.vue
+13 -7
View File
@@ -31,7 +31,12 @@ import { articlesApi, tenantAccountsApi, workspaceApi } from '@/lib/api'
import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions' import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions'
import { useArticleRegenerateAction } from '@/lib/article-regenerate-action' import { useArticleRegenerateAction } from '@/lib/article-regenerate-action'
import { copyTextToClipboard } from '@/lib/clipboard' import { copyTextToClipboard } from '@/lib/clipboard'
import { formatDateTime, getTemplateMeta, hasActiveGenerationStatus } from '@/lib/display' import {
formatDateTime,
getTemplateMeta,
hasActiveGenerationStatus,
hasActivePublishStatus,
} from '@/lib/display'
import { formatError } from '@/lib/errors' import { formatError } from '@/lib/errors'
import { isMediaPublishAccount } from '@/lib/publish-platforms' import { isMediaPublishAccount } from '@/lib/publish-platforms'
import { useCompanyStore } from '@/stores/company' import { useCompanyStore } from '@/stores/company'
@@ -45,6 +50,7 @@ const articleDrawerOpen = ref(false)
const selectedPublishArticleId = ref<number | null>(null) const selectedPublishArticleId = ref<number | null>(null)
const publishModalOpen = ref(false) const publishModalOpen = ref(false)
let recentArticlesPollingTimer: number | null = null let recentArticlesPollingTimer: number | null = null
const ARTICLE_LIST_POLL_INTERVAL_MS = 5000
const overviewQuery = useQuery({ const overviewQuery = useQuery({
queryKey: computed(() => ['workspace', 'overview', companyStore.currentBrandId]), queryKey: computed(() => ['workspace', 'overview', companyStore.currentBrandId]),
@@ -300,8 +306,9 @@ watch(
const hasGeneratingArticles = items.some((item) => const hasGeneratingArticles = items.some((item) =>
hasActiveGenerationStatus(item.generate_status), hasActiveGenerationStatus(item.generate_status),
) )
const hasPublishingArticles = items.some((item) => hasActivePublishStatus(item.publish_status))
if (hasGeneratingArticles) { if (hasGeneratingArticles || hasPublishingArticles) {
startRecentArticlesPolling() startRecentArticlesPolling()
return return
} }
@@ -325,12 +332,11 @@ function startRecentArticlesPolling(): void {
stopRecentArticlesPolling() stopRecentArticlesPolling()
return return
} }
void queryClient.invalidateQueries({ if (recentArticlesQuery.isFetching.value) {
queryKey: ['workspace', 'recent-articles', companyStore.currentBrandId], return
}) }
void queryClient.invalidateQueries({ queryKey: ['articles'] })
void recentArticlesQuery.refetch() void recentArticlesQuery.refetch()
}, 3000) }, ARTICLE_LIST_POLL_INTERVAL_MS)
} }
function stopRecentArticlesPolling(): void { function stopRecentArticlesPolling(): void {
server/internal/scheduler/publish_lease_recovery_worker.go
+59 -16
View File
@@ -13,26 +13,36 @@ const (
defaultPublishLeaseRecoveryInterval = 3 * time.Minute defaultPublishLeaseRecoveryInterval = 3 * time.Minute
defaultPublishLeaseRecoveryTimeout = 30 * time.Second defaultPublishLeaseRecoveryTimeout = 30 * time.Second
defaultPublishLeaseRecoveryLimit = 1000 defaultPublishLeaseRecoveryLimit = 1000
defaultPublishQueuedTaskTimeout = 3 * 24 * time.Hour
) )
type PublishLeaseRecoveryWorker struct { type PublishLeaseRecoveryWorker struct {
service *tenantapp.DesktopTaskService service *tenantapp.DesktopTaskService
logger *zap.Logger logger *zap.Logger
interval time.Duration interval time.Duration
timeout time.Duration timeout time.Duration
limit int limit int
queuedTaskTimeout time.Duration
} }
func NewPublishLeaseRecoveryWorker(service *tenantapp.DesktopTaskService, logger *zap.Logger) *PublishLeaseRecoveryWorker { func NewPublishLeaseRecoveryWorker(service *tenantapp.DesktopTaskService, logger *zap.Logger) *PublishLeaseRecoveryWorker {
return &PublishLeaseRecoveryWorker{ return &PublishLeaseRecoveryWorker{
service: service, service: service,
logger: logger, logger: logger,
interval: defaultPublishLeaseRecoveryInterval, interval: defaultPublishLeaseRecoveryInterval,
timeout: defaultPublishLeaseRecoveryTimeout, timeout: defaultPublishLeaseRecoveryTimeout,
limit: defaultPublishLeaseRecoveryLimit, limit: defaultPublishLeaseRecoveryLimit,
queuedTaskTimeout: defaultPublishQueuedTaskTimeout,
} }
} }
func (w *PublishLeaseRecoveryWorker) WithQueuedTaskTimeout(timeout time.Duration) *PublishLeaseRecoveryWorker {
if w != nil && timeout > 0 {
w.queuedTaskTimeout = timeout
}
return w
}
func (w *PublishLeaseRecoveryWorker) Start(ctx context.Context) { func (w *PublishLeaseRecoveryWorker) Start(ctx context.Context) {
go w.Run(ctx) go w.Run(ctx)
} }
@@ -72,15 +82,40 @@ func (w *PublishLeaseRecoveryWorker) runOnce(parent context.Context) {
return return
} }
if (result.Requeued > 0 || result.Failed > 0 || result.Uncertain > 0) && w.logger != nil { queuedTaskTimeout := w.queuedTaskTimeout
if queuedTaskTimeout <= 0 {
queuedTaskTimeout = defaultPublishQueuedTaskTimeout
}
cancelledQueued, err := w.service.CancelStaleQueuedPublishTasks(ctx, queuedTaskTimeout, w.limit)
if err != nil {
if w.logger != nil {
w.logger.Warn("publish queued task timeout cleanup failed", zap.Error(err))
}
return
}
result.CancelledQueued = cancelledQueued
if (result.Requeued > 0 || result.Failed > 0 || result.Uncertain > 0 || result.CancelledQueued > 0) && w.logger != nil {
w.logger.Info("publish lease recovery completed", w.logger.Info("publish lease recovery completed",
zap.Int("requeued_task_count", result.Requeued), zap.Int("requeued_task_count", result.Requeued),
zap.Int("failed_task_count", result.Failed), zap.Int("failed_task_count", result.Failed),
zap.Int("uncertain_task_count", result.Uncertain), zap.Int("uncertain_task_count", result.Uncertain),
zap.Int("cancelled_queued_task_count", result.CancelledQueued),
) )
} }
} }
func publishQueuedTaskTimeoutFromRun(run JobRunContext) time.Duration {
timeout := AsDuration(run.Config, "queued_task_timeout", defaultPublishQueuedTaskTimeout)
if seconds := AsInt(run.Config, "queued_task_timeout_seconds", 0); seconds > 0 {
timeout = time.Duration(seconds) * time.Second
}
if timeout <= 0 {
return defaultPublishQueuedTaskTimeout
}
return timeout
}
func (w *PublishLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunContext) (map[string]any, error) { func (w *PublishLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunContext) (map[string]any, error) {
if w == nil || w.service == nil { if w == nil || w.service == nil {
return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil
@@ -104,11 +139,19 @@ func (w *PublishLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunC
if err != nil { if err != nil {
return map[string]any{"stage": "recover"}, err return map[string]any{"stage": "recover"}, err
} }
queuedTimeout := publishQueuedTaskTimeoutFromRun(run)
cancelledQueued, err := w.service.CancelStaleQueuedPublishTasks(ctx, queuedTimeout, limit)
if err != nil {
return map[string]any{"stage": "cancel_stale_queued"}, err
}
result.CancelledQueued = cancelledQueued
return map[string]any{ return map[string]any{
"requeued_task_count": result.Requeued, "requeued_task_count": result.Requeued,
"failed_task_count": result.Failed, "failed_task_count": result.Failed,
"uncertain_task_count": result.Uncertain, "uncertain_task_count": result.Uncertain,
"duration_ms": time.Since(startedAt).Milliseconds(), "cancelled_queued_task_count": result.CancelledQueued,
"batch_size": limit, "queued_task_timeout_seconds": int64(queuedTimeout.Seconds()),
"duration_ms": time.Since(startedAt).Milliseconds(),
"batch_size": limit,
}, nil }, nil
} }
server/internal/scheduler/publish_lease_recovery_worker_test.go
+42
View File
@@ -0,0 +1,42 @@
package scheduler
import (
"testing"
"time"
)
func TestPublishQueuedTaskTimeoutFromRunDefaultsToThreeDays(t *testing.T) {
t.Parallel()
got := publishQueuedTaskTimeoutFromRun(JobRunContext{})
if got != 3*24*time.Hour {
t.Fatalf("publishQueuedTaskTimeoutFromRun() = %s, want 72h", got)
}
}
func TestPublishQueuedTaskTimeoutFromRunReadsDurationConfig(t *testing.T) {
t.Parallel()
got := publishQueuedTaskTimeoutFromRun(JobRunContext{
Config: map[string]any{
"queued_task_timeout": "72h",
},
})
if got != 72*time.Hour {
t.Fatalf("publishQueuedTaskTimeoutFromRun(duration) = %s, want 72h", got)
}
}
func TestPublishQueuedTaskTimeoutFromRunReadsSecondsConfig(t *testing.T) {
t.Parallel()
got := publishQueuedTaskTimeoutFromRun(JobRunContext{
Config: map[string]any{
"queued_task_timeout": "72h",
"queued_task_timeout_seconds": 3600,
},
})
if got != time.Hour {
t.Fatalf("publishQueuedTaskTimeoutFromRun(seconds) = %s, want 1h", got)
}
}
server/internal/tenant/app/desktop_task_service.go
+187 -3
View File
@@ -1987,9 +1987,10 @@ type recoveredDesktopTaskLease struct {
} }
type PublishLeaseRecoveryResult struct { type PublishLeaseRecoveryResult struct {
Requeued int `json:"requeued"` Requeued int `json:"requeued"`
Failed int `json:"failed"` Failed int `json:"failed"`
Uncertain int `json:"uncertain"` Uncertain int `json:"uncertain"`
CancelledQueued int `json:"cancelled_queued"`
} }
type MonitorLeaseRecoveryResult struct { type MonitorLeaseRecoveryResult struct {
@@ -2456,6 +2457,189 @@ func (s *DesktopTaskService) RecoverExpiredPublishTasks(ctx context.Context, lim
return result, nil return result, nil
} }
func staleQueuedPublishTasksCandidateSQL() string {
return `
SELECT
t.desktop_id,
t.workspace_id
FROM desktop_tasks AS t
JOIN desktop_publish_jobs AS j
ON j.desktop_id = t.job_id
AND j.tenant_id = t.tenant_id
AND j.workspace_id = t.workspace_id
WHERE t.kind = 'publish'
AND t.status = 'queued'
AND GREATEST(
COALESCE(t.enqueued_at, t.created_at),
COALESCE(j.scheduled_at, t.created_at)
) < $1
ORDER BY COALESCE(t.enqueued_at, t.created_at) ASC,
t.created_at ASC,
t.desktop_id ASC
LIMIT $2
FOR UPDATE OF t SKIP LOCKED
`
}
func publishQueueTimeoutErrorPayload(timeout time.Duration) ([]byte, error) {
timeoutSeconds := int64(timeout.Seconds())
if timeoutSeconds < 0 {
timeoutSeconds = 0
}
message := fmt.Sprintf(
"发布任务在等待通道中超过 %s未被桌面端领取，系统已自动取消，避免任务无限期停留。请确认对应桌面端在线、账号登录正常且任务通道在线后，再重新发布。",
publishQueueTimeoutDisplayDuration(timeout),
)
return marshalOptionalJSON(map[string]any{
"code": "publish_queue_timeout",
"message": message,
"source": "publish_queue_timeout",
"queued_timeout_seconds": timeoutSeconds,
"queued_timeout_duration": timeout.String(),
})
}
func publishQueueTimeoutDisplayDuration(timeout time.Duration) string {
if timeout <= 0 {
return "配置的时间"
}
minutes := int64(timeout.Round(time.Minute) / time.Minute)
if minutes <= 0 {
return "1 分钟"
}
const (
minutesPerHour = int64(60)
minutesPerDay = int64(24 * 60)
)
if minutes%minutesPerDay == 0 {
return fmt.Sprintf("%d 天", minutes/minutesPerDay)
}
if minutes%minutesPerHour == 0 {
return fmt.Sprintf("%d 小时", minutes/minutesPerHour)
}
if minutes > minutesPerHour {
return fmt.Sprintf("%d 小时 %d 分钟", minutes/minutesPerHour, minutes%minutesPerHour)
}
return fmt.Sprintf("%d 分钟", minutes)
}
func (s *DesktopTaskService) CancelStaleQueuedPublishTasks(ctx context.Context, queuedFor time.Duration, limit int) (int, error) {
if s == nil || s.pool == nil || queuedFor <= 0 {
return 0, nil
}
if limit <= 0 {
limit = 1000
}
errorJSON, err := publishQueueTimeoutErrorPayload(queuedFor)
if err != nil {
return 0, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish queue timeout payload")
}
staleBefore := time.Now().UTC().Add(-queuedFor)
tx, err := s.pool.Begin(ctx)
if err != nil {
return 0, response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery")
}
defer tx.Rollback(ctx)
rows, err := tx.Query(ctx, staleQueuedPublishTasksCandidateSQL(), staleBefore, limit)
if err != nil {
s.logWarn("stale queued publish task query failed", err)
return 0, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select stale queued publish tasks")
}
defer rows.Close()
candidates := make([]recoveredDesktopTaskLease, 0)
for rows.Next() {
var item recoveredDesktopTaskLease
if scanErr := rows.Scan(&item.TaskID, &item.WorkspaceID); scanErr != nil {
s.logWarn("stale queued publish task scan failed", scanErr)
return 0, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse stale queued publish task")
}
item.Kind = "publish"
item.Status = "aborted"
item.ErrorJSON = errorJSON
candidates = append(candidates, item)
}
if err := rows.Err(); err != nil {
s.logWarn("stale queued publish task rows failed", err)
return 0, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate stale queued publish tasks")
}
publishOutcomes := make([]*desktopPublishSyncOutcome, 0, len(candidates))
cancelled := 0
for index := range candidates {
item := &candidates[index]
row := tx.QueryRow(ctx, `
UPDATE desktop_tasks
SET status = 'aborted',
error = $2,
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
interrupted_at = COALESCE(interrupted_at, NOW()),
interrupt_reason = COALESCE(interrupt_reason, 'publish_queue_timeout'),
updated_at = NOW()
WHERE desktop_id = $1
AND kind = 'publish'
AND status = 'queued'
RETURNING desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error
`, item.TaskID, errorJSON)
var updated repository.DesktopTask
if scanErr := row.Scan(
&updated.DesktopID,
&updated.TenantID,
&updated.WorkspaceID,
&updated.Platform,
&updated.Kind,
&updated.Payload,
&updated.Status,
&updated.Result,
&updated.Error,
); scanErr != nil {
if errors.Is(scanErr, pgx.ErrNoRows) {
continue
}
s.logWarn("stale queued publish task update failed", scanErr, zap.String("task_id", item.TaskID.String()))
return cancelled, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to cancel stale queued publish task")
}
publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, &updated)
if syncErr != nil {
return cancelled, syncErr
}
if publishOutcome != nil {
publishOutcomes = append(publishOutcomes, publishOutcome)
}
cancelled++
}
if err := tx.Commit(ctx); err != nil {
return cancelled, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery")
}
s.afterRecoveredPublishOutcomes(ctx, publishOutcomes)
for _, item := range candidates {
task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
if getErr != nil {
s.logWarn("stale queued publish task reload after commit failed", getErr, zap.String("task_id", item.TaskID.String()))
continue
}
s.publishTaskEvent(ctx, task, "task_completed")
}
if cancelled > 0 && s.logger != nil {
s.logger.Info("stale queued publish tasks cancelled",
zap.Int("cancelled", cancelled),
zap.Duration("queued_for", queuedFor),
)
}
return cancelled, nil
}
func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, limit int) (MonitorLeaseRecoveryResult, error) { func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, limit int) (MonitorLeaseRecoveryResult, error) {
var result MonitorLeaseRecoveryResult var result MonitorLeaseRecoveryResult
if s == nil || s.pool == nil { if s == nil || s.pool == nil {
server/internal/tenant/app/desktop_task_service_test.go
+66
View File
@@ -6,6 +6,7 @@ import (
"regexp" "regexp"
"strings" "strings"
"testing" "testing"
"time"
"github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/tenant/repository" "github.com/geo-platform/tenant-api/internal/tenant/repository"
@@ -53,6 +54,71 @@ func TestDesktopTaskRecoverySelectQueryLeaseExpiryFiltersExpiredLeases(t *testin
} }
} }
func TestStaleQueuedPublishTasksCandidateSQLUsesExecutableQueueAge(t *testing.T) {
t.Parallel()
query := normalizeSQLForDesktopTaskTest(staleQueuedPublishTasksCandidateSQL())
for _, fragment := range []string{
"FROM desktop_tasks AS t",
"JOIN desktop_publish_jobs AS j",
"t.kind = 'publish'",
"t.status = 'queued'",
"GREATEST( COALESCE(t.enqueued_at, t.created_at), COALESCE(j.scheduled_at, t.created_at) ) < $1",
"LIMIT $2",
"FOR UPDATE OF t SKIP LOCKED",
} {
if !strings.Contains(query, fragment) {
t.Fatalf("stale queued publish query missing %q: %s", fragment, query)
}
}
}
func TestPublishQueueTimeoutErrorPayloadIncludesConfig(t *testing.T) {
t.Parallel()
payload, err := publishQueueTimeoutErrorPayload(3 * 24 * time.Hour)
if err != nil {
t.Fatalf("publishQueueTimeoutErrorPayload() error = %v", err)
}
text := string(payload)
for _, fragment := range []string{
`"code":"publish_queue_timeout"`,
`发布任务在等待通道中超过 3 天未被桌面端领取`,
`请确认对应桌面端在线、账号登录正常且任务通道在线后，再重新发布。`,
`"source":"publish_queue_timeout"`,
`"queued_timeout_seconds":259200`,
} {
if !strings.Contains(text, fragment) {
t.Fatalf("queue timeout payload missing %q: %s", fragment, text)
}
}
}
func TestPublishQueueTimeoutDisplayDurationUsesReadableChinese(t *testing.T) {
t.Parallel()
cases := []struct {
name string
timeout time.Duration
want string
}{
{name: "days", timeout: 7 * 24 * time.Hour, want: "7 天"},
{name: "hours", timeout: 36 * time.Hour, want: "36 小时"},
{name: "hours and minutes", timeout: 90 * time.Minute, want: "1 小时 30 分钟"},
{name: "minutes", timeout: 45 * time.Minute, want: "45 分钟"},
}
for _, tc := range cases {
tc := tc
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
if got := publishQueueTimeoutDisplayDuration(tc.timeout); got != tc.want {
t.Fatalf("publishQueueTimeoutDisplayDuration(%s) = %q, want %q", tc.timeout, got, tc.want)
}
})
}
}
func TestReconcileRetryPublishesTaskAvailableForAllDesktopTaskKinds(t *testing.T) { func TestReconcileRetryPublishesTaskAvailableForAllDesktopTaskKinds(t *testing.T) {
t.Parallel() t.Parallel()