Sync SaaS Data to Postgres: Upserts, pgvector, Neon & Supabase

If you are building B2B SaaS, your application is only as valuable as the data it can ingest. Your customers expect you to pull millions of records from their existing CRMs, HRIS platforms, ATSs, ticketing systems, and accounting tools directly into your application. You need a reliable way to sync SaaS data to Postgres. You do not want to build and maintain 50 custom data pipelines, nor do you want to force customer-facing integrations through an internal ELT tool designed for offline analytics, a challenge we explore in our guide to solving the bulk extraction problem. You need a unified API Postgres sync that writes directly to your multi-tenant database.

Postgres has also picked up a second job: it is no longer just the transactional database behind your app. Teams now use it to store normalized SaaS records, JSON payloads, audit trails, and vector embeddings. Serverless hosts like Neon and Supabase make it easy to run one database per product or per environment. That is why many integration pipelines now end in Postgres, Neon, or Supabase instead of only in a data warehouse.

This guide breaks down exactly how to architect a multi-tenant data sync using Truto's Sync Job V4 runtime. We will cover how to configure Postgres as a first-class datastore destination, handle idempotent upserts, execute parameterized SQL securely, and write API data to pgvector on modern serverless hosts.

The Engineering Reality of Syncing SaaS Data to Postgres

Bulk data extraction in a B2B SaaS context is fundamentally different from internal data warehousing. When you pull data for internal business intelligence, you control the source, the credentials, and the destination. When you extract data on behalf of your customers, you control none of those things.

Syncing SaaS data to Postgres means:

• Pulling paginated data from highly volatile third-party APIs.
• Normalizing wildly inconsistent provider-specific payloads into a shape your product can query.
• Writing rows into tenant-scoped Postgres tables with deterministic inserts, upserts, deletes, and checkpoints.
• Handling OAuth, aggressive rate limits, schema drift, and partial failures without corrupting customer data.

Traditional ETL pipelines fail here because they are not designed for multi-tenant, customer-facing syncs. If you have 1,000 customers and offer 50 integrations, you are managing 50,000 unique integration states. A single broken sync is not just an analytics gap; it becomes a support ticket, a failed onboarding, or a blocked enterprise deal.

Truto solves this by removing integration-specific code from your application, enabling you to ship integrations as config, not code. The platform normalizes upstream API responses into a common data model using generic JSONata expressions, and then routes that normalized data directly to a destination. By configuring Postgres as a datastore destination, Truto's distributed execution engine writes the data directly to your tables without requiring an intermediary storage bucket.

graph TD
  Upstream["Upstream API<br>(Salesforce, HubSpot)"] -->|"Extract"| SyncRuntime["Truto Sync Runtime"]
  SyncRuntime -->|"Normalize"| Mapping["JSONata Mapping"]
  Mapping --> Destination["Datastore Destination"]
  Destination -->|"upsertRows"| Postgres["Postgres Database<br>(Neon, Supabase)"]
  Postgres --> TargetTable["Target Table"]

For a broader look at the architectural differences between internal ETL and customer-facing syncs, read our guide on Handling Bulk Data Extraction and ETL Workflows Through Unified APIs and the overview of Syncing Data from SaaS Apps.

Configuring Postgres as a Sync Job Destination in Truto

A Truto Postgres sync has three moving parts:

1. A Postgres datastore that stores the connection string.
2. A Sync Job V4 resource graph that pulls from a unified API resource.
3. A datastore destination node that calls insertRows, upsertRows, deleteRows, or executeStatement.

In Truto, a datastore separates the how to connect (credentials, connection strings) from the what to do (which sync method to call, what data to write). Datastores are scoped to your multi-tenant environments, and sensitive configuration fields are encrypted at rest. You can reuse one Postgres datastore across multiple jobs—for example, one job for CRM contacts, one for companies, and one for ticket history.

A minimal datastore create body looks like this. label is required:

type: postgres
label: Postgres (production)
config:
  connection_string: ${POSTGRES_URL}
validation_methods:
  - testConnection

The connection string must point to a publicly accessible Postgres host. Truto opens a Postgres connection for the method call, runs the write, and closes it. That design is intentionally boring. There is no hidden staging bucket and no proprietary data warehouse in the middle.

The TLS Requirement and the Supabase Gotcha

Truto requires TLS for all Postgres connections unless the connection string explicitly includes sslmode=disable. The execution environment verifies TLS certificates against public Certificate Authorities (CAs) only.

This introduces a specific architectural gotcha when connecting to Supabase. Supabase's connection pooler and direct hosts chain to a private CA ("Supabase Root 2021 CA"). Because Truto's runtime cannot be provided with custom private CAs, a standard TLS handshake with Supabase will fail with a TLS_HANDSHAKE_FAILED error.

To connect Truto to Supabase, you must append sslmode=disable to your connection string. Be clear about what that means: with sslmode=disable, the synced data and the database login (user name and password) travel over the internet without TLS. Anyone on the network path can read them. Treat it as a compatibility workaround, not a security recommendation. If you use it, give Truto a dedicated database role that can write only the sync tables, and rotate its password.

Handling Data Mutations: Insert, Upsert, and Delete Rows

Once your datastore is configured, you define it as a destination node in your V4 Sync Job resource graph. Use the mutation method that matches the data contract:

Idempotent Syncs with upsertRows

The most critical method for reliable data synchronization is upsertRows. Network failures, worker crashes, and upstream API timeouts are inevitable. Your sync jobs must be idempotent—running the same sync twice should result in the exact same database state without duplicating records.

The upsertRows method maps directly to PostgreSQL's INSERT ... ON CONFLICT (...) DO UPDATE syntax, guaranteeing an atomic insert-or-update outcome under normal conditions.

Start with the database contract, not the sync job. You need a tenant-safe table:

create table app.crm_contacts (
  tenant_id text not null,
  integrated_account_id uuid not null,
  remote_id text not null,
  email text,
  first_name text,
  last_name text,
  updated_at timestamptz,
  raw jsonb not null default '{}'::jsonb,
  synced_at timestamptz not null default now(),
  primary key (tenant_id, integrated_account_id, remote_id)
);

Then build a V4 sync job whose upsertRows conflict_columns match the primary key. The job has three nodes: a request node that lists contacts, a transform node that shapes each contact into a row, and a destination node that writes the rows:

label: CRM contacts to Postgres
default_runtime_version: 4
args_schema:
  integrated_account_id:
    type: string
    format: integrated_account_id
    required: true
  tenant_id:
    type: string
    required: true
resources:
  # 1. The request node lists contacts from the unified CRM API
  - name: list_contacts
    type: request
    resource: crm/contacts
    method: list
    integrated_account_id: '{{args.integrated_account_id}}'
 
  # 2. A transform node shapes each unified contact into a table row
  - name: shape_contacts
    type: transform
    depends_on: list_contacts
    config:
      expression: |
        resources.crm.contacts.{
          "tenant_id": $$.args.tenant_id,
          "integrated_account_id": $$.args.integrated_account_id,
          "remote_id": id,
          "email": email_addresses[0].email,
          "first_name": first_name,
          "last_name": last_name,
          "updated_at": updated_at,
          "raw": $,
          "synced_at": $now()
        }[]
 
  # 3. The destination upserts those rows into Postgres
  - name: write_contacts
    type: destination
    depends_on: shape_contacts
    resources_to_persist: [shape_contacts]
    destination_type: datastore
    method: upsertRows
    config:
      id: ds_postgres_prod
      config:
        table: app.crm_contacts
        rows: '{{payload.records:json}}'
        conflict_columns:
          - tenant_id
          - integrated_account_id
          - remote_id
        update_columns:
          - email
          - first_name
          - last_name
          - updated_at
          - raw
          - synced_at

Three details in this sample matter:

• depends_on is a single node name (a string), not a list.
• A transform reads job arguments from the root of its context, so the expression uses $$.args.tenant_id. Inside the resources.crm.contacts.{...} map, the context is one contact, and a bare args.tenant_id resolves to nothing. Every argument the job reads is declared in args_schema.
• The transform emits synced_at. update_columns only updates columns that the row actually carries, so a column the transform leaves out keeps its stored value forever.

Architectural Nuances and Production Gotchas for upsertRows:

1. Batching: Rows that carry the same set of columns go into the same statement. Each statement holds at most 1,000 rows and 65,535 parameters. All statements of one call run in one transaction, so the call writes everything or nothing.
2. Do not conflict only on remote_id: Unless you enjoy cross-tenant data corruption, always include tenant_id. Most vendors only guarantee uniqueness inside one customer account.
3. Include integrated_account_id: This is vital if the same tenant can connect multiple accounts from the exact same provider.
4. Keep raw jsonb: Even when you map first-class columns, vendor APIs drift. Storing raw payloads saves you from executing painful backfill migrations later.
5. Pass arrays with :json: A bare {{payload.records}} placeholder resolves to a string. The :json suffix keeps rows an array of objects.
6. Use update_columns deliberately: If a row omits a column in the payload, it retains its stored value in the database rather than being overwritten with NULL. This is highly useful for enrichment fields your application owns.
7. Reject bad keys early: If any row is missing a column listed in conflict_columns (or the value is null), Truto rejects the whole call, not only that row. No row from that call is written.

What Happens When a Write Fails

A failed datastore write does not fail the sync job run. The run still ends completed, and the watermark moves forward. Truto does not replay the failed records. The only signal is the sync_job_run:record_error webhook. That webhook carries the error and the datastore ID, but not the failed rows.

So watch the webhook and backfill by resetting the watermark. Because the writes are idempotent upserts, a backfill that re-sends rows you already have is safe.

Standard Inserts and Deletes

If you are writing append-only data, use insertRows. Missing keys in the payload will insert NULL rather than triggering the column's default value.

For deleteRows, you provide a where object. Equality is checked per column. Passing null translates to IS NULL, and passing an array translates to an IN (...) clause. Use it for scoped cleanup, not broad destructive operations. If you need soft deletes, prefer an upsert or executeStatement that sets is_deleted = true.

Running Parameterized SQL with executeStatement

The row methods cover most sync use cases. Sometimes you need to execute complex SQL logic—such as calling stored procedures, updating JSONB deeply, or managing complex relational constraints. For this, Truto provides the executeStatement method.

The executeStatement Security Mandate

When using executeStatement, you must pass all dynamic untrusted values through the params array. Never template values directly into the statement string.

The dangerous version looks like this:

method: executeStatement
loop_on: payload.records
config:
  statement: |
    update app.crm_contacts
    set email = '{{payload.records.email}}'
    where remote_id = '{{payload.records.remote_id}}'

With loop_on, the destination runs once per record, and payload.records is the current record. The statement looks harmless until a vendor field contains a quote, a semicolon, or a malicious payload. Truto's V4 runtime resolves placeholders across the entire method configuration. If a malicious user in the upstream CRM sets their name to '); DROP TABLE app.crm_contacts;--, and you template that into the statement, the Postgres driver will execute it as a simple query, allowing stacked statements to destroy your database.

Correct Usage (Parameterized):

method: executeStatement
loop_on: payload.records
config:
  statement: |
    update app.crm_contacts
    set email = $1,
        updated_at = $2,
        synced_at = now()
    where tenant_id = $3
      and integrated_account_id = $4
      and remote_id = $5
  params:
    - '{{payload.records.email:null}}'
    - '{{payload.records.updated_at:null}}'
    - '{{args.tenant_id}}'
    - '{{args.integrated_account_id}}'
    - '{{payload.records.remote_id}}'

Note the :null suffix on the optional fields. If a placeholder path does not exist in the record, Truto leaves the literal {{...}} text in place, and Postgres then fails to cast that text to a timestamp. With :null, a missing value becomes NULL.

OWASP's SQL injection guidance is blunt: define SQL structure separately from data values, then bind values through the database API. By passing variables through params, the Postgres driver sends the query and the parameters separately over the wire, treating bound values strictly as data rather than executable SQL syntax.

Writing AI Embeddings to pgvector on Neon or Supabase

Retrieval-Augmented Generation (RAG) means B2B SaaS applications no longer sync only tabular data. They also compute vector embeddings of that data and store them for semantic search. If your application relies on AI agents, you need to write vectors into pgvector columns.

Records from a SaaS API do not arrive with vectors, and Truto runs no embedding model of its own. You have two options:

• Your own worker. The sync job keeps the source text and raw record current, and your worker writes the vectors into the same table. This is the split the samples below use.
• A JSONata transform with your own key. Truto's JSONata functions include $recursiveCharacterTextSplitter(text, { 'chunkSize': 500, 'chunkOverlap': 50 }) to chunk text and $generateEmbeddingsCohere(body, cohere_api_key) to call Cohere's embed API with your own Cohere API key. A transform can call both and pass the float arrays to the destination.

A practical schema for RAG over CRM contacts or support tickets looks like this:

create extension if not exists vector;
 
create table app.crm_contact_embeddings (
  tenant_id text not null,
  integrated_account_id uuid not null,
  remote_id text not null,
  source_text text not null,
  embedding vector(1536),
  raw jsonb not null,
  updated_at timestamptz,
  primary key (tenant_id, integrated_account_id, remote_id)
);
 
create index crm_contact_embeddings_cosine
on app.crm_contact_embeddings
using hnsw (embedding vector_cosine_ops);

The sync job upserts the text and leaves embedding alone. Because embedding is not in update_columns, an upsert never overwrites a vector your worker already wrote:

method: upsertRows
config:
  id: ds_postgres_prod
  config:
    table: app.crm_contact_embeddings
    rows: '{{payload.records:json}}'
    conflict_columns:
      - tenant_id
      - integrated_account_id
      - remote_id
    update_columns:
      - source_text
      - raw
      - updated_at

Your worker then selects rows whose embedding is null (or whose updated_at is newer than the vector), calls your embedding model, and writes the result. Clear embedding in the same statement that changes source_text if you want stale vectors to be re-embedded.

Truto's Postgres client sends objects and arrays as JSON text. JSON text suits json, jsonb, and pgvector vector columns, because pgvector parses [0.12, -0.45, 0.78] as a vector. So if your records already carry a float array, for example from a transform that calls $generateEmbeddingsCohere, an insertRows or upsertRows call writes it straight into the vector column with no casting SQL.

For Supabase-specific agent workflows, see Connect Supabase to ChatGPT: Run SQL Queries and Manage Projects.

Managing Rate Limits and API Pagination

Inside a V4 sync job run, you do not write retry code. When the upstream API returns a 429 (Too Many Requests), the run pauses, sends a sync_job_run:rate_limited webhook, and resumes by itself.

The rest of this section applies only when your own code calls Truto's Unified API directly, for example to fetch records outside a sync job. On those calls, Truto returns the upstream 429 to the caller and normalizes the provider-specific rate limit headers into standard IETF headers:

• ratelimit-limit: The maximum number of requests permitted in the window.
• ratelimit-remaining: The number of requests remaining in the current window.
• ratelimit-reset: When the window resets. Truto passes the integration's own reset value through unchanged, so the unit depends on the integration. For some it is seconds until the reset, and for others it is a Unix epoch timestamp.

On a 429, Truto also returns retry-after when it can work one out from the upstream response. Truto sets retry-after as a number of seconds, even when the upstream sent an HTTP date, so it is the header to read first.

Your code owns the retry policy, bounded backoff, and jitter.

A minimal caller-side wrapper in TypeScript looks like this. It reads retry-after first. It falls back to ratelimit-reset, and treats a value above 1e9 as an epoch timestamp in seconds:

const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
 
function waitMsFromHeaders(headers: Headers): number {
  const retryAfter = Number(headers.get('retry-after'));
  if (Number.isFinite(retryAfter) && retryAfter > 0) {
    return retryAfter * 1000;
  }
 
  const reset = Number(headers.get('ratelimit-reset'));
  if (Number.isFinite(reset) && reset > 0) {
    // Above 1e9 the integration sent an epoch timestamp, not a delay.
    const ms = reset > 1e9 ? reset * 1000 - Date.now() : reset * 1000;
    return Math.min(Math.max(ms, 1000), 15 * 60 * 1000);
  }
 
  return 1000;
}
 
export async function runWithRateLimitBackoff(run: () => Promise<Response>, attempt = 0): Promise<Response> {
  const res = await run();
 
  if (res.status !== 429 || attempt >= 5) {
    return res;
  }
 
  const baseMs = waitMsFromHeaders(res.headers);
  const jitterMs = Math.floor(Math.random() * 500);
 
  await sleep(baseMs + jitterMs);
  return runWithRateLimitBackoff(run, attempt + 1);
}

For pagination, use vendor-supported cursors where available, store watermarks per tenant, and design every page write to be idempotent. A job should be entirely safe to restart after page 37 without duplicating 36 pages of data.

Production Checklist for a Postgres Sync

Start with a Safe Sync, Then Specialize

Syncing B2B SaaS data directly into your Postgres database requires a rigorous architectural approach. You cannot rely on fragile custom scripts, and you cannot force customer-facing integrations through internal ETL tools—a distinction we cover in our analysis of reverse ETL vs unified APIs.

Do not begin with a fully generalized integration platform inside your codebase. Start with one high-value object—usually contacts, companies, tickets, or invoices. Create a tenant-scoped table. Keep the raw payload. Use upsertRows to guarantee data integrity. Add executeStatement params only when the built-in mutation methods cannot express the custom write logic. Add the pgvector worker after the base tabular sync is stable and observable.

Integrations should be declarative configuration, not a maintenance nightmare. By utilizing Truto's V4 Sync Jobs with a Postgres datastore destination, you eliminate integration-specific code and stop rewriting the same OAuth, pagination, and object mapping logic for every CRM or ATS.

If you want Truto to write unified API data directly to your Postgres, Neon, or Supabase database, start with the datastore destination pattern from our product update on syncing unified API data to your datastore. Then harden it with the rules in this guide.