From b7589017ab652cfc30bb7f305ab42000654034b0 Mon Sep 17 00:00:00 2001 From: phaichayon Date: Fri, 3 Jul 2026 09:30:57 +0700 Subject: [PATCH] task-d.7.5 --- .../task-d75-csv-commit-import-engine.md | 66 + plans/task-d.7.5.md | 493 +++++++ src/app/api/setup/import/commit/route.ts | 51 + .../setup/server/csv-commit.service.ts | 131 ++ .../setup/server/csv/commit-builder.ts | 65 + .../setup/server/csv/import-report.ts | 121 ++ .../setup/server/csv/upsert-executor.ts | 1244 +++++++++++++++++ 7 files changed, 2171 insertions(+) create mode 100644 docs/implementation/task-d75-csv-commit-import-engine.md create mode 100644 plans/task-d.7.5.md create mode 100644 src/app/api/setup/import/commit/route.ts create mode 100644 src/features/setup/server/csv-commit.service.ts create mode 100644 src/features/setup/server/csv/commit-builder.ts create mode 100644 src/features/setup/server/csv/import-report.ts create mode 100644 src/features/setup/server/csv/upsert-executor.ts diff --git a/docs/implementation/task-d75-csv-commit-import-engine.md b/docs/implementation/task-d75-csv-commit-import-engine.md new file mode 100644 index 0000000..62d2e66 --- /dev/null +++ b/docs/implementation/task-d75-csv-commit-import-engine.md @@ -0,0 +1,66 @@ +# Task D.7.5 CSV Commit Import Engine + +Date: 2026-07-03 +Status: implemented + +## Scope Delivered + +- Added protected commit API: + - `POST /api/setup/import/commit` +- Added setup CSV commit service: + - `src/features/setup/server/csv-commit.service.ts` + - `src/features/setup/server/csv/commit-builder.ts` + - `src/features/setup/server/csv/upsert-executor.ts` + - `src/features/setup/server/csv/import-report.ts` + +## Behavior + +- Commit requires `previewHash`. +- Commit re-runs D.7.4 preview on the submitted files. +- Commit rejects mismatched `previewHash`. +- Commit rejects preview results with blocking errors. +- Commit follows dependency order from `setup/templates/templates.json` through the D.7.4 preview ordering. +- `dryRun=true` returns planned actions and does not start transaction writes. +- Non-dry-run commits execute in a single database transaction. +- Supported rows use idempotent business-key upsert behavior. +- Password updates only run for existing users when `password` is supplied and `reset_password=true`; otherwise password input is ignored with a report warning. +- Import reports never return password hashes. + +## Supported Templates + +The commit executor supports the templates registered by D.7.3/D.7.4: + +- `users.csv` +- `organizations.csv` +- `memberships.csv` +- `crm-role-assignments.csv` +- `master-options.csv` +- `branches.csv` +- `product-types.csv` +- `document-sequences.csv` +- `approval-workflows.csv` +- `approval-steps.csv` +- `approval-matrix.csv` +- `customers.csv` +- `contacts.csv` +- `contact-shares.csv` +- `leads.csv` +- `opportunities.csv` +- `opportunity-parties.csv` +- `quotations.csv` +- `quotation-items.csv` +- `quotation-parties.csv` +- `quotation-topics.csv` +- `quotation-topic-items.csv` + +## Safety Notes + +- D.7.5 reuses D.7.4 parsing, template detection, validation, dependency resolution, preview actions, and `previewHash`. +- Setup import writes directly with Drizzle for this slice because no safe setup import service contract exists for every supported template. +- The import report includes `AUDIT_GAP` warnings for direct Drizzle writes; a future service wrapper can add per-row audit semantics without changing the preview contract. +- Storage, attachments, approval execution, PDFs, setup state persistence, seed manifest persistence, and destructive reset remain out of scope. + +## Verification + +- `npm run typecheck` passed. +- `npx oxlint src/features/setup/server/csv src/features/setup/server/csv-commit.service.ts src/app/api/setup/import/commit/route.ts` passed. diff --git a/plans/task-d.7.5.md b/plans/task-d.7.5.md new file mode 100644 index 0000000..7577e51 --- /dev/null +++ b/plans/task-d.7.5.md @@ -0,0 +1,493 @@ +# Task D.7.5 – CSV Commit Import Engine + +## Objective + +Implement the CSV Commit Import Engine for ALLA OS Initial Setup. + +This task must take a previously validated CSV preview result from D.7.4, verify its `previewHash`, and commit supported CSV imports safely using transactions and idempotent upsert behavior. + +D.7.5 must reuse the D.7.4 preview/validation engine and must not duplicate validation logic. + +--- + +## Review Before Implementation + +Review: + +- AGENTS.md +- plans/task-d.7.md +- plans/task-d.7.1.md +- plans/task-d.7.2.md +- plans/task-d.7.3.md +- plans/task-d.7.4.md +- docs/setup/seed-matrix.md +- docs/setup/csv-template-spec.md +- docs/setup/import-mapping.md +- docs/setup/setup-wizard-blueprint.md +- docs/setup/verification-matrix.md +- setup/templates/* +- setup/templates/templates.json +- src/features/setup/server/csv-preview.service.ts +- src/features/setup/server/csv/** +- src/app/api/setup/import/preview/route.ts +- src/db/schema.ts +- src/db/seeds/**/* +- Existing CRM/foundation services for organizations, users, master options, document sequences, approval, customers, contacts, leads, opportunities, and quotations. + +--- + +## Implementation Scope + +Create: + +src/features/setup/server/csv-commit.service.ts + +Supporting modules if needed: + +src/features/setup/server/csv/commit-builder.ts +src/features/setup/server/csv/upsert-executor.ts +src/features/setup/server/csv/import-report.ts + +Create API: + +POST /api/setup/import/commit + +--- + +## Core Rule + +D.7.5 must use the exact same parsing, template detection, validation, dependency resolution, and preview report behavior from D.7.4. + +Do not create a second validation engine. + +Commit flow: + +1. Receive CSV files and `previewHash`. +2. Re-run preview validation using D.7.4 engine. +3. Compare new preview hash with submitted `previewHash`. +4. Reject commit if hashes do not match. +5. Reject commit if preview contains blocking errors. +6. Commit valid files in dependency/import order. +7. Return structured import report. + +--- + +## Input + +Accept `multipart/form-data`. + +Required: + +- CSV files +- `previewHash` + +Optional: + +- `mode`: `strict | permissive` +- `organizationId` +- `organizationCode` +- `dryRun` +- `allowOptionCreate` +- `skipWarnings` +- `replaceExisting` + +Default behavior: + +- `mode = strict` +- `dryRun = false` +- `allowOptionCreate = false` +- `skipWarnings = false` +- `replaceExisting = false` + +--- + +## Commit Rules + +### General + +- Commit only rows classified as `create` or `update`. +- Do not commit rows classified as `error`. +- In strict mode, any blocking error prevents the whole commit. +- Warnings do not block commit unless `skipWarnings = false` and the warning is configured as blocking. +- Unsupported templates must never commit. +- Unknown templates must never commit. + +### Transaction + +- Use one database transaction for all DB-only writes in the import batch. +- If any DB write fails, rollback the whole batch. +- Storage writes are out of scope for D.7.5. +- Attachments are still unsupported/deferred. + +### Upsert + +Every importer must be idempotent. + +Re-running the same CSV should produce either: + +- `updated` +- `skipped` +- or no-op + +Never create duplicate records. + +--- + +## Supported Commit Templates + +Implement commit for templates currently supported by D.7.3/D.7.4: + +Foundation / Organization: + +- users.csv +- organizations.csv +- memberships.csv +- crm-role-assignments.csv +- master-options.csv +- branches.csv +- product-types.csv +- document-sequences.csv +- approval-workflows.csv +- approval-steps.csv +- approval-matrix.csv + +Business: + +- customers.csv +- contacts.csv +- contact-shares.csv +- leads.csv +- opportunities.csv +- opportunity-parties.csv +- quotations.csv +- quotation-items.csv +- quotation-parties.csv +- quotation-topics.csv +- quotation-topic-items.csv + +If a template cannot yet be safely committed because the service contract is unclear, mark it as blocked with a clear reason. Do not fake success. + +--- + +## Required Import Order + +Use import order from `setup/templates/templates.json`. + +Commit order must follow resolved dependency order, not upload order. + +--- + +## Importer Responsibilities + +Each importer must define: + +- template name +- target table or feature service +- business key +- create behavior +- update behavior +- skip behavior +- conflict behavior +- reference resolution +- audit behavior if existing service supports it + +--- + +## Data Mapping + +Follow: + +docs/setup/import-mapping.md + +Important examples: + +branches.csv + +maps to: + +ms_options +category = crm_branch + +product-types.csv + +maps to: + +ms_options +category = crm_product_type + +document-sequences.csv + +resolves: + +organization_code +branch_code +product_type_code +document_type +period +prefix + +quotation-topic-items.csv + +uses `topic_key` from uploaded batch to resolve committed quotation topic ids. + +--- + +## Security + +- API must be protected. +- Require `super_admin` for setup-wide import. +- Organization admin may be supported later, but D.7.5 can restrict to `super_admin`. +- Do not return raw SQL, stack traces, password hashes, or secrets. +- Raw `password` from users.csv must be hashed before storage. +- Never log raw passwords. +- Empty password must not overwrite existing password. +- Existing user password can be reset only when explicit reset flag exists. + +--- + +## Password Rules + +For `users.csv`: + +- New credential user requires either: + - `password`, or + - future invitation flow support. +- Existing user: + - empty password = keep existing password + - password + reset_password=true = update password hash + - password + reset_password=false = ignore password with warning +- Never return password or hash in report. + +--- + +## Report Contract + +Return: + +ImportReport + +Fields: + +- status: PASS | WARNING | FAIL +- generatedAt +- previewHash +- committedHash +- dryRun +- summary +- files +- errors +- warnings + +File report: + +- fileName +- template +- status +- imported +- updated +- skipped +- errors +- warnings + +Row report: + +- row +- action: create | update | skip | error +- businessKey +- entityId if safe +- message +- warnings +- errors + +Example: + +{ + "status": "PASS", + "previewHash": "...", + "committedHash": "...", + "summary": { + "imported": 10, + "updated": 4, + "skipped": 2, + "errors": 0, + "warnings": 1 + }, + "files": [] +} + +--- + +## Error Contract + +Every error must include: + +- file +- row +- column +- code +- message +- suggestedFix + +Example: + +{ + "file": "customers.csv", + "row": 12, + "column": "customer_code", + "code": "UPSERT_FAILED", + "message": "Customer could not be imported.", + "suggestedFix": "Review the customer row and retry after fixing related references." +} + +--- + +## Dry Run + +Support `dryRun=true`. + +Dry run must: + +- re-run preview +- build commit plan +- return planned actions +- not write database +- not run transaction writes + +--- + +## Audit + +If existing feature services automatically write audit logs, allow them. + +If importing directly with Drizzle because no safe service exists, do not invent audit semantics in D.7.5. + +Instead: + +- include audit gap warning in report +- document follow-up + +--- + +## Out Of Scope + +- CSV preview validation logic duplication +- CSV template creation +- Setup Wizard UI +- Seed manifest persistence +- Setup state persistence +- Rollback after successful commit +- Storage import +- Attachment import +- Email settings import +- Unsupported templates +- Destructive reset +- PDF generation +- Approval workflow execution +- Document artifact creation + +--- + +## Deliverables + +Required: + +src/features/setup/server/csv-commit.service.ts +src/features/setup/server/csv/commit-builder.ts +src/features/setup/server/csv/upsert-executor.ts +src/features/setup/server/csv/import-report.ts +src/app/api/setup/import/commit/route.ts + +Optional: + +src/features/setup/server/csv/importers/*.ts + +Recommended importer organization: + +src/features/setup/server/csv/importers/ + users.importer.ts + organizations.importer.ts + memberships.importer.ts + crm-role-assignments.importer.ts + master-options.importer.ts + document-sequences.importer.ts + approval.importer.ts + customers.importer.ts + contacts.importer.ts + leads.importer.ts + opportunities.importer.ts + quotations.importer.ts + +--- + +## Acceptance Criteria + +✓ POST /api/setup/import/commit exists. + +✓ Commit requires `previewHash`. + +✓ Commit re-runs D.7.4 preview engine. + +✓ Commit rejects mismatched preview hash. + +✓ Commit rejects preview with blocking errors. + +✓ Commit follows dependency/import order from templates.json. + +✓ Commit is transaction-safe for DB writes. + +✓ Re-running the same valid CSV does not duplicate records. + +✓ users.csv password handling is safe. + +✓ branches.csv imports into ms_options category crm_branch. + +✓ product-types.csv imports into ms_options category crm_product_type. + +✓ quotation-topic-items.csv resolves topic_key from committed/imported quotation topics. + +✓ Unsupported templates do not commit. + +✓ dryRun returns planned actions without writing DB. + +✓ Structured import report is returned. + +✓ No storage/binary attachment import is implemented. + +✓ Typecheck passes or unrelated failures are documented. + +--- + +## Suggested Verification + +Run: + +npm run typecheck + +Run scoped lint: + +npx oxlint src/features/setup/server/csv src/features/setup/server/csv-commit.service.ts src/app/api/setup/import/commit + +Manual test flow: + +1. Call POST /api/setup/import/preview with valid CSV batch. +2. Copy previewHash. +3. Call POST /api/setup/import/commit with same files and previewHash. +4. Confirm rows are created/updated. +5. Re-run same commit. +6. Confirm no duplicates. +7. Modify CSV after preview. +8. Confirm commit rejects previewHash mismatch. +9. Submit CSV with blocking errors. +10. Confirm commit fails without partial writes. + +--- + +## Follow-up + +After this task: + +D.7.6 – Dashboard Setup Wizard UI + +D.7.8 – Seed Manifest and Setup State Persistence + +D.7.7 – Scoped Reset and Reseed Engine diff --git a/src/app/api/setup/import/commit/route.ts b/src/app/api/setup/import/commit/route.ts new file mode 100644 index 0000000..65656a7 --- /dev/null +++ b/src/app/api/setup/import/commit/route.ts @@ -0,0 +1,51 @@ +import { NextRequest, NextResponse } from 'next/server'; +import { commitSetupCsvImport } from '@/features/setup/server/csv-commit.service'; +import { AuthError, requireSystemRole } from '@/lib/auth/session'; + +function isFile(value: FormDataEntryValue): value is File { + return typeof value === 'object' && 'arrayBuffer' in value && 'name' in value; +} + +function parseBoolean(value: FormDataEntryValue | null): boolean { + if (typeof value !== 'string') return false; + return value.trim().toLowerCase() === 'true'; +} + +export async function POST(request: NextRequest) { + try { + const session = await requireSystemRole('super_admin'); + const formData = await request.formData(); + const previewHash = String(formData.get('previewHash') ?? ''); + const dryRun = parseBoolean(formData.get('dryRun')); + const uploadedFiles = Array.from(formData.values()).filter(isFile); + + if (uploadedFiles.length === 0) { + return NextResponse.json( + { message: 'Upload at least one setup CSV file using multipart/form-data.' }, + { status: 400 } + ); + } + + const files = await Promise.all( + uploadedFiles.map(async (file) => ({ + fileName: file.name, + bytes: new Uint8Array(await file.arrayBuffer()) + })) + ); + + const report = await commitSetupCsvImport({ + files, + previewHash, + dryRun, + actorId: session.user.id + }); + + return NextResponse.json(report, { status: report.status === 'FAIL' ? 400 : 200 }); + } catch (error) { + if (error instanceof AuthError) { + return NextResponse.json({ message: error.message }, { status: error.status }); + } + + return NextResponse.json({ message: 'Unable commit setup CSV import' }, { status: 500 }); + } +} diff --git a/src/features/setup/server/csv-commit.service.ts b/src/features/setup/server/csv-commit.service.ts new file mode 100644 index 0000000..1dd6324 --- /dev/null +++ b/src/features/setup/server/csv-commit.service.ts @@ -0,0 +1,131 @@ +import 'server-only'; + +import { buildCsvCommitPlan } from './csv/commit-builder'; +import { buildImportReport, createImportError, type ImportReport } from './csv/import-report'; +import { CsvCommitExecutionError, executeCsvCommitPlan } from './csv/upsert-executor'; +import { previewSetupCsvImport } from './csv-preview.service'; +import type { CsvPreviewFileInput } from './csv/types'; + +export interface CommitSetupCsvImportInput { + files: CsvPreviewFileInput[]; + previewHash: string; + dryRun?: boolean; + actorId: string; +} + +export async function commitSetupCsvImport(input: CommitSetupCsvImportInput): Promise { + const preview = await previewSetupCsvImport(input.files); + const plan = buildCsvCommitPlan(preview); + + if (!input.previewHash) { + return buildImportReport({ + previewHash: preview.previewHash, + committedHash: plan.committedHash, + dryRun: Boolean(input.dryRun), + files: [ + { + fileName: '__request__', + template: 'setup-csv-commit', + status: 'FAIL', + imported: 0, + updated: 0, + skipped: 0, + errors: [ + createImportError({ + file: '__request__', + row: 1, + column: 'previewHash', + code: 'PREVIEW_HASH_REQUIRED', + message: 'previewHash is required before committing CSV import.', + suggestedFix: 'Call /api/setup/import/preview and submit the returned previewHash.' + }) + ], + warnings: [], + rows: [] + } + ] + }); + } + + if (preview.previewHash !== input.previewHash) { + return buildImportReport({ + previewHash: preview.previewHash, + committedHash: plan.committedHash, + dryRun: Boolean(input.dryRun), + files: [ + { + fileName: '__request__', + template: 'setup-csv-commit', + status: 'FAIL', + imported: 0, + updated: 0, + skipped: 0, + errors: [ + createImportError({ + file: '__request__', + row: 1, + column: 'previewHash', + code: 'PREVIEW_HASH_MISMATCH', + message: 'Submitted previewHash does not match the uploaded CSV files.', + suggestedFix: 'Re-run preview for the exact files being committed and retry with the new previewHash.' + }) + ], + warnings: [], + rows: [] + } + ] + }); + } + + if (preview.summary.error > 0) { + return buildImportReport({ + previewHash: preview.previewHash, + committedHash: plan.committedHash, + dryRun: Boolean(input.dryRun), + files: preview.files.map((file) => ({ + fileName: file.fileName, + template: file.template, + status: 'FAIL', + imported: 0, + updated: 0, + skipped: 0, + errors: file.errors, + warnings: file.warnings, + rows: file.rows.map((row) => ({ + row: row.row, + action: row.action === 'error' ? 'error' : 'skip', + businessKey: row.key, + message: row.action === 'error' ? 'Preview row has blocking errors.' : 'Commit blocked because preview contains errors.', + warnings: row.warnings, + errors: row.errors + })) + })) + }); + } + + let files; + try { + files = await executeCsvCommitPlan(plan, { + actorId: input.actorId, + dryRun: Boolean(input.dryRun) + }); + } catch (error) { + if (error instanceof CsvCommitExecutionError) { + return buildImportReport({ + previewHash: preview.previewHash, + committedHash: plan.committedHash, + dryRun: Boolean(input.dryRun), + files: error.files + }); + } + + throw error; + } + + return buildImportReport({ + previewHash: preview.previewHash, + committedHash: plan.committedHash, + dryRun: Boolean(input.dryRun), + files + }); +} diff --git a/src/features/setup/server/csv/commit-builder.ts b/src/features/setup/server/csv/commit-builder.ts new file mode 100644 index 0000000..562f659 --- /dev/null +++ b/src/features/setup/server/csv/commit-builder.ts @@ -0,0 +1,65 @@ +import { createHash } from 'node:crypto'; +import type { CsvPreviewFileResult, CsvPreviewResult, CsvPreviewRow } from './types'; + +export interface CsvCommitPlanRow { + fileName: string; + template: string; + importOrder: number; + row: CsvPreviewRow; +} + +export interface CsvCommitPlanFile { + fileName: string; + template: string; + importOrder: number; + rows: CsvCommitPlanRow[]; +} + +export interface CsvCommitPlan { + preview: CsvPreviewResult; + files: CsvCommitPlanFile[]; + rows: CsvCommitPlanRow[]; + committedHash: string; +} + +function stableRowPayload(file: CsvPreviewFileResult, row: CsvPreviewRow) { + return { + fileName: file.fileName, + importOrder: file.importOrder, + row: row.row, + key: row.key, + action: row.action, + values: row.values + }; +} + +export function buildCsvCommitPlan(preview: CsvPreviewResult): CsvCommitPlan { + const files = preview.files + .toSorted((left, right) => left.importOrder - right.importOrder || left.fileName.localeCompare(right.fileName)) + .map((file) => ({ + fileName: file.fileName, + template: file.template, + importOrder: file.importOrder, + rows: file.rows + .filter((row) => row.action !== 'error') + .toSorted((left, right) => left.row - right.row) + .map((row) => ({ + fileName: file.fileName, + template: file.template, + importOrder: file.importOrder, + row + })) + })); + + const rows = files.flatMap((file) => file.rows); + const committedHash = createHash('sha256') + .update( + JSON.stringify({ + previewHash: preview.previewHash, + rows: preview.files.flatMap((file) => file.rows.map((row) => stableRowPayload(file, row))) + }) + ) + .digest('hex'); + + return { preview, files, rows, committedHash }; +} diff --git a/src/features/setup/server/csv/import-report.ts b/src/features/setup/server/csv/import-report.ts new file mode 100644 index 0000000..39cdb64 --- /dev/null +++ b/src/features/setup/server/csv/import-report.ts @@ -0,0 +1,121 @@ +import type { CsvPreviewError, CsvPreviewWarning } from './types'; + +export type ImportStatus = 'PASS' | 'WARNING' | 'FAIL'; +export type ImportAction = 'create' | 'update' | 'skip' | 'error'; + +export interface ImportRowReport { + row: number; + action: ImportAction; + businessKey: string | null; + entityId?: string; + message: string; + warnings: CsvPreviewWarning[]; + errors: CsvPreviewError[]; +} + +export interface ImportFileReport { + fileName: string; + template: string; + status: ImportStatus; + imported: number; + updated: number; + skipped: number; + errors: CsvPreviewError[]; + warnings: CsvPreviewWarning[]; + rows: ImportRowReport[]; +} + +export interface ImportReport { + status: ImportStatus; + generatedAt: string; + previewHash: string; + committedHash: string; + dryRun: boolean; + summary: { + imported: number; + updated: number; + skipped: number; + errors: number; + warnings: number; + }; + files: ImportFileReport[]; + errors: CsvPreviewError[]; + warnings: CsvPreviewWarning[]; +} + +export function createImportError(input: { + file: string; + row: number; + column: string; + code: string; + message: string; + suggestedFix: string; +}): CsvPreviewError { + return input; +} + +export function createImportWarning(input: { + file: string; + row: number; + column: string; + code: string; + message: string; + suggestedFix: string; +}): CsvPreviewWarning { + return input; +} + +export function buildImportReport(input: { + previewHash: string; + committedHash: string; + dryRun: boolean; + files: ImportFileReport[]; +}): ImportReport { + const errors = input.files.flatMap((file) => file.errors); + const warnings = input.files.flatMap((file) => file.warnings); + const summary = input.files.reduce( + (accumulator, file) => ({ + imported: accumulator.imported + file.imported, + updated: accumulator.updated + file.updated, + skipped: accumulator.skipped + file.skipped, + errors: accumulator.errors + file.errors.length, + warnings: accumulator.warnings + file.warnings.length + }), + { imported: 0, updated: 0, skipped: 0, errors: 0, warnings: 0 } + ); + + return { + status: errors.length > 0 ? 'FAIL' : warnings.length > 0 ? 'WARNING' : 'PASS', + generatedAt: new Date().toISOString(), + previewHash: input.previewHash, + committedHash: input.committedHash, + dryRun: input.dryRun, + summary, + files: input.files, + errors, + warnings + }; +} + +export function summarizeFileRows(input: { + fileName: string; + template: string; + rows: ImportRowReport[]; + warnings?: CsvPreviewWarning[]; +}): ImportFileReport { + const errors = input.rows.flatMap((row) => row.errors); + const rowWarnings = input.rows.flatMap((row) => row.warnings); + const warnings = [...rowWarnings, ...(input.warnings ?? [])]; + + return { + fileName: input.fileName, + template: input.template, + status: errors.length > 0 ? 'FAIL' : warnings.length > 0 ? 'WARNING' : 'PASS', + imported: input.rows.filter((row) => row.action === 'create').length, + updated: input.rows.filter((row) => row.action === 'update').length, + skipped: input.rows.filter((row) => row.action === 'skip').length, + errors, + warnings, + rows: input.rows + }; +} diff --git a/src/features/setup/server/csv/upsert-executor.ts b/src/features/setup/server/csv/upsert-executor.ts new file mode 100644 index 0000000..22c7c67 --- /dev/null +++ b/src/features/setup/server/csv/upsert-executor.ts @@ -0,0 +1,1244 @@ +import 'server-only'; + +import { hash } from 'bcryptjs'; +import { randomUUID } from 'node:crypto'; +import { and, eq, isNull, sql } from 'drizzle-orm'; +import { + crmApprovalMatrices, + crmApprovalSteps, + crmApprovalWorkflows, + crmContactShares, + crmCustomerContacts, + crmCustomerOwnerHistory, + crmCustomers, + crmLeads, + crmOpportunities, + crmOpportunityCustomers, + crmQuotationCustomers, + crmQuotationItems, + crmQuotationTopicItems, + crmQuotationTopics, + crmQuotations, + crmRoleProfiles, + crmUserRoleAssignments, + documentSequences, + memberships, + msOptions, + organizations, + users +} from '@/db/schema'; +import { db } from '@/lib/db'; +import type { CsvCommitPlan, CsvCommitPlanFile, CsvCommitPlanRow } from './commit-builder'; +import { createImportError, createImportWarning, summarizeFileRows, type ImportFileReport, type ImportRowReport } from './import-report'; +import { normalizeEmail, normalizeLower, normalizeOrganizationCode, pipeList } from './value-validator'; + +type ImportTx = typeof db; + +export class CsvCommitExecutionError extends Error { + files: ImportFileReport[]; + + constructor(message: string, files: ImportFileReport[], cause: unknown) { + super(message, { cause }); + this.name = 'CsvCommitExecutionError'; + this.files = files; + } +} + +interface ExecuteContext { + actorId: string; + dryRun: boolean; + topicAliases: Map; +} + +interface UpsertResult { + action: 'create' | 'update' | 'skip'; + entityId?: string; + warnings?: ReturnType[]; +} + +function toBool(value: string | undefined, fallback = false): boolean { + if (!value) return fallback; + return value.trim().toLowerCase() === 'true'; +} + +function toInt(value: string | undefined, fallback = 0): number { + if (!value) return fallback; + return Number.parseInt(value, 10); +} + +function toNumber(value: string | undefined, fallback = 0): number { + if (!value) return fallback; + return Number(value); +} + +function toDate(value: string | undefined): Date | null { + if (!value) return null; + return new Date(`${value}T00:00:00.000Z`); +} + +function emptyToNull(value: string | undefined): string | null { + const trimmed = value?.trim(); + return trimmed ? trimmed : null; +} + +function businessKey(row: CsvCommitPlanRow): string | null { + return row.row.key; +} + +async function first(promise: Promise): Promise { + const [row] = await promise; + return row ?? null; +} + +async function findUserId(tx: ImportTx, email: string): Promise { + if (!email) return null; + const row = await first( + tx + .select({ id: users.id }) + .from(users) + .where(sql`lower(${users.email}) = ${normalizeEmail(email)}`) + .limit(1) + ); + return row?.id ?? null; +} + +async function findOrganization(tx: ImportTx, organizationCode: string): Promise<{ id: string; slug: string } | null> { + if (!organizationCode) return null; + const normalized = normalizeOrganizationCode(organizationCode); + const row = await first( + tx + .select({ id: organizations.id, slug: organizations.slug }) + .from(organizations) + .where(sql`lower(${organizations.slug}) = ${normalized.toLowerCase()} or ${organizations.id} = ${normalized}`) + .limit(1) + ); + return row ?? null; +} + +async function findOptionId( + tx: ImportTx, + organizationId: string | null, + category: string, + code: string | undefined +): Promise { + if (!organizationId || !code) return null; + const row = await first( + tx + .select({ id: msOptions.id }) + .from(msOptions) + .where( + and( + eq(msOptions.organizationId, organizationId), + eq(msOptions.category, normalizeLower(category)), + eq(msOptions.code, normalizeLower(code)), + isNull(msOptions.deletedAt) + ) + ) + .limit(1) + ); + return row?.id ?? null; +} + +async function findCustomerId(tx: ImportTx, organizationId: string | null, code: string | undefined): Promise { + if (!organizationId || !code) return null; + const row = await first( + tx + .select({ id: crmCustomers.id }) + .from(crmCustomers) + .where(and(eq(crmCustomers.organizationId, organizationId), eq(crmCustomers.code, code), isNull(crmCustomers.deletedAt))) + .limit(1) + ); + return row?.id ?? null; +} + +async function findContactId( + tx: ImportTx, + organizationId: string | null, + customerId: string | null, + values: Record +): Promise { + if (!organizationId || !customerId) return null; + const email = emptyToNull(values.contact_email ?? values.email); + const name = emptyToNull(values.contact_name); + const condition = email + ? sql`lower(${crmCustomerContacts.email}) = ${email.toLowerCase()}` + : name + ? eq(crmCustomerContacts.name, name) + : undefined; + if (!condition) return null; + const row = await first( + tx + .select({ id: crmCustomerContacts.id }) + .from(crmCustomerContacts) + .where( + and( + eq(crmCustomerContacts.organizationId, organizationId), + eq(crmCustomerContacts.customerId, customerId), + condition, + isNull(crmCustomerContacts.deletedAt) + ) + ) + .limit(1) + ); + return row?.id ?? null; +} + +async function findOpportunityId(tx: ImportTx, organizationId: string | null, code: string | undefined): Promise { + if (!organizationId || !code) return null; + const row = await first( + tx + .select({ id: crmOpportunities.id }) + .from(crmOpportunities) + .where(and(eq(crmOpportunities.organizationId, organizationId), eq(crmOpportunities.code, code), isNull(crmOpportunities.deletedAt))) + .limit(1) + ); + return row?.id ?? null; +} + +async function findLeadId(tx: ImportTx, organizationId: string | null, code: string | undefined): Promise { + if (!organizationId || !code) return null; + const row = await first( + tx + .select({ id: crmLeads.id }) + .from(crmLeads) + .where(and(eq(crmLeads.organizationId, organizationId), eq(crmLeads.code, code), isNull(crmLeads.deletedAt))) + .limit(1) + ); + return row?.id ?? null; +} + +async function findQuotationId(tx: ImportTx, organizationId: string | null, code: string | undefined): Promise { + if (!organizationId || !code) return null; + const row = await first( + tx + .select({ id: crmQuotations.id }) + .from(crmQuotations) + .where(and(eq(crmQuotations.organizationId, organizationId), eq(crmQuotations.code, code), isNull(crmQuotations.deletedAt))) + .limit(1) + ); + return row?.id ?? null; +} + +async function findWorkflowId(tx: ImportTx, organizationId: string | null, code: string | undefined): Promise { + if (!organizationId || !code) return null; + const row = await first( + tx + .select({ id: crmApprovalWorkflows.id }) + .from(crmApprovalWorkflows) + .where(and(eq(crmApprovalWorkflows.organizationId, organizationId), eq(crmApprovalWorkflows.code, normalizeLower(code)), isNull(crmApprovalWorkflows.deletedAt))) + .limit(1) + ); + return row?.id ?? null; +} + +async function findRoleProfileId(tx: ImportTx, organizationId: string | null, code: string | undefined): Promise { + if (!organizationId || !code) return null; + const row = await first( + tx + .select({ id: crmRoleProfiles.id }) + .from(crmRoleProfiles) + .where(and(eq(crmRoleProfiles.organizationId, organizationId), eq(crmRoleProfiles.code, normalizeLower(code)), isNull(crmRoleProfiles.deletedAt))) + .limit(1) + ); + return row?.id ?? null; +} + +async function resolveOrganizationId(tx: ImportTx, values: Record): Promise { + const organization = await findOrganization(tx, values.organization_code); + if (!organization) throw new Error(`Organization ${values.organization_code} not found after preview.`); + return organization.id; +} + +async function resolveUserId(tx: ImportTx, email: string): Promise { + const userId = await findUserId(tx, email); + if (!userId) throw new Error(`User ${email} not found after preview.`); + return userId; +} + +async function upsertUser(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const email = normalizeEmail(values.email); + const existing = await first(tx.select({ id: users.id }).from(users).where(sql`lower(${users.email}) = ${email}`).limit(1)); + const activeOrganization = values.active_organization_code ? await findOrganization(tx, values.active_organization_code) : null; + const resetPassword = toBool(values.reset_password); + const warnings: ReturnType[] = []; + + if (existing) { + const updateValues: Partial = { + name: values.name, + systemRole: normalizeLower(values.system_role), + image: emptyToNull(values.image_url), + activeOrganizationId: activeOrganization?.id ?? null, + updatedAt: new Date() + }; + if (values.password && resetPassword) updateValues.passwordHash = await hash(values.password, 12); + if (values.password && !resetPassword) { + warnings.push( + createImportWarning({ + file: planRow.fileName, + row: planRow.row.row, + column: 'password', + code: 'PASSWORD_IGNORED', + message: 'Password was ignored because reset_password is false.', + suggestedFix: 'Set reset_password=true when intentionally rotating a setup user password.' + }) + ); + } + await tx.update(users).set(updateValues).where(eq(users.id, existing.id)); + return { action: 'update', entityId: existing.id, warnings }; + } + + if (!values.password) throw new Error(`New user ${email} requires a password.`); + const id = randomUUID(); + await tx.insert(users).values({ + id, + name: values.name, + email, + passwordHash: await hash(values.password, 12), + systemRole: normalizeLower(values.system_role), + image: emptyToNull(values.image_url), + activeOrganizationId: activeOrganization?.id ?? null + }); + return { action: 'create', entityId: id, warnings }; +} + +async function upsertOrganization(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const slug = normalizeLower(values.slug); + const createdBy = await resolveUserId(tx, values.created_by_email); + const existing = await first(tx.select({ id: organizations.id }).from(organizations).where(eq(organizations.slug, slug)).limit(1)); + const updateValues = { + name: values.organization_name, + imageUrl: emptyToNull(values.logo_url), + plan: normalizeLower(values.plan || 'enterprise'), + createdBy, + updatedAt: new Date() + }; + if (existing) { + await tx.update(organizations).set(updateValues).where(eq(organizations.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = normalizeOrganizationCode(values.organization_code); + await tx.insert(organizations).values({ id, slug, ...updateValues }); + return { action: 'create', entityId: id }; +} + +async function upsertMembership(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const userId = await resolveUserId(tx, values.user_email); + const existing = await first( + tx.select({ id: memberships.id }).from(memberships).where(and(eq(memberships.organizationId, organizationId), eq(memberships.userId, userId))).limit(1) + ); + const branchScopeIds = await Promise.all(pipeList(values.branch_scope_codes ?? '').map((code) => findOptionId(tx, organizationId, 'crm_branch', code))); + const productTypeScopeIds = await Promise.all( + pipeList(values.product_type_scope_codes ?? '').map((code) => findOptionId(tx, organizationId, 'crm_product_type', code)) + ); + const record = { + userId, + organizationId, + role: normalizeLower(values.membership_role), + businessRole: normalizeLower(values.business_role), + permissions: pipeList(values.permissions ?? ''), + branchScopeIds: branchScopeIds.filter(Boolean) as string[], + productTypeScopeIds: productTypeScopeIds.filter(Boolean) as string[], + updatedAt: new Date() + }; + if (existing) { + await tx.update(memberships).set(record).where(eq(memberships.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(memberships).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertMasterOption(tx: ImportTx, planRow: CsvCommitPlanRow, categoryOverride?: string): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const category = categoryOverride ?? normalizeLower(values.category); + const code = normalizeLower(values.code ?? values.branch_code ?? values.product_type_code); + const label = values.label ?? values.branch_name ?? values.product_type_name; + const parentId = + values.parent_category && values.parent_code ? await findOptionId(tx, organizationId, values.parent_category, values.parent_code) : null; + const metadata = + categoryOverride === 'crm_branch' + ? { + prefix: emptyToNull(values.prefix), + address: emptyToNull(values.address), + taxId: emptyToNull(values.tax_id), + phone: emptyToNull(values.phone), + email: emptyToNull(values.email) + } + : categoryOverride === 'crm_product_type' + ? { ...(values.metadata_json ? JSON.parse(values.metadata_json) : {}), documentPrefix: emptyToNull(values.document_prefix) } + : values.metadata_json + ? JSON.parse(values.metadata_json) + : null; + const existing = await first( + tx + .select({ id: msOptions.id }) + .from(msOptions) + .where(and(eq(msOptions.organizationId, organizationId), eq(msOptions.category, category), eq(msOptions.code, code))) + .limit(1) + ); + const record = { + organizationId, + category, + code, + label, + value: emptyToNull(values.value) ?? code, + parentId, + sortOrder: toInt(values.sort_order), + isActive: toBool(values.is_active, true), + metadata, + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(msOptions).set(record).where(eq(msOptions.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(msOptions).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertCrmRoleAssignment(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const userId = await resolveUserId(tx, values.user_email); + const roleProfileId = await findRoleProfileId(tx, organizationId, values.crm_role_code); + if (!roleProfileId) throw new Error(`CRM role ${values.crm_role_code} not found.`); + const existing = await first( + tx + .select({ id: crmUserRoleAssignments.id }) + .from(crmUserRoleAssignments) + .where( + and( + eq(crmUserRoleAssignments.organizationId, organizationId), + eq(crmUserRoleAssignments.userId, userId), + eq(crmUserRoleAssignments.roleProfileId, roleProfileId) + ) + ) + .limit(1) + ); + const branchIds = await Promise.all(pipeList(values.branch_codes ?? '').map((code) => findOptionId(tx, organizationId, 'crm_branch', code))); + const productIds = await Promise.all(pipeList(values.product_type_codes ?? '').map((code) => findOptionId(tx, organizationId, 'crm_product_type', code))); + const record = { + organizationId, + userId, + roleProfileId, + branchScopeMode: normalizeLower(values.branch_scope_mode), + branchScopeIds: branchIds.filter(Boolean) as string[], + productTypeScopeMode: normalizeLower(values.product_type_scope_mode), + productTypeScopeIds: productIds.filter(Boolean) as string[], + isPrimary: toBool(values.is_primary), + isActive: toBool(values.is_active, true), + assignedBy: context.actorId, + assignedAt: new Date(), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmUserRoleAssignments).set(record).where(eq(crmUserRoleAssignments.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmUserRoleAssignments).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertDocumentSequence(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const branchId = (await findOptionId(tx, organizationId, 'crm_branch', values.branch_code)) ?? ''; + const productType = normalizeLower(values.product_type_code || 'all'); + const existing = await first( + tx + .select({ id: documentSequences.id }) + .from(documentSequences) + .where( + and( + eq(documentSequences.organizationId, organizationId), + eq(documentSequences.branchId, branchId), + eq(documentSequences.productType, productType), + eq(documentSequences.documentType, normalizeLower(values.document_type)), + eq(documentSequences.period, values.period) + ) + ) + .limit(1) + ); + const record = { + organizationId, + branchId, + productType, + documentType: normalizeLower(values.document_type), + prefix: values.prefix, + period: values.period, + currentNumber: toInt(values.current_number), + paddingLength: toInt(values.padding_length, 3), + format: values.format || '{prefix}{period}-{running}', + resetPolicy: normalizeLower(values.reset_policy || 'period'), + isActive: toBool(values.is_active, true), + updatedAt: new Date() + }; + if (existing) { + await tx.update(documentSequences).set(record).where(eq(documentSequences.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(documentSequences).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertApprovalWorkflow(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const code = normalizeLower(values.workflow_code); + const existing = await first( + tx.select({ id: crmApprovalWorkflows.id }).from(crmApprovalWorkflows).where(and(eq(crmApprovalWorkflows.organizationId, organizationId), eq(crmApprovalWorkflows.code, code))).limit(1) + ); + const record = { + organizationId, + code, + name: values.workflow_name, + entityType: normalizeLower(values.entity_type), + description: emptyToNull(values.description), + isSystem: toBool(values.is_system), + isActive: toBool(values.is_active, true), + createdBy: context.actorId, + updatedBy: context.actorId, + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmApprovalWorkflows).set(record).where(eq(crmApprovalWorkflows.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmApprovalWorkflows).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertApprovalStep(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const workflowId = await findWorkflowId(tx, organizationId, values.workflow_code); + if (!workflowId) throw new Error(`Workflow ${values.workflow_code} not found.`); + const stepNumber = toInt(values.step_number); + const existing = await first( + tx + .select({ id: crmApprovalSteps.id }) + .from(crmApprovalSteps) + .where(and(eq(crmApprovalSteps.workflowId, workflowId), eq(crmApprovalSteps.stepNumber, stepNumber))) + .limit(1) + ); + const record = { + organizationId, + workflowId, + stepNumber, + roleCode: normalizeLower(values.role_code), + roleName: values.role_name, + approvalMode: normalizeLower(values.approval_mode || 'sequential'), + isRequired: toBool(values.is_required, true), + slaHours: toInt(values.sla_hours, 24), + calendarMode: normalizeLower(values.calendar_mode || 'calendar_days'), + timeoutAction: normalizeLower(values.timeout_action || 'none'), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmApprovalSteps).set(record).where(eq(crmApprovalSteps.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmApprovalSteps).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertApprovalMatrix(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const workflowId = await findWorkflowId(tx, organizationId, values.workflow_code); + if (!workflowId) throw new Error(`Workflow ${values.workflow_code} not found.`); + const branchId = await findOptionId(tx, organizationId, 'crm_branch', values.branch_code); + const productType = emptyToNull(values.product_type_code); + const existing = await first( + tx + .select({ id: crmApprovalMatrices.id }) + .from(crmApprovalMatrices) + .where( + and( + eq(crmApprovalMatrices.organizationId, organizationId), + eq(crmApprovalMatrices.entityType, normalizeLower(values.entity_type)), + eq(crmApprovalMatrices.workflowId, workflowId), + eq(crmApprovalMatrices.priority, toInt(values.priority, 100)) + ) + ) + .limit(1) + ); + const record = { + organizationId, + entityType: normalizeLower(values.entity_type), + workflowId, + branchId, + productType, + minAmount: values.min_amount ? toNumber(values.min_amount) : null, + maxAmount: values.max_amount ? toNumber(values.max_amount) : null, + currency: emptyToNull(values.currency), + priority: toInt(values.priority, 100), + isDefault: toBool(values.is_default), + isActive: toBool(values.is_active, true), + description: emptyToNull(values.description), + createdBy: context.actorId, + updatedBy: context.actorId, + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmApprovalMatrices).set(record).where(eq(crmApprovalMatrices.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmApprovalMatrices).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertCustomer(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const existing = await first( + tx.select({ id: crmCustomers.id, ownerUserId: crmCustomers.ownerUserId }).from(crmCustomers).where(and(eq(crmCustomers.organizationId, organizationId), eq(crmCustomers.code, values.customer_code))).limit(1) + ); + const ownerUserId = values.owner_user_email ? await findUserId(tx, values.owner_user_email) : null; + const branchId = await findOptionId(tx, organizationId, 'crm_branch', values.branch_code); + const record = { + organizationId, + branchId, + code: values.customer_code, + name: values.customer_name, + abbr: emptyToNull(values.abbr), + taxId: emptyToNull(values.tax_id), + customerType: normalizeLower(values.customer_type), + customerStatus: normalizeLower(values.customer_status), + address: emptyToNull(values.address), + province: emptyToNull(values.province), + district: emptyToNull(values.district), + subDistrict: emptyToNull(values.sub_district), + postalCode: emptyToNull(values.postal_code), + country: emptyToNull(values.country), + phone: emptyToNull(values.phone), + fax: emptyToNull(values.fax), + email: emptyToNull(values.email), + website: emptyToNull(values.website), + leadChannel: emptyToNull(values.lead_channel), + customerGroup: emptyToNull(values.customer_group), + customerSubGroup: emptyToNull(values.customer_sub_group), + ownerUserId, + ownerAssignedAt: ownerUserId ? new Date() : null, + ownerAssignedBy: ownerUserId ? context.actorId : null, + notes: emptyToNull(values.notes), + isActive: toBool(values.is_active, true), + createdBy: await resolveUserId(tx, values.created_by_email), + updatedBy: context.actorId, + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmCustomers).set(record).where(eq(crmCustomers.id, existing.id)); + if (ownerUserId && ownerUserId !== existing.ownerUserId) { + await tx.insert(crmCustomerOwnerHistory).values({ + id: randomUUID(), + organizationId, + customerId: existing.id, + oldOwnerUserId: existing.ownerUserId, + newOwnerUserId: ownerUserId, + changedBy: context.actorId, + remark: 'Setup CSV import owner update' + }); + } + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmCustomers).values({ id, ...record }); + if (ownerUserId) { + await tx.insert(crmCustomerOwnerHistory).values({ + id: randomUUID(), + organizationId, + customerId: id, + oldOwnerUserId: null, + newOwnerUserId: ownerUserId, + changedBy: context.actorId, + remark: 'Setup CSV import owner assignment' + }); + } + return { action: 'create', entityId: id }; +} + +async function upsertContact(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const customerId = await findCustomerId(tx, organizationId, values.customer_code); + if (!customerId) throw new Error(`Customer ${values.customer_code} not found.`); + const existingId = await findContactId(tx, organizationId, customerId, values); + const record = { + organizationId, + customerId, + name: values.contact_name, + position: emptyToNull(values.position), + department: emptyToNull(values.department), + phone: emptyToNull(values.phone), + mobile: emptyToNull(values.mobile), + email: emptyToNull(values.email), + isPrimary: toBool(values.is_primary), + notes: emptyToNull(values.notes), + isActive: toBool(values.is_active, true), + createdBy: await resolveUserId(tx, values.created_by_email), + updatedBy: context.actorId, + deletedAt: null, + updatedAt: new Date() + }; + if (existingId) { + await tx.update(crmCustomerContacts).set(record).where(eq(crmCustomerContacts.id, existingId)); + return { action: 'update', entityId: existingId }; + } + const id = randomUUID(); + await tx.insert(crmCustomerContacts).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertContactShare(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const customerId = await findCustomerId(tx, organizationId, values.customer_code); + const contactId = await findContactId(tx, organizationId, customerId, values); + if (!contactId) throw new Error(`Contact ${values.contact_email} not found.`); + const sharedToUserId = await resolveUserId(tx, values.shared_to_user_email); + const sharedByUserId = await resolveUserId(tx, values.shared_by_user_email); + const existing = await first( + tx + .select({ id: crmContactShares.id }) + .from(crmContactShares) + .where( + and( + eq(crmContactShares.organizationId, organizationId), + eq(crmContactShares.contactId, contactId), + eq(crmContactShares.sharedToUserId, sharedToUserId) + ) + ) + .limit(1) + ); + const record = { + organizationId, + contactId, + sharedToUserId, + sharedByUserId, + isActive: toBool(values.is_active, true), + remark: emptyToNull(values.remark), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmContactShares).set(record).where(eq(crmContactShares.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmContactShares).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertLead(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const existing = await first(tx.select({ id: crmLeads.id }).from(crmLeads).where(and(eq(crmLeads.organizationId, organizationId), eq(crmLeads.code, values.lead_code))).limit(1)); + const customerId = await findCustomerId(tx, organizationId, values.customer_code); + const contactId = await findContactId(tx, organizationId, customerId, values); + const ownerMarketingUserId = values.owner_marketing_user_email ? await findUserId(tx, values.owner_marketing_user_email) : null; + const assignedSalesOwnerId = values.assigned_sales_owner_email ? await findUserId(tx, values.assigned_sales_owner_email) : null; + const record = { + organizationId, + branchId: await findOptionId(tx, organizationId, 'crm_branch', values.branch_code), + code: values.lead_code, + customerId, + contactId, + description: emptyToNull(values.description), + projectName: emptyToNull(values.project_name), + projectLocation: emptyToNull(values.project_location), + leadChannel: emptyToNull(values.lead_channel), + productType: emptyToNull(values.product_type_code), + priority: emptyToNull(values.priority), + estimatedValue: values.estimated_value ? toNumber(values.estimated_value) : null, + awarenessId: await findOptionId(tx, organizationId, 'crm_lead_awareness', values.awareness_code), + status: normalizeLower(values.status), + followupStatus: emptyToNull(values.followup_status), + lostReason: emptyToNull(values.lost_reason), + outcome: normalizeLower(values.outcome || 'open'), + ownerMarketingUserId, + assignedSalesOwnerId, + assignedAt: assignedSalesOwnerId ? new Date() : null, + assignedBy: assignedSalesOwnerId ? context.actorId : null, + assignmentRemark: emptyToNull(values.assignment_remark), + createdBy: await resolveUserId(tx, values.created_by_email), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmLeads).set(record).where(eq(crmLeads.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmLeads).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertOpportunity(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const existing = await first( + tx.select({ id: crmOpportunities.id }).from(crmOpportunities).where(and(eq(crmOpportunities.organizationId, organizationId), eq(crmOpportunities.code, values.opportunity_code))).limit(1) + ); + const customerId = await findCustomerId(tx, organizationId, values.customer_code); + if (!customerId) throw new Error(`Customer ${values.customer_code} not found.`); + const assignedToUserId = values.assigned_to_user_email ? await findUserId(tx, values.assigned_to_user_email) : null; + const record = { + organizationId, + branchId: await findOptionId(tx, organizationId, 'crm_branch', values.branch_code), + leadId: await findLeadId(tx, organizationId, values.lead_code), + code: values.opportunity_code, + customerId, + contactId: await findContactId(tx, organizationId, customerId, values), + title: values.title, + description: emptyToNull(values.description), + requirement: emptyToNull(values.requirement), + projectName: emptyToNull(values.project_name), + projectLocation: emptyToNull(values.project_location), + productType: normalizeLower(values.product_type_code), + status: normalizeLower(values.status), + outcomeStatus: normalizeLower(values.outcome_status || 'open'), + priority: normalizeLower(values.priority), + leadChannel: emptyToNull(values.lead_channel), + estimatedValue: values.estimated_value ? toNumber(values.estimated_value) : null, + chancePercent: values.chance_percent ? toInt(values.chance_percent) : null, + expectedCloseDate: toDate(values.expected_close_date), + competitor: emptyToNull(values.competitor), + source: emptyToNull(values.source), + notes: emptyToNull(values.notes), + isHotProject: toBool(values.is_hot_project), + pipelineStage: normalizeLower(values.pipeline_stage || 'lead'), + assignedToUserId, + assignedAt: assignedToUserId ? new Date() : null, + assignedBy: assignedToUserId ? context.actorId : null, + createdBy: await resolveUserId(tx, values.created_by_email), + updatedBy: await resolveUserId(tx, values.updated_by_email), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmOpportunities).set(record).where(eq(crmOpportunities.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmOpportunities).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertOpportunityParty(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const opportunityId = await findOpportunityId(tx, organizationId, values.opportunity_code); + const customerId = await findCustomerId(tx, organizationId, values.customer_code); + if (!opportunityId || !customerId) throw new Error('Opportunity party references could not be resolved.'); + const role = normalizeLower(values.role_code); + const existing = await first( + tx + .select({ id: crmOpportunityCustomers.id }) + .from(crmOpportunityCustomers) + .where(and(eq(crmOpportunityCustomers.opportunityId, opportunityId), eq(crmOpportunityCustomers.customerId, customerId), eq(crmOpportunityCustomers.role, role))) + .limit(1) + ); + const record = { organizationId, opportunityId, customerId, role, remark: emptyToNull(values.remark), deletedAt: null, updatedAt: new Date() }; + if (existing) { + await tx.update(crmOpportunityCustomers).set(record).where(eq(crmOpportunityCustomers.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmOpportunityCustomers).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertQuotation(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const existing = await first(tx.select({ id: crmQuotations.id }).from(crmQuotations).where(and(eq(crmQuotations.organizationId, organizationId), eq(crmQuotations.code, values.quotation_code))).limit(1)); + const customerId = await findCustomerId(tx, organizationId, values.customer_code); + if (!customerId) throw new Error(`Customer ${values.customer_code} not found.`); + const record = { + organizationId, + branchId: await findOptionId(tx, organizationId, 'crm_branch', values.branch_code), + code: values.quotation_code, + opportunityId: await findOpportunityId(tx, organizationId, values.opportunity_code), + customerId, + contactId: await findContactId(tx, organizationId, customerId, values), + quotationDate: toDate(values.quotation_date) ?? new Date(), + validUntil: toDate(values.valid_until), + quotationType: normalizeLower(values.quotation_type), + projectName: emptyToNull(values.project_name), + projectLocation: emptyToNull(values.project_location), + attention: emptyToNull(values.attention), + reference: emptyToNull(values.reference), + notes: emptyToNull(values.notes), + status: normalizeLower(values.status), + revision: toInt(values.revision), + parentQuotationId: await findQuotationId(tx, organizationId, values.parent_quotation_code), + revisionRemark: emptyToNull(values.revision_remark), + currency: normalizeLower(values.currency).toUpperCase(), + exchangeRate: toNumber(values.exchange_rate, 1), + subtotal: toNumber(values.subtotal), + discount: toNumber(values.discount), + discountType: emptyToNull(values.discount_type), + taxRate: toNumber(values.tax_rate), + taxAmount: toNumber(values.tax_amount), + totalAmount: toNumber(values.total_amount), + chancePercent: values.chance_percent ? toInt(values.chance_percent) : null, + isHotProject: toBool(values.is_hot_project), + competitor: emptyToNull(values.competitor), + salesmanId: values.salesman_email ? await findUserId(tx, values.salesman_email) : null, + isSent: toBool(values.is_sent), + sentAt: values.sent_at ? new Date(values.sent_at) : null, + sentVia: emptyToNull(values.sent_via), + createdBy: await resolveUserId(tx, values.created_by_email), + updatedBy: await resolveUserId(tx, values.updated_by_email), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmQuotations).set(record).where(eq(crmQuotations.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmQuotations).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertQuotationItem(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const quotationId = await findQuotationId(tx, organizationId, values.quotation_code); + if (!quotationId) throw new Error(`Quotation ${values.quotation_code} not found.`); + const itemNumber = toInt(values.item_number); + const existing = await first( + tx.select({ id: crmQuotationItems.id }).from(crmQuotationItems).where(and(eq(crmQuotationItems.quotationId, quotationId), eq(crmQuotationItems.itemNumber, itemNumber))).limit(1) + ); + const record = { + organizationId, + quotationId, + itemNumber, + productType: normalizeLower(values.product_type_code), + description: values.description, + quantity: toNumber(values.quantity), + unit: emptyToNull(values.unit), + unitPrice: toNumber(values.unit_price), + discount: toNumber(values.discount), + discountType: emptyToNull(values.discount_type), + taxRate: toNumber(values.tax_rate), + totalPrice: toNumber(values.total_price), + notes: emptyToNull(values.notes), + sortOrder: toInt(values.sort_order, itemNumber), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmQuotationItems).set(record).where(eq(crmQuotationItems.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmQuotationItems).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertQuotationParty(tx: ImportTx, planRow: CsvCommitPlanRow): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const quotationId = await findQuotationId(tx, organizationId, values.quotation_code); + const customerId = await findCustomerId(tx, organizationId, values.customer_code); + if (!quotationId || !customerId) throw new Error('Quotation party references could not be resolved.'); + const role = normalizeLower(values.role_code); + const existing = await first( + tx + .select({ id: crmQuotationCustomers.id }) + .from(crmQuotationCustomers) + .where(and(eq(crmQuotationCustomers.quotationId, quotationId), eq(crmQuotationCustomers.customerId, customerId), eq(crmQuotationCustomers.role, role))) + .limit(1) + ); + const record = { + organizationId, + quotationId, + customerId, + role, + isPrimary: toBool(values.is_primary), + remark: emptyToNull(values.remark), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmQuotationCustomers).set(record).where(eq(crmQuotationCustomers.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmQuotationCustomers).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function upsertQuotationTopic(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const quotationId = await findQuotationId(tx, organizationId, values.quotation_code); + if (!quotationId) throw new Error(`Quotation ${values.quotation_code} not found.`); + const aliasKey = `${organizationId}|${values.quotation_code}|${values.topic_key}`.toLowerCase(); + const existingId = context.topicAliases.get(aliasKey); + const existing = + existingId ?? + ( + await first( + tx + .select({ id: crmQuotationTopics.id }) + .from(crmQuotationTopics) + .where( + and( + eq(crmQuotationTopics.organizationId, organizationId), + eq(crmQuotationTopics.quotationId, quotationId), + eq(crmQuotationTopics.title, values.title), + isNull(crmQuotationTopics.deletedAt) + ) + ) + .limit(1) + ) + )?.id; + const record = { + organizationId, + quotationId, + topicType: normalizeLower(values.topic_type), + title: values.title, + sortOrder: toInt(values.sort_order), + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmQuotationTopics).set(record).where(eq(crmQuotationTopics.id, existing)); + context.topicAliases.set(aliasKey, existing); + return { action: 'update', entityId: existing }; + } + const id = randomUUID(); + await tx.insert(crmQuotationTopics).values({ id, ...record }); + context.topicAliases.set(aliasKey, id); + return { action: 'create', entityId: id }; +} + +async function upsertQuotationTopicItem(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + const values = planRow.row.values; + const organizationId = await resolveOrganizationId(tx, values); + const aliasKey = `${organizationId}|${values.quotation_code}|${values.topic_key}`.toLowerCase(); + const topicId = context.topicAliases.get(aliasKey); + if (!topicId) throw new Error(`Quotation topic ${values.topic_key} not found in committed batch.`); + const sortOrder = toInt(values.sort_order); + const existing = await first( + tx + .select({ id: crmQuotationTopicItems.id }) + .from(crmQuotationTopicItems) + .where( + and( + eq(crmQuotationTopicItems.organizationId, organizationId), + eq(crmQuotationTopicItems.topicId, topicId), + eq(crmQuotationTopicItems.sortOrder, sortOrder), + isNull(crmQuotationTopicItems.deletedAt) + ) + ) + .limit(1) + ); + const record = { + organizationId, + topicId, + content: values.content, + sortOrder, + deletedAt: null, + updatedAt: new Date() + }; + if (existing) { + await tx.update(crmQuotationTopicItems).set(record).where(eq(crmQuotationTopicItems.id, existing.id)); + return { action: 'update', entityId: existing.id }; + } + const id = randomUUID(); + await tx.insert(crmQuotationTopicItems).values({ id, ...record }); + return { action: 'create', entityId: id }; +} + +async function executeRow(tx: ImportTx, planRow: CsvCommitPlanRow, context: ExecuteContext): Promise { + switch (planRow.fileName) { + case 'users.csv': + return upsertUser(tx, planRow); + case 'organizations.csv': + return upsertOrganization(tx, planRow); + case 'memberships.csv': + return upsertMembership(tx, planRow); + case 'crm-role-assignments.csv': + return upsertCrmRoleAssignment(tx, planRow, context); + case 'master-options.csv': + return upsertMasterOption(tx, planRow); + case 'branches.csv': + return upsertMasterOption(tx, planRow, 'crm_branch'); + case 'product-types.csv': + return upsertMasterOption(tx, planRow, 'crm_product_type'); + case 'document-sequences.csv': + return upsertDocumentSequence(tx, planRow); + case 'approval-workflows.csv': + return upsertApprovalWorkflow(tx, planRow, context); + case 'approval-steps.csv': + return upsertApprovalStep(tx, planRow); + case 'approval-matrix.csv': + return upsertApprovalMatrix(tx, planRow, context); + case 'customers.csv': + return upsertCustomer(tx, planRow, context); + case 'contacts.csv': + return upsertContact(tx, planRow, context); + case 'contact-shares.csv': + return upsertContactShare(tx, planRow); + case 'leads.csv': + return upsertLead(tx, planRow, context); + case 'opportunities.csv': + return upsertOpportunity(tx, planRow, context); + case 'opportunity-parties.csv': + return upsertOpportunityParty(tx, planRow); + case 'quotations.csv': + return upsertQuotation(tx, planRow); + case 'quotation-items.csv': + return upsertQuotationItem(tx, planRow); + case 'quotation-parties.csv': + return upsertQuotationParty(tx, planRow); + case 'quotation-topics.csv': + return upsertQuotationTopic(tx, planRow, context); + case 'quotation-topic-items.csv': + return upsertQuotationTopicItem(tx, planRow, context); + default: + throw new Error(`Template ${planRow.fileName} is not supported by commit executor.`); + } +} + +function auditGapWarning(file: CsvCommitPlanFile) { + return createImportWarning({ + file: file.fileName, + row: 1, + column: '__file__', + code: 'AUDIT_GAP', + message: 'Setup CSV import writes directly with Drizzle because a safe setup import service contract is not available for this template.', + suggestedFix: 'Add a dedicated setup import service wrapper if audit semantics must be captured per imported row.' + }); +} + +export async function executeCsvCommitPlan(plan: CsvCommitPlan, input: { actorId: string; dryRun: boolean }): Promise { + const context: ExecuteContext = { + actorId: input.actorId, + dryRun: input.dryRun, + topicAliases: new Map() + }; + + if (input.dryRun) { + return plan.files.map((file) => + summarizeFileRows({ + fileName: file.fileName, + template: file.template, + warnings: [auditGapWarning(file)], + rows: file.rows.map((planRow) => ({ + row: planRow.row.row, + action: planRow.row.action === 'create' ? 'create' : planRow.row.action === 'update' ? 'update' : 'skip', + businessKey: businessKey(planRow), + message: `Dry run planned ${planRow.row.action}.`, + warnings: planRow.row.warnings, + errors: [] + })) + }) + ); + } + + return db.transaction(async (transaction) => { + const tx = transaction as unknown as ImportTx; + const reports: ImportFileReport[] = []; + for (const file of plan.files) { + const rows: ImportRowReport[] = []; + for (const planRow of file.rows) { + try { + const result = await executeRow(tx, planRow, context); + rows.push({ + row: planRow.row.row, + action: result.action, + businessKey: businessKey(planRow), + entityId: result.entityId, + message: `Row ${result.action} completed.`, + warnings: [...planRow.row.warnings, ...(result.warnings ?? [])], + errors: [] + }); + } catch (error) { + const failedRows = [ + ...rows, + { + row: planRow.row.row, + action: 'error' as const, + businessKey: businessKey(planRow), + message: 'Row import failed.', + warnings: planRow.row.warnings, + errors: [ + createImportError({ + file: file.fileName, + row: planRow.row.row, + column: '__row__', + code: 'UPSERT_FAILED', + message: error instanceof Error ? error.message : 'Row could not be imported.', + suggestedFix: 'Review the row and retry after fixing related references or conflicting data.' + }) + ] + } + ]; + rows.push({ + row: planRow.row.row, + action: 'error', + businessKey: businessKey(planRow), + message: 'Row import failed.', + warnings: planRow.row.warnings, + errors: [ + createImportError({ + file: file.fileName, + row: planRow.row.row, + column: '__row__', + code: 'UPSERT_FAILED', + message: error instanceof Error ? error.message : 'Row could not be imported.', + suggestedFix: 'Review the row and retry after fixing related references or conflicting data.' + }) + ] + }); + throw new CsvCommitExecutionError( + `Setup CSV commit failed for ${file.fileName} row ${planRow.row.row}.`, + [ + ...reports, + summarizeFileRows({ + fileName: file.fileName, + template: file.template, + warnings: [auditGapWarning(file)], + rows: failedRows + }) + ], + error + ); + } + } + reports.push( + summarizeFileRows({ + fileName: file.fileName, + template: file.template, + warnings: [auditGapWarning(file)], + rows + }) + ); + } + return reports; + }); +}