Case studiesReference integration: partner file to database

Partner CSV files to Postgres: each file imported once

How a flow picks up a partner’s price file from an SFTP server, checks the server’s identity, imports each row once even when the file is retried, adds names from a code list, upserts into Postgres, and parks a row it cannot match for correction.

Why file imports go wrong

File drops are simple until something repeats or half-finishes. The importer crashes after row 400 and the file is read again from the top. A partner re-uploads yesterday’s file. One row names a product the catalogue has never heard of. Someone stands up a server on the partner’s hostname.

The reference flow handles each of these in a different node, so the import can be retried safely and the odd row ends up in front of a person instead of stopping the file.

The flow

Nodes in the SFTP import flow, in order

sftp: price file
once per file and row
add product name
sql: upsert

biz-sftp-import, source node

{
  "match": "sftp:partner-prices",
  "sftp": {
    "host": "sftp.partner.example",
    "username": "opexflow",
    "private_key": "${secret:partner_sftp_key}",
    "host_key": "SHA256:…",
    "dir": "/outbound",
    "pattern": "PRICES_*.csv",
    "format": "csv",
    "after": "archive",
    "archive_dir": "/outbound/done",
    "interval_secs": 120
  }
}

The source polls the directory every 120 seconds for files named PRICES_*.csv and delivers one record per CSV row.
A dedupe node keys on the file’s identity plus the SKU, so a retried file does not import a row twice. Keys are kept for 30 days.
A lookup node adds the product name and category from a code list. An unknown SKU fails the row.
A SQL node upserts the row into the products table, inserting a new SKU or updating the name, category and price of an existing one.
When every row of a file has been handled, the file moves to the archive directory.

Postgres table and a sample row

CREATE TABLE products (sku text PRIMARY KEY, name text, category text, price numeric);

# file PRICES_1.csv
sku,price
SKU-100,12.50

What each node guarantees

Situation and behaviour
Situation	What happens
The server presents a different host key	The connection is refused. Host key pinning is required and trust on first use cannot be configured. The example ships a placeholder fingerprint that no server has, so it cannot run against a real server until the real one is set.
The same file is seen again	A processed-file ledger keyed by path, size and modification time delivers each file once.
The flow fails part way through a file	The whole file is retried, up to 5 times. The dedupe key is committed only after the whole downstream of a row succeeded, so rows already imported are skipped and the rest go through.
A row names an unknown SKU	The lookup fails the row into the dead-letter queue, where it can be corrected and replayed.
A row is imported twice anyway	The upsert is keyed on the SKU, so the second write updates the same row.
A file is still being written	Files younger than min_age_secs are left alone. The example does not set it. Set it when the partner uploads in place.
The file is done	It is archived or deleted only after the flow settled all its records.

To read the pinned fingerprint from a server, run ssh-keyscan -t ed25519 host | ssh-keygen -lf -. The example’s install form checks the value against the SHA256 fingerprint format.

Tested cases

The example is covered by the repository’s example tests. Every example is deployed in three shapes (defaults, optional nodes enabled, mock switched off) and must compile without an error. Its sample record then runs through the real executor as a dry run. In this example the SFTP server is a local mock directory, so a file dropped into the mock outbound folder is picked up without a server.

Deploy: the flow compiles and the only open items are the ones that belong to your systems.
Dry run: a sample row (SKU-100, price 12.50) runs through the real executor. Only a step that needs your own database may report a problem.
Not covered: a write to a real Postgres database. Postgres has no mock, so the upsert is exercised against your database.

Scope

The mock directory stands in for the SFTP server. With the mock off, the host, the user, the private key secret and the pinned fingerprint are set per installation.
The SFTP connector does not support host certificates. The SQL node supports Postgres only, one statement per call.
A mocked SFTP node is refused in a protected environment.
The product code list ships with the example. Replace it with your catalogue.