# A CSV file-watch import job

In [Chapter 1](https://rickylabs.github.io/netscript/tutorials/erp-sync/01-scaffold/) you stood up `my-erp/` with the workers and triggers plugins running. Now you wire the first real piece of the pipeline: a **file-watch trigger** that fires the moment the SAP export job drops a CSV into the hand-off folder, and a durable **background job** that parses it. This is the ingest core of the migration — the SAP export reaches you as a file, not an API call, and every file it writes must become durable background work. A file the watcher misses is a day of catalog changes Dynamics never sees; a file parsed twice is a day counted twice. Both failure modes start here, so this chapter is where the pipeline earns the word *durable*.

1. [1 · Scaffold](https://rickylabs.github.io/netscript/netscript/tutorials/erp-sync/01-scaffold/)
2. [2 · Import job](https://rickylabs.github.io/netscript/netscript/tutorials/erp-sync/02-import-job/)
3. [3 · Polyglot transform](https://rickylabs.github.io/netscript/netscript/tutorials/erp-sync/03-polyglot-transform/)
4. [4 · Queue & cron](https://rickylabs.github.io/netscript/netscript/tutorials/erp-sync/04-queue-and-cron/)
5. [5 · Deploy](https://rickylabs.github.io/netscript/netscript/tutorials/erp-sync/05-deploy/)

## What you will build

By the end of this chapter, dropping a file named `products_*.csv` into `my-erp/.data/incoming/products` will automatically enqueue and run a worker job that reads the file, parses its rows, logs a summary, and returns a structured success result — all without any HTTP call. You author two files: a **trigger** (`defineFileWatch`) and a **job** (`defineJobHandler`), then register the job so the trigger can address it by `id`.

## Before you begin

You need the workspace from [Chapter 1](https://rickylabs.github.io/netscript/tutorials/erp-sync/01-scaffold/) with **both** the workers and triggers plugins installed, and `aspire start` healthy. Confirm the prior state from the project root:

```sh
netscript plugin list
```

Expected: `workers` and `triggers` both appear. Also confirm Aspire is up — the file-watch processor and the workers runtime both depend on it:

```sh
curl <workers-endpoint>/health    # workers API, healthy JSON
curl <triggers-endpoint>/health   # triggers API, healthy JSON
```

If either is missing, return to Chapter 1 — do not start over.

## Step 1 — Create the watched folder

The trigger watches a directory; create it (and the staging path suppliers will write to) so the watcher has something to attach to:

```sh
mkdir -p .data/incoming/products
```

This is just a local folder in your workspace. In production it would be the mounted share the SAP nightly export job writes into; the trigger does not care where the bytes come from, only that a matching file appears.

## Step 2 — Scaffold the import job

A NetScript job is a function wrapped by `defineJobHandler`, given a stable `id`, and exported as the module default. Inside the handler you receive a `ctx` carrying the payload, do the work, and return a result built with `createSuccessResult` / `createFailureResult`. Create the job under the workers plugin. Let the CLI create the payload schema, handler wrapper, stable export, and registry entry:

> ns-workers / ns-triggers are shorthands you install once
>
> ns-workers
>
> and
>
> ns-triggers
>
> are names
>
> you
>
> give the workers and triggers plugin CLIs — the scaffold does not create them. Install them once, globally, and every shorthand command on this page works as written:
>
> ```bash
> deno install -gArf -n ns-workers jsr:@netscript/plugin-workers@0.0.6/cli
> deno install -gArf -n ns-triggers jsr:@netscript/plugin-triggers@0.0.6/cli
> ```
>
> Rather not install them? Each
>
> ns-<plugin> <verb …>
>
> is exactly
>
> deno x -A jsr:@netscript/plugin-<plugin>@0.0.6/cli <verb …>
>
> — run that full form instead.

```sh
ns-workers add job import-products
```

In `workers/jobs/import-products.ts`, extend the generated `PayloadSchema` with `filePath` and `fileName`, import `createFailureResult`, and replace only the starter handler body with the CSV logic below:

```ts
const handler = defineJobHandler(async (ctx) => {
  const { filePath, fileName } = PayloadSchema.parse(ctx.payload ?? {});

  // 1. Read the staged file.
  let rawContent: string;
  try {
    rawContent = await Deno.readTextFile(filePath);
  } catch (error: unknown) {
    const message = error instanceof Error ? error.message : String(error);
    return createFailureResult(`Failed to read ${fileName}: ${message}`);
  }

  // 2. Parse a simple CSV (header row + data rows). No external lib needed.
  const lines = rawContent.trim().split('\n').filter((line) => line.trim().length > 0);
  if (lines.length < 2) {
    return createFailureResult('CSV is empty or has no data rows');
  }
  const headers = lines[0].split(',').map((h) => h.trim().toLowerCase());
  const rows = lines.slice(1).map((line) => {
    const values = line.split(',').map((v) => v.trim());
    const row: Record<string, string> = {};
    headers.forEach((h, i) => row[h] = values[i] ?? '');
    return row;
  });

  // 3. Return a structured result. The runtime records it on the execution.
  return createSuccessResult({ fileName, rowCount: rows.length, headers });
});
```

The two things to read off this: the handler is an `async`/arrow function (never a bare `function`), and the stable `id` is attached with `Object.assign` so the runtime registry can address the job by a predictable string rather than its filename.

> Jobs do one thing
>
> This handler reads, parses, and reports — and stops. In a fuller ERP sync the next step (validate, upsert into a service, publish a saga message) would be its own job or a saga, keeping each unit small and independently retryable. The scaffold's
>
> --samples
>
> jobs show that fuller chain under
>
> plugins/workers/jobs/
>
> ; for this track one job is enough to prove the pipeline.

## Step 3 — Scaffold the file-watch trigger

Create the watcher from the current trigger CLI surface:

The `add-file-watch` verb uses the spaced `add file-watch` shell syntax:

```sh
ns-triggers add file-watch product-import-trigger \
  --path=.data/incoming/products \
  --pattern=products_*.csv
```

The command writes `triggers/product-import-trigger.ts` and refreshes the trigger registry. The file-watch command deliberately scaffolds a neutral handler, so replace that generated handler with the enqueue effect shown below. `defineFileWatch(handler, spec)` comes from `@netscript/plugin-triggers-core/builders`. The handler returns an **array of effects**; the one you want is `enqueueJob(jobRef, { payload })`, which hands a worker job the inbound event. The job is referenced by a small typed object — its `id`, `name`, `topic`, and `entrypoint`.

```ts
import { defineFileWatch, enqueueJob } from '@netscript/plugin-triggers-core/builders';
import type { JobDefinition } from '@netscript/plugin-workers-core';

// A reference to the worker job authored in Step 2.
const importProductsJob = {
  id: 'import-products' as JobDefinition<'import-products'>['id'],
  name: 'Import Products',
  topic: 'default',
  entrypoint: './workers/jobs/import-products.ts',
} satisfies JobDefinition<'import-products'>;

export default defineFileWatch(
  // event.payload carries filePath / fileName for the matched file.
  (event) => Promise.resolve([enqueueJob(importProductsJob, { payload: event.payload })]),
  {
    id: 'product-import-trigger',
    paths: ['.data/incoming/products'],
    patterns: ['products_*.csv'],
    on: ['create'],
    stabilityThreshold: { checkIntervalMs: 1000, stableChecks: 2 },
    description: 'Watches for product CSV files and starts the import job.',
    tags: ['file-watch', 'product', 'import'],
  },
);
```

Read the `spec` carefully — it is the whole behavior of the watcher:

**FileWatchSpec — the static fields you set on defineFileWatch**

| Name | Type | Description |
| --- | --- | --- |
| `id` | `string` | Stable identifier for this trigger, used in logs and the events feed. |
| `paths` | `string[]` | Directories to watch. Here, the .data/incoming/products folder you created in Step 1. |
| `patterns` | `string[]` | Glob patterns a file must match to fire the handler. products_*.csv ignores everything else. |
| `on` | `('create' \| 'modify' \| 'remove')[]` | Which filesystem events fire the trigger. 'create' = a new file landing. |
| `stabilityThreshold` | `{ checkIntervalMs, stableChecks }` | Debounce: wait until the file size is unchanged across stableChecks polls before firing, so a half-written upload is never parsed. |

> stabilityThreshold is not optional polish
>
> A supplier copying a large file appears on disk before it is fully written. Without a stability check the watcher would fire on the first byte and your job would read a truncated CSV. The
>
> checkIntervalMs: 1000, stableChecks: 2
>
> here waits for the size to hold steady for two one-second polls before enqueuing — set it generously for large files on slow shares.

## Step 4 — Register the job

The trigger addresses the job by `id`, which means the workers runtime needs a generated registry that maps each `id` to its handler. Generate the plugin registries:

```sh
netscript generate plugins
```

For workers, this structurally scans the project's top-level `workers/jobs/*.ts` modules and skips declared helpers such as `job-tools.ts`; it does not require an official sample filename. The command also discovers plugin-contributed jobs and triggers, then writes the generated registries the running services load. After this, `import-products` is addressable, and `product-import-trigger` is loaded by the file-watch processor. Regenerate after adding or replacing a job—do not edit `.netscript/generated/**` by hand.

> Restart the processors after generating
>
> If
>
> aspire start
>
> was up before you generated the registry, restart it (or let it hot-reload) so the workers runtime and the file-watch processor pick up the new job and trigger.

## Verify your progress

With Aspire up, drop a matching CSV into the watched folder and watch the pipeline run end to end. First create a sample file in the SAP export shape — legacy column names, prices in integer cents (Chapter 3 transforms exactly this file into Dynamics' shape):

```sh
cat > .data/incoming/products_2024.csv <<'CSV'
material_no,description,price_cents
WID-1,Widget,999
GAD-2,Gadget,1999
CSV
mv .data/incoming/products_2024.csv .data/incoming/products/products_2024.csv
```

Moving the finished file into the watched folder fires a single `create` event (and avoids the watcher seeing a half-written file). Now confirm both sides of the hand-off:

```sh
# 1. The trigger recorded the inbound file event.
curl '<triggers-endpoint>/api/v1/events?limit=10'

# 2. The job it enqueued executed.
ns-workers executions --limit=10 --json
```

Expected: the events feed lists a `product-import-trigger` event, and the executions feed shows a completed `import-products` run whose result is `{ "fileName": "products_2024.csv", "rowCount": 2, "headers": ["material_no","description","price_cents"] }`. Open the `workers` resource logs in the [Aspire dashboard](https://localhost:18888) to read the job's structured log lines.

- [ ] `.data/incoming/products/` exists and you dropped a `products_*.csv` into it.
- [ ] The triggers events feed shows a `product-import-trigger` event.
- [ ] The workers executions feed shows a completed `import-products` run with `rowCount: 2`.
- [ ] `deno task check` is clean.

> If the job never runs
>
> - **Aspire isn't up** — the file-watch processor and the workers runtime are Aspire resources. Start `aspire start` from `aspire/` and retry.
> - **The job isn't registered** — re-run `netscript generate plugins` so `import-products` is in the generated registry, then restart Aspire.
> - **Filename didn't match** — the pattern is `products_*.csv`. A file named `catalog.csv` in the same folder is ignored by design.
> - **Wrong folder** — the file must land inside `.data/incoming/products`, the directory in `paths`, not its parent.

## What you built

A file-watch trigger (`defineFileWatch`) that fires on every `products_*.csv` the SAP export drops into the hand-off folder, and a durable background job (`defineJobHandler`) that parses it — wired together by `enqueueJob` and made addressable with `netscript generate plugins`. An inbound SAP export now becomes durable background work with no HTTP in the loop. But the rows you just imported are still in the legacy shape — column names Dynamics does not use and prices in cents. Next, you build the transform stage that fixes that, and run it.

[1 · Scaffold](https://rickylabs.github.io/netscript/netscript/tutorials/erp-sync/01-scaffold/) [3 · Polyglot transform](https://rickylabs.github.io/netscript/netscript/tutorials/erp-sync/03-polyglot-transform/)