xiongyuxing/gongxue-base
1
0
Fork 0
forked from wangziqi/gongxue-base
Code Pull Requests Activity

feat: add public bank sync failure ops

Browse Source
This commit is contained in:
Codex
2026-06-30 09:12:45 +08:00
parent d1cb341351
commit 959e612211
14 changed files with 484 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/api/src/features/platform-admin/index.ts
View File

@@ -19,6 +19,7 @@ import {
platformPermissionsRoute,
platformPlansRoute,
platformQuestionBanksRoute,
publicQuestionBankSyncStatusRoute,
platformStaffRoute,
processOverdueInvoicesRoute,
questionBankGrantsRoute,
@@ -47,6 +48,7 @@ export const platformAdminRoutes: RouteDefinition[] = [
['GET', '/api/platform-admin/plans', platformPlansRoute],
['GET', '/api/platform-admin/question-banks', platformQuestionBanksRoute],
['GET', '/api/platform-admin/question-bank-grants', questionBankGrantsRoute],
['GET', '/api/platform-admin/question-bank-sync-status', publicQuestionBankSyncStatusRoute],
['PUT', '/api/platform-admin/question-bank-grants', upsertQuestionBankGrantRoute],
['GET', '/api/platform-admin/tenants', tenantsRoute],
['POST', '/api/platform-admin/tenants', createTenantRoute],

103
apps/api/src/features/platform-admin/routes.ts
View File

@@ -86,6 +86,7 @@ const PLATFORM_PERMISSION_CATALOG = [
{ key: 'platform:audit:notification', group: 'audit', label: '维护审计告警通知' },
{ key: 'platform:question_bank:read', group: 'question_bank', label: '查看公共题库' },
{ key: 'platform:question_bank:grant', group: 'question_bank', label: '授权公共题库' },
{ key: 'platform:question_bank:ops', group: 'question_bank', label: '查看公共题库同步运营' },
] as const;
function csvEscape(value: unknown) {
@@ -1014,6 +1015,108 @@ export async function questionBankGrantsRoute(ctx: RequestContext) {
return { items };
}
export async function publicQuestionBankSyncStatusRoute(ctx: RequestContext) {
await requirePlatformAdmin(ctx, 'platform:question_bank:ops');
const tenantId = listQuery(ctx, 'tenantId');
const sourceQuestionBankId = listQuery(ctx, 'sourceQuestionBankId');
const syncStatus = listQuery(ctx, 'syncStatus');
const status = listQuery(ctx, 'status');
const onlyOpenIssues = listQuery(ctx, 'onlyOpenIssues') === 'true';
const limit = intParam(ctx, 'limit', 100, 500);
if (tenantId && !UUID_RE.test(tenantId)) throw new HttpError(400, 'tenantId is invalid', 'INVALID_UUID');
if (sourceQuestionBankId && !UUID_RE.test(sourceQuestionBankId)) throw new HttpError(400, 'sourceQuestionBankId is invalid', 'INVALID_UUID');
if (syncStatus && !['pending', 'synced', 'failed'].includes(syncStatus)) {
throw new HttpError(400, 'syncStatus is invalid', 'INVALID_SYNC_STATUS');
}
if (status && !['active', 'sync_pending', 'suspended', 'archived'].includes(status)) {
throw new HttpError(400, 'status is invalid', 'INVALID_STATUS');
}
const items = await query(
`
select a.id,
a.tenant_id as "tenantId",
t.slug::text as "tenantSlug",
t.name as "tenantName",
a.source_question_bank_id as "sourceQuestionBankId",
source_qb.name as "sourceQuestionBankName",
a.grant_id as "grantId",
a.target_question_bank_id as "targetQuestionBankId",
target_qb.name as "targetQuestionBankName",
a.target_entry_id as "targetEntryId",
ce.name as "targetEntryName",
a.target_collection_id as "targetCollectionId",
qc.name as "targetCollectionName",
a.status,
a.sync_status as "syncStatus",
a.copied_question_count as "copiedQuestionCount",
a.last_synced_at as "lastSyncedAt",
a.updated_at as "updatedAt",
coalesce((a.metadata #>> '{lastSync,conflictCount}')::integer, 0) as "conflictCount",
coalesce(a.metadata #> '{lastSync,counts}', '{}'::jsonb) as "lastSyncCounts",
jsonb_build_object(
'lastSync', coalesce(a.metadata->'lastSync', '{}'::jsonb),
'publicBankSyncWorker', coalesce(a.metadata->'publicBankSyncWorker', '{}'::jsonb)
) as metadata,
coalesce(open_notifications.open_count, 0)::integer as "openNotificationCount",
coalesce(open_notifications.failure_count, 0)::integer as "openFailureNotificationCount",
coalesce(open_notifications.conflict_count, 0)::integer as "openConflictNotificationCount"
from public.tenant_question_bank_adoptions a
join public.tenants t on t.id = a.tenant_id
join public.question_banks source_qb on source_qb.id = a.source_question_bank_id
left join public.question_banks target_qb on target_qb.id = a.target_question_bank_id
left join public.content_entries ce on ce.id = a.target_entry_id and ce.tenant_id = a.tenant_id
left join public.question_collections qc on qc.id = a.target_collection_id and qc.tenant_id = a.tenant_id
left join lateral (
select count(*) filter (where n.status in ('unread', 'read')) as open_count,
count(*) filter (
where n.status in ('unread', 'read')
and n.notification_type = 'public_question_bank_sync_failed'
) as failure_count,
count(*) filter (
where n.status in ('unread', 'read')
and n.notification_type = 'public_question_bank_conflict'
) as conflict_count
from public.tenant_content_notifications n
where n.tenant_id = a.tenant_id
and n.adoption_id = a.id
and n.notification_type in ('public_question_bank_sync_failed', 'public_question_bank_conflict')
) open_notifications on true
where ($1::uuid is null or a.tenant_id = $1::uuid)
and ($2::uuid is null or a.source_question_bank_id = $2::uuid)
and ($3::text = '' or a.sync_status = $3)
and ($4::text = '' or a.status = $4)
and (
$5::boolean = false
or a.sync_status = 'failed'
or coalesce(open_notifications.open_count, 0) > 0
)
order by
case when a.sync_status = 'failed' then 0 when a.sync_status = 'pending' then 1 else 2 end,
coalesce(a.last_synced_at, a.updated_at) desc
limit $6
`,
[tenantId || null, sourceQuestionBankId || null, syncStatus, status, onlyOpenIssues, limit],
);
const summaryRows = await query<{ syncStatus: string; count: number }>(
`
select a.sync_status as "syncStatus", count(*)::integer as count
from public.tenant_question_bank_adoptions a
where ($1::uuid is null or a.tenant_id = $1::uuid)
and ($2::uuid is null or a.source_question_bank_id = $2::uuid)
group by a.sync_status
`,
[tenantId || null, sourceQuestionBankId || null],
);
return {
items,
summary: Object.fromEntries(summaryRows.map(row => [row.syncStatus, Number(row.count || 0)])),
};
}
export async function upsertQuestionBankGrantRoute(ctx: RequestContext) {
await requirePlatformAdmin(ctx, 'platform:question_bank:grant');

27
apps/api/src/features/tenant-content/public-banks.ts
View File

@@ -766,6 +766,28 @@ async function markPublicBankConflictNotificationsResolved(client: pg.PoolClient
);
}
async function markPublicBankFailureNotificationsResolved(client: pg.PoolClient, input: {
tenantId: string;
adoptionId: string;
actorUserId: string | null;
}) {
await client.query(
`
update public.tenant_content_notifications
set status = 'resolved',
read_by = coalesce(read_by, $3),
read_at = coalesce(read_at, now()),
resolved_at = coalesce(resolved_at, now()),
updated_at = now()
where tenant_id = $1
and adoption_id = $2
and notification_type = 'public_question_bank_sync_failed'
and status <> 'dismissed'
`,
[input.tenantId, input.adoptionId, input.actorUserId],
);
}
function objectValue(value: unknown): Record<string, unknown> {
return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
}
@@ -1249,6 +1271,11 @@ export async function executePublicQuestionBankSync(input: PublicQuestionBankSyn
actorUserId: auth.userId,
});
}
await markPublicBankFailureNotificationsResolved(client, {
tenantId: auth.tenantId,
adoptionId: adoption.id,
actorUserId: auth.userId,
});
await client.query(
`

167
apps/worker/src/jobs/public-banks.ts
View File

@@ -39,6 +39,14 @@ function truncate(value: unknown, max = 1900) {
return String(value ?? '').slice(0, max);
}
function publicBankFailureDedupeKey(candidate: PublicBankSyncCandidate, code: string) {
return [
'public-bank-sync-failed',
candidate.id,
crypto.createHash('sha256').update(code).digest('hex').slice(0, 16),
].join(':');
}
async function claimPublicBankSyncCandidates(limit: number, claimId: string) {
const client = await pool.connect();
try {
@@ -126,38 +134,133 @@ async function markPublicBankSyncFailed(candidate: PublicBankSyncCandidate, erro
workerId: config.publicBankSyncWorkerId,
failedAt: nowIso(),
};
await pool.query(
`
update public.tenant_question_bank_adoptions
set status = 'active',
sync_status = 'failed',
metadata = jsonb_set(
coalesce(metadata, '{}'::jsonb),
'{publicBankSyncWorker}',
coalesce(metadata->'publicBankSyncWorker', '{}'::jsonb) || $3::jsonb,
true
),
updated_at = now()
where tenant_id = $1 and id = $2
`,
[
candidate.tenantId,
candidate.id,
JSON.stringify({
lastStatus: 'failed',
lastError: details,
lastWorkerId: config.publicBankSyncWorkerId,
lastFinishedAt: nowIso(),
}),
],
);
await pool.query(
`
insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
values ($1, null, 'content.public_question_bank.sync_worker_failed', 'tenant_question_bank_adoption', $2, $3::jsonb)
`,
[candidate.tenantId, candidate.id, JSON.stringify(details)],
);
const client = await pool.connect();
try {
await client.query('begin');
await client.query(
`
update public.tenant_question_bank_adoptions
set status = 'active',
sync_status = 'failed',
metadata = jsonb_set(
coalesce(metadata, '{}'::jsonb),
'{publicBankSyncWorker}',
coalesce(metadata->'publicBankSyncWorker', '{}'::jsonb) || $3::jsonb,
true
),
updated_at = now()
where tenant_id = $1 and id = $2
`,
[
candidate.tenantId,
candidate.id,
JSON.stringify({
lastStatus: 'failed',
lastError: details,
lastWorkerId: config.publicBankSyncWorkerId,
lastFinishedAt: nowIso(),
}),
],
);
const adoptionResult = await client.query<{
source_question_bank_id: string;
target_question_bank_id: string | null;
target_entry_id: string | null;
target_collection_id: string | null;
source_question_bank_name: string | null;
target_question_bank_name: string | null;
}>(
`
select a.source_question_bank_id,
a.target_question_bank_id,
a.target_entry_id,
a.target_collection_id,
source_qb.name as source_question_bank_name,
target_qb.name as target_question_bank_name
from public.tenant_question_bank_adoptions a
join public.question_banks source_qb
on source_qb.id = a.source_question_bank_id
left join public.question_banks target_qb
on target_qb.id = a.target_question_bank_id
and target_qb.tenant_id = a.tenant_id
where a.tenant_id = $1
and a.id = $2
`,
[candidate.tenantId, candidate.id],
);
const adoption = adoptionResult.rows[0] || null;
await client.query(
`
insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
values ($1, null, 'content.public_question_bank.sync_worker_failed', 'tenant_question_bank_adoption', $2, $3::jsonb)
`,
[candidate.tenantId, candidate.id, JSON.stringify(details)],
);
if (adoption) {
await client.query(
`
insert into public.tenant_content_notifications (
tenant_id, notification_type, status, severity,
adoption_id, source_question_bank_id,
title, message, action_label, action_path,
dedupe_key, metadata, created_by
)
values (
$1, 'public_question_bank_sync_failed', 'unread', 'error',
$2, $3,
$4, $5, '查看同步状态', $6,
$7, $8::jsonb, null
)
on conflict (tenant_id, notification_type, dedupe_key)
where dedupe_key is not null
do update set status = case
when public.tenant_content_notifications.status = 'dismissed' then public.tenant_content_notifications.status
else 'unread'
end,
severity = excluded.severity,
title = excluded.title,
message = excluded.message,
action_label = excluded.action_label,
action_path = excluded.action_path,
metadata = excluded.metadata,
resolved_at = null,
updated_at = now()
`,
[
candidate.tenantId,
candidate.id,
adoption.source_question_bank_id,
`${adoption.source_question_bank_name || '公共题库'} 同步失败`,
`公共题库自动同步失败，错误码：${details.code}。请稍后重试或联系平台处理。`,
`/tenant-admin/content?adoptionId=${candidate.id}&panel=public-banks`,
publicBankFailureDedupeKey(candidate, details.code),
JSON.stringify({
adoptionId: candidate.id,
sourceQuestionBankId: adoption.source_question_bank_id,
sourceQuestionBankName: adoption.source_question_bank_name,
targetQuestionBankId: adoption.target_question_bank_id,
targetQuestionBankName: adoption.target_question_bank_name,
targetEntryId: adoption.target_entry_id,
targetCollectionId: adoption.target_collection_id,
syncStatus: 'failed',
errorCode: details.code,
errorMessage: details.message,
triggeredBy: 'worker',
workerId: config.publicBankSyncWorkerId,
failedAt: details.failedAt,
}),
],
);
}
await client.query('commit');
} catch (failure) {
await client.query('rollback');
throw failure;
} finally {
client.release();
}
}
export async function processPublicBankSyncBatch(limit = config.publicBankSyncBatchSize): Promise<PublicBankSyncWorkerResult> {