A CSV file-watch import job

In Chapter 1 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
2. 2 · Import job
3. 3 · Polyglot transform
4. 4 · Queue & cron
5. 5 · 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 with both the workers and triggers plugins installed, and aspire start healthy. Confirm the prior state from the project root:

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:

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:

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 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:

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.

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:

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.

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:

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:

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.

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):

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:

# 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 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.

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.