dataflow/docs/spec.md

Dataflow — Application Specification

Purpose

Dataflow is an internal tool for importing, cleaning, and standardizing data from multiple sources (e.g., bank transaction exports). It is intentionally simple — no queues, no schedulers, no plugins. Everything is triggered explicitly by the user through the UI or API.

---

Philosophy and Design Choices

Simple over clever

Every decision favors clarity. No abstractions for their own sake, no frameworks beyond what is necessary. If a piece of code is hard to follow, that is a sign it should be rewritten, not abstracted.

All logic lives in the database

SQL functions are the single source of truth for business logic. The API layer is a thin HTTP wrapper — it validates input, calls a function, and returns the result. It does not construct business logic in JavaScript.

SQL is written as full literal strings

Database calls in the route files use fully formed SQL strings with values interpolated directly (not parameterized). This makes every query copy-pasteable into psql for debugging. A small lit() helper in api/lib/sql.js handles quoting and escaping. This is an intentional trade-off: the tool is internal, and debuggability is worth more than the marginal injection protection parameterization provides over what lit() already does.

One SQL file per route

SQL is organized in database/ with one file per route (sources.sql, rules.sql, mappings.sql, records.sql, stacks.sql, status.sql) plus import.sql and transform.sql for the import/transform engine. This makes it easy to find the SQL behind any API endpoint — look at the route file to find the function name, then look at the matching query file for the implementation.

Explicit over implicit

Nothing happens automatically. Transformations are triggered by the user. Views are generated on demand. There are no database triggers, no background workers, no scheduled jobs.

JSONB for flexibility

Raw imported records and transformed records are stored as JSONB. This avoids schema migrations when source formats change and allows different sources to have different field sets in the same table.

---

Architecture

manage.py              — interactive CLI for setup, deployment, and management
database/
  schema.sql           — table definitions (run once or to reset)
  sources.sql          — all SQL for /api/sources
  rules.sql            — all SQL for /api/rules
  mappings.sql         — all SQL for /api/mappings
  records.sql          — all SQL for /api/records
  stacks.sql           — all SQL for /api/stacks
  status.sql           — all SQL for /api/status
  import.sql           — CSV import and the import audit trail
  transform.sql        — the rule/mapping engine
api/
  server.js            — Express server, mounts routes, auth middleware
  middleware/
    auth.js            — Basic Auth enforcement on all /api routes
  lib/
    sql.js             — lit() and arr() helpers for SQL literal building
  routes/
    sources.js         — HTTP handlers for source management
    rules.js           — HTTP handlers for rule management
    mappings.js        — HTTP handlers for mapping management
    records.js         — HTTP handlers for record queries
    stacks.js          — HTTP handlers for stack management
    status.js          — HTTP handler for deployment status
ui/
  src/
    api.js             — fetch wrapper, credential management
    App.jsx            — root: login gate, sidebar, source selector, routing
    pages/
      Login.jsx        — username/password form
      Sources.jsx      — source CRUD, field config, view generation
      Import.jsx       — CSV upload and import log
      Rules.jsx        — rule CRUD with live pattern preview
      Mappings.jsx     — mapping table with TSV import/export
      Records.jsx      — paginated, sortable view of transformed records
      Pivot.jsx        — interactive pivot table with cell inspector
      Stacks.jsx       — multi-source union views with running balance
      Remap.jsx        — bulk remap of an output field value across mappings
      Log.jsx          — global import log across all sources
    components/        — Sidebar, StatusBar
    theme.jsx          — light/dark context provider
public/                — compiled UI (output of npm run build in ui/)
docs/                  — this file, tutorial, UI and Perspective references
examples/              — bank_transactions.csv, the tutorial's sample data

---

Database Schema

Five tables in the dataflow schema:

sources

Defines a data source. The constraint_fields array specifies which fields make a record unique. config (JSONB) holds the output schema (fields array) used to generate the typed view.

records

Stores every imported record. data holds the raw import. transformed holds the enriched record after rules and mappings are applied. constraint_key is a JSONB object of the constraint field values used to detect duplicates at import time. import_id references the import_log row; deleting a log entry cascades to its records.

rules

Regex transformation rules. Each rule reads from field, applies pattern with optional flags, and writes to output_field. function_type is either extract (regexp_matches) or replace (regexp_replace). sequence controls the order rules are applied. retain keeps the raw extracted value in output_field even when a mapping overrides it.

mappings

Maps an extracted value to a standardized output object. input_value is JSONB (matches the extracted value exactly, including arrays from multi-capture-group patterns). output is a JSONB object that can contain multiple fields (e.g., {"vendor": "Walmart", "category": "Groceries"}).

import_log

Audit trail. One row per import call, recording how many records were inserted versus skipped as duplicates. info (JSONB) stores the full inserted_keys and excluded_keys arrays. Deleting a log row cascades to its records via the import_id FK.

---

Data Flow

Import

CSV file → parse in Node.js → import_records(source, data)
  → build JSONB constraint_key per record
  → compare against existing records (CTE — no unique constraint)
  → INSERT new records, skip duplicates
  → log to import_log (with inserted_keys / excluded_keys)
  → apply_transformations() runs automatically on new records

Transform

apply_transformations(source) — pure SQL CTE
  qualifying: records not yet transformed (or all, if overwrite=true)
  rx: fan out one row per regex match (LATERAL join, rules × records)
  agg_matches: collapse matches back to one value per (record, rule)
  linked: LEFT JOIN mappings to find mapped output
  rule_output: build per-rule output JSONB
  record_additions: merge all rule outputs per record in sequence order
  UPDATE records SET transformed = data || additions

The transform is fully set-based — no row-by-row loops. All records for a source are processed in a single query.

View generation

generate_source_view(source) reads config.fields from the source and builds a CREATE VIEW statement in the dfv schema. Each field is cast to its declared type (text, numeric, date). Fields with an expression are computed from other transformed fields using {field} substitution syntax.

---

SQL Functions

Each route file has a matching SQL file in database/; import.sql and transform.sql hold the engine shared by several routes.

sources.sql list_sources, get_source, create_source, update_source, delete_source, get_import_log, get_source_stats, get_source_fields, get_view_data (plpgsql — dynamic sort via EXECUTE + quote_ident), import_records, jsonb_merge + jsonb_concat_obj aggregate, apply_transformations, reprocess_records, generate_source_view

rules.sql list_rules, get_rule, create_rule, update_rule, delete_rule, preview_rule (plpgsql — conditional query for extract vs replace), test_rule (returns TABLE(rule JSONB, results JSONB))

mappings.sql list_mappings, get_mapping, create_mapping, upsert_mapping, update_mapping, delete_mapping, get_mapping_counts, get_all_values (plpgsql — extracted values with mapping join), get_unmapped_values (plpgsql — extracted values with no mapping)

records.sql list_records, get_record, search_records (JSONB containment on data and transformed), delete_record, delete_source_records

stacks.sql list_stacks, get_stack, create_stack, update_stack, delete_stack, get_stack_view_data (union of source views with field mapping and running balance), list_pivot_layouts, save_pivot_layout, delete_pivot_layout

status.sql get_status — returns deployment state (schema version, function presence, service status)

---

API

All routes are under /api. Every route requires HTTP Basic Auth. The GET /health endpoint is exempt.

Authentication: Authorization: Basic <base64(user:pass)> on every request. Credentials are verified against LOGIN_USER (plaintext username) and LOGIN_PASSWORD_HASH (bcrypt hash) in .env. There are no sessions or tokens — credentials are sent with every request.

Route summary:

Sources — api/routes/sources.js

Method	Path	Description
GET	/api/sources	List all sources
POST	/api/sources	Create source
GET	/api/sources/:name	Get source
PUT	/api/sources/:name	Update source (constraint_fields, config, global_picklist)
DELETE	/api/sources/:name	Delete source and all its data
POST	/api/sources/suggest	Suggest source config from an uploaded CSV
POST	/api/sources/:name/import	Import CSV; transformations are applied to the new records
GET	/api/sources/import-log	Import history across all sources
GET	/api/sources/:name/import-log	Import history for one source
DELETE	/api/sources/:name/import-log/:id	Delete an import batch and every record in it
POST	/api/sources/:name/transform	Apply transformations to untransformed records only
POST	/api/sources/:name/reprocess	Reapply transformations to all records
GET	/api/sources/:name/stats	Record counts
GET	/api/sources/:name/fields	All known field names and their origins
GET	/api/sources/:name/override-keys	Distinct field names used in overrides for this source
POST	/api/sources/:name/view	Generate/refresh the dfv view
GET	/api/sources/:name/view-data	Paginated, sortable, filterable view data
GET	/api/sources/:name/layouts	List saved pivot layouts
POST	/api/sources/:name/layouts	Save a pivot layout
DELETE	/api/sources/:name/layouts/:id	Delete a pivot layout

Rules — api/routes/rules.js

Method	Path	Description
GET	/api/rules/source/:name	List rules for a source
GET	/api/rules/:id	Get a rule
POST	/api/rules	Create rule
PUT	/api/rules/:id	Update rule
DELETE	/api/rules/:id	Delete rule
GET	/api/rules/preview	Preview an ad-hoc pattern against live records
GET	/api/rules/:id/test	Test a saved rule against live records

Mappings — api/routes/mappings.js

Method	Path	Description
GET	/api/mappings/source/:name	List mappings
GET	/api/mappings/:id	Get a mapping
POST	/api/mappings	Create mapping
POST	/api/mappings/bulk	Upsert multiple mappings
PUT	/api/mappings/:id	Update mapping
DELETE	/api/mappings/:id	Delete mapping
GET	/api/mappings/source/:name/all-values	All extracted values (mapped + unmapped) with counts
GET	/api/mappings/source/:name/unmapped	Only values with no mapping yet
GET	/api/mappings/source/:name/counts	Record counts per mapping
GET	/api/mappings/source/:name/export.tsv	Export extracted values as TSV
POST	/api/mappings/source/:name/import-csv	Import/update mappings from an uploaded TSV
GET	/api/mappings/global-values	Output values across all global_picklist sources (autocomplete)
GET	/api/mappings/outputs	Search output field values across all mappings
GET	/api/mappings/outputs/:col/:val	Mappings carrying a specific output field value
POST	/api/mappings/remap-field	Replace an output field value across all mappings

Records — api/routes/records.js

Method	Path	Description
GET	/api/records/source/:name	List records (paginated)
GET	/api/records/:id	Get a single record
POST	/api/records/search	Search by JSONB containment
DELETE	/api/records/:id	Delete record
DELETE	/api/records/source/:name/all	Delete all records for a source
PUT	/api/records/:id/overrides	Set manual overrides on a record
DELETE	/api/records/:id/overrides	Clear a record's overrides
POST	/api/records/bulk-overrides	Apply the same overrides to many records

Stacks — api/routes/stacks.js

Method	Path	Description
GET	/api/stacks	List all stacks
GET	/api/stacks/:name	Get a stack with its sources
POST	/api/stacks	Create stack
PUT	/api/stacks/:name	Update stack
DELETE	/api/stacks/:name	Delete stack
PUT	/api/stacks/:name/sources/:source	Add or update a source in the stack
DELETE	/api/stacks/:name/sources/:source	Remove a source from the stack
PUT	/api/stacks/:name/sources/reorder	Reorder the stack's sources
GET	/api/stacks/:name/view-sql	Preview the SQL that would build the view (dry run)
POST	/api/stacks/:name/view	Generate/refresh the dfv view
POST	/api/stacks/:name/exec-sql	Execute user-edited SQL for the view
GET	/api/stacks/:name/view-data	Paginated stacked data with running balance
GET	/api/stacks/:name/balance	Current running balance from the generated view
POST	/api/stacks/:name/calibrate	Set the balance offset from a known balance at a date
GET	/api/stacks/:name/layouts	List saved pivot layouts
POST	/api/stacks/:name/layouts	Save a pivot layout
DELETE	/api/stacks/:name/layouts/:id	Delete a pivot layout

Status — api/routes/status.js

Method	Path	Description
GET	/api/status	Deployment status
GET	/health	Health check (no auth)

---

api/lib/sql.js — SQL Literal Helpers

lit(val)  // JS value → SQL literal: 'text', TRUE, 42, NULL, '{"json":"val"}'
arr(val)  // JS array → PostgreSQL array literal: ARRAY['a','b']

Single quotes within string values are escaped by doubling them ('→''). Objects and arrays are JSON-serialized. These helpers exist so that query strings in route files are fully formed and can be copied directly into psql.

---

UI

Built with React + Vite + Tailwind CSS. Compiled output goes to public/. The server serves public/ as static files and falls back to public/index.html for all non-API routes (SPA routing).

Authentication flow:

1. On load, check sessionStorage for saved credentials. If found, re-authenticate silently via GET /api/sources.
2. On login, credentials are stored in memory (api.js module-level) and in sessionStorage (survives page refresh, cleared on tab close).
3. On 401 response, credentials are cleared and the login screen is shown.
4. localStorage persists the selected source name across sessions.

Pages:

• Sources — View and edit source configuration. Shows all known field names and their origins (raw data, schema, rules, mappings). Checkboxes control which fields are constraint fields and which appear in the output view. Supports CSV upload to auto-detect fields.

• Import — Upload a CSV to import records into the selected source. Transformations run automatically on new records. Shows import log with inserted/duplicate counts, expandable key detail, checkbox selection, and delete with confirmation.

• Rules — Create and manage regex rules. Live preview fires automatically (debounced 500ms) as pattern/field/flags are edited, showing match results against real records. Rules can be enabled/disabled by toggle.

• Mappings — Tabular mapping editor. Shows all extracted values from transformed records with record counts and sample raw data. Rows are yellow (unmapped), white (mapped), or blue (edited but unsaved). Supports TSV export and import. Columns can be added dynamically.

• Records — Paginated table showing the dfv.{source} view. Server-side sorting (column validated against information_schema.columns, interpolated with quote_ident). Dates are formatted YYYY-MM-DD for correct lexicographic sort. Regex filters can be added per column. If the view cast fails (e.g. a field typed as date contains text), the error is shown inline rather than a blank page.

• Pivot — Interactive pivot/crosstab powered by Perspective (@perspective-dev client/viewer/datagrid v4.5.1, viewer-d3fc v4.4.1 — installed via npm). Loads all rows from the source view into an in-browser Perspective worker and renders a <perspective-viewer> web component. Supports grouping, splitting, filtering, sorting, and charting interactively.

  Toolbar (above the viewer):

  • Named layouts — saved per source in the pivot_layouts DB table. Each chip recalls the full viewer state including group_by, split_by, filters, expressions, selection mode, and expand depth. A blue Save button overwrites the active layout in place; + Save as… saves to a new name. The × on each chip deletes it.
  • depth: 0 1 2 3 — collapses or expands all grouped rows to the specified hierarchy level. Implemented via view.set_depth(d) + plugin.draw(view) (the only working mechanism found — plugin_config.expand_depth and viewer.flush() alone have no effect).
  • The Perspective built-in selection mode button (Read-Only / Select Row / Select Column / Select Region) defaults to Select Region on fresh load, set directly via plugin.restore({ edit_mode: 'SELECT_REGION' }) after the viewer loads.

  Cell inspector (right panel):

  • Opens when a cell is clicked and a group_by hierarchy is active. If there is no group_by, the click is ignored — without coordinate filters the query would return the full dataset.
  • Row filtering uses a temporary Perspective view (table.view({ filter: eventFilters, expressions: config.expressions })) so that computed/expression columns in split_by are evaluated correctly. Falls back to JS-side filtering if the view query fails.
  • Shows cell coordinates (group_by › split_by values), the clicked metric with value, any user-set filters, and a table of matching raw rows.
  • Number formatting rounds to 2 decimal places by default; a −/+ control in the inspector header adjusts precision (0–8).

  Layout persistence:

  • localStorage key psp_layout_{source} saves the last viewer state on each named layout save.
  • Named layouts store { ...viewer.save(), plugin_config: plugin.save(), expand_depth } as JSONB in pivot_layouts. On recall, viewer config, plugin config (edit mode), and expand depth are all restored independently.

  See docs/perspective.md for the full technical reference on controlling Perspective programmatically.

• Stacks — Named unions of multiple sources. Each stack defines a field mapping (how source fields map to common output columns), an amount field, a date field, and an optional balance offset. The view-data endpoint unions the underlying source views and computes a running balance sorted by date. The Pivot page supports stacks as well as individual sources, with layouts stored in the same pivot_layouts table.

• Log — Global import log across all sources. Same expandable key detail and delete capability as the Import page, plus a source name column.

---

manage.py

Interactive CLI for setup and operations. Run with python3 manage.py. Requires no arguments.

Shows current status on every screen:

• Database connection (host, port, db, user) and whether it succeeds
• Whether the dataflow schema is deployed
• Whether SQL functions are deployed (detected by presence of apply_transformations)
• Login credentials (configured / not configured)
• UI build status and timestamp
• Systemd service status
• Nginx reverse proxy status

Menu options:

1. Database configuration and deployment dialog — Write or update .env with database connection settings. If connection fails, offers to create the database and user using admin credentials. Optionally deploys schema and SQL functions after .env is written.

2. Redeploy schema — Runs database/schema.sql against the configured database. Warns that this drops all data. Requires explicit confirmation.

3. Redeploy SQL functions — Runs the function files in database/ in dependency order: sources.sql, rules.sql, mappings.sql, records.sql, import.sql, transform.sql, stacks.sql, status.sql. Safe to run at any time without data loss.

4. Build UI — Runs npm run build in ui/, outputting to public/.

5. Set up nginx reverse proxy — Detects if a proxy already exists for the configured port. Writes an nginx site config to /etc/nginx/sites-enabled/. If an SSL certificate exists in /etc/letsencrypt/live/, configures HTTPS with HTTP redirect; otherwise configures HTTP only with an offer to run certbot. Requires sudo.

6. Install systemd service — Copies dataflow.service to /etc/systemd/system/, runs daemon-reload, enables on boot. Requires sudo.

7. Start / restart service — Runs systemctl start or systemctl restart depending on current state. Requires sudo.

8. Stop service — Runs systemctl stop. Requires sudo.

9. Set login credentials — Prompts for username and password, bcrypt-hashes the password via node -e "require('bcrypt')...", and writes LOGIN_USER and LOGIN_PASSWORD_HASH to .env. Requires Node.js and bcrypt npm package to be installed.

10. Uninstall — Reverses everything the other options install, in reverse order: stops/disables/removes the systemd unit, removes the nginx site and reloads nginx, drops the database and its user (prompts for admin credentials), then deletes .env, public/, and node_modules. Lists exactly what it found before doing anything and requires typing delete to proceed. The repository itself is left in place.

Key behaviors:

• All commands that will be run are printed before the user is asked to confirm.
• Actions that require sudo prompt transparently — sudo is not run with -n, so it uses cached credentials or prompts as normal.
• Nginx config files are written via sudo cp from a temp file, then sudo chmod 644 to make them world-readable for status detection.
• Certificate existence is checked with sudo test -f since /etc/letsencrypt/live/ requires root.
• Option 1's result (updated config) is passed back to the menu loop so status reflects changes without requiring a restart.

---

Environment Variables (.env)

DB_HOST              PostgreSQL host
DB_PORT              PostgreSQL port (default 5432)
DB_NAME              Database name
DB_USER              Database user
DB_PASSWORD          Database password
API_PORT             Port the Express server listens on (default 3020)
NODE_ENV             development | production
LOGIN_USER           Username for Basic Auth
LOGIN_PASSWORD_HASH  bcrypt hash of the password

---

Running the Application

# Install API dependencies
npm install

# Install UI dependencies and build
cd ui && npm install && npm run build && cd ..

# Start (development, auto-reload)
npm run dev

# Start (production)
npm start

The server binds to 0.0.0.0 on API_PORT and serves both the API and the compiled UI from public/.

---

Deploying SQL Changes

Any time SQL functions are modified, run python3 manage.py and choose "Redeploy SQL functions only". It runs every function file in dependency order — the list lives in QUERY_FILES in manage.py, which is the one place the order is defined.

To deploy a single file by hand:

PGPASSWORD=<pass> psql -h <host> -U <user> -d <db> -v ON_ERROR_STOP=1 -f database/rules.sql

Deployment is safe to repeat — every function uses CREATE OR REPLACE.

The files are the source of truth. Editing a function directly in the database, without writing the change back to its file, means the next redeploy silently reverts it.

Schema changes (schema.sql) drop and recreate the schema, deleting all data. In production, write migration scripts instead.