Integrations

Everything outside the Flow core is an integration behind one contract, the integration port (server/integrations/port.js). VISION.md, "The core and its integrations", says why. This page says how an adapter is built, how an installation turns it on and off, and how it owns its tables.

An adapter #

An adapter is a plain object that defineAdapter checks:

import { fileURLToPath } from 'node:url';

export const exampleAdapter = {
  id: 'example', // lower case, letters, digits and dashes
  name: 'Example',
  bundled: false, // true: ships on in a preset
  schema: {
    migrationsFolder: fileURLToPath(new URL('./example/migrations', import.meta.url)),
    tables: ['example_cases', 'example_sync_state'],
  },
  capabilities: {
    health: async () => ({ ok: true }), // required
    pullWorkItems: async (cursor) => { /* … */ },
  },
  start: async ({ db, log }) => { /* optional */ },
  stop: async () => { /* optional */ },
};

The capabilities are pullWorkItems, pullHistoryJobs, pushInvoice, pushBill, pullPayments, checkInvoice, listCustomers, createCustomer, findInvoices, listWork, billableJobs, listPriceItems, jobIssues, moveWork, cancelMove, personSettings, setPersonSetting, captureError, browserMonitoring, traceSpan, traceRequest, alerts and health. alerts({ organisationId }) answers what only a person can clear, for the quiet alert on the Integrations page (docs/ALERTS.md). listPriceItems({ organisationId }) answers the system's own price list from the adapter's copy, never a live call: { kind, syncedAt, items }, kind accounting or work, each item { id, code, name, unitAmountCents, taxCode, sold } (docs/PRICE_LIST.md). listWork({ organisationId }) answers the work system's jobs for the Jobs board from the adapter's own copy, never a live call: { items, statuses }, each item with its sourceSystem, sourceJobRef, status, clientName, practitioner, dates, lines and total (Seazona: open orders and orders finished in the last 30 days, never one deleted in Seazona). An item may carry link, { url, exact }, which opens the job in the work system's own web app (exact is false for only its home page), and pendingMove, { to, from, requestedAt, expiresAt, openUrl, openExact }: a stage move the work system has not made yet. jobLink({ sourceJobRef }) answers that same link for one job, or null; it never calls the system. moveWork({ sourceJobRef, toStatus, personId }) takes a person's move of a job to another stage on the Jobs board and answers what happens; an adapter whose API cannot change a stage answers a hand-off, { kind: 'hand_off', from, to, openUrl, expiresAt } (Seazona). cancelMove({ sourceJobRef }) takes back a move not yet made. personSettings({ personId }) answers each person's own choices in the integration, [{ key, label, hint, kind, value }], and setPersonSetting({ personId, key, value }) changes one (docs/JOB_BOARD_MOVES.md). jobIssues({ refs }) answers the integration's open issues for those job refs, { [ref]: [{ integration, phrase, href }] }: one short phrase and the page where a person fixes it; the Jobs board shows them on each job (Seazona: history jobs a person must decide). billableJobs({ organisationId, from, to }) answers the billable jobs in the adapter's own copy whose job day is in that range, never a live call: { jobs }, each { ref, day, subtotalCents } (subtotalCents null when a line has no amount); the daily reconcile counts them (docs/RECONCILE.md). captureError({ message, source, error }) takes each error Flow logs, and browserMonitoring() answers the public settings a browser needs. traceSpan({ name, op, attributes }, work) runs work inside a span of the current performance trace, and traceRequest() answers a function that names the request's trace by its route (server/core/tracing.js). The Sentry adapter has all four (docs/SENTRY.md). listCustomers({ organisationId }) answers the system's customers as { id, name }. createCustomer({ organisationId, contactId, actor }) creates that system's customer from the Flow contact and answers { outcome, contact }. Xero does this when Create missing clients is on (docs/CONTACTS.md). findInvoices({ organisationId, invoiceIds, numbers, flowInvoiceIds, contactId, from, to }) answers sales invoices from the adapter's own copy, never a live call, with the copy's state (complete, fresh, changedAt). The pipeline uses both (docs/JOB_INVOICE_LINKS.md). checkInvoice({ organisationId, obligation }) answers whether the accounting system holds an invoice for a billable job, in the ledger's evidence shape; it is the live check before Flow drafts a job's invoice (POST /api/invoicing/jobs/:id/draft). deleteInvoice({ organisationId, invoiceId, actor }) reads that system's invoice live and answers { state, status }: deleted, gone, approved or unavailable. The core voids an issued Flow invoice only after deleted or gone (docs/XERO.md, "Deleting a draft"). An adapter declares only the ones it has, and must declare health.

An adapter may also declare routes(context): a function that returns an Express router, with context { db, personAuth, log }. Flow mounts it at /api/<id> (registry.routes in server/integrations/port.js), after the /api rate limit. The router is built only for an adapter that is on with its tables in place; for one that is off, every request under /api/<id> gets the turned-off answer (below) with status 503. The router does its own sign-in checks with personAuth.

Two optional functions give an adapter a lifecycle:

• start({ db, log }) runs once at server start, only while the adapter is on, after its tables are in place. Keep the database handle here and start background work (a scheduler, a poller). When a person saves the adapter's settings or turns it on, it runs again with reconfigured: true, so an adapter can clear a stop that waits for a person (for example after its API client was refused).
• stop() ends that work. registry.stop() calls it for every adapter that started, newest first. Tests call it so they end cleanly.

To ship an adapter, add it to the list in server/integrations/adapters.js. Core code never imports an adapter; it asks the registry.

Turn an integration on or off #

An owner turns an integration on in Flow, in Integrations (below). A host can also turn one on with FLOW_INTEGRATIONS: the adapter ids, separated by commas (seazona,xero). Unset or empty: none. An unknown id stops boot, so a typo never leaves an integration silently off.

Set up in Flow #

flowchart LR
  owner[Owner: Integrations] -- settings and credentials --> api[PUT /api/integrations/:id]
  api -- seal credentials<br/>with the secrets key --> db[(integration_settings)]
  api --> apply[Stop the adapter<br/>create its tables<br/>start it with the new settings]
  boot[Flow starts] --> read[Read integration_settings] --> merge[Host environment over<br/>stored settings] --> start[Start each adapter that is on]

• An adapter declares setup: the settings an owner may enter (fields, each named by its environment setting), a short guide, and derivedKeys. Xero and Seazona declare theirs in their index.js.
• Only an owner sees the settings (GET /api/integrations). A change needs a verified owner sign-in (PUT /api/integrations/:id). It applies at once: no restart and no host secret to edit.
• A credential is write-only. Flow seals it with AES-256-GCM under its secrets key, bound to the integration and the setting's name, and stores only the sealed value. The screen shows whether it is set, never the value. Every change writes an integration_setting_events row with the setting names and who changed them, never a value.
• The host's own environment wins: a setting set there shows as "Set by the host" and cannot be changed in Flow.
• A field with advanced: true, and every setting the host sets, shows in the Advanced fold under Settings. The fold is open while the integration is off, so Connect shows every setting.
• An integration turned on in Flow that cannot start (a setting is missing) shows the reason and Flow keeps running. One turned on by the host stops boot, as before.
• derivedKeys (Xero's XERO_TOKEN_KEY) come from the secrets key, so no one makes them by hand.

The secrets key #

FLOW_SECRETS_KEY is the key: 32 bytes, as 64 hex characters or base64 (openssl rand -hex 32). When it is unset, Flow derives the key from SESSION_SECRET when that is at least 32 characters. With neither, Flow stores no credential and the screen says so. Changing the key, or its source, makes stored credentials and Xero's stored sign-in unreadable: an owner enters the credentials again and connects Xero again. A copy of the database alone opens no credential, because the key is never stored there.

Tables #

An installation's database holds only the tables of the integrations it has turned on (issue #13).

flowchart LR
  boot[Flow starts] --> core[Core migrations]
  core --> each{For each adapter}
  each -- on --> migrate[Run its migrations<br/>check its tables exist]
  each -- off --> skip[Touch nothing]
  migrate --> ready[ready]
  skip --> off[turned off answer]

The rules:

• The adapter owns its schema. Its tables are not in server/db/schema.js. It keeps its own drizzle migrations in schema.migrationsFolder (a meta/_journal.json and SQL files, as npx drizzle-kit generate writes them) and its own migration journal, drizzle.__drizzle_migrations_<id>.

• Every table starts with the adapter id, dashes as underscores (my-lab owns my_lab_…). defineAdapter refuses any other name, so an adapter never touches a core table or another adapter's.

• schema.tables lists every table the migrations create. After it migrates, Flow checks each one exists and stops boot if one is missing.

• Tables are created only while the adapter is on. At start, after the core migrations, registry.start(db) runs the migrations of each enabled adapter, and of no other.

• Turning an adapter on later creates its tables at the next start. There is no manual step. Until that start it answers as turned off.

• Turning an adapter off never drops its tables. Its rows stay, and turning it on again finds them.

• Removing the tables is a separate, explicit step. It refuses while the integration is in FLOW_INTEGRATIONS and while any of its tables holds a row, and removes everything in one transaction:

  DATABASE_URL=postgres://… node server/integrations/remove-tables.js <id>
  

Read an adapter's tables #

Go through the registry, never straight to the tables:

const result = await registry.whenEnabled('example', (adapter) => adapter.capabilities.pullWorkItems(cursor));

whenEnabled runs the function only when the adapter is on and its tables are in place. Otherwise it returns the adapter's turned-off answer, and code shows that answer instead of crashing:

{ "ok": false, "code": "integration_off", "integration": "example", "reason": "disabled", "error": "Example is turned off on this installation." }

registry.with(capability) (the core's fan-out) also returns only adapters that are on with their tables in place. registry.list() gives each adapter's state: disabled, ready, restart_needed, or not_started before the first start.

Outbound calls #

A provider can cut off an installation's API access after many failed calls, even inside its rate limit. So every adapter that calls an outside system sends each call through the call guard (server/integrations/call-guard.js, issues #101 and #111). The guard names no vendor. Each adapter brings its own tables and its own policy.

call ──> guard.check ──> rate limit ──> provider ──> guard.record
             │                                          │
             └── stopped: CallsStopped, no call         └── failure? update circuit,
                                                             budget, halt; report once

The guard stops calls in four ways:

A failed call is no answer, a 5xx, a 429, or any other 4xx except 404 (a 404 on one record means it was deleted). A stopped call throws CallsStopped at once (kind: 'calls_stopped', code <id>_api_halted, <id>_api_resource_stopped, <id>_api_circuit_open or <id>_api_failure_budget_spent, retryAfterMs null for a stop only a person ends). Each stop is reported once with log.error, so the connected error service gets one report per problem; a repeat is a warning. Reports carry the resource and the HTTP status, never a host, a query, a token or a response body.

To follow the pattern, an adapter:

1. Creates two tables in its own migrations, with the columns in CALL_GUARD_COLUMNS: a state table (one row per credential or tenant: key, circuit_open_until, circuit_opens, halted_reason, halted_at, stopped_resources, updated_at) and a calls table (one row per call: key, at, resource, request_path, status_code, failure_kind, duration_ms). The state is in PostgreSQL, so every process on one database shares it and a restart does not clear it.
2. Declares callGuard: { integration, name, tables: { state, calls }, policy } on the adapter. defineAdapter checks that the tables are the adapter's own.
3. Calls guard.check(key, { resource }) before each call and guard.record(key, { resource, path, statusCode | failureKind, durationMs }) after it, and guard.resume() from start() when reconfigured is true.
4. Treats CallsStopped as "wait", not as a failure: a background job does not spend its attempts on it, and a batch stops at the first one.
5. Shows guard.status(key) in its health(), and answers guardAlerts(name, await guard.status(key)) from alerts().

test/integration-call-guard.test.js fails when a shipped adapter has no call guard and is not listed as exempt with a reason, and runs the same contract suite (test/support/call-guard-contract.js) against each guarded adapter's tables and policy. Xero and Seazona use the guard. Sentry is exempt: its SDK follows Sentry's own limits.

An adapter that admits each call in its own transaction (Seazona's rate governor writes the call row when it admits a call) passes lockName so the guard takes the same lock, calls guard.stoppedNow(tx, key) inside that transaction, and passes the row's id as callId to record. Its existing tables keep their names: columns maps key and at (Seazona: client_key, admitted_at), and callDefaults fills other required columns of a new row. report: false is for an adapter that already reports each stop once at a higher level (Seazona's lanes).

Button presses #

A person's button press that calls an outside system (a copy, a recheck, a send) is one action. Pressed again and again, it spends the provider's rate limit, and can trip the call guard. So each such route also goes through the action guard (createActionGuard in server/integrations/call-guard.js), after sign-in and before the handler.

press ──> sign-in ──> action guard ──> handler ──> call guard ──> provider
                          │                                         │
                          └── too soon: 429 + wait, no call          └── answer sets the next wait

Every 429 carries Retry-After and retryAfterMs. A good run says its wait in the Flow-Cooldown-Ms header. The page shows the wait on the button (holdButton in server/ui/app.js), so a person sees when it is ready.

The waits each adapter declares:

Check prices (POST /api/invoicing/price-list/check) reads Flow's copies only, so it has no action. The Seazona routes call nothing at a press (a backfill is paced by the rate governor), so they have no action. A new route that calls an outside system at a press declares an action. The state is in memory: one Flow process serves one lab, as for the request limit (server/rate-limit.js); the call guard still protects every process. createApp({ buttonCooldowns: false }) turns the waits off for tests only (test/action-guard.test.js tests them).

Health #

GET /health lists every adapter under integrations with enabled and state. When an adapter is ready its health() answer is there too, so it must read Flow's own data and answer fast: an answer slower than 5 seconds, or an error, shows as { ok: false, error }. An integration problem never fails Flow's own health check (server/integrations/health.js).

Tests #

test/integration-schema.test.js proves the rules against PostgreSQL with a synthetic adapter (test/fixtures/sample-adapter/): a start with it off creates none of its tables, a start with it on creates all of them, turning it off keeps them and their rows, and removal refuses while a table holds a row. An adapter's own tests follow the same pattern: boot it off and assert none of its tables exist, boot it on and assert all of them do.