Authoring a sync

This page builds a complete Connect package for a fictional helpdesk API: an account URL setting, an API key, a connection test, and a tickets sync that reads only what changed since its last run and adds it to a dataset. The sections after the code explain each part. Set up the project first as described in Local development.

The API

• Every request carries the key in an x-api-key header.
• GET /v1/me returns the account behind the key.
• GET /v1/tickets returns { tickets, next_cursor }. It filters with updated_since and updated_before, sorts with sort=updated_at, and pages with limit and cursor. next_cursor is null on the last page.

The complete package

import { auth, defineIntegration, HttpError, z } from "@beetlio/connect";

const Ticket = z.object({
  id: z.string(),
  subject: z.string(),
  status: z.enum(["open", "pending", "closed"]),
  updated_at: z.iso.datetime(),
});

const TicketPage = z.object({
  tickets: z.array(Ticket),
  next_cursor: z.string().nullable(),
});

export default defineIntegration({
  key: "acme-helpdesk",
  displayName: "Acme Helpdesk",
  description: "Tickets from an Acme Helpdesk account.",
  connection: {
    origin: { type: "input", input: "baseUrl" },
    inputs: z.strictObject({
      baseUrl: z.url().meta({
        title: "API URL",
        description: "Your account's API address, for example https://yourcompany.acme.example.",
      }),
    }),
    auth: auth.apiKey({ in: "header", name: "x-api-key" }),
    async verify(ctx) {
      await ctx.json("/v1/me", z.object({ id: z.string() }));
    },
  },
  syncs: (sync) => ({
    tickets: sync({
      displayName: "Tickets",
      records: Ticket,
      inputs: z.strictObject({
        pageSize: z.number().int().min(1).max(100).default(100).meta({ title: "Page size" }),
      }),
      checkpoint: z.object({ updatedSince: z.iso.datetime() }),
      async *run(ctx) {
        const since = ctx.checkpoint?.updatedSince ?? "1970-01-01T00:00:00Z";
        const until = new Date().toISOString();
        const query = {
          updated_since: since,
          updated_before: until,
          sort: "updated_at",
          limit: ctx.config.sync.pageSize,
        };
        let count = 0;

        try {
          for await (const page of ctx.paginate({
            request: { path: "/v1/tickets", query },
            schema: TicketPage,
            next: ({ data }) =>
              data.next_cursor === null
                ? undefined
                : { path: "/v1/tickets", query: { ...query, cursor: data.next_cursor } },
          })) {
            count += page.data.tickets.length;
            yield { records: page.data.tickets };
          }
        } catch (error) {
          if (error instanceof HttpError) {
            await ctx.log.error(
              `Ticket request failed with HTTP ${error.status} (request ID ${error.requestId ?? "none"})`,
            );
          }
          throw error;
        }

        await ctx.log.info(`Read ${count} tickets updated between ${since} and ${until}`);
        yield { records: [], checkpoint: { updatedSince: until } };
      },
    }),
  }),
});

Identity

key names the integration, and each key inside syncs names a sync. Keys start with a lowercase letter, use only lowercase letters, digits and hyphens, and are at most 63 characters long. People see displayName and description when they pick the integration in the app.

Settings and credentials

Settings become form fields. Put the ones every sync shares in connection.inputs, and the ones for a single sync in that sync's inputs. Both must be z.strictObject(), and your code reads them, validated and with defaults applied, as ctx.config.connection and ctx.config.sync. .meta({ title, description }) sets the label and help text. Transforms and custom refinements cannot become form fields.

origin is where requests go: a fixed URL or, as here, a setting, which suits APIs with one URL per account. An origin is a scheme and host only, so path prefixes such as /v1 belong in each request path. The SDK's configuration guide covers the other origin types.

The auth helper declares the credential fields. auth.apiKey(...) adds a secret API key field and sends its value in the header you name, so your code never handles the key. For other schemes, see Authentication.

verify is the connection test: one small authenticated request that throws if access is wrong. Beetl runs it when someone creates the connection, and the CLI runs it when you configure a profile.

Records

records is a Zod object describing each row the sync writes, and every record you yield is validated against it. Use types that convert to JSON Schema, such as strings, numbers, booleans, enums and nullable fields; the SDK's records guide covers the rest.

A sync does not declare a primary key or a write mode. The person who adds the sync chooses Append or Replace, and Beetl does not merge records by key. See Write modes for how each one writes. Include the source's stable ID and last-modified time in each record, as the example does with id and updated_at, so a pipeline can keep the latest version of each row. The example's time window keeps one run from reading a ticket twice: a ticket edited mid-run moves past updated_before, and the next run picks it up.

Emitting records

run is an async generator. Each yield hands over one batch with records and, optionally, a checkpoint. A batch holds at most 10,000 records and must stay under 8 MiB as JSON. To split a plain list into batches, use yield* batchRecords(records).

Paths are relative to the origin, and authentication is applied to every request.

Checkpoints

A checkpoint is a small JSON value, such as a cursor or a timestamp, that tells the next run where to resume. Declare its shape with checkpoint, read the previous value from ctx.checkpoint (undefined on the first run), and yield a new one with a batch.

When a run's write commits, Beetl saves the last checkpoint the run yielded and passes it to the next run. A batch without a checkpoint keeps the previous one. A failed run saves nothing, so the next attempt starts from the same place. The example yields its checkpoint once, in a final empty batch, so the window is only marked done after every page has been read.

A checkpoint is used when the sync's write mode is Append. A Replace run always starts without one and reads everything. Changing a sync's settings in Beetl resets its checkpoint; see Checkpoints.

Logging

ctx.log has debug, info, warn and error methods. Each takes a message and returns a promise to await. Beetl shows the lines in the connection's Logs tab, and the CLI prints them to stderr. The Logs tab shows the message text, so put counts, IDs and request IDs into the message itself. Log what a run did and where it stopped, and keep each message short.

Errors

ctx.json and ctx.paginate throw an HttpError on an error status, with status, a short detail, and the provider's requestId when there is one. A response that does not match your schema throws a validation error.

By default, a GET, HEAD or OPTIONS request that gets a 408, 429, 500, 502, 503 or 504 response is tried up to 3 times before the error reaches you. Set connection.retry to change attempts, statuses, methods or delays, or false to turn retries off.

Let errors fail the run. If you catch one and return normally, a Replace run replaces the dataset with a partial result, and an incremental sync can save a checkpoint past records it never read. A thrown error fails the run and Beetl writes nothing. Log the details with ctx.log.error first, as the example does, so they show up in the run's logs.

Run it

npx beetl-connect configure . tickets
npx beetl-connect sync . tickets --output tickets.ndjson

Run sync a second time: it starts from the saved checkpoint and should return only tickets changed since the first run. Pass --mode replace to start over, as a Replace sync in Beetl does. When the output looks right, pack and upload the package.