FEATURE — v0.May1526

ETL
Pipelines.

Why it exists

Manual data entry doesn't scale. Pipelines let you pull data from external sources, transform it to match Pazator's entity model, and ingest it directly into the store. The whole thing runs in-browser — connectors fetch URLs client-side, transforms run in JavaScript, and the result goes straight into pazatorStore._data.

Pipeline Structure

{
  id:           "pipe_1717000000_abc12",
  name:         "Import bank records",
  connector:    { type: "csv", config: { url: "...", delimiter: ",", hasHeader: true } },
  transforms:   [ { type: "map", config: { mappings: { "full_name": "name", ... } } } ],
  targetStore:  "others",
  createdAt:    "2025-06-01T...",
  updatedAt:    "2025-06-01T..."
}

Connectors (4)

Every pipeline starts with one connector that defines how to fetch raw data.

CSV — Fetches a URL, splits by newline and delimiter. If hasHeader is true (default), the first row becomes object keys. Otherwise each row becomes { value: "..." }.
JSON — Fetches a URL, parses JSON, navigates a dot-separated jsonPath (e.g. data.results) to reach the array. If the path resolves to a single object, it's wrapped in an array.
SQL — POSTs a { connectionString, query } to a PZLS sync server at /api/query. Expects { success: true, rows: [...] }. The sync server URL defaults to http://localhost:3456 and is read from pazator_sync_config.
REST API — Generic HTTP fetch with configurable method, headers, and body. Expects a JSON response — array is used directly, single object is wrapped.

Transforms (5)

Transforms are applied sequentially, each receiving the output of the previous one.

Filter — Case-insensitive substring match on a field. Keeps rows where item[field] contains value.
Map Fields — Renames/remaps fields. Only mapped keys survive — any field not in the mapping is dropped. Example: { "full_name": "name", "age_years": "age" }.
Sort — Ascending or descending sort on a field. Uses localeCompare for strings.
Limit — Keeps the first N records. Simple slice(0, count).
Deduplicate — Case-insensitive dedup on a field. Keeps the first occurrence, drops subsequent matches.

Execution Flow

runPipeline(id)
  → fetchFromConnector(connector)     // returns array of objects
  → applyTransforms(data, transforms) // sequential transform pipeline
  → ingestToStore(result, targetStore) // pushes into pazatorStore._data[store]

Each run creates a history record with status (running, completed, error), input/output counts, and timestamps. The history is capped at 200 entries and stored in localStorage.pazator_pipeline_runs. The UI shows the most recent 50.

Target Stores

Processed records are pushed directly into pazatorStore._data[targetStore]. Three options: humans, others, or chats. Records without an id field get one auto-generated (<store>_<timestamp>_<index>).

There is no dedup against existing records — repeated runs append duplicates. Make sure your transforms produce clean, mergeable data.

Scheduler

Pipelines can be scheduled to run automatically. Schedules define a pipeline ID and an interval in minutes (converted to milliseconds). The scheduler uses setInterval internally and persists to pazator_pipeline_schedules. On init(), all enabled schedules are restarted.

Storage

Three localStorage keys:

pazator_pipelines — pipeline definitions
pazator_pipeline_runs — run history (capped 200)
pazator_pipeline_schedules — scheduler config

What's Missing

No dedup across runs — every execution appends. No credential management — SQL connection strings and REST API keys are stored in the pipeline config in plain text. No pipeline chaining (output of one feeding into another). The SQL connector requires a PZLS server to proxy the query — it doesn't connect to databases directly. CSV and JSON connectors are limited to public URLs (no auth headers for CSV/JSON, only REST has headers).

The transform editor in the "New Pipeline" modal only configures the connector — transforms are created as an empty array by default. Configuring transforms after creation would need to be done programmatically or through a UI that doesn't exist yet.

ETLPipelines.