feat: schedule student supervision rules

2026-06-30 17:40:03 +08:00
parent 0204c934d8
commit c7ad3450fd
21 changed files with 1679 additions and 482 deletions
@@ -56,6 +56,9 @@ export interface WorkerConfig {
  exportBatchSize: number;
  exportWorkerId: string;
  exportBackoffSeconds: number[];
+  studentSupervisionBatchSize: number;
+  studentSupervisionWorkerId: string;
+  studentSupervisionClaimStaleSeconds: number;
  exportLocalStorageRoot: string;
  exportPdfFontPath: string;
  storageDefaultProvider: string;
@@ -237,6 +240,9 @@ const loadedConfig: WorkerConfig = {
  exportBackoffSeconds: envList('WORKER_EXPORT_BACKOFF_SECONDS', '30,120,600,1800')
    .map((value: string) => Number(value))
    .filter((value: number) => Number.isFinite(value) && value > 0),
+  studentSupervisionBatchSize: envNumber('WORKER_STUDENT_SUPERVISION_BATCH_SIZE', 20),
+  studentSupervisionWorkerId: envString('WORKER_STUDENT_SUPERVISION_ID', `student-supervision-${process.pid}`),
+  studentSupervisionClaimStaleSeconds: envNumber('WORKER_STUDENT_SUPERVISION_CLAIM_STALE_SECONDS', 15 * 60),
  exportLocalStorageRoot: envString('EXPORT_LOCAL_STORAGE_ROOT', '.local-storage'),
  exportPdfFontPath: envString('EXPORT_PDF_FONT_PATH', ''),
  storageDefaultProvider: envString('STORAGE_DEFAULT_PROVIDER', 'local_dev'),
@@ -131,6 +131,18 @@ async function runOnce() {
    );
    return;
  }
+  if (job === 'student-supervision') {
+    const { closePool: closeApiPool } = await import('../../api/src/core/db.js');
+    const { processStudentSupervisionBatch } = await import('./jobs/student-supervision.js');
+    extraClosers.add(closeApiPool);
+    const result = await processStudentSupervisionBatch();
+    console.log(
+      `[worker] student-supervision batch processed=${result.processed}`
+        + ` generated=${result.generated} followups=${result.followups}`
+        + ` failed=${result.failed} skipped=${result.skipped}`,
+    );
+    return;
+  }
  throw new Error(`Unsupported worker job: ${job}`);
 }

@@ -0,0 +1,269 @@
+import crypto from 'node:crypto';
+import { pool } from '../db.js';
+import { config } from '../config.js';
+import {
+  defaultSupervisionDueAt,
+  generateStudentSupervisionFollowups,
+  nextSupervisionRunAt,
+  parseSupervisionRules,
+  parseSupervisionSchedule,
+  shanghaiTodayKey,
+  supervisionSystemAuth,
+} from '../../../api/src/features/tenant-admin/supervision.js';
+
+interface SupervisionRuleCandidate {
+  id: string;
+  tenantId: string;
+  name: string;
+  rules: Record<string, unknown>;
+  schedule: Record<string, unknown>;
+  classId: string | null;
+  assignedToUserId: string | null;
+  limit: number;
+  metadata: Record<string, unknown>;
+}
+
+export interface StudentSupervisionWorkerResult {
+  processed: number;
+  generated: number;
+  failed: number;
+  skipped: number;
+  followups: number;
+}
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function errorMessage(error: unknown) {
+  return error instanceof Error ? error.message : String(error);
+}
+
+function errorCode(error: unknown) {
+  return typeof error === 'object' && error !== null && 'code' in error
+    ? String((error as { code?: unknown }).code || 'STUDENT_SUPERVISION_WORKER_ERROR')
+    : 'STUDENT_SUPERVISION_WORKER_ERROR';
+}
+
+function truncate(value: unknown, max = 1900) {
+  return String(value ?? '').slice(0, max);
+}
+
+function stableBatchKey(rule: SupervisionRuleCandidate) {
+  const raw = [
+    'worker',
+    rule.id,
+    shanghaiTodayKey(),
+  ].join(':');
+  return raw.length <= 120 ? raw : crypto.createHash('sha256').update(raw).digest('hex');
+}
+
+async function claimStudentSupervisionRules(limit: number, claimId: string) {
+  const client = await pool.connect();
+  try {
+    await client.query('begin');
+    const result = await client.query<SupervisionRuleCandidate>(
+      `
+      with candidates as (
+        select sr.id
+        from public.tenant_student_supervision_rules sr
+        join public.tenants t on t.id = sr.tenant_id
+        where t.status = 'active'
+          and sr.status = 'active'
+          and coalesce(sr.schedule->>'enabled', 'false') = 'true'
+          and coalesce(sr.schedule->>'frequency', 'manual') <> 'manual'
+          and coalesce(sr.next_run_at, sr.created_at) <= now()
+          and coalesce(nullif(sr.metadata #>> '{studentSupervisionWorker,claimedAt}', '')::timestamptz, '1970-01-01'::timestamptz)
+            <= now() - ($2::integer * interval '1 second')
+        order by coalesce(sr.next_run_at, sr.created_at) asc, sr.created_at asc
+        limit $1
+        for update of sr skip locked
+      )
+      update public.tenant_student_supervision_rules sr
+      set metadata = jsonb_set(
+            coalesce(sr.metadata, '{}'::jsonb),
+            '{studentSupervisionWorker}',
+            coalesce(sr.metadata->'studentSupervisionWorker', '{}'::jsonb) || $3::jsonb,
+            true
+          ),
+          updated_at = now()
+      from candidates
+      where sr.id = candidates.id
+      returning sr.id,
+                sr.tenant_id as "tenantId",
+                sr.name,
+                sr.rules,
+                sr.schedule,
+                sr.class_id as "classId",
+                sr.assigned_to_user_id as "assignedToUserId",
+                sr.limit_count as "limit",
+                sr.metadata
+      `,
+      [
+        Math.max(1, Math.min(100, limit)),
+        Math.max(60, config.studentSupervisionClaimStaleSeconds),
+        JSON.stringify({
+          claimId,
+          workerId: config.studentSupervisionWorkerId,
+          claimedAt: nowIso(),
+        }),
+      ],
+    );
+    await client.query('commit');
+    return result.rows;
+  } catch (error) {
+    await client.query('rollback');
+    throw error;
+  } finally {
+    client.release();
+  }
+}
+
+async function markRuleFailed(rule: SupervisionRuleCandidate, error: unknown) {
+  const schedule = parseSupervisionSchedule(rule.schedule);
+  const nextRunAt = nextSupervisionRunAt(schedule);
+  const details = {
+    code: errorCode(error),
+    message: truncate(errorMessage(error)),
+    failedAt: nowIso(),
+    workerId: config.studentSupervisionWorkerId,
+  };
+  await pool.query(
+    `
+    update public.tenant_student_supervision_rules
+    set last_run_at = now(),
+        next_run_at = $3::timestamptz,
+        last_result = $4::jsonb,
+        metadata = jsonb_set(
+          coalesce(metadata, '{}'::jsonb),
+          '{studentSupervisionWorker}',
+          coalesce(metadata->'studentSupervisionWorker', '{}'::jsonb) || $5::jsonb,
+          true
+        ),
+        updated_at = now()
+    where tenant_id = $1 and id = $2
+    `,
+    [
+      rule.tenantId,
+      rule.id,
+      nextRunAt,
+      JSON.stringify({ status: 'failed', ...details }),
+      JSON.stringify({
+        lastStatus: 'failed',
+        lastError: details,
+        lastWorkerId: config.studentSupervisionWorkerId,
+        lastFinishedAt: nowIso(),
+      }),
+    ],
+  );
+  await pool.query(
+    `
+    insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+    values ($1, null, 'tenant.students.supervision_rule_worker_failed', 'tenant_student_supervision_rules', $2, $3::jsonb)
+    `,
+    [rule.tenantId, rule.id, JSON.stringify(details)],
+  ).catch(() => {});
+}
+
+async function processRule(rule: SupervisionRuleCandidate) {
+  const rules = parseSupervisionRules(rule.rules);
+  const schedule = parseSupervisionSchedule(rule.schedule);
+  const result = await generateStudentSupervisionFollowups(
+    supervisionSystemAuth(rule.tenantId),
+    {
+      rules,
+      classId: rule.classId,
+      assignedToUserId: rule.assignedToUserId,
+      dueAt: defaultSupervisionDueAt(),
+      batchKey: stableBatchKey(rule),
+      limit: Math.max(1, Math.min(100, Number(rule.limit || 20))),
+      metadata: {
+        ...rule.metadata,
+        source: 'student_supervision_worker',
+        ruleId: rule.id,
+        ruleName: rule.name,
+        workerId: config.studentSupervisionWorkerId,
+      },
+    },
+  );
+  const nextRunAt = nextSupervisionRunAt(schedule);
+  await pool.query(
+    `
+    update public.tenant_student_supervision_rules
+    set last_run_at = now(),
+        next_run_at = $3::timestamptz,
+        last_result = $4::jsonb,
+        metadata = jsonb_set(
+          coalesce(metadata, '{}'::jsonb),
+          '{studentSupervisionWorker}',
+          coalesce(metadata->'studentSupervisionWorker', '{}'::jsonb) || $5::jsonb,
+          true
+        ),
+        updated_at = now()
+    where tenant_id = $1 and id = $2
+    `,
+    [
+      rule.tenantId,
+      rule.id,
+      nextRunAt,
+      JSON.stringify({
+        status: result.errorCount ? 'completed_with_errors' : 'completed',
+        ranAt: nowIso(),
+        nextRunAt,
+        batchKey: result.batchKey,
+        total: result.total,
+        successCount: result.successCount,
+        errorCount: result.errorCount,
+      }),
+      JSON.stringify({
+        lastStatus: result.errorCount ? 'completed_with_errors' : 'completed',
+        lastWorkerId: config.studentSupervisionWorkerId,
+        lastFinishedAt: nowIso(),
+      }),
+    ],
+  );
+  await pool.query(
+    `
+    insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+    values ($1, null, 'tenant.students.supervision_rule_worker_completed', 'tenant_student_supervision_rules', $2, $3::jsonb)
+    `,
+    [
+      rule.tenantId,
+      rule.id,
+      JSON.stringify({
+        batchKey: result.batchKey,
+        total: result.total,
+        successCount: result.successCount,
+        errorCount: result.errorCount,
+        workerId: config.studentSupervisionWorkerId,
+      }),
+    ],
+  );
+  return result;
+}
+
+export async function processStudentSupervisionBatch(): Promise<StudentSupervisionWorkerResult> {
+  const claimId = crypto.randomUUID();
+  const rules = await claimStudentSupervisionRules(config.studentSupervisionBatchSize, claimId);
+  const result: StudentSupervisionWorkerResult = {
+    processed: 0,
+    generated: 0,
+    failed: 0,
+    skipped: 0,
+    followups: 0,
+  };
+
+  for (const rule of rules) {
+    result.processed += 1;
+    try {
+      const generated = await processRule(rule);
+      result.generated += 1;
+      result.followups += generated.successCount || 0;
+    } catch (error) {
+      result.failed += 1;
+      await markRuleFailed(rule, error);
+    }
+  }
+
+  return result;
+}