444 lines
17 KiB
TypeScript
444 lines
17 KiB
TypeScript
import { createHash } from 'node:crypto';
|
||
import {
|
||
IMPORT_ENTITY_DEFINITIONS,
|
||
type ImportEntity,
|
||
type ImportFieldDefinition,
|
||
} from '@pig/core';
|
||
import type { Database } from '@pig/db';
|
||
import {
|
||
accounts,
|
||
contacts,
|
||
demandDeals,
|
||
importIdentities,
|
||
supplyDeals,
|
||
} from '@pig/db';
|
||
import { and, eq, inArray, sql } from 'drizzle-orm';
|
||
import type { Principal } from '../lib/auth';
|
||
import { MutationError } from '../lib/mutation';
|
||
import {
|
||
MAX_IMPORT_CELL_CHARS,
|
||
MAX_IMPORT_COLUMNS,
|
||
MAX_IMPORT_ROWS,
|
||
} from './tabular-import';
|
||
|
||
export type ImportTransaction = Parameters<Parameters<Database['transaction']>[0]>[0];
|
||
|
||
export interface ImportPlanInput {
|
||
entity: ImportEntity;
|
||
sourceName: string;
|
||
headers: string[];
|
||
rows: string[][];
|
||
mapping: Record<string, string>;
|
||
keySourceColumn: string;
|
||
}
|
||
|
||
export interface ImportRowError {
|
||
field: string | null;
|
||
message: string;
|
||
}
|
||
|
||
export interface ImportPreviewRow {
|
||
rowNumber: number;
|
||
key: string;
|
||
action: 'create' | 'update' | 'error';
|
||
recordId: string | null;
|
||
values: Record<string, unknown>;
|
||
errors: ImportRowError[];
|
||
}
|
||
|
||
export interface ImportPreview {
|
||
digest: string;
|
||
rows: ImportPreviewRow[];
|
||
counts: { create: number; update: number; error: number };
|
||
}
|
||
|
||
export interface ImportCommitResult {
|
||
created: number;
|
||
updated: number;
|
||
total: number;
|
||
}
|
||
|
||
type ImportDb = Database | ImportTransaction;
|
||
|
||
export class ImportService {
|
||
constructor(private readonly db: ImportDb) {}
|
||
|
||
async preview(input: ImportPlanInput): Promise<ImportPreview> {
|
||
assertPlanBounds(input);
|
||
const definition = IMPORT_ENTITY_DEFINITIONS[input.entity];
|
||
const knownFields = new Set(definition.fields.map((field) => field.key));
|
||
const headerSet = new Set(input.headers);
|
||
for (const [target, source] of Object.entries(input.mapping)) {
|
||
if (!knownFields.has(target)) throw new MutationError('invalid_mapping', `Unknown target field: ${target}.`, 400);
|
||
if (!headerSet.has(source)) throw new MutationError('invalid_mapping', `Unknown source column: ${source}.`, 400);
|
||
}
|
||
if (!headerSet.has(input.keySourceColumn)) {
|
||
throw new MutationError('invalid_mapping', 'Select a source column as the stable import key.', 400);
|
||
}
|
||
|
||
const keyIndex = input.headers.indexOf(input.keySourceColumn);
|
||
const keys = input.rows.map((row) => (row[keyIndex] ?? '').trim());
|
||
const nonemptyKeys = [...new Set(keys.filter(Boolean))];
|
||
const identities = nonemptyKeys.length === 0
|
||
? []
|
||
: await this.db
|
||
.select()
|
||
.from(importIdentities)
|
||
.where(and(
|
||
eq(importIdentities.entity, input.entity),
|
||
eq(importIdentities.keyColumn, input.keySourceColumn),
|
||
inArray(importIdentities.keyValue, nonemptyKeys),
|
||
));
|
||
const identityByKey = new Map(identities.map((identity) => [identity.keyValue, identity]));
|
||
const duplicateKeys = findDuplicateImportKeys(keys);
|
||
|
||
const rows = input.rows.map((row, index): ImportPreviewRow => {
|
||
const key = keys[index]!;
|
||
const identity = identityByKey.get(key);
|
||
const errors: ImportRowError[] = [];
|
||
if (!key) errors.push({ field: null, message: 'The selected source key is blank.' });
|
||
if (duplicateKeys.has(key)) errors.push({ field: null, message: 'The source key is duplicated in this file.' });
|
||
const converted = convertImportRow(
|
||
input.entity,
|
||
input.headers,
|
||
row,
|
||
input.mapping,
|
||
!identity,
|
||
);
|
||
errors.push(...converted.errors);
|
||
return {
|
||
rowNumber: index + 2,
|
||
key,
|
||
action: errors.length > 0 ? 'error' : identity ? 'update' : 'create',
|
||
recordId: identity?.recordId ?? null,
|
||
values: converted.values,
|
||
errors,
|
||
};
|
||
});
|
||
|
||
await this.validateRelationships(input.entity, rows);
|
||
const counts = rows.reduce(
|
||
(total, row) => ({ ...total, [row.action]: total[row.action] + 1 }),
|
||
{ create: 0, update: 0, error: 0 },
|
||
);
|
||
const digest = planDigest(input, rows);
|
||
return { digest, rows, counts };
|
||
}
|
||
|
||
private async validateRelationships(
|
||
entity: ImportEntity,
|
||
previewRows: ImportPreviewRow[],
|
||
): Promise<void> {
|
||
if (entity !== 'contact' && entity !== 'demand_deal' && entity !== 'supply_deal') return;
|
||
const accountIds = [...new Set(previewRows
|
||
.map((row) => row.values.accountId)
|
||
.filter((value): value is string => typeof value === 'string'))];
|
||
if (accountIds.length === 0) return;
|
||
const found = await this.db
|
||
.select({ id: accounts.id, side: accounts.side })
|
||
.from(accounts)
|
||
.where(inArray(accounts.id, accountIds));
|
||
const accountsById = new Map(found.map((account) => [account.id, account]));
|
||
for (const row of previewRows) {
|
||
const accountId = row.values.accountId;
|
||
if (typeof accountId !== 'string') continue;
|
||
const account = accountsById.get(accountId);
|
||
if (!account) row.errors.push({ field: 'accountId', message: 'The account does not exist.' });
|
||
else if (entity === 'demand_deal' && account.side === 'supply') {
|
||
row.errors.push({ field: 'accountId', message: 'A demand deal needs a demand or dual-sided account.' });
|
||
} else if (entity === 'supply_deal' && account.side === 'demand') {
|
||
row.errors.push({ field: 'accountId', message: 'A supply deal needs a supply or dual-sided account.' });
|
||
}
|
||
if (row.errors.length > 0) row.action = 'error';
|
||
}
|
||
}
|
||
|
||
async commit(
|
||
input: ImportPlanInput & { previewDigest: string },
|
||
principal: Principal,
|
||
now: Date,
|
||
): Promise<ImportCommitResult> {
|
||
if (!('execute' in this.db)) throw new Error('Import commit requires a database transaction.');
|
||
await this.db.execute(
|
||
sql`select pg_advisory_xact_lock(hashtextextended(${`pig:import:${input.entity}`}, 0))`,
|
||
);
|
||
const preview = await this.preview(input);
|
||
if (preview.digest !== input.previewDigest) {
|
||
throw new MutationError(
|
||
'stale_import_preview',
|
||
'The import plan changed after preview. Run the dry run again before committing.',
|
||
409,
|
||
);
|
||
}
|
||
if (preview.counts.error > 0) {
|
||
throw new MutationError(
|
||
'invalid_import_rows',
|
||
'Fix every row error and run the dry run again before committing.',
|
||
409,
|
||
);
|
||
}
|
||
|
||
let created = 0;
|
||
let updated = 0;
|
||
for (const row of preview.rows) {
|
||
const recordId = await this.writeRecord(input.entity, row, principal, now);
|
||
if (row.action === 'create') {
|
||
created += 1;
|
||
await this.db.insert(importIdentities).values({
|
||
entity: input.entity,
|
||
keyColumn: input.keySourceColumn,
|
||
keyValue: row.key,
|
||
recordId,
|
||
importedByUserId: principal.userId,
|
||
createdAt: now,
|
||
updatedAt: now,
|
||
});
|
||
} else {
|
||
updated += 1;
|
||
await this.db
|
||
.update(importIdentities)
|
||
.set({ importedByUserId: principal.userId, updatedAt: now })
|
||
.where(and(
|
||
eq(importIdentities.entity, input.entity),
|
||
eq(importIdentities.keyColumn, input.keySourceColumn),
|
||
eq(importIdentities.keyValue, row.key),
|
||
));
|
||
}
|
||
}
|
||
return { created, updated, total: preview.rows.length };
|
||
}
|
||
|
||
private async writeRecord(
|
||
entity: ImportEntity,
|
||
row: ImportPreviewRow,
|
||
principal: Principal,
|
||
now: Date,
|
||
): Promise<string> {
|
||
const values = row.values;
|
||
if (row.action === 'create') {
|
||
if (entity === 'account') {
|
||
const [record] = await this.db.insert(accounts).values({
|
||
...values,
|
||
ownerUserId: principal.userId,
|
||
source: 'import',
|
||
confidence: 'unverified',
|
||
createdAt: now,
|
||
updatedAt: now,
|
||
} as typeof accounts.$inferInsert).returning({ id: accounts.id });
|
||
if (record) return record.id;
|
||
} else if (entity === 'contact') {
|
||
const [record] = await this.db.insert(contacts).values({
|
||
...values,
|
||
ownerUserId: principal.userId,
|
||
source: 'import',
|
||
confidence: 'unverified',
|
||
createdAt: now,
|
||
updatedAt: now,
|
||
} as typeof contacts.$inferInsert).returning({ id: contacts.id });
|
||
if (record) return record.id;
|
||
} else if (entity === 'demand_deal') {
|
||
const [record] = await this.db.insert(demandDeals).values({
|
||
...values,
|
||
ownerUserId: principal.userId,
|
||
stageChangedAt: now,
|
||
createdAt: now,
|
||
updatedAt: now,
|
||
} as typeof demandDeals.$inferInsert).returning({ id: demandDeals.id });
|
||
if (record) return record.id;
|
||
} else {
|
||
const [record] = await this.db.insert(supplyDeals).values({
|
||
...values,
|
||
ownerUserId: principal.userId,
|
||
stageChangedAt: now,
|
||
createdAt: now,
|
||
updatedAt: now,
|
||
} as typeof supplyDeals.$inferInsert).returning({ id: supplyDeals.id });
|
||
if (record) return record.id;
|
||
}
|
||
throw new MutationError('write_failed', `Import row ${row.rowNumber} was not created.`, 409);
|
||
}
|
||
|
||
const recordId = row.recordId;
|
||
if (!recordId) throw new MutationError('missing_import_identity', 'The import identity is incomplete.', 409);
|
||
if (entity === 'account') {
|
||
const [record] = await this.db.update(accounts).set({
|
||
...values,
|
||
source: 'import',
|
||
updatedAt: now,
|
||
} as Partial<typeof accounts.$inferInsert>).where(eq(accounts.id, recordId)).returning({ id: accounts.id });
|
||
if (record) return record.id;
|
||
} else if (entity === 'contact') {
|
||
const [record] = await this.db.update(contacts).set({
|
||
...values,
|
||
source: 'import',
|
||
updatedAt: now,
|
||
} as Partial<typeof contacts.$inferInsert>).where(eq(contacts.id, recordId)).returning({ id: contacts.id });
|
||
if (record) return record.id;
|
||
} else if (entity === 'demand_deal') {
|
||
const [record] = await this.db.update(demandDeals).set({
|
||
...values,
|
||
updatedAt: now,
|
||
} as Partial<typeof demandDeals.$inferInsert>).where(eq(demandDeals.id, recordId)).returning({ id: demandDeals.id });
|
||
if (record) return record.id;
|
||
} else {
|
||
const [record] = await this.db.update(supplyDeals).set({
|
||
...values,
|
||
updatedAt: now,
|
||
} as Partial<typeof supplyDeals.$inferInsert>).where(eq(supplyDeals.id, recordId)).returning({ id: supplyDeals.id });
|
||
if (record) return record.id;
|
||
}
|
||
throw new MutationError('missing_import_record', `The record for import row ${row.rowNumber} no longer exists.`, 409);
|
||
}
|
||
}
|
||
|
||
export function convertImportRow(
|
||
entity: ImportEntity,
|
||
headers: readonly string[],
|
||
row: readonly string[],
|
||
mapping: Readonly<Record<string, string>>,
|
||
isCreate: boolean,
|
||
): { values: Record<string, unknown>; errors: ImportRowError[] } {
|
||
const values: Record<string, unknown> = {};
|
||
const errors: ImportRowError[] = [];
|
||
for (const field of IMPORT_ENTITY_DEFINITIONS[entity].fields) {
|
||
const sourceColumn = mapping[field.key];
|
||
if (!sourceColumn) {
|
||
if (isCreate && field.required) errors.push({ field: field.key, message: `${field.label} must be mapped.` });
|
||
continue;
|
||
}
|
||
const sourceIndex = headers.indexOf(sourceColumn);
|
||
const raw = sourceIndex < 0 ? '' : (row[sourceIndex] ?? '').trim();
|
||
if (!raw) {
|
||
if (field.required) errors.push({ field: field.key, message: `${field.label} is required.` });
|
||
else if (field.clearable !== false) values[field.key] = null;
|
||
continue;
|
||
}
|
||
const converted = convertCell(field, raw);
|
||
if (converted.error) errors.push({ field: field.key, message: converted.error });
|
||
else values[field.key] = converted.value;
|
||
}
|
||
if (entity === 'account') {
|
||
const side = values.side;
|
||
if (values.supplierType && side === 'demand') {
|
||
errors.push({ field: 'supplierType', message: 'Supplier type requires a supply or dual-sided account.' });
|
||
}
|
||
if (values.customerSegment && side === 'supply') {
|
||
errors.push({ field: 'customerSegment', message: 'Customer segment requires a demand or dual-sided account.' });
|
||
}
|
||
}
|
||
return { values, errors };
|
||
}
|
||
|
||
function convertCell(
|
||
field: ImportFieldDefinition,
|
||
raw: string,
|
||
): { value?: unknown; error?: string } {
|
||
if (raw.length > (field.maxLength ?? MAX_IMPORT_CELL_CHARS)) {
|
||
return { error: `${field.label} exceeds ${field.maxLength ?? MAX_IMPORT_CELL_CHARS} characters.` };
|
||
}
|
||
if (field.kind === 'text') return { value: raw };
|
||
if (field.kind === 'email') {
|
||
return /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(raw)
|
||
? { value: raw.toLocaleLowerCase() }
|
||
: { error: `${field.label} is not a complete email address.` };
|
||
}
|
||
if (field.kind === 'url') {
|
||
try {
|
||
const url = new URL(raw);
|
||
return ['http:', 'https:'].includes(url.protocol)
|
||
? { value: url.toString() }
|
||
: { error: `${field.label} must use http or https.` };
|
||
} catch {
|
||
return { error: `${field.label} is not a valid URL.` };
|
||
}
|
||
}
|
||
if (field.kind === 'uuid') {
|
||
return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(raw)
|
||
? { value: raw.toLocaleLowerCase() }
|
||
: { error: `${field.label} must be a PIG UUID.` };
|
||
}
|
||
if (field.kind === 'integer') {
|
||
if (!/^-?\d+$/.test(raw)) return { error: `${field.label} must be a whole number.` };
|
||
const value = Number(raw);
|
||
if (!Number.isSafeInteger(value)) return { error: `${field.label} is outside the supported range.` };
|
||
if (field.min != null && value < field.min) return { error: `${field.label} must be at least ${field.min}.` };
|
||
if (field.max != null && value > field.max) return { error: `${field.label} must be at most ${field.max}.` };
|
||
return { value };
|
||
}
|
||
if (field.kind === 'decimal') {
|
||
const value = Number(raw);
|
||
if (!Number.isFinite(value)) return { error: `${field.label} must be a number.` };
|
||
if (field.min != null && value < field.min) return { error: `${field.label} must be at least ${field.min}.` };
|
||
if (field.max != null && value > field.max) return { error: `${field.label} must be at most ${field.max}.` };
|
||
return { value };
|
||
}
|
||
if (field.kind === 'boolean') {
|
||
const value = raw.toLocaleLowerCase();
|
||
if (['true', 'yes', '1'].includes(value)) return { value: true };
|
||
if (['false', 'no', '0'].includes(value)) return { value: false };
|
||
return { error: `${field.label} must be true/false, yes/no, or 1/0.` };
|
||
}
|
||
if (field.kind === 'date') {
|
||
const value = new Date(raw);
|
||
return Number.isNaN(value.getTime())
|
||
? { error: `${field.label} is not a valid date.` }
|
||
: { value };
|
||
}
|
||
if (field.kind === 'currency') {
|
||
return /^[A-Za-z]{3}$/.test(raw)
|
||
? { value: raw.toUpperCase() }
|
||
: { error: `${field.label} must be a three-letter currency code.` };
|
||
}
|
||
return field.options?.includes(raw)
|
||
? { value: raw }
|
||
: { error: `${field.label} must be one of: ${field.options?.join(', ')}.` };
|
||
}
|
||
|
||
export function findDuplicateImportKeys(keys: readonly string[]): Set<string> {
|
||
const once = new Set<string>();
|
||
const duplicates = new Set<string>();
|
||
for (const key of keys) {
|
||
if (!key) continue;
|
||
if (once.has(key)) duplicates.add(key);
|
||
else once.add(key);
|
||
}
|
||
return duplicates;
|
||
}
|
||
|
||
function assertPlanBounds(input: ImportPlanInput): void {
|
||
if (!input.sourceName.trim() || input.sourceName.length > 255) {
|
||
throw new MutationError('invalid_import', 'The source file name is invalid.', 400);
|
||
}
|
||
if (input.headers.length === 0 || input.headers.length > MAX_IMPORT_COLUMNS) {
|
||
throw new MutationError('invalid_import', 'Imports need 1–100 columns.', 400);
|
||
}
|
||
if (input.rows.length === 0 || input.rows.length > MAX_IMPORT_ROWS) {
|
||
throw new MutationError('invalid_import', 'Imports need 1–2,000 data rows.', 400);
|
||
}
|
||
if (input.headers.some((header) => !header || header.length > 255)) {
|
||
throw new MutationError('invalid_import', 'Source headers must be non-empty and at most 255 characters.', 400);
|
||
}
|
||
if (input.rows.some((row) => row.length > MAX_IMPORT_COLUMNS || row.some((cell) => cell.length > MAX_IMPORT_CELL_CHARS))) {
|
||
throw new MutationError('invalid_import', 'The imported table exceeds the row, column, or cell limits.', 400);
|
||
}
|
||
}
|
||
|
||
function planDigest(input: ImportPlanInput, rows: readonly ImportPreviewRow[]): string {
|
||
const mapping = Object.fromEntries(Object.entries(input.mapping).sort(([left], [right]) => left.localeCompare(right)));
|
||
return createHash('sha256').update(JSON.stringify({
|
||
entity: input.entity,
|
||
sourceName: input.sourceName,
|
||
headers: input.headers,
|
||
rows: input.rows,
|
||
mapping,
|
||
keySourceColumn: input.keySourceColumn,
|
||
decisions: rows.map((row) => ({
|
||
rowNumber: row.rowNumber,
|
||
key: row.key,
|
||
action: row.action,
|
||
recordId: row.recordId,
|
||
errors: row.errors,
|
||
})),
|
||
})).digest('hex');
|
||
}
|