This commit is contained in:
@@ -0,0 +1,443 @@
|
||||
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');
|
||||
}
|
||||
Reference in New Issue
Block a user