> For the complete documentation index, see [llms.txt](https://developer.celigo.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://developer.celigo.com/api/api-reference/data-stores/topics.md).

# Topics

Topics are durable event streams that decouple the flows that publish messages from the flows that consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each subscribed topic export step reads them at its own pace, and messages expire after the topic's `retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but creating or updating a topic and every message operation require the `topics` entitlement — without it they fail with `topics_subscription_required` (a 422 validation error on topic writes, 403 on the message endpoints) — and the license also caps how many topics the account may hold (`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).

### Topic schema

## The Topic object

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"components":{"schemas":{"Topic":{"type":"object","description":"Topic as returned by the API.","required":["_id","_userId","name","retentionMs","createdAt","lastModified"],"properties":{"_id":{"type":"string","format":"objectId","readOnly":true,"description":"Unique identifier for the topic. Topic exports and imports bind to it through\n`topic._topicId`, and it is the `_topicId` used by every `/v1/topics/{_id}/…` endpoint."},"_userId":{"type":"string","format":"objectId","readOnly":true,"description":"Account owner whose license and topic quota the topic counts against."},"name":{"type":"string","maxLength":200,"description":"Display name shown in the UI. Empty when the topic was created without one; not unique, so\nuse `_id` to identify a topic."},"description":{"type":["string","null"],"maxLength":5120,"description":"What the topic carries and which flows consume it. Absent or null when never set."},"retentionMs":{"type":"integer","minimum":3600000,"maximum":7776000000,"default":604800000,"description":"How long a published message stays readable, in milliseconds, before it is dropped from the\nstream whether or not a listener consumed it. The week-long default applies when the topic was\ncreated without one."},"externalId":{"type":"string","maxLength":1024,"description":"Caller-chosen key set at creation or update. Present only when it was supplied; filter the\nlist with `GET /v1/topics?externalId=…`. Not unique: the filtered list can hold more than\none topic, so use `_id` to address a topic."},"_integrationId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App installation that owns the topic. Present only on topics created by an\nIntegration App; absent on topics you create directly, which are shared across integrations\nthrough `_registeredTopicIds` instead."},"_connectorId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App (connector) that installed the topic. Present only on connector-managed\ntopics, which the connector — not the user — registers and deregisters."},"_templateId":{"type":"string","format":"objectId","readOnly":true,"description":"Template the topic was installed from. Present only when it originated from a template."},"_sourceId":{"type":"string","format":"objectId","readOnly":true,"description":"Topic this one was cloned or installed from. Present only on copies."},"draft":{"type":"boolean","description":"When true, the topic is an unconfirmed draft (created with `draft: true`, typically by\nthe AI-assisted builder) that auto-deletes at `draftExpiresAt`. Absent on ordinary\ntopics and once a `PUT` with `draft: false` promotes the draft."},"draftExpiresAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when an unconfirmed draft is deleted automatically — 48 hours after\ncreation. Present only while `draft` is true; cleared together with it."},"createdAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was created."},"lastModified":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was last updated by PUT or PATCH."},"deletedAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was moved to the recycle bin. Present only on the copy returned by\n`GET /v1/recycleBinTTL/topics/{_id}`; deleted topics are hidden from every other endpoint."}}}}}}
```

## List topics

> Returns the account's topics sorted by \`name\` ascending (ties by most recently modified first).\
> Topics in the recycle bin are not included. \`include\` and \`exclude\` projections are not\
> supported on this resource.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"parameters":{"After":{"name":"after","in":"query","required":false,"description":"Opaque cursor for forward pagination. Pass the value from the `Link`\nresponse header (`rel=\"next\"`) to fetch the next page.","schema":{"type":"string"}}},"schemas":{"Topic":{"type":"object","description":"Topic as returned by the API.","required":["_id","_userId","name","retentionMs","createdAt","lastModified"],"properties":{"_id":{"type":"string","format":"objectId","readOnly":true,"description":"Unique identifier for the topic. Topic exports and imports bind to it through\n`topic._topicId`, and it is the `_topicId` used by every `/v1/topics/{_id}/…` endpoint."},"_userId":{"type":"string","format":"objectId","readOnly":true,"description":"Account owner whose license and topic quota the topic counts against."},"name":{"type":"string","maxLength":200,"description":"Display name shown in the UI. Empty when the topic was created without one; not unique, so\nuse `_id` to identify a topic."},"description":{"type":["string","null"],"maxLength":5120,"description":"What the topic carries and which flows consume it. Absent or null when never set."},"retentionMs":{"type":"integer","minimum":3600000,"maximum":7776000000,"default":604800000,"description":"How long a published message stays readable, in milliseconds, before it is dropped from the\nstream whether or not a listener consumed it. The week-long default applies when the topic was\ncreated without one."},"externalId":{"type":"string","maxLength":1024,"description":"Caller-chosen key set at creation or update. Present only when it was supplied; filter the\nlist with `GET /v1/topics?externalId=…`. Not unique: the filtered list can hold more than\none topic, so use `_id` to address a topic."},"_integrationId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App installation that owns the topic. Present only on topics created by an\nIntegration App; absent on topics you create directly, which are shared across integrations\nthrough `_registeredTopicIds` instead."},"_connectorId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App (connector) that installed the topic. Present only on connector-managed\ntopics, which the connector — not the user — registers and deregisters."},"_templateId":{"type":"string","format":"objectId","readOnly":true,"description":"Template the topic was installed from. Present only when it originated from a template."},"_sourceId":{"type":"string","format":"objectId","readOnly":true,"description":"Topic this one was cloned or installed from. Present only on copies."},"draft":{"type":"boolean","description":"When true, the topic is an unconfirmed draft (created with `draft: true`, typically by\nthe AI-assisted builder) that auto-deletes at `draftExpiresAt`. Absent on ordinary\ntopics and once a `PUT` with `draft: false` promotes the draft."},"draftExpiresAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when an unconfirmed draft is deleted automatically — 48 hours after\ncreation. Present only while `draft` is true; cleared together with it."},"createdAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was created."},"lastModified":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was last updated by PUT or PATCH."},"deletedAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was moved to the recycle bin. Present only on the copy returned by\n`GET /v1/recycleBinTTL/topics/{_id}`; deleted topics are hidden from every other endpoint."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}}}},"paths":{"/v1/topics":{"get":{"summary":"List topics","description":"Returns the account's topics sorted by `name` ascending (ties by most recently modified first).\nTopics in the recycle bin are not included. `include` and `exclude` projections are not\nsupported on this resource.","operationId":"listTopics","tags":["Topics"],"parameters":[{"name":"limit","in":"query","required":false,"description":"Maximum number of topics to return per page. Omit to return all topics.","schema":{"type":"integer","minimum":1}},{"name":"sort_by","in":"query","required":false,"description":"Field to sort by. Any other field name is rejected.","schema":{"type":"string","enum":["name","lastModified"]}},{"$ref":"#/components/parameters/After"},{"name":"externalId","in":"query","required":false,"description":"Return only topics whose `externalId` equals this value. Repeat the parameter to match any of\nseveral values.","schema":{"type":"string"}}],"responses":{"200":{"description":"Array of topics.","headers":{"Link":{"description":"RFC 5988 pagination link. Includes `<...>; rel=\"next\"` with an `after` cursor when more\ntopics remain; absent on the final page.","schema":{"type":"string"}}},"content":{"application/json":{"schema":{"type":"array","items":{"$ref":"#/components/schemas/Topic"}}}}},"204":{"description":"No topics match — the account has none, or none carry the requested `externalId`."},"400":{"description":"`after` is not a cursor issued by this endpoint, or `sort_by` names an unsupported field.\nError code: `invalid_query_param`.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"422":{"description":"`limit` is not a positive integer. Error code: `invalid_query_param`.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## Create a topic

> Creates a topic and counts it against the license's \`topics.topicsAllowed\` quota. The stream\
> itself is provisioned lazily on the first publish, so a new topic reports no messages until\
> something is published to it. Pass \`\_\_integrationIds\` to register the topic on integrations in the\
> same call; otherwise register it later with the \`/v1/integrations/{\_id}/topics/…/register\`\
> operations so flows in those integrations can bind to it.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"Request":{"type":"object","description":"Writable fields for creating or updating a topic.","required":["name"],"properties":{"name":{"type":"string","maxLength":200,"description":"Display name shown in the UI and recorded as `deletedInfo.name` on audit entries. Names need\nnot be unique — exports and imports bind to the topic id, so renaming never breaks a\nsubscription. Script tags and other unsafe HTML are stripped on save."},"description":{"type":["string","null"],"maxLength":5120,"description":"What the topic carries and which flows consume it. Shown in the UI only; it plays no part in\nrouting. Send null to clear a previous value."},"retentionMs":{"type":"integer","minimum":3600000,"maximum":7776000000,"default":604800000,"description":"How long a published message stays readable, in milliseconds, before it is dropped from the\nstream whether or not a listener consumed it. The default keeps a week of messages, enough for\na listener that is paused over a weekend to catch up; the ceiling is further capped by the\naccount license's `topics.retentionCeilingDays`. Changing it on an existing topic\nre-configures the stream, so messages older than the new window are no longer retained."},"externalId":{"type":"string","maxLength":1024,"description":"Caller-chosen key for finding the topic again without storing its id — filter the list with\n`GET /v1/topics?externalId=…`. Integration Apps use it to locate a topic across installs.\nNot unique: another topic can carry the same value, so the filtered list can hold several."},"draft":{"type":"boolean","description":"When true, the topic is a draft that auto-deletes 48 hours after creation\n(`draftExpiresAt` in the response). Set at creation; an update can clear the\nflag but never set it, and a PUT that omits it leaves an existing draft as is —\nthe one field the whole-document PUT does not clear."},"__integrationIds":{"type":"array","maxItems":100,"items":{"type":"string","format":"objectId"},"description":"Integrations to register the new topic on immediately after creation, so flows in those\nintegrations can subscribe to it. Honored only on `POST /v1/topics` (ignored on PUT), never\npersisted, and not echoed back. Ids are validated before the topic is saved; registration\nruns after the save, and per-integration failures are reported in\n`__failedIntegrationRegistrations` on the 201 response without rolling back the new topic."}}},"Topic":{"type":"object","description":"Topic as returned by the API.","required":["_id","_userId","name","retentionMs","createdAt","lastModified"],"properties":{"_id":{"type":"string","format":"objectId","readOnly":true,"description":"Unique identifier for the topic. Topic exports and imports bind to it through\n`topic._topicId`, and it is the `_topicId` used by every `/v1/topics/{_id}/…` endpoint."},"_userId":{"type":"string","format":"objectId","readOnly":true,"description":"Account owner whose license and topic quota the topic counts against."},"name":{"type":"string","maxLength":200,"description":"Display name shown in the UI. Empty when the topic was created without one; not unique, so\nuse `_id` to identify a topic."},"description":{"type":["string","null"],"maxLength":5120,"description":"What the topic carries and which flows consume it. Absent or null when never set."},"retentionMs":{"type":"integer","minimum":3600000,"maximum":7776000000,"default":604800000,"description":"How long a published message stays readable, in milliseconds, before it is dropped from the\nstream whether or not a listener consumed it. The week-long default applies when the topic was\ncreated without one."},"externalId":{"type":"string","maxLength":1024,"description":"Caller-chosen key set at creation or update. Present only when it was supplied; filter the\nlist with `GET /v1/topics?externalId=…`. Not unique: the filtered list can hold more than\none topic, so use `_id` to address a topic."},"_integrationId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App installation that owns the topic. Present only on topics created by an\nIntegration App; absent on topics you create directly, which are shared across integrations\nthrough `_registeredTopicIds` instead."},"_connectorId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App (connector) that installed the topic. Present only on connector-managed\ntopics, which the connector — not the user — registers and deregisters."},"_templateId":{"type":"string","format":"objectId","readOnly":true,"description":"Template the topic was installed from. Present only when it originated from a template."},"_sourceId":{"type":"string","format":"objectId","readOnly":true,"description":"Topic this one was cloned or installed from. Present only on copies."},"draft":{"type":"boolean","description":"When true, the topic is an unconfirmed draft (created with `draft: true`, typically by\nthe AI-assisted builder) that auto-deletes at `draftExpiresAt`. Absent on ordinary\ntopics and once a `PUT` with `draft: false` promotes the draft."},"draftExpiresAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when an unconfirmed draft is deleted automatically — 48 hours after\ncreation. Present only while `draft` is true; cleared together with it."},"createdAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was created."},"lastModified":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was last updated by PUT or PATCH."},"deletedAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was moved to the recycle bin. Present only on the copy returned by\n`GET /v1/recycleBinTTL/topics/{_id}`; deleted topics are hidden from every other endpoint."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}}}},"paths":{"/v1/topics":{"post":{"summary":"Create a topic","description":"Creates a topic and counts it against the license's `topics.topicsAllowed` quota. The stream\nitself is provisioned lazily on the first publish, so a new topic reports no messages until\nsomething is published to it. Pass `__integrationIds` to register the topic on integrations in the\nsame call; otherwise register it later with the `/v1/integrations/{_id}/topics/…/register`\noperations so flows in those integrations can bind to it.","operationId":"createTopic","tags":["Topics"],"requestBody":{"required":true,"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Request"}}}},"responses":{"201":{"description":"Created topic. When `__integrationIds` was sent and some registrations failed, the body also\ncarries `__failedIntegrationRegistrations`, an array of `{_integrationId, error}` entries — the\ntopic itself was still created.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Topic"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"description":"An id in `__integrationIds` does not belong to an integration in this account. Nothing is created.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}},"422":{"description":"Validation failed. `retentionMs` outside its bounds (`min`, `max`), above the license ceiling\n(`topics_retention_exceeds_entitlement`), or not numeric (`invalid_field`); `name` longer than\n200 characters (`maxlength`); `description` or `externalId` over their size limits\n(`size_limit_exceeded`); the account at its topic quota (`topics_limit_exceeded`); an account\nwithout the Event Streams entitlement (`topics_subscription_required`); or a malformed\n`__integrationIds` (`invalid_request`, `invalid_ref`).","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## Get a topic

> Returns a topic's configuration. Read its messages with \`GET /v1/topics/{\_id}/messages\`. A topic\
> in the recycle bin is reported as not found; restore it first.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"Topic":{"type":"object","description":"Topic as returned by the API.","required":["_id","_userId","name","retentionMs","createdAt","lastModified"],"properties":{"_id":{"type":"string","format":"objectId","readOnly":true,"description":"Unique identifier for the topic. Topic exports and imports bind to it through\n`topic._topicId`, and it is the `_topicId` used by every `/v1/topics/{_id}/…` endpoint."},"_userId":{"type":"string","format":"objectId","readOnly":true,"description":"Account owner whose license and topic quota the topic counts against."},"name":{"type":"string","maxLength":200,"description":"Display name shown in the UI. Empty when the topic was created without one; not unique, so\nuse `_id` to identify a topic."},"description":{"type":["string","null"],"maxLength":5120,"description":"What the topic carries and which flows consume it. Absent or null when never set."},"retentionMs":{"type":"integer","minimum":3600000,"maximum":7776000000,"default":604800000,"description":"How long a published message stays readable, in milliseconds, before it is dropped from the\nstream whether or not a listener consumed it. The week-long default applies when the topic was\ncreated without one."},"externalId":{"type":"string","maxLength":1024,"description":"Caller-chosen key set at creation or update. Present only when it was supplied; filter the\nlist with `GET /v1/topics?externalId=…`. Not unique: the filtered list can hold more than\none topic, so use `_id` to address a topic."},"_integrationId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App installation that owns the topic. Present only on topics created by an\nIntegration App; absent on topics you create directly, which are shared across integrations\nthrough `_registeredTopicIds` instead."},"_connectorId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App (connector) that installed the topic. Present only on connector-managed\ntopics, which the connector — not the user — registers and deregisters."},"_templateId":{"type":"string","format":"objectId","readOnly":true,"description":"Template the topic was installed from. Present only when it originated from a template."},"_sourceId":{"type":"string","format":"objectId","readOnly":true,"description":"Topic this one was cloned or installed from. Present only on copies."},"draft":{"type":"boolean","description":"When true, the topic is an unconfirmed draft (created with `draft: true`, typically by\nthe AI-assisted builder) that auto-deletes at `draftExpiresAt`. Absent on ordinary\ntopics and once a `PUT` with `draft: false` promotes the draft."},"draftExpiresAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when an unconfirmed draft is deleted automatically — 48 hours after\ncreation. Present only while `draft` is true; cleared together with it."},"createdAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was created."},"lastModified":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was last updated by PUT or PATCH."},"deletedAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was moved to the recycle bin. Present only on the copy returned by\n`GET /v1/recycleBinTTL/topics/{_id}`; deleted topics are hidden from every other endpoint."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}}}},"paths":{"/v1/topics/{_id}":{"get":{"summary":"Get a topic","description":"Returns a topic's configuration. Read its messages with `GET /v1/topics/{_id}/messages`. A topic\nin the recycle bin is reported as not found; restore it first.","operationId":"getTopicById","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}}],"responses":{"200":{"description":"Topic object.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Topic"}}}},"400":{"description":"`_id` is not a 24-character hexadecimal ObjectId. Error code: `invalid_ref`.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"description":"No active topic has this id. A topic in the recycle bin answers with the message \"You are\ntrying to access a Topic that is already in the Recycle Bin…\"; restore it with\n`POST /v1/recycleBinTTL/topics/{_id}`.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## Update a topic

> Replaces the topic's writable fields with the request body. Fields left out are cleared, not\
> kept: \`description\` and \`externalId\` are removed and \`retentionMs\` reverts to seven days — send\
> the full document, or use PATCH to change \`name\` or \`description\` alone. Read-only fields in the\
> body are ignored. Changing \`retentionMs\` re-configures the stream's retention window;\
> the topic id and its subscriptions are unaffected.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"Request":{"type":"object","description":"Writable fields for creating or updating a topic.","required":["name"],"properties":{"name":{"type":"string","maxLength":200,"description":"Display name shown in the UI and recorded as `deletedInfo.name` on audit entries. Names need\nnot be unique — exports and imports bind to the topic id, so renaming never breaks a\nsubscription. Script tags and other unsafe HTML are stripped on save."},"description":{"type":["string","null"],"maxLength":5120,"description":"What the topic carries and which flows consume it. Shown in the UI only; it plays no part in\nrouting. Send null to clear a previous value."},"retentionMs":{"type":"integer","minimum":3600000,"maximum":7776000000,"default":604800000,"description":"How long a published message stays readable, in milliseconds, before it is dropped from the\nstream whether or not a listener consumed it. The default keeps a week of messages, enough for\na listener that is paused over a weekend to catch up; the ceiling is further capped by the\naccount license's `topics.retentionCeilingDays`. Changing it on an existing topic\nre-configures the stream, so messages older than the new window are no longer retained."},"externalId":{"type":"string","maxLength":1024,"description":"Caller-chosen key for finding the topic again without storing its id — filter the list with\n`GET /v1/topics?externalId=…`. Integration Apps use it to locate a topic across installs.\nNot unique: another topic can carry the same value, so the filtered list can hold several."},"draft":{"type":"boolean","description":"When true, the topic is a draft that auto-deletes 48 hours after creation\n(`draftExpiresAt` in the response). Set at creation; an update can clear the\nflag but never set it, and a PUT that omits it leaves an existing draft as is —\nthe one field the whole-document PUT does not clear."},"__integrationIds":{"type":"array","maxItems":100,"items":{"type":"string","format":"objectId"},"description":"Integrations to register the new topic on immediately after creation, so flows in those\nintegrations can subscribe to it. Honored only on `POST /v1/topics` (ignored on PUT), never\npersisted, and not echoed back. Ids are validated before the topic is saved; registration\nruns after the save, and per-integration failures are reported in\n`__failedIntegrationRegistrations` on the 201 response without rolling back the new topic."}}},"Topic":{"type":"object","description":"Topic as returned by the API.","required":["_id","_userId","name","retentionMs","createdAt","lastModified"],"properties":{"_id":{"type":"string","format":"objectId","readOnly":true,"description":"Unique identifier for the topic. Topic exports and imports bind to it through\n`topic._topicId`, and it is the `_topicId` used by every `/v1/topics/{_id}/…` endpoint."},"_userId":{"type":"string","format":"objectId","readOnly":true,"description":"Account owner whose license and topic quota the topic counts against."},"name":{"type":"string","maxLength":200,"description":"Display name shown in the UI. Empty when the topic was created without one; not unique, so\nuse `_id` to identify a topic."},"description":{"type":["string","null"],"maxLength":5120,"description":"What the topic carries and which flows consume it. Absent or null when never set."},"retentionMs":{"type":"integer","minimum":3600000,"maximum":7776000000,"default":604800000,"description":"How long a published message stays readable, in milliseconds, before it is dropped from the\nstream whether or not a listener consumed it. The week-long default applies when the topic was\ncreated without one."},"externalId":{"type":"string","maxLength":1024,"description":"Caller-chosen key set at creation or update. Present only when it was supplied; filter the\nlist with `GET /v1/topics?externalId=…`. Not unique: the filtered list can hold more than\none topic, so use `_id` to address a topic."},"_integrationId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App installation that owns the topic. Present only on topics created by an\nIntegration App; absent on topics you create directly, which are shared across integrations\nthrough `_registeredTopicIds` instead."},"_connectorId":{"type":"string","format":"objectId","readOnly":true,"description":"Integration App (connector) that installed the topic. Present only on connector-managed\ntopics, which the connector — not the user — registers and deregisters."},"_templateId":{"type":"string","format":"objectId","readOnly":true,"description":"Template the topic was installed from. Present only when it originated from a template."},"_sourceId":{"type":"string","format":"objectId","readOnly":true,"description":"Topic this one was cloned or installed from. Present only on copies."},"draft":{"type":"boolean","description":"When true, the topic is an unconfirmed draft (created with `draft: true`, typically by\nthe AI-assisted builder) that auto-deletes at `draftExpiresAt`. Absent on ordinary\ntopics and once a `PUT` with `draft: false` promotes the draft."},"draftExpiresAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when an unconfirmed draft is deleted automatically — 48 hours after\ncreation. Present only while `draft` is true; cleared together with it."},"createdAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was created."},"lastModified":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was last updated by PUT or PATCH."},"deletedAt":{"type":"string","format":"date-time","readOnly":true,"description":"Timestamp when the topic was moved to the recycle bin. Present only on the copy returned by\n`GET /v1/recycleBinTTL/topics/{_id}`; deleted topics are hidden from every other endpoint."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}},"404-not-found":{"description":"Not found. The requested resource does not exist or is not visible to the caller.\n","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}},"paths":{"/v1/topics/{_id}":{"put":{"summary":"Update a topic","description":"Replaces the topic's writable fields with the request body. Fields left out are cleared, not\nkept: `description` and `externalId` are removed and `retentionMs` reverts to seven days — send\nthe full document, or use PATCH to change `name` or `description` alone. Read-only fields in the\nbody are ignored. Changing `retentionMs` re-configures the stream's retention window;\nthe topic id and its subscriptions are unaffected.","operationId":"updateTopic","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}}],"requestBody":{"required":true,"content":{"application/json":{"schema":{"$ref":"#/components/schemas/Request"}}}},"responses":{"200":{"description":"Updated topic.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Topic"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"$ref":"#/components/responses/404-not-found"},"422":{"description":"Validation failed — the same codes as on create: `min`, `max`,\n`topics_retention_exceeds_entitlement`, `invalid_field` for `retentionMs`; `maxlength` for\n`name`; `size_limit_exceeded` for `description` or `externalId`;\n`topics_subscription_required` when the account lacks the Event Streams entitlement.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## Delete a topic

> Moves the topic to the recycle bin, where it is kept for 30 days before being purged, and removes\
> it from \`\_registeredTopicIds\` on every integration that had it registered; restoring it does not\
> re-register it. Its retained messages are kept and become readable again if the topic is\
> restored, and are gone for good once the topic is purged. Deletion is refused while any export or\
> import still binds to the topic through \`topic.\_topicId\`, or a flow still contains such a step —\
> delete or repoint those first, and purge them from the recycle bin before purging the topic.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}},"404-not-found":{"description":"Not found. The requested resource does not exist or is not visible to the caller.\n","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}},"schemas":{"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}}},"paths":{"/v1/topics/{_id}":{"delete":{"summary":"Delete a topic","description":"Moves the topic to the recycle bin, where it is kept for 30 days before being purged, and removes\nit from `_registeredTopicIds` on every integration that had it registered; restoring it does not\nre-register it. Its retained messages are kept and become readable again if the topic is\nrestored, and are gone for good once the topic is purged. Deletion is refused while any export or\nimport still binds to the topic through `topic._topicId`, or a flow still contains such a step —\ndelete or repoint those first, and purge them from the recycle bin before purging the topic.","operationId":"deleteTopic","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}}],"responses":{"204":{"description":"Topic moved to the recycle bin."},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"$ref":"#/components/responses/404-not-found"},"422":{"description":"An export, import, or flow still references the topic. Error code `dependencies_not_deleted`, one entry per blocking resource.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## Patch a topic

> Applies a JSON Patch document (RFC 6902) to the topic. Only \`/name\` and \`/description\` are\
> patchable; every other path — including \`/retentionMs\` and \`/externalId\` — is rejected, so use\
> PUT to change those. Unlike PUT, fields not named in the patch are left untouched.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"JsonPatchRequest":{"type":"array","description":"A JSON Patch document (RFC 6902). Send an array of patch\noperations on whitelisted fields — all other paths are rejected\nwith 422.","minItems":1,"items":{"$ref":"#/components/schemas/JsonPatchOperation"}},"JsonPatchOperation":{"type":"object","description":"A single JSON Patch operation (RFC 6902).","required":["op","path"],"properties":{"op":{"type":"string","enum":["replace","add","remove"],"description":"The operation to perform."},"path":{"type":"string","description":"JSON Pointer (RFC 6901) to the field to patch. Only\nwhitelisted paths are accepted — unlisted paths return\n`422` with `\"<path> is not a whitelisted property\"`."},"value":{"description":"The new value to set. Required for `replace` and `add`, omit for `remove`."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}},"404-not-found":{"description":"Not found. The requested resource does not exist or is not visible to the caller.\n","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}},"paths":{"/v1/topics/{_id}":{"patch":{"summary":"Patch a topic","description":"Applies a JSON Patch document (RFC 6902) to the topic. Only `/name` and `/description` are\npatchable; every other path — including `/retentionMs` and `/externalId` — is rejected, so use\nPUT to change those. Unlike PUT, fields not named in the patch are left untouched.","operationId":"patchTopic","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}}],"requestBody":{"required":true,"content":{"application/json":{"schema":{"$ref":"#/components/schemas/JsonPatchRequest"}}}},"responses":{"204":{"description":"Topic patched. No body returned."},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"$ref":"#/components/responses/404-not-found"},"422":{"description":"A path other than `/name` or `/description` was patched (\"… is not a whitelisted property\"),\nor the new value failed validation (for example `maxlength` on `/name`).","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## List retained messages

> Reads the messages currently retained on a topic, oldest first, without affecting any\
> listener's read position — use it to inspect what producers are publishing or to confirm a\
> purge. Pages are cursor-based: follow \`nextPageURL\` until it is absent. Query parameters other\
> than the four below are ignored.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"TopicMessagesResponse":{"type":"object","description":"One page of retained messages.","required":["messages"],"properties":{"messages":{"type":"array","description":"Messages in the requested window, oldest first. Empty when the topic has never received a\nmessage, when everything has aged out of `retentionMs`, or when the continuation cursor is\nalready at the end of the stream.","items":{"$ref":"#/components/schemas/TopicMessage"}},"nextPageURL":{"type":"string","description":"Relative URL of the next page, carrying the same filters plus an `afterOffset` cursor.\nFollow it as-is rather than building the cursor yourself. Absent on the last page."}}},"TopicMessage":{"type":"object","description":"One retained message, in publish order.","required":["timestamp","payload"],"properties":{"timestamp":{"type":"string","format":"date-time","description":"Timestamp when the message was appended to the stream; the value `time_gt` and `time_lte` filter on."},"payload":{"description":"The message exactly as it was published — normally the JSON object sent in `messages[]`.\nA value that was not valid JSON is returned verbatim as a string."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}},"404-not-found":{"description":"Not found. The requested resource does not exist or is not visible to the caller.\n","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}},"paths":{"/v1/topics/{_id}/messages":{"get":{"summary":"List retained messages","description":"Reads the messages currently retained on a topic, oldest first, without affecting any\nlistener's read position — use it to inspect what producers are publishing or to confirm a\npurge. Pages are cursor-based: follow `nextPageURL` until it is absent. Query parameters other\nthan the four below are ignored.","operationId":"listTopicMessages","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}},{"name":"limit","in":"query","required":false,"description":"Maximum number of messages per page.","schema":{"type":"integer","minimum":1,"maximum":100,"default":50}},{"name":"afterOffset","in":"query","required":false,"description":"Continuation cursor from the previous page's `nextPageURL`; returns messages after that\nposition. Take it from `nextPageURL` rather than computing it.","schema":{"type":"string"}},{"name":"time_gt","in":"query","required":false,"description":"Only messages published strictly after this time, as Unix epoch milliseconds.","schema":{"type":"integer","format":"int64"}},{"name":"time_lte","in":"query","required":false,"description":"Only messages published at or before this time, as Unix epoch milliseconds.","schema":{"type":"integer","format":"int64"}}],"responses":{"200":{"description":"One page of messages. `messages` is empty when nothing is retained in the requested window.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/TopicMessagesResponse"}}}},"400":{"description":"A query parameter is malformed: `limit` outside 1–100 (`invalid_limit`), `time_gt` or\n`time_lte` not epoch milliseconds (`invalid_time_gt`, `invalid_time_lte`), or `afterOffset`\nnot a cursor issued by this endpoint (`invalid_after_offset`).","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"$ref":"#/components/responses/404-not-found"}}}}}}
```

## Purge retained messages

> Removes every retained message from the topic while keeping the topic, its retention setting,\
> and its subscriptions; the purge cannot be undone. Positions keep counting: the next message\
> published takes the next offset, subscribed listeners receive it as before, and a listener\
> whose saved position was purged resumes from the earliest message still retained. Records an\
> audit entry with \`event: purge\` and \`fieldChanges\[0].fieldPath: messages\`. To remove the topic\
> itself, use \`DELETE /v1/topics/{\_id}\`.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"PurgeMessagesResponse":{"type":"object","description":"Result of purging a topic's retained messages.","required":["deleted"],"properties":{"deleted":{"type":"boolean","description":"When true, the topic's stream was cleared — also reported when it held no messages at the time.\nFalse only for a topic that has never received a message, so there was no stream to clear."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}},"404-not-found":{"description":"Not found. The requested resource does not exist or is not visible to the caller.\n","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}},"paths":{"/v1/topics/{_id}/messages":{"delete":{"summary":"Purge retained messages","description":"Removes every retained message from the topic while keeping the topic, its retention setting,\nand its subscriptions; the purge cannot be undone. Positions keep counting: the next message\npublished takes the next offset, subscribed listeners receive it as before, and a listener\nwhose saved position was purged resumes from the earliest message still retained. Records an\naudit entry with `event: purge` and `fieldChanges[0].fieldPath: messages`. To remove the topic\nitself, use `DELETE /v1/topics/{_id}`.","operationId":"purgeTopicMessages","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}}],"responses":{"200":{"description":"Purge result.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/PurgeMessagesResponse"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"$ref":"#/components/responses/404-not-found"}}}}}}
```

## Publish messages to a topic

> Appends messages to the topic in the order given, so every subscribed topic export receives them\
> on its next read. The first publish to a new topic provisions its stream, which can take a few\
> seconds longer than later calls. Oversized messages are rejected individually while the rest of\
> the batch is still published — always check \`responses\[].statusCode\`.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"PublishRequest":{"type":"object","description":"Messages to append to the topic.","required":["messages"],"properties":{"messages":{"type":"array","description":"Messages to append, in order; each becomes one record that listeners receive. An empty array\nis accepted and publishes nothing. Every message is serialized as JSON and may be at most\n1 MiB; oversized messages are reported per entry in the response and the rest are still\npublished.","items":{"type":"object","description":"One message. Any JSON object; the platform stores it as-is and returns it as `payload` when read back."}}}},"PublishResponse":{"type":"object","description":"Per-message outcome of a publish call. The HTTP status is 200 whenever the request body was\nvalid; inspect each entry's `statusCode` to see which messages were appended.","required":["responses"],"properties":{"responses":{"type":"array","description":"One entry per element of `messages`, in the same order. Empty when `messages` was empty.","items":{"type":"object","required":["statusCode"],"properties":{"statusCode":{"type":"integer","description":"Outcome for this message: 200 when it was appended, 422 when it was rejected before\npublishing (for example because it exceeds the size limit), 500 when the stream\nrefused the batch it belonged to."},"_json":{"type":"object","description":"Present when `statusCode` is 200. Currently always an empty object."},"errors":{"type":"array","description":"Present when `statusCode` is not 200. Why this message was not published.","items":{"type":"object","required":["code","message"],"properties":{"code":{"type":"string","description":"Machine-readable reason. `topic_publish_event_too_large` when the serialized\nmessage exceeds 1 MiB; `topic_publish_failed` when the stream rejected the batch."},"message":{"type":"string","description":"Human-readable explanation, naming the message index."},"source":{"type":"string","description":"Layer that produced the error — `resource` for size rejections, `internal` for stream failures."},"resolved":{"type":"boolean","description":"Always false here; publish errors are not retried by the platform."},"occurredAt":{"type":"integer","description":"Unix epoch milliseconds when the error was raised."}}}}}}}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}},"404-not-found":{"description":"Not found. The requested resource does not exist or is not visible to the caller.\n","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}},"paths":{"/v1/topics/{_id}/publish":{"post":{"summary":"Publish messages to a topic","description":"Appends messages to the topic in the order given, so every subscribed topic export receives them\non its next read. The first publish to a new topic provisions its stream, which can take a few\nseconds longer than later calls. Oversized messages are rejected individually while the rest of\nthe batch is still published — always check `responses[].statusCode`.","operationId":"publishTopicMessages","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}}],"requestBody":{"required":true,"content":{"application/json":{"schema":{"$ref":"#/components/schemas/PublishRequest"}}}},"responses":{"200":{"description":"Per-message results in request order. A 200 means the request was accepted, not that every\nmessage was published.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/PublishResponse"}}}},"400":{"description":"`messages` is missing or not an array. Error code: `invalid_messages`.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"$ref":"#/components/responses/404-not-found"}}}}}}
```

## Get a listener's read position

> Reports how far a topic export (listener) has read on this topic — where its next run resumes.\
> Use it to check whether a listener is keeping up, or whether it has a read position at all: it\
> has none while it has not subscribed, or while its topic has never received a message. The export\
> must belong to the account but need not subscribe to this topic; an export that does not simply\
> has no cursor.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"TopicExportOffsets":{"type":"object","description":"Read position of one topic export (listener) on a topic.","required":["hasCursor","offset"],"properties":{"hasCursor":{"type":"boolean","description":"When true, the listener is subscribed to this topic and its next run resumes from\n`offset`. The position appears when the listener's flow goes from disabled to enabled,\nbefore anything is read: at the end of the stream under `topic.startPosition: latest`,\nat the first retained message under `earliest`. On a topic nothing has ever been\npublished to, it appears only with the first publish, and the listener still receives\nthat first message. False in two cases: the listener has never subscribed — it was saved\nwithout `type: \"stream\"`, its flow has not gone from disabled to enabled since it was\nsaved (a flow created enabled never subscribes), or it does not listen to this topic; or\nits topic has never received a message, until the first publish. A listener whose flow\nis disabled keeps its position. When the message at that position is no longer retained\n— it expired past `retentionMs` or was purged — the listener still reports true with the\nold `offset`, and its next run resumes from the earliest message still retained,\nwhatever its `startPosition`."},"offset":{"type":"integer","minimum":-1,"description":"Zero-based position of the next message the export will read. `-1` when `hasCursor` is\nfalse. The position counts from the start of the stream and keeps counting after a purge,\nso while no message has expired or been purged the retained message count minus `offset`\nis the listener's backlog."}}},"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}}}},"paths":{"/v1/topics/{_id}/exports/{_exportId}/offsets":{"get":{"summary":"Get a listener's read position","description":"Reports how far a topic export (listener) has read on this topic — where its next run resumes.\nUse it to check whether a listener is keeping up, or whether it has a read position at all: it\nhas none while it has not subscribed, or while its topic has never received a message. The export\nmust belong to the account but need not subscribe to this topic; an export that does not simply\nhas no cursor.","operationId":"getTopicExportOffsets","tags":["Topics"],"parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}},{"name":"_exportId","in":"path","required":true,"description":"Id of the topic export (`adaptorType: TopicExport`) whose position to read.","schema":{"type":"string","format":"objectId"}}],"responses":{"200":{"description":"Read position of the export on this topic.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/TopicExportOffsets"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"description":"The topic or the export does not exist in this account.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## Get audit log for a topic

> Returns the change history for a topic, newest first: who changed what and when, including\
> message purges, recycle-bin deletes and restores. Works for a topic in the recycle bin; once the\
> topic is purged, query \`GET /v1/audit?resourceType=topic&\_resourceId={\_id}\` instead.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"Error":{"type":"object","description":"Standard error response envelope returned by integrator.io APIs.","properties":{"errors":{"type":"array","description":"List of errors that occurred while processing the request.","items":{"type":"object","properties":{"code":{"oneOf":[{"type":"string"},{"type":"integer"}],"description":"Machine-readable error code. Usually a string like\n`invalid_ref`, `missing_required_field`, or `unauthorized`;\nmay be an **integer** when the error mirrors an upstream HTTP\nstatus (e.g. `500`) — most commonly returned by connection-ping\nand adaptor-proxy responses."},"message":{"type":"string","description":"Human-readable description of the error."},"field":{"type":"string","description":"Optional pointer to the document field that caused the error.\nUsed by structural validation errors (`missing_required_field`,\n`invalid_ref`) to indicate which field is at fault\n(e.g. `_id`, `type`, `http.baseURI`)."},"source":{"type":"string","description":"Optional origin layer for the error — e.g. `application` when\nthe error came from the remote system the adaptor called,\n`connector` when the adaptor itself rejected the request."}},"required":["message"]}}},"required":["errors"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}},"404-not-found":{"description":"Not found. The requested resource does not exist or is not visible to the caller.\n","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}},"paths":{"/v1/topics/{_id}/audit":{"get":{"summary":"Get audit log for a topic","description":"Returns the change history for a topic, newest first: who changed what and when, including\nmessage purges, recycle-bin deletes and restores. Works for a topic in the recycle bin; once the\ntopic is purged, query `GET /v1/audit?resourceType=topic&_resourceId={_id}` instead.","operationId":"getTopicAudit","tags":["Topics"],"parameters":[{"name":"_id","in":"path","description":"Topic id.","required":true,"schema":{"type":"string","format":"objectId"}},{"name":"limit","in":"query","description":"Maximum number of audit entries to return per page.","required":false,"schema":{"type":"integer","minimum":1,"maximum":1000,"default":1000}},{"name":"after","in":"query","required":false,"description":"Opaque pagination cursor for the next page. Take it from the `after` value in the `Link`\nresponse header's `rel=\"next\"` URL rather than constructing it.","schema":{"type":"string"}},{"name":"from","in":"query","required":false,"description":"Only return entries at or after this timestamp.","schema":{"type":"string","format":"date-time"}},{"name":"to","in":"query","required":false,"description":"Only return entries at or before this timestamp.","schema":{"type":"string","format":"date-time"}},{"name":"action","in":"query","required":false,"description":"Filter by the change type. Maps to the `event` field on each entry; a value outside the\nplatform's audit vocabulary is rejected with 422 `invalid_audit_field`.","schema":{"type":"string","enum":["create","update","delete","view","restore","purge","signin","signout","purgelogs"]}},{"name":"source","in":"query","required":false,"description":"Filter by how the change was initiated.","schema":{"type":"string","enum":["ui","api","system","connector","script","sso"]}},{"name":"_byUserId","in":"query","required":false,"description":"Filter to changes performed by a single user.","schema":{"type":"string","format":"objectId"}}],"responses":{"200":{"description":"Array of audit entries, newest first.","headers":{"Link":{"description":"RFC 5988 pagination link. Includes `<...>; rel=\"next\"` with an `after` cursor when more\nentries exist; absent on the final page.","schema":{"type":"string"}}},"content":{"application/json":{"schema":{"type":"array","items":{"type":"object","properties":{"_id":{"type":"string","format":"objectId","description":"Unique identifier for this audit entry."},"resourceType":{"type":"string","description":"Always `topic` for this endpoint."},"_resourceId":{"type":"string","format":"objectId","description":"Id of the topic this entry belongs to."},"event":{"type":"string","description":"Type of change. A topic's history runs create → update… → delete → restore or\npurge; `purge` also records each `DELETE /v1/topics/{_id}/messages`, which is\ntold apart by `fieldChanges[0].fieldPath` being `messages`.","enum":["create","update","delete","restore","purge"]},"source":{"type":"string","description":"How the change was initiated. A change made through the Celigo CLI is recorded as\n`cli` — deletes, restores from the recycle bin, and message purges included.","enum":["api","ui","system","cli"]},"time":{"type":"string","format":"date-time","description":"Timestamp when the change occurred."},"byUser":{"type":"object","description":"User who made the change.","properties":{"_id":{"type":"string","format":"objectId","description":"User id."},"email":{"type":"string","format":"email","description":"User email address."},"name":{"type":"string","description":"User display name."}}},"fieldChanges":{"type":"array","description":"Fields that were modified. Empty on `create`, `delete`, and `restore`; a single\n`messages` entry without values on a message purge.","items":{"type":"object","properties":{"fieldPath":{"type":"string","description":"Dot-notation path of the changed field, or `messages` for a message purge."},"oldValue":{"description":"Previous value (absent when the field was unset)."},"newValue":{"description":"New value (absent when the field was cleared)."}}}},"deletedInfo":{"type":"object","description":"Present on every entry once the topic has been deleted; carries the topic's name at\nthe time it was deleted so purged topics stay identifiable.","properties":{"name":{"type":"string","description":"Name of the topic when it was deleted."}}}}}}}}},"204":{"description":"No audit entries exist for this topic."},"400":{"description":"`limit` is not a positive integer. Error code: `invalid_query_param`.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"},"404":{"$ref":"#/components/responses/404-not-found"},"422":{"description":"`action` or `source` is outside the audit vocabulary (`invalid_audit_field`), or `from` / `to`\nis not an ISO 8601 timestamp (`invalid_timestamp_format`).","content":{"application/json":{"schema":{"$ref":"#/components/schemas/Error"}}}}}}}}}
```

## List dependencies of a topic

> Returns the resources that reference the topic: \`exports\` and \`imports\` bound to it through\
> \`topic.\_topicId\`, \`flows\` that contain those steps, and \`integrations\` that list it in\
> \`\_registeredTopicIds\`. Exports and imports listed here block \`DELETE /v1/topics/{\_id}\` until they\
> are removed or repointed. An unknown topic id returns an empty map rather than 404.

```json
{"openapi":"3.2.0","info":{"title":"Topics","version":"1.0.0"},"tags":[{"name":"Topics","description":"Topics are durable event streams that decouple the flows that publish messages from the flows\nthat consume them: a topic import step (or `POST /v1/topics/{_id}/publish`) appends messages, each\nsubscribed topic export step reads them at its own pace, and messages expire after the topic's\n`retentionMs`. Event Streams is a licensed feature: listing topics works on every account, but\ncreating or updating a topic and every message operation require the `topics` entitlement —\nwithout it they fail with `topics_subscription_required` (a 422 validation error on topic writes,\n403 on the message endpoints) — and the license also caps how many topics the account may hold\n(`topicsAllowed`) and the longest retention allowed (`retentionCeilingDays`).\n\n## Topic schema\n\n{% openapi-schemas spec=\"topic\" schemas=\"Topic\" grouped=\"true\" %}"}],"servers":[{"url":"https://api.integrator.io","description":"Production (US / default region)"},{"url":"https://api.eu.integrator.io","description":"Production (EU region)"},{"url":"https://api.au.integrator.io","description":"Production (AU region)"},{"url":"https://api.ca.integrator.io","description":"Production (CA region)"}],"security":[{"bearerAuth":[]}],"components":{"securitySchemes":{"bearerAuth":{"type":"http","scheme":"bearer"}},"schemas":{"DependencyResponse":{"type":"object","description":"Map of dependent-resource types to arrays of dependency entries.\nKeys are plural resource type strings (e.g. `flows`, `imports`,\n`connections`). An empty object `{}` means no dependents.\n","additionalProperties":{"type":"array","items":{"$ref":"#/components/schemas/DependencyEntry"}}},"DependencyEntry":{"type":"object","description":"A single resource that depends on the queried resource.","properties":{"id":{"type":"string","description":"Unique identifier of the dependent resource."},"name":{"type":"string","description":"Display name of the dependent resource."},"type":{"type":"string","description":"Sub-type of the dependent resource when its collection distinguishes one, such as\n`stream` for a listener export. Only present for resource types that carry a `type`."},"paths":{"type":"array","description":"Dot-notation paths within the dependent resource that reference\nthe target resource. `[*]` denotes array elements.","items":{"type":"string"}},"accessLevel":{"type":"string","description":"The caller's access level on the dependent resource."},"dependencyIds":{"type":"object","description":"Map of resource types to arrays of IDs that this dependent\nresource references on the target. Keys are singular or plural\nresource type strings; values are arrays of ID strings.","additionalProperties":{"type":"array","items":{"type":"string"}}}},"required":["id","name","paths","accessLevel","dependencyIds"]}},"responses":{"401-unauthorized":{"description":"Unauthorized. The request lacks a valid bearer token, or the provided token\nfailed to authenticate.\n\nNote: the 401 response is produced by the auth middleware **before** the\nrequest reaches the endpoint handler, so it does **not** follow the\nstandard `{errors: [...]}` envelope. Instead the body is a bare\n`{message: string}` object with no `code`, no `errors` array. Callers\nhandling 401s should key off the HTTP status and the `message` string,\nnot try to destructure an `errors[]`.","content":{"application/json":{"schema":{"type":"object","properties":{"message":{"type":"string","description":"Human-readable description of the auth failure. Known values:\n- `\"Unauthorized\"` — no `Authorization` header on the request.\n- `\"Bearer Authentication Failed\"` — header present but token\n  is invalid, revoked, or expired."}},"required":["message"]}}}}}},"paths":{"/v1/topics/{_id}/dependencies":{"get":{"operationId":"listTopicDependencies","tags":["Topics"],"summary":"List dependencies of a topic","description":"Returns the resources that reference the topic: `exports` and `imports` bound to it through\n`topic._topicId`, `flows` that contain those steps, and `integrations` that list it in\n`_registeredTopicIds`. Exports and imports listed here block `DELETE /v1/topics/{_id}` until they\nare removed or repointed. An unknown topic id returns an empty map rather than 404.","parameters":[{"name":"_id","in":"path","required":true,"description":"Topic id.","schema":{"type":"string","format":"objectId"}}],"responses":{"200":{"description":"Dependency map. Keys are resource-type strings; values are arrays of dependency entries.\nReturns `{}` when nothing references the topic.","content":{"application/json":{"schema":{"$ref":"#/components/schemas/DependencyResponse"}}}},"401":{"$ref":"#/components/responses/401-unauthorized"}}}}}}
```