Core concepts

You can filter and transform data before it’s delivered to your application. You also have a couple options for how to consume messages:

​

Source tables

A source table in Sequin is a table in your database. Consumers filter rows or changes from a source table for processing.

​

Rows and changes

Consumers can process a table’s rows or changes. A table’s rows are the actual rows in the table and are redelivered whenever a row is changed or updated. Changes are captured via the Postgres Write Ahead Log (WAL):

The WAL contains information about a specific change to a table. This includes the OLD values for updated and deleted rows as well as the opeartion (e.g. insert, update, or delete) of the change. But the WAL is ephemeral and only retains changes for a short period of time.

The table, of course, retains all its rows indefinitely.

You’ll choose to have a consumer process a table’s rows when you want it to process the latest version of a given row. Conversely, you’ll choose to have a consumer process a table’s changes when either you want data specific to the change (like OLD) or you want to process hard deletes.

Here’s a table summarizing the differences between rows and changes:

_{* Replays for changes: Limited to Sequin’s debugger feature, which only runs when you enable it.
^ Maximum consumer offline time: The maximum amount of time a consumer can be offline before it will begin missing changes. Because the WAL is ephemeral, Sequin will not retain changes for more than 7 days. Record consumers do not have this limitation.}

The right choice depends on the consumer. You can have one consumer that processes a table’s rows and another consumer that processes that table’s changes.

​

Shape of a row

Consumers process rows and changes as JSON objects.

A row has this shape:

{
  record: {
    [key: string]: any;
  };
  metadata: {
    table_schema: string;
    table_name: string;
    consumer: {
      id: string;
      name: string;
    };
  };
}

For example:

{
  "record": {
    "id": 1,
    "name": "Paul Atreides",
    "title": "Duke of Arrakis",
    "spice_allocation": 1000,
    "is_kwisatz_haderach": true
  },
  "metadata": {
    "table_schema": "public",
    "table_name": "house_atreides_members",
    "consumer": {
      "id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
      "name": "dune_characters_consumer"
    }
  }
}

​

Shape of a change

A change has this shape:

{
  record: {
    [key: string]: any;
  };
  // `changes` is only present for update operations
  changes: {
    [key: string]: any;
  } | null;
  action: 'insert' | 'update' | 'delete';
  metadata: {
    table_schema: string;
    table_name: string;
    commit_timestamp: string;
    consumer: {
      id: string;
      name: string;
    };
  };
}

For updates, changes will contain the OLD values for columns that were updated (i.e. unchanged columns will not be included in changes). changes will be null for insert and delete operations.

record will contain the NEW values for insert and update operations. record will contain the OLD values for delete operations.

For example, an insert:

{
  "record": {
    "id": 2,
    "name": "Chani",
    "title": "Fremen Warrior",
    "spice_allocation": 500,
    "is_sayyadina": true
  },
  "changes": null,
  "action": "insert",
  "metadata": {
    "table_schema": "public",
    "table_name": "fremen_members",
    "commit_timestamp": "2023-10-15T14:30:00Z",
    "consumer": {
      "id": "e2f9a3b1-7c6d-4b5a-9f8e-1d2c3b4a5e6f",
      "name": "arrakis_population_consumer"
    }
  }
}

An update:

{
  "record": {
    "id": 1,
    "name": "Paul Atreides",
    "title": "Emperor of the Known Universe",
    "spice_allocation": 10000,
    "is_kwisatz_haderach": true
  },
  "changes": {
    "title": "Duke of Arrakis",
    "spice_allocation": 1000
  },
  "action": "update",
  "metadata": {
    "table_schema": "public",
    "table_name": "house_atreides_members",
    "commit_timestamp": "2023-10-16T09:45:00Z",
    "consumer": {
      "id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
      "name": "dune_characters_consumer"
    }
  }
}

And a delete:

{
  "record": {
    "id": 3,
    "name": "Baron Vladimir Harkonnen",
    "title": "Baron of House Harkonnen",
    "spice_allocation": 5000,
    "is_mentat": false
  },
  "changes": null,
  "action": "delete",
  "metadata": {
    "table_schema": "public",
    "table_name": "house_harkonnen_members",
    "commit_timestamp": "2023-10-17T18:20:00Z",
    "consumer": {
      "id": "a1b2c3d4-e5f6-4a5b-8c7d-9e0f1a2b3c4d",
      "name": "landsraad_members_consumer"
    }
  }
}

​

Consumers

A consumer processes either rows or changes from a source table. Consumers allow you to filter, transform, and send messages from a table to your application or another service.

A consumer specifies:

• The source table to process
• The type of data to process from the source table (rows or changes)
• The SQL to filter the data
• (Soon) Transforms to apply to the data
• How workers will consume messages

​

Source table

When setting up a consumer, you’ll be prompted to specify the source table to process:

At the moment, each consumer can process a single source table.

After selecting a source table, Sequin will prompt you to specify the type of data to process. You can choose between processing rows or changes.

​

Filtering

Your consumer can specify one or more filters to process a subset of a table’s rows or changes. The filters support SQL operators:

When your consumer is processing changes, it can also specify the specific operations to filter for:

​

Sorting (row consumers only)

When a consumer is processing a table’s rows, it must specify a sort column. Sequin will use this sort column to order rows and detect changes.

It’s important you choose the right sort column. The right sort column is either:

• A timestamp like updated_at that your system updates whenever a row is inserted or updated.
• A sequence like seq or index that your system increments whenever a row is inserted or updated.

If your table does not have a timestamp or integer column that is updated on insert/update, you can add one:

-- Add the updated_at column
alter table your_table add column updated_at timestamp default now();

-- Update existing rows (if needed)
-- This will add *some* sort to existing rows, which may be desirable.
-- (Note if your table has millions of rows, this could take a while)
update your_table set updated_at = inserted_at;

-- Create a function to update the timestamp
create or replace function update_timestamp()
returns trigger as $$
begin
  new.updated_at = now();
  return new;
end;
$$ language plpgsql;

-- Create a trigger to call the function
create trigger set_timestamp
before update on your_table
for each row
execute function update_timestamp();

-- Create a sequence
create sequence your_table_seq;

-- Add the seq column
alter table your_table add column seq integer default nextval('your_table_seq');

-- Update existing rows (if needed)
-- (Note if your table has millions of rows, this could take a while)
update your_table set seq = nextval('your_table_seq');

create or replace function update_seq()
returns trigger as $$
begin
  new.seq = nextval('your_table_seq');
  return new;
end;
$$ language plpgsql;

create trigger set_seq
before update on your_table
for each row
execute function update_seq();

​

Transforming (coming soon)

You can transform messages in a stream using a Lua function. This is useful for transforming data into a format more suitable for your application.

​

Consumption

Consumers can push messages to your application via webhooks (push consumers), or you can pull messages from the consumer using Sequin’s HTTP API (pull consumers).

To provide exactly-once processing, Sequin requires that every message delivered to a consumer is acknowledged.

​

Push consumers

When a consumer is configured for push consumption, Sequin will send POST requests to the endpoint you provide:

A message is acknowledge by push consumers when Sequin receives a 200 OK response to a webhook. If any other response is received, the message is not acknowledged and will be retried.

If webhook endpoint does not respond before the request timeout elapses, Sequin will cancel the request and retry.

​

Pull consumers

You can pull messages from the consumer using Sequin’s HTTP API. When a consumer is configured for pull consumption, Sequin will provision a unique endpoint for the consumer. Your workers will pull messages from this endpoint:

After receiving a message from a pull consumer, your system will send a follow-up acknowledgement request when the message has finished processing (i.e. POST /consumers/:consumer_id/ack).

If the message is not acknowledged within the pull consumer’s visability timeout, Sequin will make the message available to other workers.

​

Replication slots

To detect changes in your database, Sequin uses a replication slot and a publication. The publication determines which tables are captured by the replication slot.

Note that Sequin uses the replication slot to process both a table’s rows and changes.

Replication slots allow Sequin to provide transactional guarantees and never miss a change, even if Sequin is disconnected from your database for a period of time.

​

Guarantees

​

Transactional enqueue

Sequin guarantees that every insert, update, and delete is captured, respecting the order of operations in any transaction.

​

Strict ordering

Sequin guarantees that messages with the same primary key are processed in order for a given stream and consumer.

This is useful for applications that need to process changes to a single record in order. For example, if you’re processing changes to a user’s profile, you want to make sure that the changes are applied in order.

To achieve this, Sequin will not deliver messages to a consumer until the previous messages for that primary key has been acknowledged.

​

Exactly-one processing

Sequin guarantees that messages are delivered at least once and must be acknowledged by a consumer. This means that no messages are lost in transit.