# Data Import — Implementation Plan

**Spec:** docs/specs/2026-05-31-data-import.md  ·  **Slug:** data-import  ·  **Wave:** 7
**Depends on:** customers-module, foundation-auth-rbac, foundation-design-system, invoices-core, system-communications-notifications, zync-subscription

## Goal
Let tenants bulk-import Customers, Invoices, Products/Services, and Time Entries from CSV/XLSX files at `/settings/import` (OWNER/ADMIN, Business+ tier). Files upload to R2 via presigned URL, are enqueued to a Cloudflare Queue (`IMPORT_QUEUE`), and process row-by-row with **partial success** semantics — each row commits in its own transaction, bad rows are recorded and downloadable as an error CSV. Progress and history surface in the UI; an `import_completed` notification fires on finish.

## Architecture
- **Two new tables** (`import_jobs`, `import_job_results`) added to `packages/db` (Drizzle). `import_jobs.tenant_id` → `tenants(id)`; `import_jobs.created_by` → `users(id)`; `import_job_results.import_job_id` → `import_jobs(id)`.
- **API (Hono, apps/zync-api)**: routes under `/api/imports/*`. Upload presign → R2 (existing `STORAGE` bucket), job create (header parse + row count), preview (sync first-5 validation), start (enqueue to `IMPORT_QUEUE`), poll, list, error CSV download, and static template CSV download. All gated by `requirePermission('settings:import:write')` + `requireTier('business')`.
- **Queue consumer (apps/zync-api worker)**: `import.process` handler streams the file from R2, applies `column_mapping`, validates each row per type, and commits each row/invoice-group in an independent Neon transaction. Writes per-row outcomes to `import_job_results`, increments counters on `import_jobs`, then sets `completed`/`failed` and calls `createNotification` with `type='import_completed'`.
- **Upstream consumed**: `customers` (INSERT with `ON CONFLICT (tenant_id, email) DO NOTHING`, `address` JSONB), `invoices` + `invoice_lines` + `invoice_sequences` (`nextInvoiceNumber`), `products` (`unit_price`/`currency`/`unit`/`is_active`/`tax_rate_id`, matched against `vat_rates`), `time_entries` (`started_at`/`stopped_at`/`duration_seconds`/`project_id`/`user_id`/`billable`), `projects` (name match), `tasks` (title match), `tenant_memberships`/`users` (member email match), `audit_log` (in-transaction audit), `createNotification`, `requireTier`/`meetsMinimumTier`/`requirePermission`, `useTierGate`/`useUpgradeModal`, design-system `Button`/`Card`/`DataTable`/`Badge`/`Dialog`/`EmptyState`/`Progress`/`Select`/`toast`.
- **Frontend (apps/zync-app, Vite+React)**: `/settings/import` page = tier-gated locked state (freelancer) OR history table + a 5-stage wizard (Select type → Upload → Map columns → Preview → Processing/Results). Column auto-detect runs client-side against a static alias table; mapping is applied server-side.

## Tech Stack
- **apps/zync-api** — Hono routes + queue consumer; `papaparse` (CSV) and `xlsx` (XLSX, customers only) in the Worker bundle; AWS SigV4 presign for R2 (`@aws-sdk/s3-request-presigner` + `@aws-sdk/client-s3`, or Worker-native presign helper) against the `STORAGE` R2 bucket.
- **packages/db** — Drizzle table definitions + migration.
- **packages/types** — shared import enums/types; extend `NotificationType` union with `'import_completed'`.
- **apps/zync-app** — React wizard + history list; `@tanstack/react-query` polling.
- **Cloudflare bindings**: `STORAGE` (existing R2), `IMPORT_QUEUE` (new Queue producer+consumer), Hyperdrive→Neon (`DB`).

## Wave Plan
| Sub-wave | Tasks | Files touched | Parallelizable? |
|----------|-------|---------------|-----------------|
| A — Schema & types | 1, 2 | packages/db, packages/types | Task 2 after 1 |
| B — Infra & shared logic | 3, 4, 5 | wrangler.toml, packages/types (aliases/validators), apps/zync-api lib | 3 parallel; 4,5 parallel |
| C — API routes | 6, 7, 8, 9, 10, 11, 12 | apps/zync-api routes | After A+B; 6→7 ordering, rest parallel |
| D — Queue worker | 13 | apps/zync-api queue consumer | After A, B, and 6–9 contracts |
| E — Frontend | 14, 15, 16, 17, 18, 19 | apps/zync-app | After C contracts; 14→15..19 |
| F — Notification & cron hook | 20, 21 | packages/types, cron registration | After 13 |

## Tasks

### Task 1: Database schema — `import_jobs` & `import_job_results`
**Blocks:** 2, 6, 13  ·  **Blocked by:** —
**Files:**
- Create: `packages/db/src/schema/imports.ts`
- Modify: `packages/db/src/schema/index.ts` (export new schema)
- Create: `packages/db/migrations/<timestamp>_data_import.sql`
**Steps:**
- [ ] Define both tables in Drizzle matching the DDL below (UUID PKs, UUID FKs, TIMESTAMPTZ, JSONB, BOOLEAN-free counters as INT).
- [ ] Add the two listed indexes and the partial error index.
- [ ] Export `importJobs` and `importJobResults` from the db package barrel.
- [ ] Generate the SQL migration and confirm it applies cleanly to the Neon dev branch.
**Schema / Interfaces:**
```sql
CREATE TABLE import_jobs (
  id                UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id         UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
  created_by        UUID NOT NULL REFERENCES users(id),
  type              TEXT NOT NULL CHECK (type IN ('customers','invoices','products','time_entries','bulk_action')),
  status            TEXT NOT NULL DEFAULT 'pending'
                      CHECK (status IN ('pending','processing','completed','failed')),
  original_filename TEXT NOT NULL,
  r2_key            TEXT NOT NULL,
  column_mapping    JSONB,
  total_rows        INT,
  rows_processed    INT NOT NULL DEFAULT 0,
  success_count     INT NOT NULL DEFAULT 0,
  skipped_count     INT NOT NULL DEFAULT 0,
  error_count       INT NOT NULL DEFAULT 0,
  error_message     TEXT,
  created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
  started_at        TIMESTAMPTZ,
  completed_at      TIMESTAMPTZ
);
CREATE INDEX import_jobs_tenant_created ON import_jobs (tenant_id, created_at DESC);

CREATE TABLE import_job_results (
  id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  import_job_id   UUID NOT NULL REFERENCES import_jobs(id) ON DELETE CASCADE,
  tenant_id       UUID NOT NULL,
  row_number      INT NOT NULL,
  status          TEXT NOT NULL CHECK (status IN ('success','skipped','error')),
  message         TEXT,
  original_data   TEXT,
  entity_id       UUID,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX import_job_results_job ON import_job_results (import_job_id, row_number);
CREATE INDEX import_job_results_errors ON import_job_results (import_job_id)
  WHERE status IN ('skipped','error');
```
**Acceptance:**
- [ ] Migration applies on Neon dev branch; both tables + 3 indexes exist.
- [ ] FK cascade: deleting an `import_jobs` row deletes its `import_job_results` rows.

### Task 2: Shared types in `@zync/types`
**Blocks:** 6, 13, 14  ·  **Blocked by:** 1
**Files:**
- Create: `packages/types/src/imports.ts`
- Modify: `packages/types/src/index.ts` (re-export)
- Modify: `packages/types/src/notifications.ts` (add `'import_completed'` to `NotificationType`)
**Steps:**
- [ ] Define `ImportType`, `ImportStatus`, `ImportRowStatus`, `ColumnMapping`, `ImportJob`, `ImportJobResult`, `ImportPreviewRow`, `ImportPreviewResponse`.
- [ ] Add `'import_completed'` to the canonical `NotificationType` union.
- [ ] Export all from the package barrel.
**Schema / Interfaces:**
```ts
export type ImportType = 'customers' | 'invoices' | 'products' | 'time_entries';
export type ImportStatus = 'pending' | 'processing' | 'completed' | 'failed';
export type ImportRowStatus = 'success' | 'skipped' | 'error';
export type ColumnMapping = Record<string, string | null>; // { "File Col": "zync_field" | null }

export interface ImportJob {
  id: string; type: ImportType; status: ImportStatus;
  original_filename: string; total_rows: number | null;
  rows_processed: number; success_count: number; skipped_count: number; error_count: number;
  error_message: string | null; created_at: string; completed_at: string | null;
}
export interface ImportPreviewRow {
  row: number; status: 'valid' | 'warning' | 'error';
  data: Record<string, string>; message?: string;
}
export interface ImportPreviewResponse { rows: ImportPreviewRow[]; total_rows: number; }
```
**Acceptance:**
- [ ] `@zync/types` compiles; `NotificationType` includes `'import_completed'`.

### Task 3: Cloudflare bindings — `IMPORT_QUEUE`
**Blocks:** 12, 13  ·  **Blocked by:** —
**Files:**
- Modify: `apps/zync-api/wrangler.toml`
- Modify: `apps/zync-api/src/env.ts` (or `Env` type) to add `IMPORT_QUEUE: Queue`
**Steps:**
- [ ] Add `[[queues.producers]]` `binding = "IMPORT_QUEUE"`, `queue = "import-process"`.
- [ ] Add `[[queues.consumers]]` `queue = "import-process"` with `max_batch_size = 1`, `max_retries = 0` (job is idempotent; do not retry whole job — partial results already committed), `max_batch_timeout = 5`.
- [ ] Add `IMPORT_QUEUE: Queue<ImportJobMessage>` to the `Env` interface; confirm `STORAGE` (R2) already present.
**Acceptance:**
- [ ] `wrangler types` / typecheck recognizes `IMPORT_QUEUE` and `STORAGE` on `Env`.

### Task 4: Column alias table + auto-detect (shared, client+server)
**Blocks:** 13, 16  ·  **Blocked by:** —
**Files:**
- Create: `packages/types/src/import-aliases.ts`
**Steps:**
- [ ] Implement `normalizeHeader(s)`: lowercase, strip all non-alphanumeric (Latin + Hebrew aware), collapse.
- [ ] Define `CUSTOMER_ALIASES` and `INVOICE_ALIASES` maps (Zync field → recognized names, including Hebrew) verbatim from the spec alias tables.
- [ ] Implement `autoDetectMapping(type, columns: string[]): ColumnMapping` returning best match per file column or `null`.
- [ ] Define `ZYNC_FIELDS` per type (the selectable target fields list for the mapping UI), and `REQUIRED_FIELDS` per type.
**Schema / Interfaces:**
```ts
export const CUSTOMER_ALIASES: Record<string, string[]> = {
  name: ['name','full name','fullname','customer name','client name','שם','שם לקוח'],
  email: ['email','e-mail','email address','אימייל','דואל','דוא"ל'],
  phone: ['phone','phone number','mobile','tel','טלפון','נייד'],
  company: ['company','company name','business','organization','חברה','שם חברה'],
  address_street: ['street','address','street address','רחוב','כתובת'],
  address_city: ['city','עיר'],
  address_state: ['state','region','מחוז'],
  address_zip: ['zip','zip code','postal code','מיקוד'],
  address_country: ['country','ארץ','מדינה'],
  notes: ['notes','comments','remarks','הערות'],
};
export const INVOICE_ALIASES: Record<string, string[]> = {
  customer_name: ['customer','customer name','client','client name','שם לקוח'],
  customer_email: ['customer email','client email','אימייל לקוח'],
  invoice_number: ['invoice number','invoice #','invoice id','מספר חשבונית'],
  issue_date: ['issue date','date','invoice date','תאריך הנפקה','תאריך'],
  due_date: ['due date','payment date','תאריך פירעון'],
  status: ['status','סטטוס'],
  line_description: ['description','item','item description','תיאור','פריט'],
  line_qty: ['qty','quantity','כמות'],
  line_unit_price: ['unit price','price','rate','מחיר יחידה','מחיר'],
  line_tax_rate: ['tax rate','vat','vat rate','מע"מ','שיעור מס'],
};
export const REQUIRED_FIELDS: Record<ImportType, string[][]> = {
  customers: [['name']],            // each inner array = an OR-group, all groups must be satisfied
  invoices: [['customer_name','customer_email']],
  products: [['name'],['price']],
  time_entries: [['date'],['user_email']],
};
```
**Acceptance:**
- [ ] `autoDetectMapping('customers', ['Full Name','Email Address','Mobile','Ref Code'])` maps the first three, leaves `Ref Code` `null`.
- [ ] Hebrew header `שם לקוח` auto-maps to `name` (customers) / `customer_name` (invoices).

### Task 5: Per-type row validators (shared logic in api lib)
**Blocks:** 8, 13  ·  **Blocked by:** —
**Files:**
- Create: `apps/zync-api/src/lib/imports/validators.ts`
- Create: `apps/zync-api/src/lib/imports/parse.ts`
**Steps:**
- [ ] `parse.ts`: `parseCsvHeader(buf)`, `countCsvDataRows(buf)`, `parseCsvRows(buf)` (papaparse), `parseXlsxRows(buf)` (xlsx, sheet 1, first non-empty row = header). Empty (all-blank) rows silently skipped; cells beyond mapped columns truncated.
- [ ] `validators.ts`: pure functions `validateCustomerRow`, `validateInvoiceLineRow`, `validateProductRow`, `validateTimeEntryRow` returning `{ status: 'valid'|'warning'|'error', data, message? }`. Duplicate checks (email, sku) and FK lookups (project/task/member) are passed in as resolver callbacks so the same code runs in preview and worker.
- [ ] Implement the validation rules below verbatim per type.
- [ ] RFC 5322 simplified email regex; `parseDate` accepting `YYYY-MM-DD` and `DD/MM/YYYY`; `parseBoolean` (`true`/`false`/`1`/`0`, default per field); tax-rate normalization (`0..1` kept; integer `>1` divided by 100).
**Schema / Interfaces:**
```ts
// Customers: name required (non-empty trim) else error "Name is required".
//   email optional; if present must match regex else error "Invalid email format";
//   duplicate (tenant_id,email) → warning "Duplicate email: {email}" (skip).
//   notes truncated to 5000 chars silently.
// Invoices (per line): customer_name OR customer_email required else error "Customer required".
//   customer_email if present must be valid. issue_date if present must parse (YYYY-MM-DD) else error;
//   absent → import date. due_date if present must parse and not before issue_date.
//   status in draft|sent|paid|cancelled (ci) else 'draft'. line_unit_price required numeric
//   (optional leading '-') else error. line_qty default 1, must be >0 if present.
//   line_tax_rate: 0..1 or integer% (→ /100); default tenant default vat rate.
//   line_description truncated to 500 chars.
// Products: name required; price required numeric non-negative else error; currency valid ISO-4217
//   (default ILS) else error; unit free text max 50; tax_rate integer 0..100 (default 17)
//   else error; sku max 100, duplicate (tenant_id,sku) → warning "Duplicate SKU: {sku}" (skip);
//   active true/false/1/0 (default true).
// Time entries: date required parse, future > 1y → error; start_time/end_time HH:MM 24h,
//   end after start else error; duration_minutes required positive int if start+end not both given
//   (ignored if both given); project_name must match existing project (ci exact) else
//   error "Project not found: {name}"; task_name requires project_name, matched by name within
//   project else error "Task not found: {name} in project {project}"; user_email must match a tenant
//   member else error "User not found: {email}"; description truncated 1000; billable default true.
```
**Acceptance:**
- [ ] Unit-level: a customer row with empty name → `error` "Name is required"; valid duplicate email → `warning`.
- [ ] Invoice tax-rate `18` normalizes to `0.18`; `0.18` stays `0.18`.
- [ ] Time entry with only `duration_minutes` (no times) is valid; with `end_time` before `start_time` → error.

### Task 6: `POST /api/imports/upload` — presigned R2 PUT
**Blocks:** 14  ·  **Blocked by:** 1, 3
**Files:**
- Create: `apps/zync-api/src/routes/imports/index.ts` (router mount)
- Create: `apps/zync-api/src/routes/imports/upload.ts`
- Create: `apps/zync-api/src/lib/imports/r2-key.ts`
**Steps:**
- [ ] Mount router at `/api/imports` behind `authMiddleware`, `requirePermission('settings:import:write')`, `requireTier('business')`.
- [ ] Zod-validate body `{ filename, content_type, size_bytes }` (require-zod-validation-in-routes).
- [ ] Reject `size_bytes > 10*1024*1024` → `413 "FILE_TOO_LARGE"`; reject `content_type` not in `['text/csv','application/vnd.openxmlformats-officedocument.spreadsheetml.sheet']` → `415`.
- [ ] `buildR2Key(tenantId, filename)`: `{tenantId}/imports/{crypto.randomUUID()}-{sanitize(filename)}` where sanitize = lowercase, spaces→`-`, strip non-alphanumeric except `.` and `-`.
- [ ] Generate a SigV4 presigned PUT URL (5 min) for `STORAGE` and return `{ upload_url, r2_key, expires_in: 300 }`.
**Schema / Interfaces:**
```ts
// Request: { filename: string; content_type: string; size_bytes: number }
// Response 200: { upload_url: string; r2_key: string; expires_in: 300 }
```
**Acceptance:**
- [ ] Oversized request → 413; bad content-type → 415; valid → 200 with a PUT URL resolving to the `STORAGE` bucket and the canonical key shape.
- [ ] Freelancer tenant → 403 from `requireTier`.

### Task 7: `POST /api/imports` — create job (header parse + row count)
**Blocks:** 16  ·  **Blocked by:** 6, 5
**Files:**
- Create: `apps/zync-api/src/routes/imports/create.ts`
**Steps:**
- [ ] Zod body `{ type: ImportType, r2_key, original_filename }`; assert `r2_key` is prefixed with the caller's `{tenantId}/imports/` (defense against cross-tenant key injection).
- [ ] `STORAGE.get(r2_key)`; if missing/unparseable → `422 PARSE_ERROR`.
- [ ] Parse header row only + count data rows (`parse.ts`). If invoices and file is XLSX → `415` (XLSX not supported for invoices in v1). If data rows > 5000 → `422 ROW_LIMIT_EXCEEDED`.
- [ ] INSERT `import_jobs` (status `pending`, `total_rows`, `created_by = session.user_id`, `tenant_id`); write `audit_log` in the same transaction (require-audit-in-transaction).
- [ ] Return `201 { id, status:'pending', columns: string[] }`.
**Schema / Interfaces:**
```ts
// Request: { type: ImportType; r2_key: string; original_filename: string }
// Response 201: { id: string; status: 'pending'; columns: string[] }
// Errors: 422 ROW_LIMIT_EXCEEDED | 422 PARSE_ERROR | 415 (xlsx+invoices)
```
**Acceptance:**
- [ ] 6001-row file → 422 `ROW_LIMIT_EXCEEDED`; corrupt file → 422 `PARSE_ERROR`.
- [ ] Valid CSV → 201 with `columns` = raw header names; `import_jobs` row created with `total_rows`.

### Task 8: `POST /api/imports/:id/preview` — sync first-5 validation
**Blocks:** 17  ·  **Blocked by:** 5, 7
**Files:**
- Create: `apps/zync-api/src/routes/imports/preview.ts`
**Steps:**
- [ ] Load job (tenant-scoped via `tenantQuery`); 404 if not owned.
- [ ] Zod body `{ mapping: ColumnMapping }`.
- [ ] Read first 5 data rows from R2, apply mapping → typed objects, run the matching validator with live resolver callbacks (duplicate-email lookup against `customers`, sku lookup against `products`, project/task/member lookups for time entries).
- [ ] Return `{ rows: ImportPreviewRow[], total_rows }`. Status per row: `valid`/`warning`/`error` with `message`.
**Schema / Interfaces:**
```ts
// Request: { mapping: ColumnMapping }   (null value = ignore column)
// Response 200: { rows: { row, status:'valid'|'warning'|'error', data, message? }[], total_rows }
```
**Acceptance:**
- [ ] A file whose row 5 has blank mapped `name` returns that row with `status:'error'`, `message:'Name is required'`.
- [ ] Duplicate email in first 5 rows returns `status:'warning'`, `message:'Duplicate email: {email}'`.

### Task 9: `POST /api/imports/:id/start` — lock mapping + enqueue
**Blocks:** 17, 13  ·  **Blocked by:** 7, 3
**Files:**
- Create: `apps/zync-api/src/routes/imports/start.ts`
**Steps:**
- [ ] Load tenant-scoped job; if status ≠ `pending` → `409`.
- [ ] Zod body `{ mapping: ColumnMapping }`. Validate required fields are mapped per `REQUIRED_FIELDS[type]` (each OR-group satisfied) else `422 REQUIRED_FIELD_UNMAPPED`. Reject duplicate target mappings except `notes` (concatenation allowed).
- [ ] UPDATE `import_jobs` SET `column_mapping`, `status='processing'`, `started_at=now()`; write `audit_log` in-transaction.
- [ ] `IMPORT_QUEUE.send({ import_job_id, tenant_id, type, r2_key, column_mapping })`.
- [ ] Return `202 { id, status:'processing' }`.
**Schema / Interfaces:**
```ts
// Request: { mapping: ColumnMapping }
// Response 202: { id, status:'processing' }
// Errors: 409 (already started/completed) | 422 REQUIRED_FIELD_UNMAPPED
export interface ImportJobMessage {
  import_job_id: string; tenant_id: string; type: ImportType;
  r2_key: string; column_mapping: ColumnMapping;
}
```
**Acceptance:**
- [ ] Missing required mapping → 422 `REQUIRED_FIELD_UNMAPPED`; second call on a `processing` job → 409.
- [ ] Success enqueues exactly one `IMPORT_QUEUE` message and flips status to `processing`.

### Task 10: `GET /api/imports/:id` & `GET /api/imports` — poll + list
**Blocks:** 15, 18  ·  **Blocked by:** 7
**Files:**
- Create: `apps/zync-api/src/routes/imports/get.ts`
- Create: `apps/zync-api/src/routes/imports/list.ts`
**Steps:**
- [ ] `GET /:id`: tenant-scoped; return the progress shape (id, type, status, original_filename, total_rows, rows_processed, success_count, skipped_count, error_count, created_at, completed_at, error_message). 404 if not owned.
- [ ] `GET /`: cursor pagination via `encodeCursor`/`decodeCursor`, `clampLimit(limit, 20, 100)`, order `created_at DESC`. Return `{ jobs, next_cursor }` using `buildPaginated`.
**Schema / Interfaces:**
```ts
// GET /api/imports/:id 200 -> ImportJob + rows_processed/success/skipped/error counts (full progress)
// GET /api/imports?cursor&limit 200 -> { jobs: ImportJob[]; next_cursor: string | null }
```
**Acceptance:**
- [ ] Poll returns live counters while `processing`; `next_cursor` paginates at 20/page, max 100.
- [ ] Cross-tenant job id → 404.

### Task 11: `GET /api/imports/:id/errors.csv` — error report download
**Blocks:** 18  ·  **Blocked by:** 10
**Files:**
- Create: `apps/zync-api/src/routes/imports/errors-csv.ts`
**Steps:**
- [ ] Tenant-scoped job load; require `status='completed'` AND `error_count + skipped_count > 0` else `404`/`409`.
- [ ] SELECT `import_job_results` WHERE `status IN ('skipped','error')` ORDER BY `row_number`.
- [ ] Stream CSV `row_number,original_data,status,message` (CSV-escape `original_data` and `message`).
- [ ] Headers: `Content-Type: text/csv`, `Content-Disposition: attachment; filename="import-errors-{id}.csv"`.
**Acceptance:**
- [ ] Completed job with errors returns a CSV whose rows match the failed/skipped `import_job_results`; a clean job returns 404.

### Task 12: `GET /api/imports/templates/:type` — static template CSVs
**Blocks:** 16  ·  **Blocked by:** —
**Files:**
- Create: `apps/zync-api/src/routes/imports/templates.ts`
- Create: `apps/zync-api/src/lib/imports/templates.ts` (inlined CSV string constants)
**Steps:**
- [ ] `type` ∈ `customers|invoices|products|time-entries` else 404.
- [ ] Serve the inlined template CSV (verbatim content below) as `Content-Type: text/csv`, `Content-Disposition: attachment; filename="{type}-template.csv"`, `Cache-Control: public, max-age=3600`.
- [ ] This route is public-cacheable; still mount behind `authMiddleware` (templates are not tenant-specific but the page is gated).
**Schema / Interfaces:**
```
# customers-template.csv
name,email,phone,company,address_street,address_city,address_zip,address_country,notes
Rivka Cohen,rivka@example.com,050-1234567,Cohen Ltd,הרצל 1,תל אביב,6120101,IL,לקוח VIP

# invoices-template.csv
customer_name,customer_email,invoice_number,issue_date,due_date,status,line_description,line_qty,line_unit_price,line_tax_rate
Cohen Ltd,rivka@example.com,INV-001,2026-01-15,2026-02-15,draft,ייעוץ עסקי,2,500,0.18
Cohen Ltd,rivka@example.com,INV-001,,,,תוכנה,1,1200,0.18

# products-template.csv
name,description,price,currency,unit,tax_rate,sku,active
Web Design,Full website design and build,3500,ILS,item,17,SKU-001,true
Hourly Consulting,Strategy consulting,350,ILS,hour,17,SKU-002,true

# time-entries-template.csv
date,start_time,end_time,duration_minutes,project_name,task_name,user_email,description,billable
2026-05-31,09:00,11:30,,Project Alpha,API integration,dev@example.com,Backend work,true
2026-05-31,,,90,Project Beta,,pm@example.com,Planning session,false
```
**Acceptance:**
- [ ] Each of the 4 types downloads its exact template with the `attachment` disposition and 1-hour cache header.

### Task 13: Queue consumer — `import.process` worker
**Blocks:** 20  ·  **Blocked by:** 1, 2, 3, 4, 5, 9
**Files:**
- Create: `apps/zync-api/src/queues/import-process.ts`
- Modify: `apps/zync-api/src/index.ts` (wire `queue(batch, env)` handler dispatch on `IMPORT_QUEUE`)
**Steps:**
- [ ] `processImport(message, env)`: fetch job; if `status ≠ 'processing'` → ack + return (idempotent).
- [ ] Stream file via `STORAGE.get(r2_key)`; parse CSV (or XLSX for customers). On parse failure → `import_jobs.status='failed'`, `error_message`, fire `import_completed` notification, ack, return.
- [ ] Apply `column_mapping` to each row → typed object (multiple file cols → same `notes` joined with `'; '`).
- [ ] **Customers/Products/Time entries:** per data row, open an independent Neon transaction:
  - valid → INSERT (see per-type inserts below) + `audit_log` + `import_job_results(status='success', entity_id)`; `rows_processed+1`, `success_count+1`.
  - warning/skip → `import_job_results(status='skipped', message, original_data)`; `rows_processed+1`, `skipped_count+1`.
  - error → `import_job_results(status='error', message, original_data)`; `rows_processed+1`, `error_count+1`.
  - Each row commits independently; a failing row never rolls back prior rows.
- [ ] **Invoices:** read all rows into memory, group by `invoice_number` + customer key; sort by original row number; accumulate consecutive same-`invoice_number` rows (empty `invoice_number` continuation = previous invoice's line). Commit each group atomically (`invoices` + `invoice_lines`). If any line in the group fails validation → one `error` result for the whole group. `rows_processed` advances per CSV row; one `import_job_results` row per invoice group.
  - Unknown customer → create minimal `customers` row (`name`, `email` if present) in the same group transaction and link via `customer_id`.
  - Missing `invoice_number` → assign via `nextInvoiceNumber(db, tenantId, 'invoice')` against `invoice_sequences`.
- [ ] On completion: `import_jobs.status='completed'`, `completed_at=now()`; call `createNotification` `type='import_completed'`. Ack.
- [ ] On unhandled exception: `import_jobs.status='failed'`, `error_message=<message>`; ack (do not retry whole job).
**Schema / Interfaces:**
```ts
// Per-type inserts (canonical upstream columns):
// customers:
//   INSERT INTO customers (id, tenant_id, name, email, phone, company, address, notes, status)
//     VALUES (gen_random_uuid(), $tenant, $name, $email, $phone, $company,
//             jsonb_build_object('street',$street,'city',$city,'state',$state,'zip',$zip,'country',$country),
//             $notes, 'active')
//     ON CONFLICT (tenant_id, email) DO NOTHING RETURNING id;
//   -- no row returned + email present => skipped 'Duplicate email: {email}'.
// products:
//   resolve tax_rate% -> tax_rate_id via vat_rates (match rate); currency default 'ILS';
//   unit default 'item'->stored as given; active -> is_active.
//   INSERT INTO products (id, tenant_id, name, description, unit_price, currency, unit, tax_rate_id, is_active);
//   duplicate sku (tenant_id, sku) -> skipped 'Duplicate SKU: {sku}'.
// time_entries:
//   resolve project_id (projects.name ci), task_id (tasks.title within project), user_id (tenant member email);
//   compute started_at/stopped_at from date+start_time/end_time, else duration from duration_minutes;
//   duration_seconds = (end-start) or duration_minutes*60.
//   INSERT INTO time_entries (id, tenant_id, user_id, task_id, project_id, description,
//     started_at, stopped_at, duration_seconds, source, billable)
//     VALUES (gen_random_uuid(), $tenant, $user, $task, $project, $desc,
//             $started_at, $stopped_at, $duration_seconds, 'manual', $billable);
// invoices (per group):
//   INSERT INTO invoices (id, tenant_id, customer_id, invoice_number, status, currency,
//     issue_date, due_date, vat_rate, subtotal, vat_amount, total, source, created_by);
//   INSERT INTO invoice_lines (id, invoice_id, tenant_id, description, quantity, unit_price,
//     line_total, taxable, position) per accumulated line; recompute subtotal/vat/total.
//   status mapped: draft->DRAFT, sent->SENT, paid->PAID, cancelled->VOID.
export async function processImport(message: ImportJobMessage, env: Env): Promise<void>;
```
**Acceptance:**
- [ ] 500-row customer file with 12 duplicate emails + 6 missing-name rows yields `success_count=482, skipped_count=12, error_count=6`, status `completed`.
- [ ] Killing the worker mid-batch and re-delivering does not double-insert committed rows (idempotent guard on status).
- [ ] Invoice CSV with 2 rows sharing `INV-001` (second row blank invoice_number) produces ONE invoice with TWO `invoice_lines`.
- [ ] Unparseable file → status `failed`, `error_message` set, one `import_completed` notification.

### Task 14: Frontend — page shell, tier gate & locked state
**Blocks:** 15  ·  **Blocked by:** 2
**Files:**
- Create: `apps/zync-app/src/pages/settings/import/ImportPage.tsx`
- Modify: `apps/zync-app/src/router.tsx` (route `/settings/import`, OWNER/ADMIN guard)
**Steps:**
- [ ] Route `/settings/import` inside the settings layout; restrict to `OWNER`/`ADMIN` (redirect/403 otherwise).
- [ ] Use `useTierGate('business')`: if below Business, render the locked `Card` ("Data import is available on the Business plan…") with an "Upgrade to Business" button that calls `useUpgradeModal()` (spec 36).
- [ ] If Business+, render the history list (Task 15) + "New Import" button that opens the wizard (Task 16).
**Acceptance:**
- [ ] Freelancer sees locked card; clicking Upgrade opens the upsell modal. Business+ sees history + New Import.
- [ ] Non-OWNER/ADMIN member cannot reach the route.

### Task 15: Frontend — import history list
**Blocks:** —  ·  **Blocked by:** 10, 14
**Files:**
- Create: `apps/zync-app/src/pages/settings/import/ImportHistoryTable.tsx`
- Create: `apps/zync-app/src/hooks/useImports.ts`
**Steps:**
- [ ] `useImports()` (react-query) → `GET /api/imports` with cursor pagination (20/page).
- [ ] `DataTable`: columns Type, File, Started, Status (`Badge`: pending grey / processing blue+spinner / completed green / failed red), Results (`success / total`).
- [ ] For completed jobs with `error_count+skipped_count>0`, render "⬇ Errors (N)" linking to `GET /api/imports/:id/errors.csv`.
- [ ] For `failed` jobs, render "View error" expanding `error_message` inline.
- [ ] `EmptyState` when no imports yet (from error-empty-states catalog).
**Acceptance:**
- [ ] History renders statuses with correct badge colors; error link downloads the CSV; failed job expands its message; pagination loads next 20.

### Task 16: Frontend — wizard shell, Stage 1 (type) & Stage 2 (upload)
**Blocks:** 17  ·  **Blocked by:** 4, 6, 7, 12, 14
**Files:**
- Create: `apps/zync-app/src/pages/settings/import/ImportWizard.tsx`
- Create: `apps/zync-app/src/pages/settings/import/steps/SelectType.tsx`
- Create: `apps/zync-app/src/pages/settings/import/steps/UploadFile.tsx`
**Steps:**
- [ ] Stepper (5 stages) at top of a centred 720px content `Card`; `aria-current="step"` on the active step; states ●/○/✓.
- [ ] Stage 1: four type cards (Customers, Invoices, Products/Services, Time Entries) with descriptions + "Download template" link to `GET /api/imports/templates/:type`. Invoices card: "Supports: CSV only" + tooltip "Excel import for invoices will be added in a future update." Exactly one selectable; Continue disabled until selected.
- [ ] Stage 2: drag-drop + Browse dropzone. Client validation: extension matches accepted types (XLSX only for customers), size ≤ 10 MB; on failure show inline error and set dropzone border to `var(--accent)`.
- [ ] Upload sequence: `POST /api/imports/upload` → PUT file to returned `upload_url` → `POST /api/imports` → store `{ import_job_id, columns }`, advance to Stage 3.
- [ ] Error toasts: network fail "Upload failed. Please try again."; 413 "File exceeds 10 MB limit."; 422 ROW_LIMIT "File contains more than 5,000 rows…"; 415 "Unsupported file type."
**Acceptance:**
- [ ] Selecting Invoices disables XLSX; uploading a 11 MB file is blocked client-side; a valid CSV advances to mapping with `columns` populated.

### Task 17: Frontend — Stage 3 (mapping) & Stage 4 (preview)
**Blocks:** 18  ·  **Blocked by:** 8, 9, 16
**Files:**
- Create: `apps/zync-app/src/pages/settings/import/steps/MapColumns.tsx`
- Create: `apps/zync-app/src/pages/settings/import/steps/PreviewValidate.tsx`
**Steps:**
- [ ] Mapping table File column → Zync field; run `autoDetectMapping(type, columns)` on mount; show "We recognized X of Y columns automatically."
- [ ] Each column: a `Select` of available Zync fields + "Ignore this column"; auto-matched rows show "✓ Auto-matched", unmapped "⚠ Unmapped", ignored "✓ Ignored".
- [ ] Enforce required fields (`REQUIRED_FIELDS[type]`) before Continue; duplicate target mapping is an inline error except `notes` (allowed, concatenated).
- [ ] On Continue → `POST /api/imports/:id/preview` with mapping; render Stage 4.
- [ ] Stage 4: table of first ≤5 rows with per-row status; summary "N valid · N warning · N error". Row backgrounds: error `var(--danger-bg)`, warning `oklch(97% 0.015 80)`, valid `var(--bg)`. Inline message under error/warning rows. "Confirm & Start Import" always enabled → `POST /api/imports/:id/start`, then Stage 5. "← Back" returns to mapping preserving config.
**Acceptance:**
- [ ] Required-field-unmapped blocks Continue with a message; preview shows the spec's 3-valid/1-warning/1-error sample correctly colored; Confirm enqueues the job.

### Task 18: Frontend — Stage 5 (processing & results)
**Blocks:** —  ·  **Blocked by:** 10, 11, 17
**Files:**
- Create: `apps/zync-app/src/pages/settings/import/steps/Processing.tsx`
- Create: `apps/zync-app/src/pages/settings/import/steps/Results.tsx`
**Steps:**
- [ ] Processing: poll `GET /api/imports/:id` every 3s; `Progress` bar width = `rows_processed/total_rows*100`; copy "This may take a minute. You can leave this page…"; "View import history" link. Respect `prefers-reduced-motion` (no indeterminate animation when set). Stop polling on unmount.
- [ ] On `completed`/`failed`, render Results: "✓ N imported", "⚠ N skipped", "✗ N failed"; "⬇ Download error report (CSV)" only if `error_count+skipped_count>0` → `GET /api/imports/:id/errors.csv`.
- [ ] Footer actions: "Import another file" (resets wizard) and "Go to {Customers|Invoices|Products|Time}" deep-link by type.
**Acceptance:**
- [ ] Progress advances via 3s polling and stops on completion; results counts match the job; error CSV button hidden when there are zero errors/skips.

### Task 19: i18n strings & RTL
**Blocks:** —  ·  **Blocked by:** 14, 15, 16, 17, 18
**Files:**
- Modify: `packages/i18n/src/locales/en/import.json`
- Modify: `packages/i18n/src/locales/he/import.json`
**Steps:**
- [ ] Add all wizard/history/locked-state/toast/notification strings under an `import.*` namespace in `en` and `he`.
- [ ] Add `notification.import_completed.title` / `.body` keys (referenced by the worker's `createNotification` `title_key`/`body_key`).
- [ ] Verify the wizard renders correctly under RTL (`dir="rtl"`) using logical properties (`ps-*`/`pe-*`); no hardcoded left/right.
**Acceptance:**
- [ ] No raw English literals in the import UI; Hebrew locale renders RTL with mirrored stepper and aligned tables.

### Task 20: Notification wiring — `import_completed`
**Blocks:** —  ·  **Blocked by:** 13, 2
**Files:**
- Modify: `apps/zync-api/src/queues/import-process.ts` (call site)
- Modify: `packages/notifications/src/templates/en/import-completed.ts`
- Modify: `packages/notifications/src/templates/he/import-completed.ts`
**Steps:**
- [ ] Worker calls `createNotification({ tenantId, userId: job.created_by, type:'import_completed', titleKey:'notification.import_completed.title', bodyKey:'notification.import_completed.body', params:{ type, success_count, skipped_count, error_count }, entityType:'import', entityId: import_job_id })`.
- [ ] Add email templates per locale for `import_completed`.
**Acceptance:**
- [ ] On job completion the importing user receives an in-app `import_completed` notification with correct counts; it appears in the notification center.

### Task 21: Retention hook — register tables/R2 with `data-retention-purge`
**Blocks:** —  ·  **Blocked by:** 1
**Files:**
- Modify: `apps/zync-api/src/cron/data-retention-purge.ts` (registry of purgeable entities)
**Steps:**
- [ ] Register `import_job_results` (purge rows older than 90 days), `import_jobs` (purge rows older than 90 days), and R2 deletion of each purged job's `r2_key` from the `STORAGE` bucket in the same cron run.
- [ ] Order: delete R2 object, then `import_jobs` row (cascade removes `import_job_results`).
**Acceptance:**
- [ ] A cron dry-run reports `import_jobs`/`import_job_results` older than 90 days as purge candidates and lists their `r2_key`s for deletion.