Sequin Sink Consumers API

openapi: 3.0.3
info:
  title: Sequin Management API
  description: >-
    The Sequin Management API configures and operates a Sequin Postgres change
    data capture (CDC) deployment programmatically. Sequin is open source and
    self-hostable, and also runs as Sequin Cloud. Most resources can be defined
    declaratively in a sequin.yaml file; this API exposes the same resources -
    Postgres database connections (sources), sink consumers (destinations),
    HTTP endpoints, and backfills - over REST. The Sequin Stream sink also
    exposes an HTTP pull consumption surface (receive / ack / nack). All
    requests require a Bearer token found in the Sequin console under "Manage
    account". Not all Sequin resources are available in the Management API.
  version: '1.0'
  contact:
    name: Sequin
    url: https://sequinstream.com
  license:
    name: MIT
    url: https://github.com/sequinstream/sequin/blob/main/LICENSE
servers:
  - url: https://api.sequinstream.com/api
    description: Sequin Cloud
  - url: http://localhost:7376/api
    description: Local development (self-hosted)
security:
  - bearerAuth: []
tags:
  - name: Postgres Databases
    description: Source Postgres database connections Sequin replicates from.
  - name: Sink Consumers
    description: Destinations that stream Postgres changes to external systems.
  - name: HTTP Endpoints
    description: Reusable HTTP endpoint destinations used by webhook sinks.
  - name: Backfills
    description: Replay existing Postgres rows into a sink.
  - name: Stream Pull
    description: HTTP pull consumption for the Sequin Stream sink.
paths:
  /postgres_databases:
    get:
      operationId: listPostgresDatabases
      tags:
        - Postgres Databases
      summary: List Postgres database connections
      description: Lists all Postgres database connections in your account.
      responses:
        '200':
          description: A list of Postgres database connections.
          content:
            application/json:
              schema:
                type: object
                properties:
                  data:
                    type: array
                    items:
                      $ref: '#/components/schemas/PostgresDatabase'
        '401':
          $ref: '#/components/responses/Unauthorized'
    post:
      operationId: createPostgresDatabase
      tags:
        - Postgres Databases
      summary: Create a Postgres database connection
      description: Creates a new Postgres database connection for Sequin to replicate from.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/PostgresDatabaseInput'
      responses:
        '200':
          description: The created Postgres database connection.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/PostgresDatabase'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '422':
          $ref: '#/components/responses/ValidationError'
  /postgres_databases/{id_or_name}:
    parameters:
      - $ref: '#/components/parameters/IdOrName'
    get:
      operationId: getPostgresDatabase
      tags:
        - Postgres Databases
      summary: Retrieve a Postgres database connection
      description: Retrieves a Postgres database connection by its ID or name.
      responses:
        '200':
          description: The requested Postgres database connection.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/PostgresDatabase'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
    patch:
      operationId: updatePostgresDatabase
      tags:
        - Postgres Databases
      summary: Update a Postgres database connection
      description: Updates an existing Postgres database connection.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/PostgresDatabaseInput'
      responses:
        '200':
          description: The updated Postgres database connection.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/PostgresDatabase'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
        '422':
          $ref: '#/components/responses/ValidationError'
    delete:
      operationId: deletePostgresDatabase
      tags:
        - Postgres Databases
      summary: Delete a Postgres database connection
      description: Deletes a Postgres database connection.
      responses:
        '200':
          description: Deletion confirmation.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/DeleteResponse'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
  /postgres_databases/{id_or_name}/test-connection:
    parameters:
      - $ref: '#/components/parameters/IdOrName'
    post:
      operationId: testPostgresDatabaseConnection
      tags:
        - Postgres Databases
      summary: Test a Postgres database connection
      description: Tests connectivity and replication permissions for a database connection.
      responses:
        '200':
          description: Connection test result.
          content:
            application/json:
              schema:
                type: object
                properties:
                  success:
                    type: boolean
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
  /postgres_databases/{id_or_name}/refresh-tables:
    parameters:
      - $ref: '#/components/parameters/IdOrName'
    post:
      operationId: refreshPostgresDatabaseTables
      tags:
        - Postgres Databases
      summary: Refresh table cache
      description: Refreshes Sequin's cached view of the database's tables and columns.
      responses:
        '200':
          description: Refresh confirmation.
          content:
            application/json:
              schema:
                type: object
                properties:
                  success:
                    type: boolean
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
  /sinks:
    get:
      operationId: listSinkConsumers
      tags:
        - Sink Consumers
      summary: List sink consumers
      description: >-
        Lists all sink consumers. A sink consumer is a destination for Postgres
        changes (Kafka, SQS, SNS, Kinesis, Redis, NATS, RabbitMQ, Elasticsearch,
        Typesense, GCP Pub/Sub, Azure Event Hubs, webhooks, Sequin Stream, and
        more), including its source, filters, transforms, and health.
      responses:
        '200':
          description: A list of sink consumers.
          content:
            application/json:
              schema:
                type: object
                properties:
                  data:
                    type: array
                    items:
                      $ref: '#/components/schemas/SinkConsumer'
        '401':
          $ref: '#/components/responses/Unauthorized'
    post:
      operationId: createSinkConsumer
      tags:
        - Sink Consumers
      summary: Create a sink consumer
      description: Creates a new sink consumer that streams Postgres changes to a destination.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/SinkConsumerInput'
      responses:
        '200':
          description: The created sink consumer.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/SinkConsumer'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '422':
          $ref: '#/components/responses/ValidationError'
  /sinks/{id_or_name}:
    parameters:
      - $ref: '#/components/parameters/IdOrName'
    get:
      operationId: getSinkConsumer
      tags:
        - Sink Consumers
      summary: Get a sink consumer
      description: Retrieves a sink consumer by its ID or name.
      responses:
        '200':
          description: The requested sink consumer.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/SinkConsumer'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
    patch:
      operationId: updateSinkConsumer
      tags:
        - Sink Consumers
      summary: Update a sink consumer
      description: Updates an existing sink consumer.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/SinkConsumerInput'
      responses:
        '200':
          description: The updated sink consumer.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/SinkConsumer'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
        '422':
          $ref: '#/components/responses/ValidationError'
    delete:
      operationId: deleteSinkConsumer
      tags:
        - Sink Consumers
      summary: Delete a sink consumer
      description: Deletes a sink consumer.
      responses:
        '200':
          description: Deletion confirmation.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/DeleteResponse'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
  /destinations/http_endpoints:
    get:
      operationId: listHttpEndpoints
      tags:
        - HTTP Endpoints
      summary: List HTTP endpoints
      description: Lists all reusable HTTP endpoint destinations used by webhook sinks.
      responses:
        '200':
          description: A list of HTTP endpoints.
          content:
            application/json:
              schema:
                type: object
                properties:
                  data:
                    type: array
                    items:
                      $ref: '#/components/schemas/HttpEndpoint'
        '401':
          $ref: '#/components/responses/Unauthorized'
    post:
      operationId: createHttpEndpoint
      tags:
        - HTTP Endpoints
      summary: Create an HTTP endpoint
      description: Creates a reusable HTTP endpoint destination.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/HttpEndpointInput'
      responses:
        '200':
          description: The created HTTP endpoint.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/HttpEndpoint'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '422':
          $ref: '#/components/responses/ValidationError'
  /destinations/http_endpoints/{id_or_name}:
    parameters:
      - $ref: '#/components/parameters/IdOrName'
    get:
      operationId: getHttpEndpoint
      tags:
        - HTTP Endpoints
      summary: Get an HTTP endpoint
      description: Retrieves an HTTP endpoint by its ID or name.
      responses:
        '200':
          description: The requested HTTP endpoint.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/HttpEndpoint'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
    patch:
      operationId: updateHttpEndpoint
      tags:
        - HTTP Endpoints
      summary: Update an HTTP endpoint
      description: Updates an existing HTTP endpoint.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/HttpEndpointInput'
      responses:
        '200':
          description: The updated HTTP endpoint.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/HttpEndpoint'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
        '422':
          $ref: '#/components/responses/ValidationError'
    delete:
      operationId: deleteHttpEndpoint
      tags:
        - HTTP Endpoints
      summary: Delete an HTTP endpoint
      description: Deletes an HTTP endpoint.
      responses:
        '200':
          description: Deletion confirmation.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/DeleteResponse'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
  /sinks/{sink_id_or_name}/backfills:
    parameters:
      - $ref: '#/components/parameters/SinkIdOrName'
    get:
      operationId: listBackfills
      tags:
        - Backfills
      summary: List backfills
      description: Lists the backfills for a specific sink.
      responses:
        '200':
          description: A list of backfills.
          content:
            application/json:
              schema:
                type: object
                properties:
                  data:
                    type: array
                    items:
                      $ref: '#/components/schemas/Backfill'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
    post:
      operationId: createBackfill
      tags:
        - Backfills
      summary: Create a backfill
      description: >-
        Creates a new backfill for a specific sink, replaying existing Postgres
        rows into the destination. Supports full backfills or partial backfills
        scoped to specific tables.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/BackfillInput'
      responses:
        '200':
          description: The created backfill.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Backfill'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
        '422':
          $ref: '#/components/responses/ValidationError'
  /sinks/{sink_id_or_name}/backfills/{backfill_id}:
    parameters:
      - $ref: '#/components/parameters/SinkIdOrName'
      - name: backfill_id
        in: path
        required: true
        description: The ID of the backfill.
        schema:
          type: string
    get:
      operationId: getBackfill
      tags:
        - Backfills
      summary: Get a backfill
      description: Retrieves a backfill by ID for a specific sink.
      responses:
        '200':
          description: The requested backfill.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Backfill'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
    patch:
      operationId: updateBackfill
      tags:
        - Backfills
      summary: Update a backfill
      description: Updates a backfill (for example, to pause or cancel it).
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/BackfillInput'
      responses:
        '200':
          description: The updated backfill.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Backfill'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
    delete:
      operationId: deleteBackfill
      tags:
        - Backfills
      summary: Delete a backfill
      description: Deletes a backfill.
      responses:
        '200':
          description: Deletion confirmation.
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/DeleteResponse'
        '401':
          $ref: '#/components/responses/Unauthorized'
        '404':
          $ref: '#/components/responses/NotFound'
  /http_pull_consumers/{consumer_name}/receive:
    parameters:
      - name: consumer_name
        in: path
        required: true
        description: The name of the Sequin Stream sink / consumer group.
        schema:
          type: string
    get:
      operationId: receiveMessages
      tags:
        - Stream Pull
      summary: Receive messages
      description: >-
        Retrieves a batch of available messages from a Sequin Stream consumer
        group. Each message carries an ack_id used to later acknowledge or
        negatively acknowledge it.
      parameters:
        - name: batch_size
          in: query
          required: false
          description: Maximum number of messages to return.
          schema:
            type: integer
            default: 1
      responses:
        '200':
          description: A batch of messages.
          content:
            application/json:
              schema:
                type: object
                properties:
                  data:
                    type: array
                    items:
                      $ref: '#/components/schemas/StreamMessage'
        '401':
          $ref: '#/components/responses/Unauthorized'
  /http_pull_consumers/{consumer_name}/ack:
    parameters:
      - name: consumer_name
        in: path
        required: true
        description: The name of the Sequin Stream sink / consumer group.
        schema:
          type: string
    post:
      operationId: ackMessages
      tags:
        - Stream Pull
      summary: Acknowledge messages
      description: >-
        Acknowledges messages by ack_id so the consumer group will not deliver
        them again.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/AckRequest'
      responses:
        '200':
          description: Acknowledgement result.
          content:
            application/json:
              schema:
                type: object
                properties:
                  success:
                    type: boolean
        '401':
          $ref: '#/components/responses/Unauthorized'
  /http_pull_consumers/{consumer_name}/nack:
    parameters:
      - name: consumer_name
        in: path
        required: true
        description: The name of the Sequin Stream sink / consumer group.
        schema:
          type: string
    post:
      operationId: nackMessages
      tags:
        - Stream Pull
      summary: Negatively acknowledge messages
      description: >-
        Negatively acknowledges messages by ack_id so they become available for
        redelivery immediately.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/AckRequest'
      responses:
        '200':
          description: Nack result.
          content:
            application/json:
              schema:
                type: object
                properties:
                  success:
                    type: boolean
        '401':
          $ref: '#/components/responses/Unauthorized'
components:
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      description: >-
        Bearer token found in the Sequin console under "Manage account". Passed
        as `Authorization: Bearer YOUR_API_TOKEN`.
  parameters:
    IdOrName:
      name: id_or_name
      in: path
      required: true
      description: The ID (UUID) or unique name of the resource.
      schema:
        type: string
    SinkIdOrName:
      name: sink_id_or_name
      in: path
      required: true
      description: The ID (UUID) or unique name of the sink.
      schema:
        type: string
  responses:
    Unauthorized:
      description: Missing or invalid Bearer token.
      content:
        application/json:
          schema:
            $ref: '#/components/schemas/Error'
    NotFound:
      description: The requested resource was not found.
      content:
        application/json:
          schema:
            $ref: '#/components/schemas/Error'
    ValidationError:
      description: The request payload failed validation.
      content:
        application/json:
          schema:
            $ref: '#/components/schemas/Error'
  schemas:
    Error:
      type: object
      properties:
        error:
          type: object
          properties:
            code:
              type: string
            message:
              type: string
            details:
              type: object
              additionalProperties: true
    DeleteResponse:
      type: object
      properties:
        id:
          type: string
        deleted:
          type: boolean
    PostgresDatabaseInput:
      type: object
      required:
        - name
        - hostname
        - database
        - username
      properties:
        name:
          type: string
          description: Unique name for the database connection.
        hostname:
          type: string
        port:
          type: integer
          default: 5432
        database:
          type: string
        username:
          type: string
        password:
          type: string
        ssl:
          type: boolean
        ipv6:
          type: boolean
        pool_size:
          type: integer
    PostgresDatabase:
      allOf:
        - $ref: '#/components/schemas/PostgresDatabaseInput'
        - type: object
          properties:
            id:
              type: string
              format: uuid
    SinkConsumerInput:
      type: object
      required:
        - name
        - database
        - destination
      properties:
        name:
          type: string
        status:
          type: string
          enum:
            - active
            - disabled
            - paused
        database:
          type: string
          description: Name or ID of the source Postgres database.
        table:
          type: string
          description: Source table in schema.table form.
        filters:
          type: array
          items:
            type: object
            additionalProperties: true
        transform:
          type: string
          description: Name of a transform to apply to messages.
        destination:
          type: object
          description: >-
            The destination configuration. The `type` field selects the sink
            kind - e.g. kafka, sqs, sns, kinesis, redis_stream, redis_string,
            nats, rabbitmq, elasticsearch, typesense, meilisearch, gcp_pubsub,
            azure_event_hub, webhook, sequin_stream, s2.
          properties:
            type:
              type: string
              enum:
                - kafka
                - sqs
                - sns
                - kinesis
                - redis_stream
                - redis_string
                - nats
                - rabbitmq
                - elasticsearch
                - typesense
                - meilisearch
                - gcp_pubsub
                - azure_event_hub
                - webhook
                - sequin_stream
                - s2
          additionalProperties: true
    SinkConsumer:
      allOf:
        - $ref: '#/components/schemas/SinkConsumerInput'
        - type: object
          properties:
            id:
              type: string
              format: uuid
            health:
              type: object
              additionalProperties: true
    HttpEndpointInput:
      type: object
      required:
        - name
        - url
      properties:
        name:
          type: string
        url:
          type: string
          format: uri
        headers:
          type: object
          additionalProperties:
            type: string
        encrypted_headers:
          type: object
          additionalProperties:
            type: string
    HttpEndpoint:
      allOf:
        - $ref: '#/components/schemas/HttpEndpointInput'
        - type: object
          properties:
            id:
              type: string
              format: uuid
    BackfillInput:
      type: object
      properties:
        state:
          type: string
          enum:
            - active
            - paused
            - cancelled
        tables:
          type: array
          description: Optional list of tables to scope a partial backfill.
          items:
            type: string
        sort_column:
          type: string
    Backfill:
      allOf:
        - $ref: '#/components/schemas/BackfillInput'
        - type: object
          properties:
            id:
              type: string
              format: uuid
            sink:
              type: string
            rows_ingested_count:
              type: integer
            inserted_at:
              type: string
              format: date-time
    StreamMessage:
      type: object
      properties:
        ack_id:
          type: string
        action:
          type: string
          enum:
            - insert
            - update
            - delete
            - read
        record:
          type: object
          additionalProperties: true
        changes:
          type: object
          additionalProperties: true
        metadata:
          type: object
          additionalProperties: true
    AckRequest:
      type: object
      required:
        - ack_ids
      properties:
        ack_ids:
          type: array
          items:
            type: string