Skip to content

couchbase://

Producer + consumer for Couchbase Server, speaking the native binary protocol via github.com/couchbase/gocb/v2 — the official Couchbase Go SDK. No CGO, no JVM, no libcouchbase.

URI shape

couchbase://<host>[:<port>][,host2,...]/<bucket>[/<scope>[/<collection>]]?<params>

Multi-host bootstrap (CCCP): couchbase://cb1:11210,cb2:11210/orders?username=app&password=….

If scope / collection are omitted they default to _default.

Operations (op=)

Producer — set via the op= URI param or per-message via the HelixCouchbaseOp header.

op Input body Output body Common use
insert JSON bytes / map[string]any unchanged Fail-on-duplicate write.
upsert (default) JSON bytes / map[string]any unchanged Idempotent write.
replace JSON bytes / map[string]any unchanged Update existing doc; honours HelixCouchbaseCAS for optimistic locking.
get nil map[string]any Reads by id.
remove nil nil Deletes by id; honours HelixCouchbaseCAS.
touch nil nil Resets TTL (HelixCouchbaseTTL).
query nil or map[string]any (bindings) rows or single row or count N1QL. Shape via resultSetType=.
lookupIn nil map[string]any (path→value) Sub-document read.
mutateIn map[string]any nil Sub-document write.

Consumer — polling N1QL:

trigger:
  uri: couchbase://cb:11210/orders?query=SELECT * FROM orders WHERE status='new'&delay=2s

Each row becomes one inbound exchange.

Body shapes

Mutation inputs (insert/upsert/replace):

Body type Behaviour
[]byte Treated as raw JSON.
string Treated as raw JSON.
map[string]any Marshalled to JSON.
mutateIn only: map[string]any where keys = paths Each path → value is written under subdocPaths=.

Read outputs (get/lookupIn/query):

op Output
get map[string]any (parsed JSON)
lookupIn map[string]any keyed by sub-doc path
query (default) []map[string]any (all rows)
query + resultSetType=one map[string]any (first row, or nil if empty)
query + resultSetType=count int64

Headers

Inputs (per-message overrides):

Header Overrides
HelixCouchbaseOp URI op=
HelixCouchbaseID URI id=
HelixCouchbaseQuery URI query=
HelixCouchbaseTTL URI expiry=
HelixCouchbaseCAS Optimistic-lock CAS for replace / remove

Outputs:

Header Set on Meaning
HelixCouchbaseCAS get / mutations Doc CAS (uint64).
HelixCouchbaseExpiry get RFC3339 TTL deadline, when set.
HelixCouchbaseApplied mutations "true" on success.
HelixCouchbaseRowCount query / consumer Materialised row count.
HelixCouchbaseBucket consumer Source bucket.

Parameters

Quick reference:

Group Keys
Operation op, id, query, expiry, durabilityLevel
Sub-doc subdocOp, subdocPaths
Query scanConsistency, resultSetType
Security username, password, tlsCAFile, tlsCertFile, tlsKeyFile, tlsInsecureSkipVerify, tlsMinVersion, tlsMaxVersion
Timeouts connectTimeout, kvTimeout, queryTimeout
Consumer delay, initialDelay, repeatCount
Headers headerIncludePattern, headerExcludePattern

Examples

Upsert orders from a Kafka topic

workflows:
  - id: ingest-orders
    trigger:
      uri: kafka://orders.in?bootstrapServers=kafka:9092&groupId=order-writer
    steps:
      - target: "couchbase://cb1:11210,cb2:11210/orders/_default/_default?username=${CB_USER}&password=${CB_PASS}&op=upsert&durabilityLevel=majority"

Body is the Kafka value (JSON bytes); the document key comes from the HelixCouchbaseID header set upstream (or id= URI param if every message uses the same key — unlikely).

Read with optimistic locking

workflows:
  - id: update-stock
    trigger:
      uri: http-server://0.0.0.0:8080/api/decrement
    steps:
      # 1. Load doc + CAS into headers.
      - target: "couchbase://cb:11210/inventory?op=get"
      # 2. Mutate body.
      - transform: "{ stock: payload.stock - 1, ...payload }"
      # 3. Conditional write — fails if doc changed under us.
      - target: "couchbase://cb:11210/inventory?op=replace"

The HelixCouchbaseCAS header set by step 1 is read by step 3 — gocb sends it as ReplaceOptions.Cas, returning CasMismatch if the doc changed since the get.

N1QL polling consumer

workflows:
  - id: process-new-orders
    trigger:
      uri: couchbase://cb:11210/orders?username=app&password=…&query=SELECT META().id, * FROM orders WHERE status='new'&delay=5s&scanConsistency=requestPlus
    steps:
      - target: "log://new-order"

Each query row becomes one exchange (body = map[string]any). Use scanConsistency=requestPlus for read-your-writes (slower but strongly consistent).

Sub-document write

- transform:
    constant: |
      {
        "fulfillment.status": "shipped",
        "fulfillment.updated_at": "2026-06-15T10:00:00Z"
      }
- target: "couchbase://cb:11210/orders?id=order-42&op=mutateIn&subdocOp=upsert&subdocPaths=fulfillment.status,fulfillment.updated_at"

The body's keys become sub-doc paths; only those fields hit the wire.

Failure modes

Error Cause Fix
couchbase: empty URI body URI missing host. couchbase://host:port/bucket.
couchbase: URI missing /bucket path URI has host but no bucket. Append /<bucket>.
DocumentExistsError op=insert and the doc key already exists. Use upsert if you want idempotent, or catch + handle.
DocumentNotFoundError op=get/replace/remove/touch on a missing key. Pre-check, or use op=upsert with conflict handling.
CasMismatch op=replace/remove with HelixCouchbaseCAS and the doc was modified concurrently. Re-read, re-apply (optimistic retry loop).
couchbase: consumer requires query= URI param Tried to use couchbase:// as a trigger without query=. Add query=SELECT ….
couchbase: unknown URI parameter warning Typo in a URI param. helix components couchbase lists every accepted key.

Performance notes

  • Connection pooling. One *gocb.Cluster is shared across every endpoint with the same (authority, username, TLS) tuple. No per-message connect cost.
  • Consumer poll interval. Default 500ms is aggressive — raise to 5s–30s for low-volume streams; lower to 100ms only when latency matters and you understand the load on the Query service.
  • Durability cost. majorityAndPersistActive and persistToMajority are 5–20× slower than none/majority because they wait for disk fsync. Use only for data you cannot afford to re-process.
  • Sub-doc vs full-doc. mutateIn writes touch only the named paths over the network; for a 1MB doc updating one field is ~1000× cheaper than replace.