Integrations

# Integrations

## Overview

The module lives in `features/integrations/`. It provides Amble's
third-party integration framework: a registry for code-backed connector
definitions, a declarative connector spec compiler, connection resolution,
raw-object ledger sync, webhook verification, scheduled sync actions, and admin
UI/server actions.

The integration framework converts external provider records into Amble
entities, metrics, documents, and operation proposals. It is intentionally
tenant-scoped: connector specs live in `integration_connectors`, provider
connection rows live in `agent_connections`, and sync run history lives in
`integration_runs`.

## Key Concepts

**IntegrationDefinition** - the runtime sync contract from
`features/integrations/framework/source-contract.ts`:

```ts
export interface IntegrationDefinition<TCursor = unknown, TRaw = unknown> {
  slug: string;
  label: string;
  connectionPresetIds: string[];
  resources: IntegrationResourceDefinition<TRaw>[];
  operations?: IntegrationOperationDefinition[];
  pull(
    args: IntegrationPullArgs<TCursor>,
  ): Promise<PulledIntegrationObject<TRaw>[]>;
}
```

The `pull()` function fetches provider objects. Each resource definition maps
those objects into entities, metrics, documents, or conflicts.

**Definition registry** - `features/integrations/framework/definition-registry.ts`
stores code-backed definitions in memory:

```ts
export function registerIntegrationDefinition<TCursor, TRaw>(
  definition: IntegrationDefinition<TCursor, TRaw>,
): IntegrationDefinition<TCursor, TRaw>;

export function getIntegrationDefinition(
  slug: string,
): IntegrationDefinition<unknown, unknown> | null;

export function listIntegrationDefinitions(): IntegrationDefinition<
  unknown,
  unknown
>[];
```

`features/integrations/register-builtin-definitions.ts` imports provider
definition files for their registration side effects. The generic admin sync
route imports that barrel so cold serverless invocations can resolve Acuity,
Plaid, QuickBooks, Ramp, reMarkable, and social connectors.

**Declarative connector specs** - tenant-authored specs are validated by
`ConnectorSpecSchema` in `features/integrations/declarative/schema.ts` and
compiled by `compileConnectorSpec()`:

```ts
export function compileConnectorSpec(
  spec: ConnectorSpec,
): IntegrationDefinition<unknown, unknown>;
```

Declarative specs can define REST, MCP tool, feed, or web requests; pagination;
entity or metric targets; field maps; schedules; webhook verification; and MCP
exposure flags.

**Connection resolution** - `resolveIntegrationConnections()` reads
`agent_connections` for a tenant, optional workspace, preset IDs, group key, and
optional connection ID. It returns `ResolvedIntegrationConnection[]` values
without secrets. If a definition has no `connectionPresetIds`, the framework
returns a synthetic keyless connection.

**Descriptors** - `IntegrationDescriptor` in
`features/integrations/framework/descriptor.ts` is the UI-facing metadata shape:
`id`, `label`, `externalSourcePrefix`, `connectionPresetIds`, `syncEndpoint`,
optional `lastSyncSettingKey`, and resource field mappings.

## Architecture And Data Flow

### Definition Resolution

`resolveIntegrationDefinition(tenantId, slug)` first checks the code-backed
registry. If no code definition is registered, it loads an enabled
`integration_connectors` row for the tenant, validates `row.spec` with
`ConnectorSpecSchema`, compiles it with `compileConnectorSpec()`, and caches the
compiled definition by `tenant_id`, `slug`, and connector `version`.

```ts
export async function resolveIntegrationDefinition(
  tenantId: string,
  slug: string,
  options: { db?: ResolverDb } = {},
): Promise<IntegrationDefinition<unknown, unknown> | null>;
```

`listAvailableIntegrations(tenantId)` returns enabled tenant connector specs
plus all code-backed definitions currently registered in memory.

### Manual Sync Route

The generic admin route is
`POST /api/admin/integrations/[slug]/sync`. It is implemented in
`app/api/admin/integrations/[slug]/sync/route.ts`.

The route:

1. Calls `requireAdmin()` and reads the active workspace if present.
2. Resolves the integration definition by slug.
3. Resolves matching connections for the tenant and workspace.
4. Runs `runMultiConnectionSync()` with `triggeredBy: "manual"`.
5. Returns `aggregateMultiConnectionSummary(summary)`.

QuickBooks connections are additionally filtered with `hasQuickBooksRealmId()`.

### Self-serve finance connections

Sprinter and Marbella register the same tenant-neutral custom page at
`/p/finance-connections`. It projects only safe QuickBooks company and Plaid
bank labels, provider health, and tenant-scoped run freshness. It never returns
credential ciphertext, provider access tokens, raw provider payloads, or
protected connection references.

Authenticated viewers can inspect the roster. Add, per-account sync, and
reauthorization/update affordances render only when
`hasIntegrationConnectAccess()` succeeds; the matching server routes enforce
`requireIntegrationConnectAccess()` again. That policy admits admins and
admin-permission holders, plus member-tier users only when the tenant has
enabled `integrations_self_serve`.

The self-serve sync endpoint,
`POST /api/integrations/[slug]/sync`, validates a strict bounded request and
delegates to the same definition resolver, exact tenant/workspace/connection
resolver, readiness filter, and `runMultiConnectionSync()` used by Admin. The
customer surface is not a second integration engine.

### Raw Sync And Ledger

`runRawIntegrationSync()` in `features/integrations/framework/run-raw-sync.ts`
starts an `integration_runs` row, calls the definition's `pull()`, captures
provider objects through the raw ledger, maps each object, writes entities via
`upsertEntityKeyed()`, records metrics through `recordIntegrationMetrics()`,
and updates connection health when using the default store.

`runMultiConnectionSync()` fans that raw sync across each resolved connection.
One failing connection is isolated into `perConnection[].error`; other
connections still run, and totals aggregate successful runs.

### Webhook Receipt

Inbound integration webhooks arrive at:

```txt
POST /api/webhooks/integrations/[tenantSlug]/[slug]
```

The route in `app/api/webhooks/integrations/[tenantSlug]/[slug]/route.ts`
resolves the tenant and enabled `integration_connectors` row, validates the
connector spec, verifies the request with either a provider verifier or the
configured `standard_webhooks` / `hmac_sha256` verifier, maps the provider event
name through `spec.webhook.triggers`, and emits:

```ts
await inngestActions.send({
  id: deliveryId,
  name: EVENT_NAMES.INTEGRATIONS_WEBHOOK_RECEIVED,
  data: {
    tenantId: tenant.id,
    slug: spec.slug,
    resource,
    deliveryId,
  },
});
```

`features/inngest/functions/integration-webhook-received.ts` consumes that
event, gates with `isCronAllowed()`, scopes the definition to the resource from
the webhook trigger, resolves connections, and runs
`runMultiConnectionSync({ mode: "apply", triggeredBy: "webhook" })`.

### Scheduled Sync

The scheduled sync action is the action definition
`integrations.scheduledSync` in
`features/actions/library/builtins/integrations.ts`:

```ts
export const integrationsScheduledSyncActionDefinition =
  registerActionDefinition(
    defineAction({
      key: "integrations.scheduledSync",
      version: "1",
      name: "Scheduled integration sync",
      description:
        "Run a registered integration sync for every matching connection.",
      // ...
    }),
  );
```

`features/integrations/server/schedule-editor-actions.ts` creates or updates a
cron `actions` row with `definition_key: "integrations.scheduledSync"`,
`trigger_type: "cron"`, `trigger_config: { expression: cronExpression }`, and
`input_config: { slug: connectorSlug }`.

> **Trigger-config key note:** `trigger_config` is written under two keys
> across the codebase — the schedule editor, the canonical Zod schema
> (`CronTriggerConfigSchema`), `upsert-task`, and loops all persist
> `expression`, while direct-DB writers persist `schedule`. The cron scanner in
> `features/inngest/functions/action-cron.ts` is the single runtime source of
> truth, and it resolves `trigger_config.schedule ?? trigger_config.expression`
> — so schedule-editor rows fire regardless of which key was persisted, and no
> back-fill migration is required.

When cron fires, [Inngest](/docs/features/inngest) routes the action through
`action-tick`, executes the action definition, resolves the definition and
connections, and runs `runMultiConnectionSync()` in `apply` mode.

## Connection Readiness Gating

Some connectors (QuickBooks) persist a connection row before the provider
handshake supplies the identity that makes the row syncable — a QuickBooks
OAuth stub carries the `quickbooks-online` preset but no Intuit `realmId`
until the callback completes. Rather than let a half-finished row enter the
sync fan-out (and either throw inside `pull()` or, worse, silently collapse
against another org via a coincidental match), `IntegrationDefinition` carries
an optional per-connection gate:

```ts
connectionReady?(connection: {
  id: string;
  config: Record<string, unknown>;
}): { ready: true } | { ready: false; reason: string };
```

`runMultiConnectionSync()` (`features/integrations/framework/run-multi-connection-sync.ts`)
calls `connectionReady` for every resolved connection before it reaches
`pull()`. A not-ready connection is **skipped**, not treated as a failure: it
is counted in `totals.skipped`, recorded in `perConnection[].skippedReason`,
and surfaced by `aggregateMultiConnectionSummary()` as a `connection`-resource
conflict so operators see it in the sync results panel instead of it silently
vanishing. Because every fan-out entry point (the admin sync route, MCP tool,
scheduled action, webhook worker) funnels through `runMultiConnectionSync`,
this is the single gate for the invariant.

`features/integrations/quickbooks/definition.ts` implements the gate:

```ts
connectionReady: (connection) =>
  hasQuickBooksRealmId(connection.config)
    ? { ready: true }
    : {
        ready: false,
        reason:
          "QuickBooks company is not linked yet (no realmId). Reconnect to finish setup.",
      },
```

## QuickBooks: The realmId Invariant

A QuickBooks connection row is only syncable once its `config.realmId` (the
Intuit company id) is captured — that only happens after the OAuth callback
completes. Several layers enforce this end-to-end:

- **`hasQuickBooksRealmId()` / `isRealmlessQuickBooksConfig()` / `isPendingQuickBooksOAuthStub()`**
  (`features/integrations/quickbooks/realm.ts`) are the shared predicates used
  by the connect route, the sync route, the connections resolver, and the
  revocation path — previously duplicated three ways.
- **Connection writers downgrade realm-less rows to `inactive`.**
  `createConnection()` and `updateConnection()`
  (`features/agents/server/connection-actions.ts`) never let a QuickBooks
  config without a `realmId` land at `active` — the sync reader throws
  `"missing config.realmId"` on it otherwise.
- **Fail-closed OAuth completion.** `completeConnectionOAuth()`
  (`features/agents/server/oauth-actions.ts`) checks the preset's declared
  `callbackConfigParams` (e.g. `realmId`) BEFORE exchanging the authorization
  code. If the provider callback did not include a declared param, the
  connection is written to `status: "error"` with an actionable
  `last_error` and the exchange never happens — a connected-looking row that
  can never sync is worse than a visibly-errored one.
- **Twin adoption on re-auth.** `findCallbackIdentityTwin()` looks for an
  existing connection of the same tenant + preset that already carries the
  captured identity (e.g. the same `realmId`). When a twin exists, the OAuth
  exchange targets the twin's row (preserving its id, sync cursors, and
  history) instead of promoting the stub into a duplicate that the
  provider-identity unique index would reject. The now-redundant initiator is
  cleaned up through the safe-disconnect helper (see below), never a plain
  delete, because a re-auth on an existing live connection can still be
  holding a live grant.
- **Structured error codes survive OAuth failures.** `buildOAuthResponseError()`
  attaches the provider's wire `.error` code (e.g. Intuit's `invalid_grant`) to
  the thrown `Error` as `.code`. Health classification
  (`connectionHealthFromError` in `features/integrations/framework/connection-health.ts`)
  reads `.code` first and case-insensitively — `error_description` is
  human prose and must never be the basis for classifying `needs_reauth` vs.
  `error`.

## Provider-Identity Uniqueness

A QuickBooks company (`config->>'realmId'`) and a Plaid Item
(`config->>'itemId'`) may each be linked at most once per tenant. Duplicate
rows don't double-sync — the raw ledger dedups on `connection_key` — but they
break resolution: the Plaid webhook resolver throws unconditionally on a
multi-match, and the QBO collapse throws on an exactly-tied pair.

Partial unique indexes back this invariant at the DB layer
(`supabase/migrations/20260724010000_agent_connections_provider_identity_uniqueness.sql`):

```sql
create unique index if not exists uq_agent_connections_qbo_realm
  on public.agent_connections (tenant_id, ((config->>'realmId')))
  where (config->>'presetId') = 'quickbooks-online'
    and coalesce(config->>'realmId', '') <> '';

create unique index if not exists uq_agent_connections_plaid_item
  on public.agent_connections (tenant_id, ((config->>'itemId')))
  where (config->>'presetId') in ('plaid-bank-account', 'plaid')
    and coalesce(config->>'itemId', '') <> '';
```

The predicates exclude half-finished OAuth stubs (no `realmId`/`itemId` yet) —
those are legitimate transient duplicates during connect/reconnect. Writers
upsert-by-identity first (`connect-writer`'s `reuseBy`, OAuth completion's
twin adoption); the indexes are the DB backstop for any writer that bypasses
them. A `23505` violation on insert is mapped to an operator-safe "This
account is already connected" message in
`features/integrations/server/connect-writer.ts` rather than surfacing the
raw constraint error.

## Safe Disconnect And Revocation Ordering

Deleting an `agent_connections` row is the only way a user-facing disconnect
happens, but the DB row is not the only place a provider grant lives — an
Intuit refresh token stays valid for ~100 days after the row is gone, and a
Plaid Item stays linked (and billed) until `/item/remove` is called. Deleting
the row first and revoking after (or not at all) orphans the grant.

`revokeThenDeleteConnection()`
(`features/integrations/server/safe-disconnect.ts`) enforces
**claim → revoke → version-guarded delete**:

1. Claim the row with an optimistic-concurrency (`updated_at`-matched) update
   to `status: "inactive"` / `status_code: "disconnect_pending"`.
2. Call `revokeProviderAccess()` (`features/integrations/server/revoke-provider-access.ts`)
   against the claimed snapshot — Intuit token revocation for QuickBooks,
   `itemRemove()` for Plaid. A provider response that proves the grant is
   already dead (Intuit's `invalid_grant`/`invalid_token`, Plaid's
   `ITEM_NOT_FOUND`) counts as safe-to-delete without having "revoked"
   anything new.
3. Delete the row ONLY if revocation reported `ok: true`, guarded by the same
   `updated_at` the revoke call used — any concurrent write (token rotation,
   another disconnect) bumps `updated_at` via the `set_updated_at` trigger and
   aborts the delete rather than removing credentials newer than the ones just
   revoked.

A failed revocation (`revoke_failed`) or a detected rotation (`rotated`) keeps
the row — visible and retryable — instead of silently orphaning the grant.
`deleteConnection()` in `features/agents/server/connection-actions.ts` and the
OAuth twin-adoption cleanup in `oauth-actions.ts` both route through this
helper rather than a plain delete.

## QuickBooks Transient Retry

`createQuickBooksClient().query()` (`features/integrations/quickbooks/client.ts`)
retries on the Intuit statuses that are safe to retry — 429, 500, 502, 503,
504 — honoring a `Retry-After` header when present, clamped to 10s so a
hostile or buggy header can never stall a sync for minutes. Non-transient
statuses still fail fast. Without this, a single throttle or transient 5xx on
any page of the snapshot pull failed the whole connection and flipped its
health to `error`.

## Plaid Webhook Registration And Item Removal

`resolvePlaidWebhookUrl()` (`features/integrations/plaid/client.ts`) resolves
the absolute, https-only URL Plaid should deliver Item webhooks to (an
explicit `PLAID_WEBHOOK_URL` override, or the deployment base URL plus
`/api/webhooks/plaid`; plain-http resolves to `null`, matching prior
no-webhook behavior in local dev). `createLinkToken()` threads this URL into
`linkTokenCreate()` as `webhook` — without it, Plaid never delivers
`SYNC_UPDATES_AVAILABLE` or `ITEM_LOGIN_REQUIRED` for the new Item, and the
verified webhook receiver stays dead code.

`PlaidProviderClient` also exposes `itemRemove()`, wired into
`revokeProviderAccess()` for Plaid disconnects — the safe-disconnect flow
above calls it before deleting a Plaid connection row.

`readPlaidServerConfig()` fails closed in production: when `VERCEL_ENV` is
`"production"` but `PLAID_ENV` resolves to anything other than
`"production"`, it throws a fixed message that never echoes the misconfigured
value — silently defaulting to sandbox would point live operators at test
data.

## Concrete Provider Example: Plaid

Plaid is a code-backed integration under `features/integrations/plaid/`.

`features/integrations/plaid/definition.ts` registers
`plaidIntegrationDefinition` with the generic registry. At runtime its `pull()`
resolves a tenant/workspace Plaid connection, decrypts the access token inside
`resolvePlaidConnection()`, builds a definition with `buildPlaidDefinition()`,
and delegates to that definition's `pull()`.

`features/integrations/plaid/sync.ts` defines:

```ts
export function buildPlaidDefinition(
  input: BuildPlaidDefinitionInput,
): IntegrationDefinition<PlaidSyncCursor, PlaidRaw>;

export async function syncPlaid(args: SyncPlaidArgs): Promise<PlaidSyncSummary>;
```

The Plaid definition pulls accounts and transactions, maps resources with
`plaidTranslations()`, stores redacted raw-object captures, applies field
provenance and field policies, persists the transaction cursor after successful
`apply` syncs, and records removed transactions as skipped conflicts because
entity archive propagation is deferred in that code path.

## Key APIs

```ts
export {
  compileConnectorSpec,
  deriveIntegrationDescriptor,
  UnsupportedSpecFeatureError,
} from "./declarative/compile";
export { ConnectorSpecSchema } from "./declarative/schema";
export type { ConnectorSpec } from "./declarative/schema";
export {
  getIntegrationDescriptor,
  listIntegrationDescriptors,
} from "./registry";
export * from "./framework";
```

`features/integrations/index.ts` is the public module surface.

```ts
export async function runMultiConnectionSync<TCursor = unknown, TRaw = unknown>(
  args: RunMultiConnectionSyncArgs<TCursor, TRaw>,
): Promise<MultiConnectionSyncSummary>;
```

Use this when one integration definition should sync every matching connection.

```ts
export async function runRawIntegrationSync<TCursor = unknown, TRaw = unknown>(
  args: RunRawIntegrationSyncArgs<TCursor, TRaw>,
): Promise<RawIntegrationSyncSummary>;
```

Use this inside provider-specific syncs when the caller has already resolved a
single connection or wants direct control over cursor, store, lookup, or writer
injection.

```ts
export function registerWebhookVerifier(
  slug: string,
  verifier: WebhookVerifier,
): void;
```

Use this for provider-specific webhook verification. Plaid registers its
verifier from `features/integrations/plaid/webhook-verifier.ts`.

## Concrete Provider Example: Ramp

Ramp (`features/integrations/ramp/`) is a client-credentials connector — no
user OAuth and no webhooks; a polling sync with a durable transactions cursor
is the design.

**Connect.** `POST /api/admin/integrations/ramp/connect` (behind
`requireIntegrationConnectAccess()`) accepts `{ clientId, clientSecret,
label? }`, validates the pair with one cheap `listTransactions` call against
Ramp before persisting, then writes a tenant-owned, already-active
`agent_connections` row via `createIntegrationConnection()` — credentials are
sealed with `encryptProviderCredentials()` (v2 AAD-bound envelope) and read
back with `decryptProviderCredentials()` in `definition.ts`. The UI entry is
`RampConnectButton` (a credential dialog), mounted on the admin Integrations
page and the self-serve finance cockpit. Disconnect goes through the generic
connections routes; there is no Ramp-specific delete.

**Sync.** `rampIntegrationDefinition` (registered, resolved by the generic
`[slug]` admin sync route, the self-serve sync route, `runIntegrationSync`,
and scheduled actions) pulls transactions page-by-page, enriching each with
its receipts and accounting sync status. Both resources
(`ramp:ramp-transaction`, `ramp:ramp-receipt-status`) are **ledger**-
materialized (`rampMaterialization()` in `constants.ts`): records come to
rest in `integration_objects` and never mint entities, so no tenant entity
types are required. The capture policy projects the persisted payload through
`rampPersistedPayload()` — an allowlist that keeps only the keys the
normalizers and the persisted-ledger tools read; free-form provider fields
(memos, card metadata) never persist verbatim. Provider ids stay plaintext
because the `cardholderId` tool filter compares them directly. The
transactions cursor persists on connection config with a compare-and-set
guard (`persistCursor` in `definition.ts`).

**Read.** `rampTransactions` and `rampReceiptStatus`
(`features/tools/finance/synced-record-tools.ts`) read the persisted ledger.
`rampTransactions` returns normalized
`amountCents`/`direction`/`currency`/`date`/`merchantName`/`cardholderName`
per row (derived by reusing `normalizeRampTransaction` on the persisted
enriched payload) plus a top-level `amountSemantics` note — Ramp reports card
spend as **positive** (outflow), the opposite of Plaid's `amountMinor`
normalization, so cross-provider sums must check each tool's semantics.

**Cockpit.** `ramp` is a `FINANCE_CONNECTION_PROVIDER_IDS` member, so the
self-serve finance cockpit (`/p/finance-connections` surfaces) lists Ramp
connections with health, freshness, manual sync, and disconnect — the same
surface QuickBooks and Plaid use.

## Agent Instructions

To add a code-backed integration:

1. Create a provider directory under `features/integrations/<slug>/`.
2. Implement an `IntegrationDefinition` with `slug`, `label`,
   `connectionPresetIds`, `resources`, and `pull()`.
3. Register it with `registerIntegrationDefinition()` in `definition.ts`.
4. Add the definition side-effect import to
   `features/integrations/register-builtin-definitions.ts` when the generic
   admin sync route must resolve it.
5. Add an `IntegrationDescriptor` and register it in
   `features/integrations/registry.ts` if the admin UI should show mappings and
   sync controls.
6. If webhooks are supported, add a verifier in
   `features/integrations/webhooks/verify.ts` or a provider-specific file, and
   make the connector spec's `webhook.triggers` map provider events to resource
   names.

To add a declarative integration, create or update an `integration_connectors`
row through the admin connector actions or manage-integration tool. The spec
must pass `ConnectorSpecSchema`; enabled specs are resolved and compiled at
runtime.

Do not put provider credentials in descriptors, samples, ledgers, or API
responses. Resolve and decrypt credentials only in server-only connection
modules, and keep every DB read or write filtered by `tenant_id`.