Syncs

A sync imports data from an external system into a dataset using a connector.

Define a sync

Export a sync from syncs/. Choose a connector, read from its client, and return rows that match the target dataset's schema. Use pipelines for joins and transformations.

// syncs/invoices.ts
import { defineSync } from "@sixb/core"
import { erp } from "../connectors/erp"
import { invoices } from "../datasets/invoices"

export const importInvoices = defineSync("import-invoices")
  .from(erp)
  .read((client) => client.listInvoices())
  .intoDataset(invoices)

Run automatically

Defining a sync does not start it. Attach a schedule with .when(...) to run it on a timer or in response to an event:

import { hourlyInvoices } from "../schedules/invoices"

export const importInvoices = defineSync("import-invoices")
  .when(hourlyInvoices)
  .from(erp)
  .read((client) => client.listInvoices())
  .intoDataset(invoices)

One run of a sync runs at a time: a new run waits for the one in progress, and triggers that arrive meanwhile join it.

Choose a sync mode

Set mode in defineSync("import-invoices", { mode: "append" }) to control how rows are written:

For datasets with sequenceBy, snapshots keep the newest source rows and retain records omitted from the response. Removal requires an explicit sequenced delete.

To read only changes since the previous run, use a checkpoint. See Incremental syncs.

Sync connected accounts

For OAuth connectors, Sixb reads each connected account automatically. Use connection.account.id when rows need to identify their source account:

import { defineSync } from "@sixb/core"
import { social } from "../connectors/social"
import { videos } from "../datasets/videos"

export const importVideos = defineSync("import-videos")
  .from(social)
  .read(async (client, { connection }) => {
    const rows = await client.listVideos()
    return rows.map((video) => ({
      id: video.id,
      title: video.title,
      accountId: connection.account.id,
    }))
  })
  .intoDataset(videos)

If record IDs can repeat across accounts, include the account ID in the dataset's primary key. Incremental checkpoints are saved separately for each connection and account.

Personal connections are read too. connection.owner is { type: "user", id } for one and { type: "project" } otherwise. To show each person only their own rows, write the owner's id on each row, map it to a ref.user() property with userRef.fromId, and grant can.view(Type).where((row) => row.p.owner.eq(currentUser())). Grant a workflow's group can.view(Type) without the condition when it needs to see everyone's.

When a person's account needs reconnecting, a merge or append sync skips it: the run succeeds, the account keeps its checkpoint and rows, and the next run after it reconnects resumes from there. A snapshot sync fails instead, because leaving the account out would drop its rows. A project connection that needs reconnecting fails the run in every mode.

A suspended member's connections are never read. A merge or append sync keeps the rows it already read from them, still owned by that member; a snapshot sync no longer lists them, so its next run drops their rows. To delete or reassign kept rows, start a workflow from events.users().suspended().

Import files

Use blobs.put() to store a file and include its returned reference in a fileRef column. mediaType is optional; when you pass it, it must be a single media type such as application/pdf. An async generator lets the sync return rows as files are fetched:

import { defineSync } from "@sixb/core"
import { erp } from "../connectors/erp"
import { documents } from "../datasets/documents"

export const importDocuments = defineSync("import-documents")
  .from(erp)
  .read(async function* (client, { blobs }) {
    for (const doc of await client.listDocuments()) {
      const fileRef = await blobs.put({
        body: await client.fetchDocumentBytes(doc.id),
        fileName: `${doc.title}.pdf`,
        mediaType: "application/pdf",
      })

      yield { id: doc.id, title: doc.title, fileRef }
    }
  })
  .intoDataset(documents)