import type pg from 'pg'; import { HttpError, type RequestContext } from '../../core/http.js'; import { queryOne, transaction } from '../../core/db.js'; import { readJsonBody, requiredString, stringParam } from '../../core/request.js'; import { requireTenantContentEditor, type TenantContentAuth } from './auth.js'; type ImportType = 'questions' | 'vocabulary' | 'handbook' | 'scoreline' | 'videos'; interface ImportJobForPostCheck { id: string; importType: ImportType; status: string; totalCount: number; validCount: number; errorCount: number; warningCount: number; targetCollectionId: string | null; summary: Record; } interface PostCheckIssue { severity: 'error' | 'warning'; code: string; message: string; details?: Record; } interface PostCheckSummary { status: 'passed' | 'warning' | 'failed'; checkedAt: string; jobId: string; importType: ImportType; counts: Record; issues: PostCheckIssue[]; } async function loadCompletedJob(client: pg.PoolClient, auth: TenantContentAuth, jobId: string) { const job = await client.query( ` select id, import_type as "importType", status, total_count as "totalCount", valid_count as "validCount", error_count as "errorCount", warning_count as "warningCount", target_collection_id as "targetCollectionId", summary from public.content_import_jobs where tenant_id = $1 and id = $2 limit 1 `, [auth.tenantId, jobId], ); const item = job.rows[0]; if (!item) throw new HttpError(404, 'Import job not found', 'IMPORT_JOB_NOT_FOUND'); if (!['completed', 'completed_with_errors'].includes(item.status)) { throw new HttpError(409, 'Import job is not completed yet', 'IMPORT_JOB_NOT_COMPLETED'); } return item; } async function baseCounts(client: pg.PoolClient, tenantId: string, jobId: string) { const result = await client.query<{ item_count: string; valid_item_count: string; processed_item_count: string; target_missing_count: string; target_id_count: string; inserted_count: string; updated_count: string; skipped_count: string; failed_count: string; }>( ` select count(*)::text as item_count, count(*) filter (where status in ('valid', 'inserted', 'updated', 'skipped'))::text as valid_item_count, count(*) filter (where status in ('inserted', 'updated', 'skipped'))::text as processed_item_count, count(*) filter (where status in ('inserted', 'updated', 'skipped') and target_id is null)::text as target_missing_count, count(target_id)::text as target_id_count, count(*) filter (where status = 'inserted')::text as inserted_count, count(*) filter (where status = 'updated')::text as updated_count, count(*) filter (where status = 'skipped')::text as skipped_count, count(*) filter (where status = 'failed')::text as failed_count from public.content_import_items where tenant_id = $1 and job_id = $2 `, [tenantId, jobId], ); const row = result.rows[0]; return { itemCount: Number(row?.item_count || 0), validItemCount: Number(row?.valid_item_count || 0), processedItemCount: Number(row?.processed_item_count || 0), targetMissingCount: Number(row?.target_missing_count || 0), targetIdCount: Number(row?.target_id_count || 0), insertedCount: Number(row?.inserted_count || 0), updatedCount: Number(row?.updated_count || 0), skippedCount: Number(row?.skipped_count || 0), failedCount: Number(row?.failed_count || 0), }; } function addBaseIssues(job: ImportJobForPostCheck, counts: Record, issues: PostCheckIssue[]) { if (counts.itemCount !== job.totalCount) { issues.push({ severity: 'warning', code: 'IMPORT_ITEM_COUNT_MISMATCH', message: 'Import item count does not match job totalCount', details: { itemCount: counts.itemCount, totalCount: job.totalCount }, }); } if (counts.processedItemCount !== job.validCount) { issues.push({ severity: 'error', code: 'IMPORT_PROCESSED_COUNT_MISMATCH', message: 'Processed import item count does not match validCount', details: { processedItemCount: counts.processedItemCount, validCount: job.validCount }, }); } if (counts.targetMissingCount > 0) { issues.push({ severity: 'error', code: 'IMPORT_TARGET_ID_MISSING', message: 'Some processed import items do not have a target_id', details: { targetMissingCount: counts.targetMissingCount }, }); } if (counts.failedCount > 0) { issues.push({ severity: 'error', code: 'IMPORT_ITEM_FAILED', message: 'Some import items are marked failed', details: { failedCount: counts.failedCount }, }); } if (job.errorCount > 0 && job.status === 'completed_with_errors') { issues.push({ severity: 'warning', code: 'IMPORT_COMPLETED_WITH_VALIDATION_ERRORS', message: 'Import completed with validation errors because allowPartial was used', details: { errorCount: job.errorCount }, }); } } async function countExistingTargets(client: pg.PoolClient, tenantId: string, jobId: string, tableName: string) { const result = await client.query<{ existing_count: string }>( ` select count(distinct item.target_id)::text as existing_count from public.content_import_items item join ${tableName} target on target.tenant_id = item.tenant_id and target.id = item.target_id where item.tenant_id = $1 and item.job_id = $2 and item.status in ('inserted', 'updated', 'skipped') and item.target_id is not null `, [tenantId, jobId], ); return Number(result.rows[0]?.existing_count || 0); } async function questionChecks(client: pg.PoolClient, auth: TenantContentAuth, job: ImportJobForPostCheck, counts: Record) { const issues: PostCheckIssue[] = []; const existingTargetCount = await countExistingTargets(client, auth.tenantId, job.id, 'public.questions'); counts.existingTargetCount = existingTargetCount; const versions = await client.query<{ count: string }>( ` select count(distinct q.id)::text as count from public.content_import_items item join public.questions q on q.tenant_id = item.tenant_id and q.id = item.target_id join public.question_versions v on v.tenant_id = q.tenant_id and v.id = q.current_version_id and v.question_id = q.id where item.tenant_id = $1 and item.job_id = $2 and item.status in ('inserted', 'updated', 'skipped') `, [auth.tenantId, job.id], ); counts.questionWithCurrentVersionCount = Number(versions.rows[0]?.count || 0); if (counts.existingTargetCount !== counts.processedItemCount) { issues.push({ severity: 'error', code: 'QUESTION_TARGET_MISSING', message: 'Some imported question targets are missing', details: { existingTargetCount: counts.existingTargetCount, processedItemCount: counts.processedItemCount }, }); } if (counts.questionWithCurrentVersionCount !== counts.processedItemCount) { issues.push({ severity: 'error', code: 'QUESTION_VERSION_MISSING', message: 'Some imported questions do not have a current version', details: { questionWithCurrentVersionCount: counts.questionWithCurrentVersionCount, processedItemCount: counts.processedItemCount }, }); } if (job.targetCollectionId) { const collectionItems = await client.query<{ count: string }>( ` select count(distinct item.target_id)::text as count from public.content_import_items item join public.question_collection_items qci on qci.tenant_id = item.tenant_id and qci.question_id = item.target_id and qci.collection_id = $3 where item.tenant_id = $1 and item.job_id = $2 and item.status in ('inserted', 'updated', 'skipped') `, [auth.tenantId, job.id, job.targetCollectionId], ); counts.collectionBindingCount = Number(collectionItems.rows[0]?.count || 0); if (counts.collectionBindingCount !== counts.processedItemCount) { issues.push({ severity: 'error', code: 'QUESTION_COLLECTION_BINDING_MISSING', message: 'Some imported questions are not bound to the target collection', details: { collectionBindingCount: counts.collectionBindingCount, processedItemCount: counts.processedItemCount, collectionId: job.targetCollectionId }, }); } } return issues; } async function vocabularyChecks(client: pg.PoolClient, auth: TenantContentAuth, job: ImportJobForPostCheck, counts: Record) { const issues: PostCheckIssue[] = []; counts.existingTargetCount = await countExistingTargets(client, auth.tenantId, job.id, 'public.vocabulary_units'); const words = await client.query<{ count: string }>( ` select count(w.id)::text as count from public.content_import_items item join public.vocabulary_units unit on unit.tenant_id = item.tenant_id and unit.id = item.target_id left join public.vocabulary_words w on w.tenant_id = unit.tenant_id and w.unit_id = unit.id where item.tenant_id = $1 and item.job_id = $2 and item.status in ('inserted', 'updated', 'skipped') `, [auth.tenantId, job.id], ); counts.vocabularyWordCount = Number(words.rows[0]?.count || 0); if (counts.existingTargetCount !== counts.processedItemCount) { issues.push({ severity: 'error', code: 'VOCABULARY_UNIT_TARGET_MISSING', message: 'Some imported vocabulary units are missing', details: { existingTargetCount: counts.existingTargetCount, processedItemCount: counts.processedItemCount }, }); } if (counts.processedItemCount > 0 && counts.vocabularyWordCount <= 0) { issues.push({ severity: 'error', code: 'VOCABULARY_WORDS_MISSING', message: 'Imported vocabulary units have no words' }); } return issues; } async function handbookChecks(client: pg.PoolClient, auth: TenantContentAuth, job: ImportJobForPostCheck, counts: Record) { const issues: PostCheckIssue[] = []; counts.existingTargetCount = await countExistingTargets(client, auth.tenantId, job.id, 'public.handbook_subjects'); const children = await client.query<{ chapter_count: string; entry_count: string }>( ` select count(distinct c.id)::text as chapter_count, count(distinct e.id)::text as entry_count from public.content_import_items item join public.handbook_subjects subject on subject.tenant_id = item.tenant_id and subject.id = item.target_id left join public.handbook_chapters c on c.tenant_id = subject.tenant_id and c.subject_id = subject.id left join public.handbook_entries e on e.tenant_id = c.tenant_id and e.chapter_id = c.id where item.tenant_id = $1 and item.job_id = $2 and item.status in ('inserted', 'updated', 'skipped') `, [auth.tenantId, job.id], ); counts.handbookChapterCount = Number(children.rows[0]?.chapter_count || 0); counts.handbookEntryCount = Number(children.rows[0]?.entry_count || 0); if (counts.existingTargetCount !== counts.processedItemCount) { issues.push({ severity: 'error', code: 'HANDBOOK_SUBJECT_TARGET_MISSING', message: 'Some imported handbook subjects are missing', details: { existingTargetCount: counts.existingTargetCount, processedItemCount: counts.processedItemCount }, }); } if (counts.processedItemCount > 0 && counts.handbookEntryCount <= 0) { issues.push({ severity: 'warning', code: 'HANDBOOK_ENTRIES_EMPTY', message: 'Imported handbook subjects do not contain entries' }); } return issues; } async function scorelineChecks(client: pg.PoolClient, auth: TenantContentAuth, job: ImportJobForPostCheck, counts: Record) { const kinds = await client.query<{ kind: string; processed_count: string; existing_count: string }>( ` with processed as ( select target_id, normalized_payload ->> 'kind' as kind from public.content_import_items where tenant_id = $1 and job_id = $2 and status in ('inserted', 'updated', 'skipped') ) select 'field' as kind, count(*) filter (where p.kind = 'field')::text as processed_count, count(f.id)::text as existing_count from processed p left join public.scoreline_fields f on p.kind = 'field' and f.tenant_id = $1 and f.id = p.target_id union all select 'school', count(*) filter (where p.kind = 'school')::text, count(s.id)::text from processed p left join public.scoreline_schools s on p.kind = 'school' and s.tenant_id = $1 and s.id = p.target_id union all select 'major', count(*) filter (where p.kind = 'major')::text, count(m.id)::text from processed p left join public.scoreline_majors m on p.kind = 'major' and m.tenant_id = $1 and m.id = p.target_id union all select 'record', count(*) filter (where p.kind = 'record')::text, count(r.id)::text from processed p left join public.scoreline_records r on p.kind = 'record' and r.tenant_id = $1 and r.id = p.target_id `, [auth.tenantId, job.id], ); const issues: PostCheckIssue[] = []; for (const row of kinds.rows) { const processed = Number(row.processed_count || 0); const existing = Number(row.existing_count || 0); counts[`scoreline${row.kind[0].toUpperCase()}${row.kind.slice(1)}ProcessedCount`] = processed; counts[`scoreline${row.kind[0].toUpperCase()}${row.kind.slice(1)}ExistingCount`] = existing; if (processed !== existing) { issues.push({ severity: 'error', code: 'SCORELINE_TARGET_MISSING', message: `Some imported scoreline ${row.kind} targets are missing`, details: { kind: row.kind, processed, existing }, }); } } counts.existingTargetCount = kinds.rows.reduce((sum, row) => sum + Number(row.existing_count || 0), 0); return issues; } async function videoChecks(client: pg.PoolClient, auth: TenantContentAuth, job: ImportJobForPostCheck, counts: Record) { const issues: PostCheckIssue[] = []; counts.existingTargetCount = await countExistingTargets(client, auth.tenantId, job.id, 'public.video_explanations'); const bindings = await client.query<{ count: string }>( ` select count(qv.id)::text as count from public.content_import_items item join public.video_explanations video on video.tenant_id = item.tenant_id and video.id = item.target_id left join public.question_videos qv on qv.tenant_id = video.tenant_id and qv.video_id = video.id where item.tenant_id = $1 and item.job_id = $2 and item.status in ('inserted', 'updated', 'skipped') `, [auth.tenantId, job.id], ); counts.videoBindingCount = Number(bindings.rows[0]?.count || 0); if (counts.existingTargetCount !== counts.processedItemCount) { issues.push({ severity: 'error', code: 'VIDEO_TARGET_MISSING', message: 'Some imported videos are missing', details: { existingTargetCount: counts.existingTargetCount, processedItemCount: counts.processedItemCount }, }); } if (counts.processedItemCount > 0 && counts.videoBindingCount <= 0) { issues.push({ severity: 'warning', code: 'VIDEO_BINDINGS_EMPTY', message: 'Imported videos do not have question bindings' }); } return issues; } async function domainChecks(client: pg.PoolClient, auth: TenantContentAuth, job: ImportJobForPostCheck, counts: Record) { switch (job.importType) { case 'questions': return questionChecks(client, auth, job, counts); case 'vocabulary': return vocabularyChecks(client, auth, job, counts); case 'handbook': return handbookChecks(client, auth, job, counts); case 'scoreline': return scorelineChecks(client, auth, job, counts); case 'videos': return videoChecks(client, auth, job, counts); default: return []; } } function finalStatus(issues: PostCheckIssue[]): PostCheckSummary['status'] { if (issues.some(issue => issue.severity === 'error')) return 'failed'; if (issues.some(issue => issue.severity === 'warning')) return 'warning'; return 'passed'; } async function runPostCheck(auth: TenantContentAuth, jobId: string) { return transaction(async client => { const job = await loadCompletedJob(client, auth, jobId); const counts = await baseCounts(client, auth.tenantId, job.id); const issues: PostCheckIssue[] = []; addBaseIssues(job, counts, issues); issues.push(...await domainChecks(client, auth, job, counts)); const summary: PostCheckSummary = { status: finalStatus(issues), checkedAt: new Date().toISOString(), jobId: job.id, importType: job.importType, counts, issues, }; await client.query( ` update public.content_import_jobs set summary = coalesce(summary, '{}'::jsonb) || $3::jsonb, updated_at = now() where tenant_id = $1 and id = $2 `, [auth.tenantId, job.id, JSON.stringify({ importPostCheck: summary })], ); await client.query( ` insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details) values ($1, $2, $3, 'content_import_job', $4, $5::jsonb) `, [ auth.tenantId, auth.userId, `content.import.${job.importType}.post_checked`, job.id, JSON.stringify({ status: summary.status, issueCount: summary.issues.length, counts: summary.counts }), ], ); return summary; }); } export async function importPostCheckRoute(ctx: RequestContext) { const auth = await requireTenantContentEditor(ctx); const body = await readJsonBody(ctx); const jobId = requiredString(body, 'jobId'); const item = await runPostCheck(auth, jobId); return { item }; } export async function importPostCheckStatusRoute(ctx: RequestContext) { const auth = await requireTenantContentEditor(ctx); const jobId = stringParam(ctx, 'jobId'); if (!jobId) throw new HttpError(400, 'jobId is required', 'REQUIRED_FIELD'); const job = await queryOne<{ id: string; summary: Record }>( 'select id, summary from public.content_import_jobs where tenant_id = $1 and id = $2 limit 1', [auth.tenantId, jobId], ); if (!job) throw new HttpError(404, 'Import job not found', 'IMPORT_JOB_NOT_FOUND'); return { item: { jobId: job.id, importPostCheck: job.summary?.importPostCheck || null, }, }; }