forked from wangziqi/gongxue-base
perf: stabilize real pocketbase import
This commit is contained in:
Codex
@@ -61,6 +61,14 @@ const nonCollectionJsonFiles = new Set([
|
||||
'storage-manifest.json',
|
||||
]);
|
||||
|
||||
const importBatchSize = Math.max(100, intFromEnv(process.env.PB_IMPORT_BATCH_SIZE, 1000));
|
||||
|
||||
function intFromEnv(value: unknown, fallback: number): number {
|
||||
if (value === null || value === undefined || value === '') return fallback;
|
||||
const parsed = Number(value);
|
||||
return Number.isFinite(parsed) && parsed > 0 ? Math.trunc(parsed) : fallback;
|
||||
}
|
||||
|
||||
function asArray(input: unknown): JsonRecord[] {
|
||||
if (Array.isArray(input)) return input as JsonRecord[];
|
||||
if (input && typeof input === 'object' && Array.isArray((input as { items?: unknown[] }).items)) {
|
||||
@@ -358,47 +366,84 @@ async function createRun() {
|
||||
|
||||
async function importRaw(runId: string, collection: string, records: JsonRecord[]) {
|
||||
let count = 0;
|
||||
for (const record of records) {
|
||||
const legacyId = text(record.id);
|
||||
if (!legacyId) continue;
|
||||
const sanitized = sanitizeRecord(record);
|
||||
const issueRows: Array<{
|
||||
legacyId: string;
|
||||
issue: Issue;
|
||||
}> = [];
|
||||
|
||||
for (let offset = 0; offset < records.length; offset += importBatchSize) {
|
||||
const batch = records.slice(offset, offset + importBatchSize);
|
||||
const legacyIds: string[] = [];
|
||||
const recordPayloads: string[] = [];
|
||||
|
||||
for (const record of batch) {
|
||||
const legacyId = text(record.id);
|
||||
if (!legacyId) continue;
|
||||
legacyIds.push(legacyId);
|
||||
recordPayloads.push(JSON.stringify(sanitizeRecord(record)));
|
||||
for (const detectedIssue of detectIssues(collection, record)) {
|
||||
issueRows.push({ legacyId, issue: detectedIssue });
|
||||
}
|
||||
}
|
||||
|
||||
if (!legacyIds.length) continue;
|
||||
await pool.query(
|
||||
`
|
||||
insert into public.pb_raw_records (run_id, tenant_id, collection_name, legacy_id, record)
|
||||
values ($1, $2, $3, $4, $5)
|
||||
select $1::uuid, $2::uuid, $3::text, raw.legacy_id, raw.record_payload::jsonb
|
||||
from unnest($4::text[], $5::text[]) as raw(legacy_id, record_payload)
|
||||
on conflict (run_id, collection_name, legacy_id)
|
||||
do update set record = excluded.record, imported_at = now()
|
||||
`,
|
||||
[runId, tenantId, collection, legacyId, JSON.stringify(sanitized)],
|
||||
[runId, tenantId, collection, legacyIds, recordPayloads],
|
||||
);
|
||||
|
||||
for (const issue of detectIssues(collection, record)) {
|
||||
await pool.query(
|
||||
`
|
||||
insert into public.pb_import_issues (
|
||||
run_id, tenant_id, collection_name, legacy_id, severity,
|
||||
issue_code, message, field_path, raw_value_sample
|
||||
)
|
||||
values ($1,$2,$3,$4,$5,$6,$7,$8,$9)
|
||||
`,
|
||||
[
|
||||
runId,
|
||||
tenantId,
|
||||
collection,
|
||||
legacyId,
|
||||
issue.severity,
|
||||
issue.issueCode,
|
||||
issue.message,
|
||||
issue.fieldPath || null,
|
||||
issue.rawValueSample || null,
|
||||
],
|
||||
);
|
||||
}
|
||||
count += 1;
|
||||
count += legacyIds.length;
|
||||
}
|
||||
|
||||
await insertImportIssueRows(runId, collection, issueRows);
|
||||
return count;
|
||||
}
|
||||
|
||||
async function insertImportIssueRows(
|
||||
runId: string,
|
||||
collection: string,
|
||||
rows: Array<{ legacyId: string | null; issue: Issue }>,
|
||||
) {
|
||||
for (let offset = 0; offset < rows.length; offset += importBatchSize) {
|
||||
const batch = rows.slice(offset, offset + importBatchSize);
|
||||
if (!batch.length) continue;
|
||||
await pool.query(
|
||||
`
|
||||
insert into public.pb_import_issues (
|
||||
run_id, tenant_id, collection_name, legacy_id, severity,
|
||||
issue_code, message, field_path, raw_value_sample
|
||||
)
|
||||
select $1::uuid, $2::uuid, $3::text, issue.legacy_id, issue.severity,
|
||||
issue.issue_code, issue.message, issue.field_path, issue.raw_value_sample
|
||||
from unnest(
|
||||
$4::text[],
|
||||
$5::text[],
|
||||
$6::text[],
|
||||
$7::text[],
|
||||
$8::text[],
|
||||
$9::text[]
|
||||
) as issue(legacy_id, severity, issue_code, message, field_path, raw_value_sample)
|
||||
`,
|
||||
[
|
||||
runId,
|
||||
tenantId,
|
||||
collection,
|
||||
batch.map(row => row.legacyId),
|
||||
batch.map(row => row.issue.severity),
|
||||
batch.map(row => row.issue.issueCode),
|
||||
batch.map(row => row.issue.message),
|
||||
batch.map(row => row.issue.fieldPath || null),
|
||||
batch.map(row => row.issue.rawValueSample || null),
|
||||
],
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async function issue(runId: string, collection: string, legacyId: unknown, issueCode: string, message: string, severity: Issue['severity'] = 'warning') {
|
||||
await pool.query(
|
||||
`
|
||||
@@ -463,6 +508,117 @@ async function currentQuestionVersionId(questionId: string | null): Promise<stri
|
||||
return result;
|
||||
}
|
||||
|
||||
async function bulkUserIdsByLegacy(values: unknown[]): Promise<Map<string, string>> {
|
||||
const legacyValues = [...new Set(values.map(value => text(value)).filter((value): value is string => Boolean(value)))];
|
||||
const result = new Map<string, string>();
|
||||
const missing: string[] = [];
|
||||
|
||||
for (const legacyValue of legacyValues) {
|
||||
if (userLookupCache.has(legacyValue)) {
|
||||
const cached = userLookupCache.get(legacyValue);
|
||||
if (cached) result.set(legacyValue, cached);
|
||||
} else {
|
||||
missing.push(legacyValue);
|
||||
}
|
||||
}
|
||||
|
||||
for (let offset = 0; offset < missing.length; offset += importBatchSize) {
|
||||
const batch = missing.slice(offset, offset + importBatchSize);
|
||||
if (!batch.length) continue;
|
||||
const rows = await pool.query<{ legacy_id: string; id: string }>(
|
||||
`
|
||||
select legacy_id, id
|
||||
from public.platform_users
|
||||
where legacy_id = any($1::text[])
|
||||
`,
|
||||
[batch],
|
||||
);
|
||||
const found = new Map(rows.rows.map(row => [row.legacy_id, row.id]));
|
||||
for (const legacyValue of batch) {
|
||||
const id = found.get(legacyValue) || null;
|
||||
userLookupCache.set(legacyValue, id);
|
||||
if (id) result.set(legacyValue, id);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
async function bulkLegacyIds(tableName: string, values: unknown[]): Promise<Map<string, string>> {
|
||||
const legacyValues = [...new Set(values.map(value => text(value)).filter((value): value is string => Boolean(value)))];
|
||||
const result = new Map<string, string>();
|
||||
const missing: string[] = [];
|
||||
|
||||
if (!legacyLookupTables.has(tableName)) throw new Error(`Unsafe legacy lookup table: ${tableName}`);
|
||||
|
||||
for (const legacyValue of legacyValues) {
|
||||
const cacheKey = `${tableName}:${legacyValue}`;
|
||||
if (legacyLookupCache.has(cacheKey)) {
|
||||
const cached = legacyLookupCache.get(cacheKey);
|
||||
if (cached) result.set(legacyValue, cached);
|
||||
} else {
|
||||
missing.push(legacyValue);
|
||||
}
|
||||
}
|
||||
|
||||
for (let offset = 0; offset < missing.length; offset += importBatchSize) {
|
||||
const batch = missing.slice(offset, offset + importBatchSize);
|
||||
if (!batch.length) continue;
|
||||
const rows = await pool.query<{ legacy_id: string; id: string }>(
|
||||
`
|
||||
select legacy_id, id
|
||||
from public.${tableName}
|
||||
where tenant_id = $1 and legacy_id = any($2::text[])
|
||||
`,
|
||||
[tenantId, batch],
|
||||
);
|
||||
const found = new Map(rows.rows.map(row => [row.legacy_id, row.id]));
|
||||
for (const legacyValue of batch) {
|
||||
const id = found.get(legacyValue) || null;
|
||||
legacyLookupCache.set(`${tableName}:${legacyValue}`, id);
|
||||
if (id) result.set(legacyValue, id);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
async function bulkQuestionVersionIds(questionIds: Array<string | null>): Promise<Map<string, string>> {
|
||||
const values = [...new Set(questionIds.filter((value): value is string => Boolean(value)))];
|
||||
const result = new Map<string, string>();
|
||||
const missing: string[] = [];
|
||||
|
||||
for (const questionId of values) {
|
||||
if (questionVersionLookupCache.has(questionId)) {
|
||||
const cached = questionVersionLookupCache.get(questionId);
|
||||
if (cached) result.set(questionId, cached);
|
||||
} else {
|
||||
missing.push(questionId);
|
||||
}
|
||||
}
|
||||
|
||||
for (let offset = 0; offset < missing.length; offset += importBatchSize) {
|
||||
const batch = missing.slice(offset, offset + importBatchSize);
|
||||
if (!batch.length) continue;
|
||||
const rows = await pool.query<{ id: string; current_version_id: string | null }>(
|
||||
`
|
||||
select id, current_version_id
|
||||
from public.questions
|
||||
where tenant_id = $1 and id = any($2::uuid[])
|
||||
`,
|
||||
[tenantId, batch],
|
||||
);
|
||||
const found = new Map(rows.rows.map(row => [row.id, row.current_version_id]));
|
||||
for (const questionId of batch) {
|
||||
const versionId = found.get(questionId) || null;
|
||||
questionVersionLookupCache.set(questionId, versionId);
|
||||
if (versionId) result.set(questionId, versionId);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
async function upsertSecret(secretScope: string, secretKey: string, secretValue: unknown, provider: string | null = null) {
|
||||
if (!importSecretValues || secretValue === null || secretValue === undefined || secretValue === '') return;
|
||||
await pool.query(
|
||||
@@ -876,45 +1032,86 @@ async function normalizeQuestions(records: JsonRecord[]) {
|
||||
}
|
||||
|
||||
async function normalizeUserAnswerRecords(runId: string, records: JsonRecord[]) {
|
||||
const affectedWrongPairs = new Set<string>();
|
||||
const userIdsByLegacy = await bulkUserIdsByLegacy(records.map(record => record.userId));
|
||||
const questionIdsByLegacy = await bulkLegacyIds('questions', records.map(record => record.questionId));
|
||||
const questionVersionIdsByQuestion = await bulkQuestionVersionIds([...questionIdsByLegacy.values()]);
|
||||
const issueRows: Array<{ legacyId: string | null; issue: Issue }> = [];
|
||||
|
||||
for (const r of records) {
|
||||
const userId = await userIdByLegacy(r.userId);
|
||||
if (!userId) {
|
||||
await issue(
|
||||
runId,
|
||||
'user_answer_records',
|
||||
r.id,
|
||||
'answer_user_not_found',
|
||||
`Answer record cannot be imported because userId was not resolved: ${text(r.userId) || '(empty)'}`,
|
||||
'critical',
|
||||
);
|
||||
continue;
|
||||
for (let offset = 0; offset < records.length; offset += importBatchSize) {
|
||||
const batch = records.slice(offset, offset + importBatchSize);
|
||||
const userIds: string[] = [];
|
||||
const questionIds: Array<string | null> = [];
|
||||
const questionVersionIds: Array<string | null> = [];
|
||||
const legacyIds: string[] = [];
|
||||
const legacyQuestionIds: Array<string | null> = [];
|
||||
const legacyCategoryIds: Array<string | null> = [];
|
||||
const selectedOptionsPayloads: string[] = [];
|
||||
const answerPayloads: string[] = [];
|
||||
const isCorrectValues: Array<boolean | null> = [];
|
||||
const answeredAtValues: string[] = [];
|
||||
const createdAtValues: string[] = [];
|
||||
|
||||
for (const r of batch) {
|
||||
const legacyIdValue = text(r.id);
|
||||
if (!legacyIdValue) continue;
|
||||
|
||||
const legacyUserId = text(r.userId);
|
||||
const userId = legacyUserId ? userIdsByLegacy.get(legacyUserId) || null : null;
|
||||
if (!userId) {
|
||||
const hasLegacyUserId = Boolean(legacyUserId);
|
||||
issueRows.push({
|
||||
legacyId: legacyIdValue,
|
||||
issue: {
|
||||
severity: 'warning',
|
||||
issueCode: hasLegacyUserId ? 'answer_user_not_found' : 'answer_user_empty',
|
||||
message: hasLegacyUserId
|
||||
? `Answer record skipped because legacy userId no longer exists in exported users: ${legacyUserId}`
|
||||
: 'Answer record skipped because legacy userId is empty.',
|
||||
},
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
const legacyQuestionId = text(r.questionId);
|
||||
const questionId = legacyQuestionId ? questionIdsByLegacy.get(legacyQuestionId) || null : null;
|
||||
if (!questionId) {
|
||||
issueRows.push({
|
||||
legacyId: legacyIdValue,
|
||||
issue: {
|
||||
severity: 'warning',
|
||||
issueCode: 'answer_question_not_found',
|
||||
message: `Answer record kept with legacy_question_id only because questionId was not resolved: ${legacyQuestionId || '(empty)'}`,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const isCorrect = r.isCorrect === null || r.isCorrect === undefined || r.isCorrect === ''
|
||||
? null
|
||||
: boolValue(r.isCorrect);
|
||||
const answeredAt = nullableDateText(r.answeredAt, r.updated, r.created);
|
||||
const selectedOptions = arrayValue(r.selectedOptions);
|
||||
const answerPayload = {
|
||||
source: 'pocketbase.user_answer_records',
|
||||
selectedOptions,
|
||||
legacyCategoryId: text(r.categoryId),
|
||||
legacyQuestionId,
|
||||
importedAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
userIds.push(userId);
|
||||
questionIds.push(questionId);
|
||||
questionVersionIds.push(questionId ? questionVersionIdsByQuestion.get(questionId) || null : null);
|
||||
legacyIds.push(legacyIdValue);
|
||||
legacyQuestionIds.push(legacyQuestionId);
|
||||
legacyCategoryIds.push(text(r.categoryId));
|
||||
selectedOptionsPayloads.push(JSON.stringify(selectedOptions));
|
||||
answerPayloads.push(JSON.stringify(answerPayload));
|
||||
isCorrectValues.push(isCorrect);
|
||||
answeredAtValues.push(answeredAt);
|
||||
createdAtValues.push(nullableDateText(r.created, answeredAt));
|
||||
}
|
||||
|
||||
const questionId = await legacyId('questions', r.questionId);
|
||||
if (!questionId) {
|
||||
await issue(
|
||||
runId,
|
||||
'user_answer_records',
|
||||
r.id,
|
||||
'answer_question_not_found',
|
||||
`Answer record kept with legacy_question_id only because questionId was not resolved: ${text(r.questionId) || '(empty)'}`,
|
||||
'warning',
|
||||
);
|
||||
}
|
||||
|
||||
const isCorrect = r.isCorrect === null || r.isCorrect === undefined || r.isCorrect === ''
|
||||
? null
|
||||
: boolValue(r.isCorrect);
|
||||
const answeredAt = nullableDateText(r.answeredAt, r.updated, r.created);
|
||||
const answerPayload = {
|
||||
source: 'pocketbase.user_answer_records',
|
||||
selectedOptions: arrayValue(r.selectedOptions),
|
||||
legacyCategoryId: text(r.categoryId),
|
||||
legacyQuestionId: text(r.questionId),
|
||||
importedAt: new Date().toISOString(),
|
||||
};
|
||||
if (!legacyIds.length) continue;
|
||||
|
||||
await pool.query(
|
||||
`
|
||||
@@ -923,9 +1120,29 @@ async function normalizeUserAnswerRecords(runId: string, records: JsonRecord[])
|
||||
legacy_question_id, legacy_category_id, selected_options,
|
||||
answer_payload, is_correct, answered_at, created_at
|
||||
)
|
||||
values ($1,$2,$3,$4,$5,$6,$7,$8::jsonb,$9::jsonb,$10,
|
||||
coalesce(nullif($11::text,'')::timestamptz, now()),
|
||||
coalesce(nullif($12::text,'')::timestamptz, now())
|
||||
select $1::uuid, answer.user_id::uuid, answer.question_id::uuid,
|
||||
answer.question_version_id::uuid, answer.legacy_id,
|
||||
answer.legacy_question_id, answer.legacy_category_id,
|
||||
answer.selected_options::jsonb, answer.answer_payload::jsonb,
|
||||
answer.is_correct::boolean,
|
||||
coalesce(nullif(answer.answered_at::text,'')::timestamptz, now()),
|
||||
coalesce(nullif(answer.created_at::text,'')::timestamptz, now())
|
||||
from unnest(
|
||||
$2::text[],
|
||||
$3::text[],
|
||||
$4::text[],
|
||||
$5::text[],
|
||||
$6::text[],
|
||||
$7::text[],
|
||||
$8::text[],
|
||||
$9::text[],
|
||||
$10::boolean[],
|
||||
$11::text[],
|
||||
$12::text[]
|
||||
) as answer(
|
||||
user_id, question_id, question_version_id, legacy_id,
|
||||
legacy_question_id, legacy_category_id, selected_options,
|
||||
answer_payload, is_correct, answered_at, created_at
|
||||
)
|
||||
on conflict (tenant_id, legacy_id) do update set
|
||||
user_id = excluded.user_id,
|
||||
@@ -940,42 +1157,40 @@ async function normalizeUserAnswerRecords(runId: string, records: JsonRecord[])
|
||||
`,
|
||||
[
|
||||
tenantId,
|
||||
userId,
|
||||
questionId,
|
||||
await currentQuestionVersionId(questionId),
|
||||
r.id,
|
||||
text(r.questionId),
|
||||
text(r.categoryId),
|
||||
JSON.stringify(arrayValue(r.selectedOptions)),
|
||||
JSON.stringify(answerPayload),
|
||||
isCorrect,
|
||||
answeredAt,
|
||||
nullableDateText(r.created, answeredAt),
|
||||
userIds,
|
||||
questionIds,
|
||||
questionVersionIds,
|
||||
legacyIds,
|
||||
legacyQuestionIds,
|
||||
legacyCategoryIds,
|
||||
selectedOptionsPayloads,
|
||||
answerPayloads,
|
||||
isCorrectValues,
|
||||
answeredAtValues,
|
||||
createdAtValues,
|
||||
],
|
||||
);
|
||||
|
||||
if (isCorrect === false && questionId) affectedWrongPairs.add(`${userId}:${questionId}`);
|
||||
}
|
||||
|
||||
for (const pair of affectedWrongPairs) {
|
||||
const [userId, questionId] = pair.split(':');
|
||||
await pool.query(
|
||||
`
|
||||
insert into public.wrong_questions (tenant_id, user_id, question_id, wrong_count, last_wrong_at, resolved_at)
|
||||
select $1, $2, $3, count(*)::integer, max(answered_at), null
|
||||
from public.answer_records
|
||||
where tenant_id = $1
|
||||
and user_id = $2
|
||||
and question_id = $3
|
||||
and is_correct = false
|
||||
on conflict (tenant_id, user_id, question_id)
|
||||
do update set wrong_count = excluded.wrong_count,
|
||||
last_wrong_at = excluded.last_wrong_at,
|
||||
resolved_at = null
|
||||
`,
|
||||
[tenantId, userId, questionId],
|
||||
);
|
||||
}
|
||||
await insertImportIssueRows(runId, 'user_answer_records', issueRows);
|
||||
|
||||
await pool.query(
|
||||
`
|
||||
insert into public.wrong_questions (tenant_id, user_id, question_id, wrong_count, last_wrong_at, resolved_at)
|
||||
select tenant_id, user_id, question_id, count(*)::integer, max(answered_at), null
|
||||
from public.answer_records
|
||||
where tenant_id = $1
|
||||
and legacy_id is not null
|
||||
and question_id is not null
|
||||
and is_correct = false
|
||||
group by tenant_id, user_id, question_id
|
||||
on conflict (tenant_id, user_id, question_id)
|
||||
do update set wrong_count = excluded.wrong_count,
|
||||
last_wrong_at = excluded.last_wrong_at,
|
||||
resolved_at = null
|
||||
`,
|
||||
[tenantId],
|
||||
);
|
||||
}
|
||||
|
||||
async function normalizeUsers(records: JsonRecord[]) {
|
||||
@@ -2336,17 +2551,35 @@ async function normalizeQuestionVideos(records: JsonRecord[]) {
|
||||
);
|
||||
}
|
||||
|
||||
function normalizeReportType(value: unknown): string {
|
||||
const type = text(value)?.toLowerCase();
|
||||
if (!type) return 'other';
|
||||
if (['question_error', 'answer_error', 'explanation_bad', 'typo'].includes(type)) return 'question_error';
|
||||
if (type === 'content_error') return 'content_error';
|
||||
if (type === 'video_error') return 'video_error';
|
||||
if (type === 'asset_error') return 'asset_error';
|
||||
if (type === 'system_bug') return 'system_bug';
|
||||
if (type === 'suggestion') return 'suggestion';
|
||||
return 'other';
|
||||
}
|
||||
|
||||
function normalizeReportStatus(value: unknown): string {
|
||||
const status = text(value)?.toLowerCase();
|
||||
if (['pending', 'accepted', 'rejected', 'resolved', 'closed'].includes(status || '')) return status || 'pending';
|
||||
return 'pending';
|
||||
}
|
||||
|
||||
async function normalizeReports(records: JsonRecord[]) {
|
||||
for (const r of records) {
|
||||
await pool.query(
|
||||
`
|
||||
insert into public.reports (
|
||||
tenant_id, legacy_id, question_id, user_id, type, description,
|
||||
status, created_at, updated_at
|
||||
status, metadata, created_at, updated_at
|
||||
)
|
||||
values ($1,$2,$3,$4,$5,$6,$7,
|
||||
coalesce(nullif($8::text,'')::timestamptz, now()),
|
||||
coalesce(nullif($9::text,'')::timestamptz, now())
|
||||
values ($1,$2,$3,$4,$5,$6,$7,$8::jsonb,
|
||||
coalesce(nullif($9::text,'')::timestamptz, now()),
|
||||
coalesce(nullif($10::text,'')::timestamptz, now())
|
||||
)
|
||||
on conflict (tenant_id, legacy_id) do update set
|
||||
question_id = excluded.question_id,
|
||||
@@ -2354,6 +2587,7 @@ async function normalizeReports(records: JsonRecord[]) {
|
||||
type = excluded.type,
|
||||
description = excluded.description,
|
||||
status = excluded.status,
|
||||
metadata = excluded.metadata,
|
||||
updated_at = excluded.updated_at
|
||||
`,
|
||||
[
|
||||
@@ -2361,9 +2595,14 @@ async function normalizeReports(records: JsonRecord[]) {
|
||||
r.id,
|
||||
await legacyId('questions', r.questionId),
|
||||
await userIdByLegacy(r.userId),
|
||||
text(r.type),
|
||||
normalizeReportType(r.type),
|
||||
text(r.description),
|
||||
text(r.status) || 'pending',
|
||||
normalizeReportStatus(r.status),
|
||||
JSON.stringify({
|
||||
source: 'pocketbase.reports',
|
||||
legacyType: text(r.type),
|
||||
legacyStatus: text(r.status),
|
||||
}),
|
||||
dateText(r.created),
|
||||
dateText(r.updated),
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user