Protocols

Ironflow supports two streaming protocols for real-time event subscriptions:

Protocol	Endpoint	Use Case
WebSocket	ws://localhost:9123/ws	Browser clients, bidirectional communication
ConnectRPC	http://localhost:9123/ironflow.v1.PubSubService/*	Server-to-server, gRPC-compatible clients

Both protocols support pattern subscriptions, replay, CEL filtering, and consumer groups. Manual acknowledgment is not implemented on either transport: the ConnectRPC bidirectional stream returns Unimplemented, and the WebSocket ack frame is parsed and logged but changes no delivery behavior. See Acknowledgment below before designing against it.

---

WebSocket Protocol

Connect directly to ws://localhost:9123/ws for browser clients or custom integrations.

Message Types

Type	Direction	Description
subscribe	Client → Server	Request subscription
unsubscribe	Client → Server	Cancel subscription
ack	Client → Server	Event acknowledgment
subscription_result	Server → Client	Subscription confirmation
subscribed	Server → Client	Legacy subscription confirmation
event	Server → Client	Event delivery
subscription_error	Server → Client	Subscription-specific error

Subscribe Request

{
  "type": "subscribe",
  "subscription": {
    "pattern": "system.run.>",
    "options": {
      "replay": 10,
      "includeMetadata": true,
      "filter": "data.status == \"completed\"",
      "consumerGroup": "my-group",
      "ackMode": "manual",
      "topicOnly": false
    }
  }
}

topicOnly (default false) delivers each frame without its data payload — use it when the subscriber only needs to know that an event fired. The event is still marshaled and sent per client, so the saving is the payload itself, not the frame. data is present but null on suppressed frames.

Metadata is unaffected: it stays governed by includeMetadata.

startAfterSequence (optional, no default) resumes delivery after a stream sequence the client previously received on an event’s meta.sequence (#1848). It is a position, not a count, so a reconnecting client neither repeats nor skips work the way replay does. Sequence 0 is a legitimate “from the beginning of the stream”, so the field is omitted rather than zero when unset. Two rejections, both INVALID_ARGUMENT:

• combined with replay — they are different starting points, so the server refuses rather than merging them
• combined with consumerGroup — a group is durable and server-positioned, and a client cursor would move the position for every other member

Setting it routes the subscription to a dedicated JetStream consumer instead of the subject’s shared one. Delivery stays at-least-once: a resumed stream can repeat the frame that was in flight when the connection dropped.

Subscription Result

{
  "type": "subscription_result",
  "results": [
    {
      "pattern": "system.run.>",
      "status": "ok",
      "subscriptionId": "sub_abc123"
    }
  ]
}

Event Delivery

{
  "type": "event",
  "subscriptionId": "sub_abc123",
  "eventId": "evt_xyz789",
  "topic": "system.run.xyz.created",
  "data": {
    "runId": "xyz",
    "functionId": "process-order",
    "status": "running"
  },
  "meta": {
    "timestamp": "2024-01-15T10:30:00Z",
    "sequence": 12345
  }
}

When includeMetadata: true is set in the subscribe options, the meta field is included in each event delivery. For user-emitted events, meta also contains a custom field with any metadata attached at emit time:

{
  "type": "event",
  "subscriptionId": "sub_abc123",
  "eventId": "evt_xyz789",
  "topic": "events:order.placed",
  "data": { "orderId": "123" },
  "meta": {
    "timestamp": "2024-01-15T10:30:00Z",
    "sequence": 12346,
    "custom": {
      "source": "checkout",
      "traceId": "abc-123"
    }
  }
}

Unsubscribe Request

{
  "type": "unsubscribe",
  "subscriptionId": "sub_abc123"
}

Acknowledgment (Manual Ack Mode)

Accepted, not acted on

The server parses the ack frame, validates ackType, and logs it. It does not reach the subscription manager, so nak triggers no redelivery and term suppresses none. ackMode: "manual" is likewise accepted without changing how events are delivered. Treat every subscription as auto-ack until this is wired.

{
  "type": "ack",
  "eventId": "evt_xyz789",
  "ackType": "ack"
}

ackType	Description
ack	Successfully processed
nak	Failed, request redelivery
term	Failed permanently, do not redeliver

An optional redeliverDelay (milliseconds) may accompany a nak. It is read off the frame and logged with the same caveat as above — no redelivery is scheduled.

Error Response

{
  "type": "subscription_error",
  "subscriptionId": "sub_abc123",
  "code": "INVALID_PATTERN",
  "message": "invalid namespace prefix",
  "retrying": false
}

---

ConnectRPC Protocol

ConnectRPC provides gRPC-compatible streaming over HTTP. Use the PubSubService for subscriptions.

Service Definition

service PubSubService {
  // Emit an event (alias for Trigger with emit semantics)
  rpc Emit(EmitRequest) returns (EmitResponse);

  // Server-streaming subscription (auto-ack)
  rpc Subscribe(SubscribeRequest) returns (stream SubscriptionEvent);

  // Bidirectional streaming (manual ack)
  rpc SubscribeBidirectional(stream SubscriptionAck) returns (stream SubscriptionEvent);

  // Consumer group management
  rpc CreateConsumerGroup(CreateConsumerGroupRequest) returns (ConsumerGroup);
  rpc GetConsumerGroup(GetConsumerGroupRequest) returns (ConsumerGroup);
  rpc ListConsumerGroups(ListConsumerGroupsRequest) returns (ListConsumerGroupsResponse);
  rpc UpdateConsumerGroup(UpdateConsumerGroupRequest) returns (ConsumerGroup);
  rpc DeleteConsumerGroup(DeleteConsumerGroupRequest) returns (google.protobuf.Empty);

  // Consumer group streaming
  rpc JoinConsumerGroup(JoinConsumerGroupRequest) returns (stream SubscriptionEvent);

  // Developer pub/sub (does NOT trigger workflow functions)
  rpc Publish(PublishRequest) returns (PublishResponse);
  rpc ListTopics(ListTopicsRequest) returns (ListTopicsResponse);
  rpc GetTopicStats(GetTopicStatsRequest) returns (GetTopicStatsResponse);
}

Subscribe Request

Endpoint: POST /ironflow.v1.PubSubService/Subscribe

Request:

{
  "pattern": "system.run.>",
  "options": {
    "replay": 10,
    "include_metadata": true,
    "filter": "data.status == \"completed\"",
    "consumer_group": "my-group"
  }
}

Resuming after a disconnect — start_after_sequence is the ConnectRPC spelling of the WebSocket startAfterSequence option above. It is a uint64, so Connect’s JSON encoding carries it as a string:

{
  "pattern": "system.run.>",
  "options": {
    "include_metadata": true,
    "start_after_sequence": "12345"
  }
}

It takes the same two invalid_argument rejections — combined with replay, or combined with consumer_group — so it cannot appear in the request above. It is optional in the proto so that sequence 0 (“from the beginning of the stream”) stays distinguishable from unset.

SubscribeOptions also declares namespace, ack_mode, and backpressure. The Subscribe handler does not read any of the three: the namespace comes from the pattern’s own prefix, and manual ack is unimplemented. ack_mode and backpressure are honored on consumer groups, where they are set on the group itself through CreateConsumerGroup / UpdateConsumerGroup. There is no topic_only on this transport — that option is WebSocket-only.

Response (streaming):

{
  "subscription_id": "sub_abc123",
  "event_id": "evt_xyz789",
  "topic": "system.run.xyz.created",
  "data": {
    "runId": "xyz",
    "status": "running"
  },
  "metadata": {
    "timestamp": "2024-01-15T10:30:00.000Z",
    "source": "system",
    "namespace": "default"
  },
  "sequence": "12345",
  "delivery_attempt": 1
}

When include_metadata: true is set and the event has user-defined metadata (attached at emit time), it is included in the metadata object under the custom key:

{
  "subscription_id": "sub_abc123",
  "event_id": "evt_xyz789",
  "topic": "events:order.placed",
  "data": { "orderId": "123" },
  "metadata": {
    "timestamp": "2024-01-15T10:30:00.000Z",
    "source": "user",
    "namespace": "default",
    "custom": {
      "traceId": "abc-123",
      "source": "checkout"
    }
  },
  "sequence": "12346",
  "delivery_attempt": 1
}

Join Consumer Group

Endpoint: POST /ironflow.v1.PubSubService/JoinConsumerGroup

Request:

{
  "namespace": "default",
  "group_name": "order-processors",
  "consumer_id": "consumer-1",
  "client_id": "my-app-instance"
}

Response: Same streaming format as Subscribe.

Bidirectional Streaming (Manual Ack)

Not Yet Implemented

SubscribeBidirectional is defined in the proto but currently returns Unimplemented. The WebSocket transport is not a workaround — its ack frame is inert too. For at-least-once processing with redelivery today, use a consumer group.

Endpoint: POST /ironflow.v1.PubSubService/SubscribeBidirectional

Ack Types (for future use):

Value	Description
ACK_TYPE_ACK	Successfully processed
ACK_TYPE_NAK	Failed, request redelivery
ACK_TYPE_TERM	Failed permanently, do not redeliver

Using with cURL

# Subscribe to events (streaming response)
curl -X POST http://localhost:9123/ironflow.v1.PubSubService/Subscribe \
  -H "Content-Type: application/json" \
  -d '{
    "pattern": "system.run.>",
    "options": {
      "replay": 10,
      "include_metadata": true
    }
  }'

# Join a consumer group
curl -X POST http://localhost:9123/ironflow.v1.PubSubService/JoinConsumerGroup \
  -H "Content-Type: application/json" \
  -d '{
    "group_name": "order-processors"
  }'

---

Error Codes

These string codes are the WebSocket vocabulary — they arrive in the code field of a subscription_error or of an errored subscription_result entry. ConnectRPC does not use them: it returns standard Connect codes (invalid_argument, internal, not_found, unimplemented) with the detail in the error message.

Code	Description	Recoverable
INVALID_PATTERN	Pattern syntax error or invalid namespace prefix, or a malformed subscribe/unsubscribe frame	No
INVALID_ARGUMENT	Option combination the server refuses: startAfterSequence with replay, or startAfterSequence with consumerGroup	No
LIMIT_EXCEEDED	Too many subscriptions or replay limit exceeded	No
INTERNAL_ERROR	Server-side failure, or pub/sub not configured	Maybe

Two more codes are declared in the wire vocabulary but are not sent by the server today. Clients that already branch on them can keep doing so; nothing will produce them:

Code	Status
INVALID_NAMESPACE	Reserved. Namespace-prefix errors arrive as INVALID_PATTERN instead.
NATS_DISCONNECT	Reserved. The server does not surface NATS connection loss on this channel.

Note

RESUBSCRIBE_FAILED is not a server error code — it is generated client-side by the SDK when a subscription fails to restore after reconnection.

---

Choosing a Protocol

Consideration	WebSocket	ConnectRPC
Browser support	Native	Requires library
Bidirectional	Transport-level only — the ack frame is inert	Separate RPC, Unimplemented
gRPC ecosystem	No	Yes
Proxy compatibility	May need configuration	Standard HTTP/2
SDK support	Browser, Node, Go SDKs	Browser, Go SDKs

Recommendation:

• Use @ironflow/browser for browser clients — ConnectRPC is the default; call detectTransport() and pass its result to configure() when WebSocket fallback is needed
• Use ConnectRPC for server-to-server communication (via Go SDK or direct HTTP)