Defining Integrations

An integration is a TypeScript declaration of how Bijection talks to one external system: which HTTP requests it may make, what each response proves, which collections it reads and how often. You write it with defineIntegration in a file inside your bijection/ directory. This integration reads customers from a provider that exposes an ordered change feed at GET changes?after=<checkpoint>:

bijection/crm.ts

import { defineIntegration } from "bijection/server";
import { v } from "bijection/values";

const changesPage = v.object({
  items: v.array(
    v.union(
      v.object({
        op: v.literal("U"),
        id: v.string(),
        seq: v.string(),
        name: v.string(),
      }),
      v.object({ op: v.literal("D"), id: v.string(), seq: v.string() }),
    ),
  ),
  through: v.string(),
  head: v.string(),
  more: v.boolean(),
});

export const crm = defineIntegration({
  http: {
    identity: {
      handler: { member: ["identify"] },
      method: "GET",
      route: [{ kind: "literal", value: "identity" }],
      query: {},
      headers: {},
      response: {
        statuses: {
          200: {
            kind: "json",
            validator: v.object({
              account_id: v.string(),
              incarnation: v.string(),
            }),
          },
        },
      },
      evidence: {
        kind: "identity",
        account: { path: ["account_id"], encoding: "text" },
        incarnation: ["incarnation"],
      },
    },
    changes: {
      handler: { member: ["collections", "customers", "sync", "read"] },
      method: "GET",
      route: [{ kind: "literal", value: "changes" }],
      query: { after: { kind: "input", input: "checkpoint" } },
      headers: {},
      response: { statuses: { 200: { kind: "json", validator: changesPage } } },
      evidence: {
        kind: "feed",
        collection: "customers",
        items: ["items"],
        discriminator: ["op"],
        variants: {
          U: { kind: "replace", key: ["id"], position: ["seq"] },
          D: { kind: "delete", key: ["id"], position: ["seq"] },
        },
        through: ["through"],
        head: ["head"],
        has_more: ["more"],
      },
    },
  },
  async identify(ctx) {
    const response = await ctx.http.request("identity");
    if (!response.body) throw new Error("identity response carried no body");
    return { ...response.body, evidence: response.evidence };
  },
  collections: {
    customers: {
      schema: { external_id: v.string(), name: v.string() },
      protocol: {
        kind: "change_feed",
        position: { domain: "customer_changes", encoding: "uint64_decimal" },
        checkpoint: {
          initial: "0",
          continuity: "complete_prefix",
          retention: { kind: "stated", unit: "days", value: 90 },
        },
        records: "full_or_delete",
        atomic_group: "one_change",
      },
      sync: {
        every: { minutes: 5 },
        async read(ctx, _input: { checkpoint: string }) {
          const response = await ctx.http.request("changes");
          if (!response.body) throw new Error("changes response carried no body");
          return {
            evidence: response.evidence,
            changes: response.body.items.map((change, item_index) =>
              change.op === "U"
                ? {
                    kind: "replace" as const,
                    key: change.id,
                    position: change.seq,
                    value: { external_id: change.id, name: change.name.trim() },
                    evidence: response.evidence,
                    item_index,
                  }
                : {
                    kind: "delete" as const,
                    key: change.id,
                    position: change.seq,
                    evidence: response.evidence,
                    item_index,
                  },
            ),
            next_checkpoint: response.body.through,
            observed_head: response.body.head,
            has_more: response.body.more,
          };
        },
      },
    },
  },
  commands: {},
});

It has four parts:

http declares every request the integration may make.
identify reads which provider account a connection speaks for.
collections declares each kind of record, its schema, and its sync.
commands declares the changes the integration can send back to the provider. This one declares none.

Nothing in the declaration names a host or a credential. The provider’s base URL and credential belong to the connection, so the same definition works against a test account and a production account.

Declaring an integration

defineIntegration must be imported by name from bijection/server, called directly, and assigned to an exported const:

bijection/crm.ts

import { defineIntegration } from "bijection/server";

export const crm = defineIntegration({ … });

The file’s path and the export name form the integration’s address, crm.js:crm here. Connections and tables refer to the integration by this address, so export each integration exactly once and keep it in the same module. Moving a definition to another module requires its installations to be recreated. The collections become properties of the returned object. crm.customers is the handle a synced table binds to. A collection can’t be named after a built-in property of the integration, such as commands.

Definitions from another directory

A definition written outside your bijection/ directory, for example in a shared package, has no address in your deployment. Admit it from a module of your deployment with admitIntegration:

bijection/storefront.ts

import { admitIntegration } from "bijection/server";
import { storefront as declaration } from "../shared/storefront";

export const storefront = admitIntegration(declaration);

The admitted definition takes the admitting module’s address, storefront.js:storefront, and .source(...) binds its collections exactly as it binds a local definition. admitIntegration accepts only the imported definition itself, and a definition can be admitted from one module only.

HTTP contracts

Each entry of http is a named contract for one request. Your handlers make a request by naming its contract:

const response = await ctx.http.request("changes");

The host builds the request from the contract, sends it to the connection’s base URL with the connection’s credential, and retains the response before your code sees it. Your code never chooses the host, the path outside the declared route, or the credential.

Field	Meaning
`handler`	The one handler allowed to use this contract, such as `["identify"]`
`method`	`GET`, `HEAD`, `POST`, `PUT`, `PATCH` or `DELETE`
`route`	Path segments appended to the connection’s base URL
`query`	Query parameters, by name
`headers`	Request headers, by name
`body`	Optional request body
`parameters`	Optional validators for values your code passes to `ctx.http.request`
`response`	The statuses the provider may answer, and the body each one carries
`evidence`	What an admitted response proves

Handlers

handler.member names the handler that owns the contract:

Handler	`member`
`identify`	`["identify"]`
A collection’s sync	`["collections", "<collection>", "sync", "read"]`
A command’s send	`["commands", "<command>", "send"]`
A command’s reconcile	`["commands", "<command>", "reconcile"]`

A handler can only make requests through its own contracts. A request through a contract owned by another handler is refused.

Routes, queries and bodies

Route segments, query values, header values and body fields are templates. Use { kind: "literal", value } for a fixed value and { kind: "input", input } for a value the host supplies:

checkpoint: the change feed position the sync resumes from.
resume: the provider cursor a listing resumes from.
parameter: a value your code passes, checked against parameters.
operation_id, target_key, target_version, argument: values a command sends.

To pass values from your code, declare parameters and address them by path:

bijection/crm.ts

search: {
  handler: { member: ["collections", "invoices", "sync", "read"] },
  method: "GET",
  route: [{ kind: "literal", value: "invoices" }],
  parameters: { status: v.string() },
  query: { status: { kind: "input", input: "parameter", path: ["status"] } },
  headers: {},
  response: { statuses: { 200: { kind: "json", validator: v.any() } } },
  evidence: { … },
},

const response = await ctx.http.request("search", { status: "open" });

Responses

response.statuses lists each status the provider may answer and the body it carries:

{ kind: "json", validator }: the body must parse and match the validator.
{ kind: "none" }: no body.
{ kind: "opaque" }: bytes that are retained but not interpreted, such as an HTML error page.

xml, jsonl and a few provider-specific shapes are also available. A status that is not listed is still retained, but proves only that the provider answered. ctx.http.request resolves to the status, the admitted headers, the typed body and an evidence reference to the retained response. The body type follows the validators you declared. For syncs, response.failures classifies error statuses so a failed read is retried or reported correctly:

response: {
  statuses: {
    200: { kind: "json", validator: v.any() },
    429: { kind: "opaque" },
  },
  failures: [{ status: 429, kind: "transient" }],
},

The kinds are transient, unavailable, insufficient_scope, authentication, signature and throttled.

Evidence

evidence says what an admitted response proves, in terms of paths into its body. Bijection checks your handler’s result against it: a record your code returns must cite the response it came from and its position in that response, and its key must match the key the evidence reads. A handler can’t publish a record a response never contained.

`kind`	Used by
`identity`	`identify`: where the response names the account
`feed`	Change feed collections: items, keys, positions and progress
`records`	Listings and single-record reads
`command`	Command sends and reconciles: the outcome the response states

Other kinds, such as incremental, traversal, projections and query_result, describe more specialized provider APIs. The HttpEvidence type in bijection/server lists them all.

Verifying the account

identify reads which provider account a connection speaks for. It runs when you verify a connection, and the result is the identity every later sync is checked against.

bijection/crm.ts

async identify(ctx) {
  const response = await ctx.http.request("identity");
  if (!response.body) throw new Error("identity response carried no body");
  return { ...response.body, evidence: response.evidence };
},

It returns:

account_id: the provider’s identifier for the account, read at the path the identity evidence declares in account.
incarnation: optional. The provider’s own reset epoch, where it has one.
evidence: the response that proved it.

If the provider names no account anywhere, declare identity evidence without account and return only { evidence }. Returning an account_id the evidence doesn’t declare is refused.

Collections

Each collection declares a schema, a protocol and a sync.

Schema

schema is an object of validators, the same ones you use in defineTable. It is the exact shape of the table that binds to the collection. The field name source_id is reserved. Keep exact provider values exact. Store large identifiers, decimal amounts and positions as strings or v.int64(), and don’t convert them through JavaScript numbers on the way in.

Protocols

protocol states what the provider’s API actually guarantees. Bijection relies on it to decide what a sync may publish, so declare what the provider documents, not what you hope for.

Change feed
Record observation

change_feed is an ordered history of changes. Each change is a full replacement or a deletion of one record, at a position the provider states.

protocol: {
  kind: "change_feed",
  position: { domain: "customer_changes", encoding: "uint64_decimal" },
  checkpoint: {
    initial: "0",
    continuity: "complete_prefix",
    retention: { kind: "stated", unit: "days", value: 90 },
  },
  records: "full_or_delete",
  atomic_group: "one_change",
},

position.encoding is uint64_decimal or int64_decimal. Positions travel as decimal strings.
checkpoint.initial is where the first sync starts.
checkpoint.retention is how long the provider keeps a position resumable: { kind: "stated", unit, value } with a unit of minutes, hours or days, or { kind: "unknown" }.

record_observation is for APIs that list or read records without an ordered change history.

protocol: {
  kind: "record_observation",
  version: { kind: "opaque" },
  coverage: { kind: "enumeration" },
  resume: { kind: "opaque_cursor" },
},

version is { kind: "opaque" } when the provider gives each record a version token, such as an ETag, and { kind: "none" } otherwise.
coverage is { kind: "enumeration" } when a finished listing covers the whole collection, and { kind: "none" } when it proves nothing about records it didn’t return.
resume is { kind: "opaque_cursor" } when the provider pages with a cursor, and { kind: "none" } otherwise.

The sync

sync declares how often the collection is read and how a response becomes records:

every is the interval: { seconds }, { minutes } or { hours }, at most 366 days. Bijection’s scheduler runs it; you don’t need a cron job. A schedule requests work: it doesn’t guarantee a completed sync every interval while the provider is down.
read(ctx, input) makes requests through its own contracts and returns what they observed. A single read may make at most 16 requests.

For a change feed, read receives { checkpoint } and returns the changes, each citing its response and its index within it, plus the provider’s progress:

return {
  evidence: response.evidence,
  changes: [/* { kind: "replace", key, position, value, evidence, item_index }
               or { kind: "delete", key, position, evidence, item_index } */],
  next_checkpoint: response.body.through,
  observed_head: response.body.head,
  has_more: response.body.more,
};

For a paged listing, read receives the provider cursor as resume and returns the records it observed. This integration lists invoices a page at a time. The contract, under http:

bijection/billing.ts

invoices: {
  handler: { member: ["collections", "invoices", "sync", "read"] },
  method: "GET",
  route: [{ kind: "literal", value: "invoices" }],
  query: { limit: { kind: "literal", value: "100" }, cursor: { kind: "input", input: "resume" } },
  headers: {},
  response: { statuses: { 200: { kind: "json", validator: v.any() } } },
  evidence: {
    kind: "records",
    collection: "invoices",
    items: ["items"],
    presence: { kind: "fixed", record: "present" },
    key: { kind: "field", path: ["id"], encoding: "text" },
    version: { kind: "body", path: ["version"] },
    end_cursor: ["next_cursor"],
    resume: { kind: "stated", from: { kind: "body", path: ["next_cursor"] } },
  },
},

And the collection, under collections:

bijection/billing.ts

invoices: {
  schema: { invoice_id: v.string(), status: v.string(), total: v.string() },
  protocol: {
    kind: "record_observation",
    version: { kind: "opaque" },
    coverage: { kind: "enumeration" },
    resume: { kind: "opaque_cursor" },
  },
  sync: {
    every: { minutes: 15 },
    async read(ctx) {
      const response = await ctx.http.request("invoices");
      if (response.status !== 200 || response.body === null)
        throw new Error("invoice listing did not complete");
      const cursor = response.body.next_cursor as string | undefined;
      return {
        evidence: response.evidence,
        records: response.body.items.map((row: any, item_index: number) => ({
          kind: "present" as const,
          key: row.id,
          version: row.version,
          evidence: response.evidence,
          item_index,
          value: { invoice_id: row.id, status: row.status, total: row.total },
        })),
        complete: !cursor,
        ...(cursor ? { next_resume: cursor } : {}),
      };
    },
  },
},

Bijection keeps the returned next_resume, commits each page with the records it admitted, and fills the resume input of the next request with it until the listing is complete.

A read returns candidate records. It never writes to a table itself. Bijection checks the result against the retained responses and publishes it into the synced table in one transaction per page. A failed page publishes nothing: it is retried later, or marked blocked until you resume it (see Connecting and syncing).

Deletions

A record is removed from the synced table only when the provider says it is gone: a delete change in a feed, or a record your read returns with kind: "deleted". A record that is simply missing from a listing is kept.

Commands

commands declares the changes the integration can send back to the provider. Each command names the collection it targets, its typed args, the provider’s delivery and conditional-write contract, and send and reconcile handlers with their own HTTP contracts. Your application never calls a command directly. It requests one as a durable external call from a mutation, and Bijection sends it, reconciles its outcome and records it. See Operations.

Setup fields

Setup fields are in beta.

Some providers need more than a base URL and a credential, such as an account subdomain. Declare these values once with setup:

bijection/helpdesk.ts

export const helpdesk = defineIntegration({
  setup: [
    {
      key: "subdomain",
      supplier: "administrator",
      label: "Helpdesk subdomain",
      help: "The part of your helpdesk address before .example.com.",
      pattern: "[a-z0-9-]+",
      prefix: "https://",
      suffix: ".example.com/api",
      fills: "base_url",
    },
  ],
  …
});

The console’s connection form and bijection integration configure --show-setup are derived from this declaration, and the backend refuses values that don’t match it. A field with fills: "base_url" renders the connection’s base URL, so you don’t pass --base-url. Setup values are never secrets: a field marked secret: true only documents what the provider needs, and the value itself goes into a credential.

Provider rate limits

Declare the provider’s request allowance with budget. It belongs to the connection, so every collection and command of that connection spends from one bucket, and a request the bucket can’t hold waits until it can:

bijection/crm.ts

export const crm = defineIntegration({
  budget: { perMinute: 60 },
  …
});

For a provider that states its allowance as a cost bucket, declare capacity with restorePerSecond or restorePerMinute instead, and optionally requestCost and the response fields under stated where the provider reports its own figures.

backoff is in beta.

backoff declares where the provider states when it will accept another request, such as a Retry-After header on a 429:

bijection/crm.ts

backoff: {
  statuses: [429],
  signals: [{ at: { kind: "header", name: "retry-after" }, is: "retryAfter" }],
  maxWaitSeconds: 300,
},

A wait longer than maxWaitSeconds holds the connection until an operator resumes it. A stated wait never makes Bijection resend a command.

Get Started

Platform

Data and Operations

Guides

Client Libraries

Tools

API Reference

Defining Integrations

Declaring an integration

Definitions from another directory

HTTP contracts

Handlers

Routes, queries and bodies

Responses

Evidence

Verifying the account

Collections

Schema

Protocols

The sync

Deletions

Commands

Setup fields

Provider rate limits

​Declaring an integration

​Definitions from another directory

​HTTP contracts

​Handlers

​Routes, queries and bodies

​Responses

​Evidence

​Verifying the account

​Collections

​Schema

​Protocols

​The sync

​Deletions

​Commands

​Setup fields

​Provider rate limits

Declaring an integration

Definitions from another directory

HTTP contracts

Handlers

Routes, queries and bodies

Responses

Evidence

Verifying the account

Collections

Schema

Protocols

The sync

Deletions

Commands

Setup fields

Provider rate limits