Connectors

A connector is Sixb's reusable connection to an external system. You give it config once, and syncs, functions, and app code resolve the same connected client by name.

Reach for a connector when your project needs to read from an external system, call a third-party API, or share one database/API client across several places — while keeping credentials and connection setup out of the code that uses them. A connector does not describe data shape: use a dataset for tables and an ontology for objects.

Define a connector

defineConnector(id, adapter) returns an inert ConnectorDefinition. It does not open a connection — the runtime does that on first use. An adapter needs a type and a connect, and may add disconnect for cleanup.

The Acme platform pulls customers, invoices, employees, and departments from a mock ERP. The connector returns a small typed client tailored to the calls the syncs make.

import { defineConnector } from "@sixb/core"
import { createAcmeErpClient } from "../lib/acme-erp"

export const acmeErpConnector = defineConnector("acme-erp", {
  type: "acme-erp",
  connect() {
    return createAcmeErpClient()
  },
})

That is the whole connector. Sixb now has one named place to get an ERP client, and createAcmeErpClient() decides which methods exist. Its AcmeErpClient exposes one list* call per ERP table — the syncs use this subset:

export interface AcmeErpClient {
  listCustomers(): Promise<readonly ErpCustomerRow[]>
  listInvoices(): Promise<readonly ErpInvoiceRow[]>
  listEmployees(): Promise<readonly ErpEmployeeRow[]>
  listDepartments(): Promise<readonly ErpDepartmentRow[]>
  // ...plus listProjects, listTasks, and other tables
}

A custom connector can return any client shape. Keep it small and tailored to the calls your project actually makes.

Use it from a sync

Syncs receive the connected client from .from(...). This keeps the sync focused on one job: read rows and write them into a dataset.

import { defineSync } from "@sixb/core"
import { acmeErpConnector } from "../connectors/acme-erp"
import { erpInvoicesDataset } from "../datasets/erp"

export const syncErpInvoices = defineSync("sync-erp-invoices")
  .from(acmeErpConnector)
  .read((client) => client.listInvoices())
  .intoDataset(erpInvoicesDataset)

See Syncs for scheduling and chaining. To map the raw rows into Invoice and Customer objects, see Projections.

Use a connector directly

You can also resolve a connector straight from the runtime — useful in functions or scripts.

import { acmeErpConnector } from "../connectors/acme-erp"
import { sixb } from "../sixb.config"

export async function loadInvoices() {
  const runtime = await sixb
  const client = await runtime.connector(acmeErpConnector)

  return client.listInvoices()
}

The first call opens the connection. Later calls reuse the same client for that runtime.

Footgun: pass the registered instance. runtime.connector(...) rejects a definition that is not the exact instance registered with the runtime. A freshly constructed defineConnector(...) with the same id throws Connector '<id>' is not the registered definition instance. Always import and pass the exported connector, not a local copy.

ConnectorContext

connect receives a ConnectorContext so adapters can scope logs, build cache keys, or cancel long-running startup work.

export const acmeErpConnector = defineConnector("acme-erp", {
  type: "acme-erp",
  connect(context) {
    // Forward context.signal to any client that supports cancellation.
    return createAcmeErpClient()
  },
})

Add disconnect(client) to close handles when the runtime tears the connector down. The mock ERP client has nothing to clean up — this shows the hook for a client that does:

export const acmeErpConnector = defineConnector("acme-erp", {
  type: "acme-erp",
  connect() {
    return createAcmeErpClient()
  },
  disconnect(client) {
    return client.close()
  },
})

Built-in adapters

When a system fits a common protocol, use a packaged adapter instead of writing connect yourself.

If the ERP were a real Postgres database, the connector would be one line:

import { defineConnector } from "@sixb/core"
import { sql } from "@sixb/connector-sql"

export const acmeErpConnector = defineConnector("acme-erp", sql(process.env.DATABASE_URL!))

export const syncErpInvoices = defineSync("sync-erp-invoices")
  .from(acmeErpConnector)
  .read((db) => db`select * from invoices`)
  .intoDataset(erpInvoicesDataset)

rest(options)

Pass a baseUrl and optional auth, timeout, rate-limit, and retry settings.

import { rest } from "@sixb/connector-rest"

rest({
  baseUrl: "https://erp.acme.internal",
  headers: { authorization: `Bearer ${process.env.ACME_ERP_TOKEN}` },
})

sql(connection) and sftp(connection)

sql takes a connection string, a URL, or a Bun SQL.Options object; the connected client is the native Bun SQL client, shared across Postgres, MySQL, and SQLite. sftp takes an ssh2 ConnectConfig; its SftpClient exposes list, stat, exists, ensureDir, open, read, write, rename, delete, mkdir, and rmdir. open(path, { signal? }) returns a backpressured ReadableStream<Uint8Array> for large files; read(path) remains the buffered convenience for small files. Both adapters close their client on disconnect.

Hosted-service connectors

Sixb also ships typed connectors for common SaaS and platform APIs. Each exports a factory you pass to defineConnector, and most ship a matching webhook helper for the Webhooks below.

The pattern is the same as any adapter — defineConnector(id, factory(options)), then resolve it by name in syncs and app code:

import { defineConnector } from "@sixb/core"
import { github } from "@sixb/connector-github"

export const githubConnector = defineConnector("github", github({ token: process.env.GITHUB_TOKEN! }))

Each factory's connected client and full options are documented in its package README.

Webhooks

A connector adapter may declare inbound webhooks alongside connect. Define them with defineWebhook(...) so the server routes incoming requests to the connector. Pass a schema to .json(...) to validate the payload and give the handler a typed body. The ERP posts invoice lifecycle events:

import { defineConnector, defineWebhook } from "@sixb/core"
import { createAcmeErpClient } from "../lib/acme-erp"

export const acmeErpConnector = defineConnector("acme-erp", {
  type: "acme-erp",
  webhooks: [
    defineWebhook("invoice-events")
      .post()
      .json({ parse: parseInvoiceWebhookEvent })
      .verify(({ request }) => {
        if (request.headers.get("x-acme-signature") !== process.env.ACME_WEBHOOK_SECRET) {
          throw new Error("[AcmeErp] Invalid webhook signature")
        }
      })
      .idempotencyKey(({ request, body }) => request.headers.get("x-acme-delivery") ?? body.deliveryId)
      .handle(({ body }) => {
        console.log(`[AcmeErp] Received ${body.type} for ${body.invoiceId}`)
      }),
  ],
  connect() {
    return createAcmeErpClient()
  },
})

parseInvoiceWebhookEvent is a plain function that validates unknown and returns the typed event (it gives body its type, invoiceId, and deliveryId fields). Bare .json() is also valid when you do not need a typed body.

Discovery and registration

Put connector definitions in connectors/ and export them. createSixb() discovers them automatically — see Project structure.

your-project/
  connectors/
    acme-erp.ts
  datasets/
    erp.ts
  syncs/
    erp.ts
  sixb.config.ts

You can also register connectors explicitly:

import { createSixb } from "@sixb/core"
import { acmeErpConnector } from "./connectors/acme-erp"

export const sixb = createSixb({
  connectors: [acmeErpConnector],
})

Connector ids must be unique within a runtime — a duplicate id throws at registration.

Core principles

• Start with a packaged adapter; write a custom connect only when the system needs special behavior.
• Keep one connector focused on one external system.
• Keep credentials and connection setup inside the connector.
• Keep data mapping in syncs, pipelines, or projections.

Where connectors fit