pushflo

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 14, 2026 License: MIT Imports: 16 Imported by: 0

README

pushflo-go

CI Go Reference Go Report Card

A Go SDK for PushFlo, plus an end-to-end encryption envelope layer so payloads stay opaque to the relay.

This first pass ships two things:

  • pushflo — a subscriber client (WebSocket, auto-reconnect, heartbeat) mirroring the ergonomics of the official @pushflodev/sdk client.
  • pushflo/envelope — a transport-agnostic package that seals a payload to a recipient's key and signs it, so PushFlo (or any pub/sub relay) only ever moves ciphertext it cannot read, forge, or replay.

Status: v0. The wire protocol (WebSocket frames, auth, heartbeat, REST publish) is confirmed against the official @pushflodev/sdk v1.1.0 source and verified against the live service by the integration test suite. The public API may still evolve before v1.

Install

Requires Go 1.25 or newer.

go get github.com/PushFlo/pushflo-go

Subscriber client

ctx := context.Background()
client, err := pushflo.NewClient(ctx, pushflo.ClientOptions{
    PublishKey: "pub_xxxxxxxxxxxxx",
})
if err != nil { log.Fatal(err) }
defer client.Destroy()

client.OnConnectionChange(func(s pushflo.ConnectionState) {
    log.Println("connection:", s)
})

if err := client.Connect(ctx); err != nil { log.Fatal(err) }

sub, _ := client.Subscribe("orders-user-123", pushflo.SubscriptionHandlers{
    OnMessage: func(m pushflo.Message) {
        log.Printf("[%s] %s", m.EventType, string(m.Content))
    },
})
defer sub.Unsubscribe()

Options mirror the TS SDK: BaseURL, AutoConnect, Debug, ConnectionTimeout, HeartbeatInterval, AutoReconnect, MaxReconnectAttempts (0 = infinite), ReconnectDelay, MaxReconnectDelay. Channel-slug helpers (IsValidChannelSlug, ToChannelSlug, ValidateChannelSlug) and typed errors (ConnectionError, AuthenticationError, NetworkError, ValidationError) are included.

Encryption envelope

The relay is treated as untrusted transport. Each payload is sealed to the recipient's X25519 public key (NaCl sealed box) and signed with the sender's Ed25519 key. The recipient verifies the signature, checks freshness and replay, then decrypts.

 cloud (sender)                         PushFlo (relay)            executor (recipient)
 ─────────────                          ───────────────           ────────────────────
 Seal(payload) ──► {mid,ts,ch,ct,sig} ──► opaque bytes ──► Open() ─► payload
   sign Ed25519                            can't read /              verify sig
   seal to X25519 pub                      forge / replay            check ts + replay
                                                                     decrypt X25519

What each side holds after examples/keygen:

  • Cloud: executor's RECIPIENT_PUBLIC + its own SIGNER_PRIVATE.
  • Executor: its own RECIPIENT_PRIVATE + cloud's SIGNER_PUBLIC.

No secret is ever shared with PushFlo.

Sender:

sealer, _ := envelope.NewSealer(envelope.SealerConfig{
    RecipientPublic: executorPub,       // [32]byte
    SignerPrivate:   cloudSignPriv,     // ed25519.PrivateKey
    SignerKeyID:     "cloud-1",
})
env, _ := sealer.SealJSON(job, envelope.Meta{Channel: "executor-1"})
wire, _ := env.Marshal()                // publish `wire` as the channel message

Recipient:

opener, _ := envelope.NewOpener(envelope.OpenerConfig{
    RecipientPublic:  executorPub,
    RecipientPrivate: executorPriv,
    Signers:          map[string]ed25519.PublicKey{"cloud-1": cloudSignPub},
    MaxClockSkew:     2 * time.Minute,
    Replay:           envelope.NewMemoryReplayCache(10 * time.Minute),
})

env, _ := envelope.Parse(message.Content)
var job Job
if err := opener.OpenJSON(env, &job); err != nil {
    // ErrBadSignature / ErrReplay / ErrExpired / ErrDecrypt / ErrUnknownSigner
    return
}
What the envelope guarantees
  • Confidentiality — only the recipient's private key can open the sealed box.
  • Authenticity & integrity — every field (message id, timestamp, channel, content type) is bound into the Ed25519 signature, so the relay cannot alter routing metadata or the ciphertext undetected. This matters because the executor runs HTTP calls on command: an unauthenticated job would be an SSRF-with-your-credentials hole.
  • Replay protection — message id + timestamp; ReplayCache rejects duplicates. Set the cache TTL ≥ MaxClockSkew. For a fleet or across restarts, back it with Redis/DB by implementing the ReplayCache interface.
What it does not hide

Message sizes, timing, and which channels are active remain visible to the relay. Pad or batch if that metadata matters.

Key rotation

KeyID (recipient epoch) and SignKeyID travel in the clear so you can rotate: give the Opener a Signers map with multiple ids, publish under the new id, then retire the old key.

Publisher

Publishing uses a secret (sec_) key. PublishSealed is the symmetric counterpart to the envelope Opener — it encrypts, signs, and publishes in one call, so the relay only sees ciphertext.

pub, _ := pushflo.NewPublisher(pushflo.PublisherOptions{SecretKey: "sec_xxx"})

// plaintext publish
pub.Publish(ctx, "orders", map[string]any{"status": "shipped"},
    pushflo.WithEventType("order.shipped"))

// encrypted publish (e.g. an executor returning a result to the cloud)
pub.PublishSealed(ctx, "results-executor-1", resultSealer, result, envelope.Meta{})

The publisher retries retryable failures (network errors, 429, 5xx) with jittered backoff; 4xx is returned immediately, and 401/403 become an AuthenticationError. API error bodies ({"error", "code"}) are surfaced in the error message, and the PublishResult carries the full server response (id, channel, event type, publisher client id, delivered count, timestamp).

Privileged channel/project operations use a mgmt_ key against the Console API (console.pushflo.dev) and live in a separate Management type (management.go), deliberately kept out of the publish path. It is currently a stub returning ErrNotImplemented: the method set and privilege boundary exist, but the REST calls are TODO until something needs to drive them programmatically.

Reference executor

examples/executor is the v1 agent in miniature: subscribe → decrypt job → run the HTTP request(s) locally with full timing capture (DNS/connect/TLS/TTFB via httptrace), redirects not auto-followed, per-request TLS-verify and timeout controls, structured error classification (dns/connect/tls/ timeout/http), and a size-capped base64 body. Egress to loopback, private, and link-local addresses is blocked by default (checked after DNS resolution, so rebinding can't bypass it); set EXECUTOR_ALLOW_PRIVATE=1 to probe internal targets deliberately.

go run ./examples/keygen                 # print key material
PUSHFLO_PUBLISH_KEY=pub_... \
CONTROL_CHANNEL=executor-1 \
RECIPIENT_PUBLIC=... RECIPIENT_PRIVATE=... SIGNER_PUBLIC=... \
  go run ./examples/executor

Wire protocol

The protocol lives isolated in protocol.go (WebSocket) and the top of publisher.go (REST) and is confirmed against the official @pushflodev/sdk v1.1.0 source:

  • Auth — the API key travels as a ?token= query parameter on the WebSocket upgrade URL, and as Authorization: Bearer for REST. Note that URLs can end up in proxy/server logs; pub_ keys are low-privilege and revocable, but rotate them if a log leak is suspected.
  • Connect — the client only counts as connected after the server's connected frame (which carries the client id), not on socket open.
  • Heartbeat — an app-level {"type":"ping"} every interval, answered by {"type":"pong"}. A connection silent for 2x the interval is treated as dead and torn down (triggering auto-reconnect), so half-open TCP connections are detected promptly.
Integration tests

Unit tests run hermetically. To additionally verify against the live service:

PUSHFLO_PUBLISH_KEY=pub_... PUSHFLO_SECRET_KEY=sec_... \
  go test -run Integration -v .

They cover connect/auth, invalid-key rejection, a publish→subscribe round trip, and a sealed end-to-end envelope through the real relay (PUSHFLO_TEST_CHANNEL overrides the default go-sdk-integration channel).

Security

See SECURITY.md for the vulnerability disclosure policy, the envelope threat model, and key-handling guidance.

License

MIT — matching the upstream SDK.

Documentation

Overview

Package pushflo is a Go client SDK for the PushFlo real-time messaging service (https://pushflo.dev), mirroring the ergonomics of the official TypeScript SDK.

It provides:

  • Client: a WebSocket subscriber with auto-reconnect, heartbeat, and per-channel subscriptions (publish keys, "pub_...").
  • Publisher: REST publishing with retry and jittered backoff (secret keys, "sec_..."), including PublishSealed for end-to-end encrypted payloads.
  • Management: the privileged API surface (management keys, "mgmt_..."), kept separate from the publish path.

The subpackage envelope adds a transport-agnostic end-to-end encryption layer (X25519 sealed box + Ed25519 signature) so the relay only ever carries ciphertext it cannot read, forge, or replay.

The WebSocket wire frames and REST endpoint are reconstructed from the public SDK's behavior and are isolated in protocol.go and publisher.go; confirm them against the PushFlo server before production use.

Index

Constants

View Source 
const DefaultBaseURL = "https://api.pushflo.dev"

DefaultBaseURL is the PushFlo API base. The WS endpoint is derived from it.

View Source 
const DefaultConsoleURL = "https://console.pushflo.dev"

DefaultConsoleURL is the PushFlo Console API base, used for channel and project management (the realtime API at DefaultBaseURL only handles publish/subscribe and message history).

Variables

View Source 
var ErrNotImplemented = errors.New("pushflo: management API not implemented yet")

ErrNotImplemented is returned by the management surface until it is built out.

Functions

func IsValidChannelSlug 

func IsValidChannelSlug(s string) bool

IsValidChannelSlug reports whether s is a valid channel slug.

func ToChannelSlug 

func ToChannelSlug(s string) string

ToChannelSlug converts an arbitrary string into a valid slug: lowercased, invalid characters replaced with hyphens, consecutive hyphens collapsed, leading/trailing hyphens trimmed, and truncated to 128 characters.

Types

type AuthenticationError 

type AuthenticationError struct {
	// contains filtered or unexported fields
}

AuthenticationError signals an invalid or missing API key. Not retryable.

type Channel 

type Channel struct {
	Name         string `json:"name"`
	Slug         string `json:"slug"`
	Description  string `json:"description,omitempty"`
	IsPrivate    bool   `json:"isPrivate"`
	MessageCount int    `json:"messageCount"`
}

type ChannelInput 

type ChannelInput struct {
	Name        string `json:"name"`
	Slug        string `json:"slug"`
	Description string `json:"description,omitempty"`
	IsPrivate   bool   `json:"isPrivate,omitempty"`
}

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client is a PushFlo subscriber, mirroring PushFloClient from the TS SDK. It is safe for concurrent use.

func NewClient 

func NewClient(ctx context.Context, opts ClientOptions) (*Client, error)

NewClient builds a Client. If AutoReconnect is desired it should be set on opts; NewClient enables it by default. When opts.AutoConnect is true it dials using ctx before returning.

func (*Client) Connect 

func (c *Client) Connect(ctx context.Context) error

Connect establishes the connection. It returns after the server's "connected" frame has been received (mirroring the official SDK, which does not treat socket-open as connected), or with an error on failure/timeout. Subsequent drops are handled by the reconnect loop when AutoReconnect is on.

func (*Client) Destroy 

func (c *Client) Destroy()

Destroy tears down the client and clears all subscriptions and handlers.

func (*Client) Disconnect 

func (c *Client) Disconnect() error

Disconnect closes the connection and suppresses auto-reconnect. The client can be reused by calling Connect again.

func (*Client) IsConnected 

func (c *Client) IsConnected() bool

IsConnected reports whether the client is currently connected.

func (*Client) OnConnected 

func (c *Client) OnConnected(fn func(clientID string))

OnConnected registers a handler fired when a connection is established. The server-assigned client ID is passed when available.

func (*Client) OnConnectionChange 

func (c *Client) OnConnectionChange(fn func(ConnectionState))

OnConnectionChange registers a callback fired on every state transition.

func (*Client) OnDisconnected 

func (c *Client) OnDisconnected(fn func(reason string))

OnDisconnected registers a handler fired when the connection drops.

func (*Client) OnError 

func (c *Client) OnError(fn func(error))

OnError registers a handler for runtime errors.

func (*Client) OnMessage 

func (c *Client) OnMessage(fn func(Message))

OnMessage registers a global handler for every message on any channel.

func (*Client) State 

func (c *Client) State() ConnectionState

State returns the current connection state.

func (*Client) Subscribe 

func (c *Client) Subscribe(channel string, h SubscriptionHandlers) (*Subscription, error)

Subscribe subscribes to a channel. The channel slug is validated locally before any network call. If the client is connected the subscribe frame is sent immediately; otherwise it is (re)sent automatically on the next connect.

type ClientOptions 

type ClientOptions struct {
	PublishKey string // required, "pub_..."

	BaseURL              string        // default DefaultBaseURL
	AutoConnect          bool          // if true, NewClient dials immediately
	Debug                bool          // verbose logging via Logger
	ConnectionTimeout    time.Duration // default 30s
	HeartbeatInterval    time.Duration // default 25s
	AutoReconnect        bool          // default true
	MaxReconnectAttempts int           // 0 = infinite (default)
	ReconnectDelay       time.Duration // initial backoff, default 1s
	MaxReconnectDelay    time.Duration // backoff ceiling, default 30s
	Logger               *log.Logger   // default log.Default()
}

ClientOptions configures a subscriber Client. Only PublishKey is required; zero-valued fields fall back to the defaults noted below.

type ConnectionError 

type ConnectionError struct {
	// contains filtered or unexported fields
}

ConnectionError signals a WebSocket connection problem. Retryable.

type ConnectionState 

type ConnectionState string

ConnectionState mirrors the states exposed by the TypeScript SDK. 

const (
	StateDisconnected ConnectionState = "disconnected"
	StateConnecting   ConnectionState = "connecting"
	StateConnected    ConnectionState = "connected"
	StateError        ConnectionState = "error"
)

type Error 

type Error struct {
	Code      string
	Message   string
	Retryable bool
	// Err is an optional wrapped cause.
	Err error
}

Error is the base error type, mirroring PushFloError in the TS SDK. Use errors.As to inspect the concrete kind.

func (*Error) Error 

func (e *Error) Error() string

func (*Error) Unwrap 

func (e *Error) Unwrap() error

type Management 

type Management struct {
	// contains filtered or unexported fields
}

Management performs privileged channel/project/history operations using a management (mgmt_) key. It is intentionally separate from Publisher so the high-privilege key and its larger API surface stay out of the common publish path — an executor or a publish-only service never links this.

This is a deliberate stub: the method set mirrors the TS SDK server client so the privilege boundary and call sites exist now, but the REST calls are TODO until there is a real need to drive these programmatically.

func NewManagement 

func NewManagement(opts ManagementOptions) (*Management, error)

NewManagement builds a Management client.

func (*Management) CreateChannel 

func (m *Management) CreateChannel(ctx context.Context, in ChannelInput) (*Channel, error)

func (*Management) CreateProject 

func (m *Management) CreateProject(ctx context.Context, in Project) (*Project, error)

func (*Management) DeleteChannel 

func (m *Management) DeleteChannel(ctx context.Context, slug string) error

func (*Management) GetChannel 

func (m *Management) GetChannel(ctx context.Context, slug string) (*Channel, error)

func (*Management) GetMessageHistory 

func (m *Management) GetMessageHistory(ctx context.Context, channel string, page, pageSize int) ([]Message, *Pagination, error)

func (*Management) ListChannels 

func (m *Management) ListChannels(ctx context.Context, page, pageSize int) ([]Channel, *Pagination, error)

func (*Management) ListProjects 

func (m *Management) ListProjects(ctx context.Context) ([]Project, *Pagination, error)

func (*Management) UpdateChannel 

func (m *Management) UpdateChannel(ctx context.Context, slug string, in ChannelInput) (*Channel, error)

type ManagementOptions 

type ManagementOptions struct {
	ManagementKey string // required, "mgmt_..."
	// BaseURL defaults to DefaultConsoleURL: channel/project CRUD lives on
	// the Console API, not the realtime API (confirmed against the official
	// SDK's PushFloServer).
	BaseURL string
	Timeout time.Duration
	Debug   bool
}

ManagementOptions configures a Management client.

type Message 

type Message struct {
	ID        string          `json:"id"`
	Channel   string          `json:"channel"`
	EventType string          `json:"eventType"`
	ClientID  string          `json:"clientId"`
	Content   json.RawMessage `json:"content"`
	Timestamp time.Time       `json:"timestamp"`
}

Message is a message received on a channel.

Content is kept as json.RawMessage so callers decide how to decode it — this is also the field the envelope layer treats as the encrypted blob when you wrap payloads (see the envelope package).

func (Message) Unmarshal 

func (m Message) Unmarshal(v any) error

Unmarshal decodes the message Content into v.

type NetworkError 

type NetworkError struct {
	// contains filtered or unexported fields
}

NetworkError signals an HTTP/transport failure. Retryability varies.

type Pagination 

type Pagination struct {
	Page       int `json:"page"`
	PageSize   int `json:"pageSize"`
	Total      int `json:"total"`
	TotalPages int `json:"totalPages"`
}

type Project 

type Project struct {
	Name            string `json:"name"`
	Slug            string `json:"slug"`
	Description     string `json:"description,omitempty"`
	ChannelCount    int    `json:"channelCount"`
	CredentialCount int    `json:"credentialCount"`
}

type PublishOption 

type PublishOption func(*publishConfig)

PublishOption customizes a single publish call.

func WithEventType 

func WithEventType(t string) PublishOption

WithEventType sets the message event type.

type PublishResult 

type PublishResult struct {
	ID          string `json:"id"`
	ChannelSlug string `json:"channelSlug"`
	EventType   string `json:"eventType"`
	ClientID    string `json:"clientId"`
	// Delivered is the number of subscribers the message reached.
	Delivered int `json:"delivered"`
	// CreatedAt is an ISO 8601 timestamp of when the message was published.
	CreatedAt string `json:"createdAt"`
}

PublishResult is returned on a successful publish.

type Publisher 

type Publisher struct {
	// contains filtered or unexported fields
}

Publisher publishes messages to channels using a secret (sec_) key.

func NewPublisher 

func NewPublisher(opts PublisherOptions) (*Publisher, error)

NewPublisher builds a Publisher.

func (*Publisher) Publish 

func (p *Publisher) Publish(ctx context.Context, channel string, content any, opts ...PublishOption) (*PublishResult, error)

Publish sends content (marshaled to JSON) to a channel.

func (*Publisher) PublishRaw 

func (p *Publisher) PublishRaw(ctx context.Context, channel string, content json.RawMessage, opts ...PublishOption) (*PublishResult, error)

PublishRaw publishes a pre-marshaled JSON content payload.

func (*Publisher) PublishSealed 

func (p *Publisher) PublishSealed(ctx context.Context, channel string, sealer *envelope.Sealer, v any, meta envelope.Meta, opts ...PublishOption) (*PublishResult, error)

PublishSealed encrypts and signs v with the given Sealer, then publishes the resulting envelope as the message content. This is the symmetric counterpart to envelope.Opener on the receiving side — the relay only sees ciphertext.

type PublisherOptions 

type PublisherOptions struct {
	SecretKey     string        // required, "sec_..."
	BaseURL       string        // default DefaultBaseURL
	Timeout       time.Duration // per-request timeout, default 30s
	RetryAttempts int           // additional attempts on retryable errors, default 3
	Debug         bool
	HTTPClient    *http.Client // optional; overrides Timeout if set
}

PublisherOptions configures a Publisher. Only SecretKey is required.

type SlugValidation 

type SlugValidation struct {
	Valid      bool
	Error      string
	Suggestion string
}

SlugValidation is the detailed result of ValidateChannelSlug.

func ValidateChannelSlug 

func ValidateChannelSlug(s string) SlugValidation

ValidateChannelSlug validates s and, when invalid, returns a reason and a best-effort corrected suggestion.

type Subscription 

type Subscription struct {
	// contains filtered or unexported fields
}

Subscription represents an active channel subscription.

func (*Subscription) Channel 

func (s *Subscription) Channel() string

Channel returns the subscribed channel slug.

func (*Subscription) Unsubscribe 

func (s *Subscription) Unsubscribe() error

Unsubscribe cancels the subscription.

type SubscriptionHandlers 

type SubscriptionHandlers struct {
	OnMessage      func(Message)
	OnError        func(error)
	OnSubscribed   func()
	OnUnsubscribed func()
}

SubscriptionHandlers are the per-subscription callbacks.

type ValidationError 

type ValidationError struct {
	// contains filtered or unexported fields
}

ValidationError signals invalid input (e.g. a bad channel slug). Not retryable.

Source Files

View all Source files

Directories

Path Synopsis
Package envelope adds an end-to-end encryption layer on top of any transport (PushFlo channels, HTTP, a message queue, ...).
 Package envelope adds an end-to-end encryption layer on top of any transport (PushFlo channels, HTTP, a message queue, ...).
examples
executor command
Command executor is a reference API-test executor: it subscribes to a control channel, decrypts each pushed job with the envelope layer, runs the HTTP request(s) locally with full timing capture, and prints a structured result.
 Command executor is a reference API-test executor: it subscribes to a control channel, decrypts each pushed job with the envelope layer, runs the HTTP request(s) locally with full timing capture, and prints a structured result.
keygen command
Command keygen generates key material for the encrypted envelope layer.
 Command keygen generates key material for the encrypted envelope layer.
subscribe command
Command subscribe is a minimal PushFlo subscriber.
 Command subscribe is a minimal PushFlo subscriber.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.