feat: notify tenants about public bank syncs

apps/api/src/features/tenant-content/index.ts

@@ -49,9 +49,11 @@ import {
   adoptPublicQuestionBankRoute,
   publicQuestionBankConflictsRoute,
   publicQuestionBanksRoute,
+  tenantContentNotificationsRoute,
   resolvePublicQuestionBankConflictsRoute,
   resolvePublicQuestionBankConflictRoute,
   syncPublicQuestionBankRoute,
+  updateTenantContentNotificationStatusRoute,
 } from './public-banks.js';
 import {
   bindQuestionVideoRoute,
@@ -82,6 +84,8 @@ import {
 export const tenantContentRoutes: RouteDefinition[] = [
   ['GET', '/api/tenant-content/content-entries', contentEntriesAdminRoute],
   ['GET', '/api/tenant-content/public-question-banks', publicQuestionBanksRoute],
+  ['GET', '/api/tenant-content/notifications', tenantContentNotificationsRoute],
+  ['POST', '/api/tenant-content/notifications/status', updateTenantContentNotificationStatusRoute],
   ['POST', '/api/tenant-content/public-question-banks/adopt', adoptPublicQuestionBankRoute],
   ['POST', '/api/tenant-content/public-question-banks/sync', syncPublicQuestionBankRoute],
   ['GET', '/api/tenant-content/public-question-banks/conflicts', publicQuestionBankConflictsRoute],

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

@@ -1,3 +1,4 @@
+import crypto from 'node:crypto';
 import type pg from 'pg';
 import { HttpError, type RequestContext } from '../../core/http.js';
 import { intParam, optionalString, readJsonBody, requiredString, stringParam } from '../../core/request.js';
@@ -43,6 +44,24 @@ interface AdoptionRow {
   updatedAt: string;
 }
 
+interface ContentNotificationRow {
+  id: string;
+  notificationType: string;
+  status: string;
+  severity: string;
+  adoptionId: string | null;
+  sourceQuestionBankId: string | null;
+  title: string;
+  message: string;
+  actionLabel: string | null;
+  actionPath: string | null;
+  metadata: Record<string, unknown>;
+  createdAt: string;
+  updatedAt: string;
+  readAt: string | null;
+  resolvedAt: string | null;
+}
+
 interface SourceQuestionSnapshot {
   id: string;
   type: string;
@@ -569,6 +588,131 @@ function syncCounts(results: SyncQuestionResult[]) {
   };
 }
 
+function publicBankNotificationHash(parts: unknown[]) {
+  return crypto.createHash('sha256').update(JSON.stringify(parts)).digest('hex').slice(0, 40);
+}
+
+function publicBankChangedResults(results: SyncQuestionResult[]) {
+  return results
+    .filter(item => item.action === 'inserted' || item.action === 'updated' || item.action === 'conflict')
+    .map(item => ({
+      sourceQuestionId: item.sourceQuestionId,
+      action: item.action,
+      sourceHash: item.sourceHash,
+      previousSourceHash: item.previousSourceHash,
+      targetHash: item.targetHash,
+    }))
+    .sort((left, right) => `${left.sourceQuestionId}:${left.action}`.localeCompare(`${right.sourceQuestionId}:${right.action}`));
+}
+
+async function upsertPublicBankSyncNotification(client: pg.PoolClient, input: {
+  auth: PublicQuestionBankSyncAuth;
+  adoption: AdoptionRow;
+  grant: EligibleBankRow;
+  results: SyncQuestionResult[];
+  counts: ReturnType<typeof syncCounts>;
+  conflicts: SyncQuestionResult[];
+  triggeredBy: 'manual' | 'worker';
+  workerId?: string | null;
+}) {
+  const changedResults = publicBankChangedResults(input.results);
+  if (!changedResults.length) return;
+
+  const hasConflicts = input.conflicts.length > 0;
+  const notificationType = hasConflicts ? 'public_question_bank_conflict' : 'public_question_bank_synced';
+  const severity = hasConflicts ? 'warning' : 'success';
+  const dedupeKey = [
+    'public-bank',
+    input.adoption.id,
+    notificationType,
+    publicBankNotificationHash(changedResults),
+  ].join(':');
+  const title = hasConflicts
+    ? `${input.grant.sourceQuestionBankName} 有同步冲突`
+    : `${input.grant.sourceQuestionBankName} 已同步`;
+  const message = hasConflicts
+    ? `平台公共题库有 ${input.conflicts.length} 道题与租户本地改写冲突，请处理后再发布给学生。`
+    : `平台公共题库已同步：新增 ${input.counts.inserted} 题，更新 ${input.counts.updated} 题。`;
+  const actionPath = hasConflicts
+    ? `/tenant-admin/content?adoptionId=${input.adoption.id}&panel=public-bank-conflicts`
+    : `/tenant-admin/content?adoptionId=${input.adoption.id}&panel=public-banks`;
+  const actionLabel = hasConflicts ? '处理冲突' : '查看同步结果';
+
+  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, $2, 'unread', $3,
+      $4, $5,
+      $6, $7, $8, $9,
+      $10, $11::jsonb, $12
+    )
+    on conflict (tenant_id, notification_type, dedupe_key)
+    where dedupe_key is not null
+    do update set severity = excluded.severity,
+                  title = excluded.title,
+                  message = excluded.message,
+                  action_label = excluded.action_label,
+                  action_path = excluded.action_path,
+                  metadata = excluded.metadata,
+                  updated_at = now()
+    `,
+    [
+      input.auth.tenantId,
+      notificationType,
+      severity,
+      input.adoption.id,
+      input.grant.sourceQuestionBankId,
+      title,
+      message,
+      actionLabel,
+      actionPath,
+      dedupeKey,
+      JSON.stringify({
+        adoptionId: input.adoption.id,
+        sourceQuestionBankId: input.grant.sourceQuestionBankId,
+        sourceQuestionBankName: input.grant.sourceQuestionBankName,
+        targetQuestionBankId: input.adoption.targetQuestionBankId,
+        targetEntryId: input.adoption.targetEntryId,
+        targetCollectionId: input.adoption.targetCollectionId,
+        syncStatus: hasConflicts ? 'conflict' : 'synced',
+        counts: input.counts,
+        changedResults: changedResults.slice(0, 50),
+        triggeredBy: input.triggeredBy,
+        workerId: input.workerId || null,
+      }),
+      input.auth.userId,
+    ],
+  );
+}
+
+async function markPublicBankConflictNotificationsResolved(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_conflict'
+      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> : {};
 }
@@ -1072,6 +1216,14 @@ export async function executePublicQuestionBankSync(input: PublicQuestionBankSyn
       ],
     );
 
+    if (!conflicts.length) {
+      await markPublicBankConflictNotificationsResolved(client, {
+        tenantId: auth.tenantId,
+        adoptionId: adoption.id,
+        actorUserId: auth.userId,
+      });
+    }
+
     await client.query(
       `
       insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
@@ -1096,6 +1248,17 @@ export async function executePublicQuestionBankSync(input: PublicQuestionBankSyn
       ],
     );
 
+    await upsertPublicBankSyncNotification(client, {
+      auth,
+      adoption: updated.rows[0],
+      grant,
+      results: syncResult.results,
+      counts: syncResult.counts,
+      conflicts,
+      triggeredBy: input.triggeredBy || 'manual',
+      workerId: input.workerId || null,
+    });
+
     return {
       item: updated.rows[0],
       sync: {
@@ -1391,6 +1554,14 @@ export async function resolvePublicQuestionBankConflictRoute(ctx: RequestContext
       ],
     );
 
+    if (!remainingConflicts.length) {
+      await markPublicBankConflictNotificationsResolved(client, {
+        tenantId: auth.tenantId,
+        adoptionId: adoption.id,
+        actorUserId: auth.userId,
+      });
+    }
+
     await client.query(
       `
       insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
@@ -1432,6 +1603,25 @@ function parsePublicQuestionBankConflictResolution(value: string): PublicQuestio
   throw new HttpError(400, 'resolution must be accept_platform or keep_local', 'PUBLIC_BANK_CONFLICT_RESOLUTION_INVALID');
 }
 
+function parseNotificationStatus(value: string) {
+  if (['unread', 'read', 'dismissed', 'resolved'].includes(value)) return value;
+  throw new HttpError(400, 'status must be unread, read, dismissed, or resolved', 'CONTENT_NOTIFICATION_STATUS_INVALID');
+}
+
+function parseNotificationUpdateStatus(value: string) {
+  if (['read', 'dismissed', 'resolved'].includes(value)) return value;
+  throw new HttpError(400, 'status must be read, dismissed, or resolved', 'CONTENT_NOTIFICATION_STATUS_INVALID');
+}
+
+function uuidString(value: unknown) {
+  const candidate = typeof value === 'string' ? value.trim() : '';
+  if (!candidate) return '';
+  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(candidate)) {
+    throw new HttpError(400, 'notificationIds must contain UUID values', 'CONTENT_NOTIFICATION_ID_INVALID');
+  }
+  return candidate;
+}
+
 function sourceQuestionIdsFromBody(value: unknown) {
   if (!Array.isArray(value)) return [];
   const seen = new Set<string>();
@@ -1747,6 +1937,14 @@ async function resolvePublicQuestionBankConflictsBatch(auth: TenantContentAuth,
       ],
     );
 
+    if (!remainingConflicts.length) {
+      await markPublicBankConflictNotificationsResolved(client, {
+        tenantId: auth.tenantId,
+        adoptionId: adoption.id,
+        actorUserId: auth.userId,
+      });
+    }
+
     for (const result of results) {
       await client.query(
         `
@@ -1823,3 +2021,107 @@ export async function resolvePublicQuestionBankConflictsRoute(ctx: RequestContex
 
   return { item };
 }
+
+export async function tenantContentNotificationsRoute(ctx: RequestContext) {
+  const auth = await requireTenantContentEditor(ctx);
+  const rawStatus = stringParam(ctx, 'status');
+  const status = rawStatus ? parseNotificationStatus(rawStatus) : '';
+  const rawType = stringParam(ctx, 'notificationType');
+  const adoptionId = stringParam(ctx, 'adoptionId');
+  const limit = intParam(ctx, 'limit', 50, 100);
+
+  const items = await query<ContentNotificationRow>(
+    `
+    select id,
+           notification_type as "notificationType",
+           status,
+           severity,
+           adoption_id as "adoptionId",
+           source_question_bank_id as "sourceQuestionBankId",
+           title,
+           message,
+           action_label as "actionLabel",
+           action_path as "actionPath",
+           metadata,
+           created_at as "createdAt",
+           updated_at as "updatedAt",
+           read_at as "readAt",
+           resolved_at as "resolvedAt"
+    from public.tenant_content_notifications
+    where tenant_id = $1
+      and ($2::text = '' or status = $2)
+      and ($3::text = '' or notification_type = $3)
+      and ($4::uuid is null or adoption_id = $4::uuid)
+    order by created_at desc
+    limit $5
+    `,
+    [auth.tenantId, status, rawType, adoptionId || null, limit],
+  );
+
+  const summaryRows = await query<{ status: string; count: number }>(
+    `
+    select status, count(*)::integer as count
+    from public.tenant_content_notifications
+    where tenant_id = $1
+      and ($2::text = '' or notification_type = $2)
+    group by status
+    `,
+    [auth.tenantId, rawType],
+  );
+  const summary = Object.fromEntries(summaryRows.map(row => [row.status, Number(row.count || 0)]));
+
+  return { items, summary };
+}
+
+export async function updateTenantContentNotificationStatusRoute(ctx: RequestContext) {
+  const auth = await requireTenantContentEditor(ctx);
+  const body = await readJsonBody(ctx);
+  const status = parseNotificationUpdateStatus(requiredString(body, 'status'));
+  const ids = Array.isArray(body.notificationIds)
+    ? body.notificationIds
+        .map(item => uuidString(item))
+        .filter(Boolean)
+        .slice(0, 100)
+    : [];
+  if (!ids.length) {
+    throw new HttpError(400, 'notificationIds is required', 'CONTENT_NOTIFICATION_IDS_REQUIRED');
+  }
+
+  const updated = await query<ContentNotificationRow>(
+    `
+    update public.tenant_content_notifications
+    set status = $3,
+        read_by = case when $3 in ('read', 'dismissed', 'resolved') then coalesce(read_by, $4) else read_by end,
+        read_at = case when $3 in ('read', 'dismissed', 'resolved') then coalesce(read_at, now()) else read_at end,
+        resolved_at = case when $3 = 'resolved' then coalesce(resolved_at, now()) else resolved_at end,
+        updated_at = now()
+    where tenant_id = $1
+      and id = any($2::uuid[])
+    returning id,
+              notification_type as "notificationType",
+              status,
+              severity,
+              adoption_id as "adoptionId",
+              source_question_bank_id as "sourceQuestionBankId",
+              title,
+              message,
+              action_label as "actionLabel",
+              action_path as "actionPath",
+              metadata,
+              created_at as "createdAt",
+              updated_at as "updatedAt",
+              read_at as "readAt",
+              resolved_at as "resolvedAt"
+    `,
+    [auth.tenantId, ids, status, auth.userId],
+  );
+
+  return {
+    item: {
+      status,
+      requestedCount: ids.length,
+      updatedCount: updated.length,
+      notifications: updated,
+    },
+  };
+}