Google Pub/Sub

Server Source code Package

Server-side event publishing to Google Cloud Pub/Sub topics. Each event is serialized as JSON, optionally keyed for per-message ordering, and published via the official @google-cloud/pubsub SDK. Supports Application Default Credentials, service account JSON, or pre-configured client injection. Per-rule topic, ordering-key, and attribute overrides. Idempotent topic provisioning with EU storage policy by default.

The destination ships inside @walkeros/server-destination-gcp alongside BigQuery; install the package once and import destinationPubSub for Pub/Sub.

Where this fits

Pub/Sub is a server destination in the walkerOS flow:

Publishes events to a Pub/Sub topic for fan-out, asynchronous processing, cross-region delivery, and integration with downstream consumers (Dataflow, Cloud Functions, Cloud Run, custom subscribers).

Installation

npm install @walkeros/server-destination-gcp

Integrated

import { startFlow } from '@walkeros/collector';
import { destinationPubSub } from '@walkeros/server-destination-gcp';

await startFlow({
destinations: {
  pubsub: {
    code: destinationPubSub,
    config: {
      settings: {
        projectId: 'YOUR_PROJECT_ID',
        topic: 'events',
      },
    },
  },
},
});

Bundled

Add to your flow.json destinations:

"destinations": {
"pubsub": {
  "package": "@walkeros/server-destination-gcp",
  "import": "destinationPubSub",
  "config": {
    "settings": {
      "projectId": "YOUR_PROJECT_ID",
      "topic": "events"
    }
  }
}
}

See bundled mode setup | CLI reference

Configuration

This destination uses the standard destination config wrapper (consent, data, env, id, ...). For the shared fields see destination configuration. Package-specific fields live under config.settings and are listed below.

Settings

Property	Type	Description	More
client	any	Google Cloud Pub/Sub client instance
projectId*	string	Google Cloud Project ID
topic*	string	Pub/Sub topic short name (like events). The full resource name projects/<projectId>/topics/<topic> is built by the SDK.
credentials	any	Service account credentials as a JSON string or an object with client_email and private_key. Default: Application Default Credentials (ADC). (deprecated: use config.credentials)
apiEndpoint	string	Override Pub/Sub API endpoint. Useful for the local emulator (like localhost:8085).
orderingKey	any	Mapping value resolved per-event. Truthy enables per-key ordering for the publish.
attributes	any	Default per-event attribute map merged into every published message. Mapping.Map shape.

* Required fields

Mapping

Per-event rules under config.mapping. For the standard rule fields (consent, condition, data, batch, name, policy) see mapping.

Property	Type	Description	More
topic	string	Per-rule topic override. Falls back to settings.topic if absent.
orderingKey	any	Per-rule ordering-key Mapping.Value. Overrides settings.orderingKey when set.
attributes	any	Per-rule attribute Mapping.Map merged on top of settings.attributes.

Authentication

Three modes, evaluated in order:

1. Application Default Credentials (ADC). Nothing to configure beyond projectId. Works on GCP-native runtimes (Cloud Run, Cloud Functions, GKE, GCE) where the runtime service account is attached, and locally with gcloud auth application-default login.
2. Service account JSON. Pass credentials via config.credentials, either as a parsed object or a JSON string (the destination JSON-parses strings). Pair with $env.NAME to inject from an environment variable. (The package-specific settings.credentials still works but is deprecated.)
3. Pre-configured client. Pass an existing PubSub SDK instance as settings.client. Useful for shared clients across destinations or custom transport configuration.

When settings.projectId and the credentials' project_id both resolve, the top-level settings.projectId wins. The runtime publish target stays unambiguous.

Ordering

orderingKey is a Mapping.Value resolved per event. Truthy resolved values enable per-key ordering, the topic handle is constructed with messageOrdering: true, and the resolved key flows on the publish.

{
  "settings": {
    "projectId": "my-project",
    "topic": "user-events",
    "orderingKey": "user.id"
  }
}

Per-rule overrides live under mapping.<entity>.<action>.settings.orderingKey.

If a publish fails for an ordered key, Pub/Sub permanently halts subsequent publishes for that key until topic.resumePublishing(key) is called. The destination handles this automatically: on publish failure for an ordered key it calls resumePublishing immediately and re-throws so the caller observes the original error.

Attributes

settings.attributes is a Mapping.Map of attribute name to mapping value. Each value is resolved per event and stringified. Per-rule overrides under mapping.<entity>.<action>.settings.attributes are merged onto the defaults.

{
  "settings": {
    "projectId": "my-project",
    "topic": "events",
    "attributes": {
      "tenant": "user.tenant_id",
      "schema_version": { "value": "v4" }
    }
  }
}

Setup

Provision the topic once per environment:

walkeros setup destination.pubsub

Idempotent: re-running on an existing topic logs setup: topic exists and runs drift detection. Drift on messageStoragePolicy.allowedPersistenceRegions, messageRetentionDuration, kmsKeyName, or labels emits WARN setup.drift {...} and never auto-mutates. Migrations are an operator decision.

config.setup:

• false (default): no provisioning. Operator must run setup explicitly.
• true: provisions with safe defaults (EU multi-region storage policy).
• { messageStoragePolicy, messageRetentionDuration, kmsKeyName, labels }: object form for explicit overrides.

Default storage regions: ['europe-west1', 'europe-west3', 'europe-west4'] (EU multi-region). Override messageStoragePolicy.allowedPersistenceRegions for projects with org policies that restrict geography.

The provisioning identity needs pubsub.topics.create and pubsub.topics.get on the project. The runtime publish identity needs roles/pubsub.publisher on the topic itself.

Subscription provisioning is owned by the Pub/Sub source, not this destination.

Emulator

The official Pub/Sub emulator runs locally for development:

gcloud beta emulators pubsub start --host-port=localhost:8085
export PUBSUB_EMULATOR_HOST=localhost:8085

The SDK automatically picks up PUBSUB_EMULATOR_HOST. For explicit configuration, set settings.apiEndpoint (e.g. localhost:8085).

Troubleshooting

• NOT_FOUND on publish: the topic does not exist. Run walkeros setup destination.<id> once.
• PERMISSION_DENIED / UNAUTHENTICATED: the runtime service account lacks roles/pubsub.publisher on the topic, or ADC is not configured. The destination logs the actionable message including the topic and project IDs.
• Ordering stuck: a previous publish for an ordering key failed. The destination calls resumePublishing automatically; if stalls persist, check publish-side error logs for the original failure.

Next steps

• Pub/Sub source for consuming events from a Pub/Sub topic
• GCP package overview for BigQuery and shared GCP setup