Help us shape the future of Go! Take the 2026 Go Developer Survey

jetcast

package module
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Oct 11, 2026 License: MIT Imports: 20 Imported by: 0

README

jetcast

CI Go Reference

中文

jetcast pushes server events to browsers that connect directly to NATS over WebSocket. The server publishes named events to channels; clients subscribe to channels, with application-defined authorization for private channels.

  • Browsers on NATS, authorization enforced by NATS. Clients authenticate through the NATS auth callout with your own tokens; every connection gets a JWT that allows exactly its channels and its own namespace. Clients live in their own NATS account, which limits their connections, subscriptions and payloads.
  • Channel authorization. srv.Channel("orders.{id}", authorize) decides at subscribe time. Stable, broad grants (a user's own feed, a team's channels) can be given at connect time instead, so NATS routes them without touching your servers.
  • Missed events are recovered. Events are retained in a JetStream stream for a few minutes. Clients detect gaps, including a lost last event, and fetch what they missed after a reconnect. When recovery is impossible they tell you, so you reload instead of silently showing stale data.
  • origin suppression, revocation and refresh. Exclude the sender of an action, revoke a user's or session's connections (kicking clients that do not cooperate), and let clients pick up new grants.
  • No assumptions about deployment. The library takes a *nats.Conn. NATS can be embedded in your binary, standalone or clustered; WebSocket can be proxied by your Go server, nginx or a load balancer.

Requires nats-server 2.14.4+ (2.15 recommended) with JetStream, and Go 1.26+.

Version 0.3 changes the Go publication API and TypeScript connection type. Read the migration examples when upgrading from 0.2.

Install

go get github.com/runforyou-ai/jetcast
npm install @runforyou/jetcast

Server

srv, err := jetcast.NewServer(nc, jetcast.ServerOptions{
	Account:       "CLIENT",     // the NATS account the callout places clients in
	CalloutSigner: issuerKey,    // account key pair configured as auth_callout issuer
	Admin:         jetcast.SystemAdmin(sysConn), // optional: kick clients on Disconnect
	ManageStreams: true,         // create the event stream and registry bucket
})

// Who is connecting? Use your session tokens.
srv.Authenticate(func(ctx context.Context, r jetcast.AuthRequest) (jetcast.User, error) {
	sess, err := sessions.Find(ctx, r.Token)
	if err != nil {
		return jetcast.User{}, err
	}
	return jetcast.User{ID: "u" + sess.UserID, Session: sess.ID, ExpiresAt: sess.ExpiresAt,
		Info: map[string]any{"name": sess.UserName}}, nil
})

// Optional: channels a user may subscribe to directly for the whole connection.
srv.Grants(func(ctx context.Context, u jetcast.User) ([]string, error) {
	return []string{"users." + u.ID + ".>"}, nil
})

// Private channels authorized at subscribe time.
srv.Channel("orders.{id}", func(ctx context.Context, u jetcast.User, p jetcast.Params) (bool, error) {
	return orders.CanView(ctx, u.ID, p["id"])
})

err = srv.Start(ctx)
defer srv.Close()

Broadcast from any process that shares the configuration (a Server embeds a Publisher):

pub, _ := jetcast.NewPublisher(nc, jetcast.Config{})
_, err := pub.Broadcast(ctx, jetcast.Event{
	Name:     "order.shipped",
	Channels: []jetcast.Channel{jetcast.Private("orders.42")},
	Data:     order,                  // JSON-encoded
	Origin:   jetcast.SocketID(r),    // origin suppression: the X-Socket-ID of the request
})

Revocation and refresh:

srv.Refresh(ctx, jetcast.ByUser("u42"))                 // picked up new grants
srv.Leave(ctx, jetcast.Private("orders.42"), jetcast.ByUser("u42"))
srv.Disconnect(ctx, jetcast.BySession("u42", sessionID)) // after invalidating the session

Browser

import { connect, UnauthorizedError } from "@runforyou/jetcast"

const client = await connect({
  servers: "wss://example.com/nats",
  getToken: async () => {
    const res = await fetch("/api/realtime-token")
    if (res.status === 401) throw new UnauthorizedError()
    return res.text()
  },
})

client.channel("news").listen("article.published", (article) => render(article))

const orders = client.private("orders.42")
orders.listen("order.shipped", (order) => update(order))
orders.onState(({ state, recovered }) => {
  if (state === "subscribed" && recovered === false) reloadOrder()
})
await orders.ready()

// origin suppression: send the socket ID with your API requests.
fetch("/api/orders/42/ship", { method: "POST", headers: { "X-Socket-ID": client.socketId } })

Go programs such as background agents use client, which has the same API.

How it works
Connect The client opens a WebSocket connection with a fresh random socket ID and your token. NATS asks jetcast through the auth callout; jetcast calls Authenticate and Grants, registers the socket in a JetStream key-value bucket and returns a JWT allowing public channels, granted channels and the socket's own subjects.
Subscribe Public and granted channels are subscribed directly. Other private channels are authorized by your Channel callback; the node that authorized them relays their events to the connection.
Broadcast Events are written to a JetStream stream, which republishes them to channel subjects with their sequence and the sequence of the channel's previous event.
Recover Clients track sequences per channel. A sequence that does not continue the previous one, a reconnect, or a periodic head check triggers recovery from the stream. Recovery is proven complete when nothing after the client's position was evicted; otherwise the client reports recovered: false.

Details: design (Chinese) and deployment.

Delivery semantics

  • Events are delivered at most once per connection and recovered within the retention window (5 minutes by default). Use your own data, not jetcast, as the source of truth: reload when a subscription reports recovered: false.
  • Authorization is checked when a connection is made (grants) and when a channel is subscribed (authorizers), then periodically for relayed channels. A relay that cannot be authorized again within twice ReauthorizeInterval, because the authorizer keeps failing or the node is too busy, is dropped and the client subscribes again. Use Disconnect to cut access immediately. Configure a ConnectionAdmin for forced disconnection. If Disconnect returns an error, the application must retry it to complete enforcement; retries include already-revoked connections. Persist the retry intent when enforcement must survive an application restart.
  • origin suppression suppresses the sender's listeners; it is not a security mechanism. The socket ID is reported by the client and not verified, and events carry only a digest of it, so other subscribers cannot learn socket IDs.

Roadmap

  • Presence channels (here, joining, leaving) and client events (whisper).
  • React hooks.
  • Queued and after-commit broadcasting with jetq.
  • History queries, a mode without auth callout, OpenTelemetry.

License

MIT

Documentation

Overview

Package jetcast broadcasts server events to browsers and other clients that connect directly to NATS. Applications publish explicitly named events and authorize subscriptions to private channels.

Application code broadcasts events on channels; clients listen on channels. Public channels need no authorization. Private channels are authorized by the application: either at connect time, by granting channel patterns that NATS enforces directly, or at subscribe time, by channel authorizers whose approved subscriptions the server relays to the client. Events are retained in a JetStream stream for a short time, so clients detect missed events and recover them, and learn when recovery is impossible.

A Server runs on every application node: it answers the NATS auth callout, handles client requests and relays private channels. A Publisher only broadcasts and suits processes that do not serve clients.

See docs/design.md for the protocol.

Example 
package main

import (
	"context"
	"errors"
	"log"
	"net/http"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nkeys"
	"github.com/runforyou-ai/jetcast"
)

func main() {
	nc, err := nats.Connect("nats://localhost:4222", nats.UserInfo("app", "app"))
	if err != nil {
		log.Fatal(err)
	}
	issuer, _ := nkeys.FromSeed([]byte("SA...")) // the auth_callout issuer's account seed

	srv, err := jetcast.NewServer(nc, jetcast.ServerOptions{
		Account:       "CLIENT", // the account the callout places clients in
		CalloutSigner: issuer,
		ManageStreams: true,
	})
	if err != nil {
		log.Fatal(err)
	}
	srv.Authenticate(func(ctx context.Context, r jetcast.AuthRequest) (jetcast.User, error) {
		if r.Token != "secret-for-u42" {
			return jetcast.User{}, errors.New("unknown token")
		}
		return jetcast.User{ID: "u42"}, nil
	})
	srv.Grants(func(ctx context.Context, u jetcast.User) ([]string, error) {
		return []string{"users." + u.ID + ".>"}, nil
	})
	_ = srv.Channel("orders.{id}", func(ctx context.Context, u jetcast.User, p jetcast.Params) (bool, error) {
		return p["id"] == "42", nil
	})
	if err := srv.Start(context.Background()); err != nil {
		log.Fatal(err)
	}
	defer srv.Close()

	http.HandleFunc("POST /orders/{id}/ship", func(w http.ResponseWriter, r *http.Request) {
		_, err := srv.Broadcast(r.Context(), jetcast.Event{
			Name:     "order.shipped",
			Channels: []jetcast.Channel{jetcast.Private("orders." + r.PathValue("id"))},
			Data:     map[string]string{"id": r.PathValue("id")},
			Origin:   jetcast.SocketID(r),
		})
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
		}
	})
}

Index

Examples

Constants

View Source 
const (
	HeaderEvent   = "Jetcast-Event"
	HeaderID      = "Jetcast-Id"
	HeaderOrigin  = "Jetcast-Origin"
	HeaderChannel = "Jetcast-Channel"
	HeaderSid     = "Jetcast-Sid"
	HeaderStatus  = "Jetcast-Status"

	// Headers added by JetStream RePublish and rebuilt on recovery.
	HeaderSequence     = "Nats-Sequence"
	HeaderLastSequence = "Nats-Last-Sequence"
	HeaderTimeStamp    = "Nats-Time-Stamp"
)

Message headers set by jetcast. Applications cannot set them.

View Source 
const (
	PathDirect = "direct"
	PathRelay  = "relay"
)

Delivery paths of a private channel.

View Source 
const (
	CodeDenied      = "denied"
	CodeInvalid     = "invalid"
	CodeUnavailable = "unavailable"
	CodeOverloaded  = "overloaded"
)

Error codes returned to clients.

View Source 
const (
	ReasonInitial   = "initial"
	ReasonExpired   = "expired"
	ReasonEpoch     = "epoch"
	ReasonTooFar    = "too_far"
	ReasonEphemeral = "ephemeral"
)

Reasons for recovered=false.

View Source 
const (
	CtlRefresh     = "refresh"
	CtlDisconnect  = "disconnect"
	CtlLeave       = "leave"
	CtlDenied      = "denied"
	CtlInterrupted = "interrupted"
)

Control message types sent to a connection.

View Source 
const (
	ConnectionWebSocket = "WEBSOCKET"
	ConnectionStandard  = "STANDARD"
)

Connection types for User.ConnectionTypes.

View Source 
const SocketIDHeader = "X-Socket-ID"

SocketIDHeader is the HTTP header clients use to send their socket ID with application requests.

Variables

View Source 
var (
	// ErrInvalidChannel reports a malformed channel name or pattern.
	ErrInvalidChannel = errors.New("jetcast: invalid channel")
	// ErrInvalidID reports a malformed user, session or socket ID.
	ErrInvalidID = errors.New("jetcast: invalid ID")
	// ErrInvalidConfig reports invalid configuration.
	ErrInvalidConfig = errors.New("jetcast: invalid configuration")
)

Functions

func MatchPattern 

func MatchPattern(pattern, name string) bool

MatchPattern reports whether the literal channel name matches pattern with NATS wildcard semantics: "*" matches one token, a final ">" one or more.

Example 
package main

import (
	"fmt"

	"github.com/runforyou-ai/jetcast"
)

func main() {
	fmt.Println(jetcast.MatchPattern("users.42.>", "users.42.notifications"))
	fmt.Println(jetcast.MatchPattern("teams.*.board", "teams.7.board"))
	fmt.Println(jetcast.MatchPattern("users.42.>", "users.43.notifications"))
}

Output:
true
true
false

func SocketID 

func SocketID(r *http.Request) string

SocketID returns the client's socket ID from an HTTP request, or "" when it is missing or malformed. Pass it as Event.Origin to exclude the sender.

func ValidateChannelName 

func ValidateChannelName(name string) error

ValidateChannelName reports whether name is a literal channel name: one to eight dot-separated tokens of ASCII letters, digits, "-" or "_", at most 200 bytes in total.

func ValidateID 

func ValidateID(id string) error

ValidateID reports whether id is a valid user or session ID: 1 to 64 ASCII letters, digits, "-" or "_".

func ValidatePattern 

func ValidatePattern(pattern string) error

ValidatePattern reports whether pattern is a valid channel pattern: like a channel name, but tokens may be "*" and the last token may be ">".

func ValidateSocketID 

func ValidateSocketID(id string) error

ValidateSocketID reports whether id is a well-formed socket ID: 22 base62 characters.

Types

type AuthRequest 

type AuthRequest struct {
	// Token is the credential the client sent, typically from getToken().
	Token string
	// Type is the connection type: "websocket" for browsers, "nats" for
	// standard clients.
	Type string
	// Host is the client address as seen by the NATS server.
	Host string
}

AuthRequest describes a connection being authenticated.

type AuthenticateFunc 

type AuthenticateFunc func(ctx context.Context, req AuthRequest) (User, error)

AuthenticateFunc authenticates a connection. It returns an error to reject it. It must not have side effects that break when called again for the same token.

type AuthorizeFunc 

type AuthorizeFunc func(ctx context.Context, u User, p Params) (bool, error)

AuthorizeFunc decides whether a user may subscribe to a private channel matching the pattern it was registered with.

type BroadcastResult 

type BroadcastResult struct {
	ID       string
	Channels []ChannelResult
}

BroadcastResult is the outcome of a broadcast. Broadcasting on several channels is not atomic: some channels may fail while others succeed.

func (BroadcastResult) Err 

func (r BroadcastResult) Err() error

Err joins the errors of failed channels.

type Channel 

type Channel struct {
	Kind Kind
	Name string
}

Channel identifies a channel by kind and name.

func ParseChannel 

func ParseChannel(s string) (Channel, error)

ParseChannel parses the "<kind>.<name>" form returned by Channel.String.

func Private 

func Private(name string) Channel

Private returns the private channel with the given name.

func Public 

func Public(name string) Channel

Public returns the public channel with the given name.

func (Channel) String 

func (c Channel) String() string

String returns "<kind>.<name>", the form used in headers and keys.

func (Channel) Validate 

func (c Channel) Validate() error

Validate reports whether c has a valid kind and a literal channel name.

type ChannelResult 

type ChannelResult struct {
	Channel Channel
	// Sequence is the stream sequence of the retained event, 0 for
	// ephemeral channels.
	Sequence uint64
	Err      error
}

ChannelResult is the outcome of broadcasting on one channel.

type Config 

type Config struct {
	// Prefix is the first token of every subject, "jetcast" by default.
	// Applications sharing a NATS account use distinct prefixes.
	Prefix string
	// Stream is the name of the event stream, "JETCAST" by default. The
	// connection registry is the key-value bucket "<Stream>_CONN".
	Stream string
	// Ephemeral lists channel patterns, applying to both kinds, whose events
	// are not retained: they are published directly, carry no sequence and
	// cannot be recovered. Typing indicators are a typical example.
	Ephemeral []string
}

Config is the configuration shared by Server and Publisher. Every process of one application uses the same Config.

type ConnectionAdmin 

type ConnectionAdmin interface {
	// Kick closes the connection with the given client ID on the NATS server
	// with the given server ID.
	Kick(ctx context.Context, serverID string, cid uint64) error
}

ConnectionAdmin forcibly disconnects client connections. Disconnect uses it to close connections of clients that ignore the disconnect request.

func SystemAdmin 

func SystemAdmin(sys *nats.Conn) ConnectionAdmin

SystemAdmin returns a ConnectionAdmin that kicks connections through the NATS system account: sys must be a connection to the system account.

type Control 

type Control struct {
	Type    string `json:"type"`
	Sid     string `json:"sid,omitempty"`
	Channel string `json:"channel,omitempty"`
	Reason  string `json:"reason,omitempty"`
}

Control is a control message sent to a connection.

type DisconnectResult 

type DisconnectResult struct {
	// Connections is the number of registered connections of the target.
	Connections int
	// Revoked is the number of connection records marked revoked.
	Revoked int
	// Kicked is the number of connections kicked through the admin.
	Kicked int
	// Failed counts connections that could not be revoked or kicked.
	Failed int
	// Enforced is false when no ConnectionAdmin is configured: clients were
	// asked to close but not forced.
	Enforced bool
}

DisconnectResult reports what Disconnect did.

type ErrorBody 

type ErrorBody struct {
	Code    string `json:"code"`
	Message string `json:"message,omitempty"`
}

ErrorBody is the error member of a response.

type Event 

type Event struct {
	// ID identifies the event; a random ID is used when empty. Retrying a
	// broadcast with the same ID within the stream's duplicate window does not
	// deliver it twice.
	ID string
	// Name is the required, nonempty event name clients listen for, such as "order.shipped".
	Name string
	// Channels lists the channels the event is broadcast on.
	Channels []Channel
	// Data is the payload: []byte and json.RawMessage are sent as is, other
	// values are encoded as JSON.
	Data any
	// Origin is the socket ID of the client that caused the event. Clients
	// with that socket receive the event without invoking their listeners.
	// Events carry a digest of it, so subscribers do
	// not learn other connections' socket IDs. The socket ID comes from the
	// client, typically the X-Socket-ID header, and is not verified: suppression
	// only spares the sender a redundant update and is not a security
	// mechanism.
	Origin string
}

Event is a broadcast event.

type GrantsFunc 

type GrantsFunc func(ctx context.Context, u User) ([]string, error)

GrantsFunc returns the private channel patterns a user may subscribe to directly for the lifetime of a connection. Every channel matching a pattern must be visible to the user. Revoking a grant requires Disconnect.

type HeadsRequest 

type HeadsRequest struct {
	Epoch    string   `json:"epoch"`
	Channels []string `json:"channels"`
}

HeadsRequest asks for the latest sequences of channels.

type HeadsResponse 

type HeadsResponse struct {
	Error  *ErrorBody        `json:"error,omitempty"`
	Epoch  string            `json:"epoch,omitempty"`
	First  uint64            `json:"first"`
	Last   uint64            `json:"last"`
	Heads  map[string]uint64 `json:"heads,omitempty"`
	Denied []string          `json:"denied,omitempty"`
}

HeadsResponse answers a heads request.

type HelloResponse 

type HelloResponse struct {
	Error     *ErrorBody      `json:"error,omitempty"`
	User      string          `json:"user,omitempty"`
	Info      json.RawMessage `json:"info,omitempty"`
	Grants    []string        `json:"grants,omitempty"`
	ExpiresAt int64           `json:"expiresAt,omitempty"` // Unix milliseconds
	Epoch     string          `json:"epoch,omitempty"`
	MaxAgeMs  int64           `json:"maxAgeMs,omitempty"`
	RenewMs   int64           `json:"renewMs,omitempty"`
	Node      string          `json:"node,omitempty"`
	Prefix    string          `json:"prefix,omitempty"`
	// Origin is the Jetcast-Origin value of events caused by this connection.
	Origin string `json:"origin,omitempty"`
	// MaxRequests is the number of requests a connection may have in flight;
	// clients wait for a slot instead of exceeding it.
	MaxRequests int `json:"maxRequests,omitempty"`
}

HelloResponse answers a hello request.

type History 

type History struct {
	// MaxAge is how long events are retained, 5 minutes by default. Clients
	// that lose events for longer cannot recover them.
	MaxAge time.Duration
	// MaxBytes caps the stream size, 1 GiB by default.
	MaxBytes int64
	// Replicas is the stream replica count, 1 by default.
	Replicas int
	// Storage is the stream storage, file by default.
	Storage jetstream.StorageType
}

History configures event retention.

type Kind 

type Kind string

Kind distinguishes public from private channels. 

const (
	// KindPublic channels can be subscribed by every connection.
	KindPublic Kind = "pub"
	// KindPrivate channels require authorization.
	KindPrivate Kind = "prv"
)

type LeaveRequest 

type LeaveRequest struct {
	Sid string `json:"sid"`
}

LeaveRequest ends a relay.

type Limits 

type Limits struct {
	// RelaysPerConnection caps relayed channels per connection, 100 by default.
	RelaysPerConnection int
	// ConcurrentRequests caps in-flight requests per connection, 8 by
	// default. Clients learn it in hello and wait instead of exceeding it.
	ConcurrentRequests int
	// RelaysPerNode caps relayed subscriptions on the node, 100000 by default.
	RelaysPerNode int
	// RecoverCount and RecoverBytes bound one recovery batch, 100 events and
	// 512 KiB by default; a batch always contains at least one event.
	RecoverCount int
	RecoverBytes int
	// HeadsChannels caps the channels of one heads request, 200 by default.
	HeadsChannels int
	// ConcurrentCallouts caps auth callout requests handled at once on the
	// node, 32 by default. Further requests wait in a queue four times as
	// long; requests that do not fit or wait too long are dropped, and the
	// client retries.
	ConcurrentCallouts int
}

Limits bounds what one connection can use on one node.

type Params 

type Params map[string]string

Params holds the values of the named tokens of a channel pattern.

type Publisher 

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

Publisher broadcasts events. It is safe for concurrent use.

func NewPublisher 

func NewPublisher(nc *nats.Conn, cfg Config) (*Publisher, error)

NewPublisher returns a publisher on nc. The connection must be allowed to publish to the ingress and event subjects of the configured prefix.

func (*Publisher) Broadcast 

func (p *Publisher) Broadcast(ctx context.Context, ev Event) (BroadcastResult, error)

Broadcast publishes ev on each of its channels. Retained channels are written to the stream and Broadcast waits for JetStream to acknowledge them; ephemeral channels are published directly. The returned error joins the errors of failed channels.

type RecoverRequest 

type RecoverRequest struct {
	Channel string `json:"channel"`
	Epoch   string `json:"epoch"`
	Pos     uint64 `json:"pos"`
	UpTo    uint64 `json:"upTo,omitempty"`
}

RecoverRequest asks for the events of a channel after Pos and before UpTo. UpTo of zero means up to the end of the stream.

type RecoverResult 

type RecoverResult struct {
	Error     *ErrorBody `json:"error,omitempty"`
	Recovered bool       `json:"recovered"`
	Reason    string     `json:"reason,omitempty"`
	More      bool       `json:"more,omitempty"`
	Next      uint64     `json:"next"` // position reached by this batch
	Count     int        `json:"count"`
	Epoch     string     `json:"epoch,omitempty"`
	Head      uint64     `json:"head"`     // latest event sequence of the channel
	Position  uint64     `json:"position"` // last sequence of the stream
}

RecoverResult ends a recovery batch. Events precede it on the reply subject.

type RenewRequest 

type RenewRequest struct {
	Sids []string `json:"sids"`
}

RenewRequest renews relay leases on one node.

type RenewResponse 

type RenewResponse struct {
	Error   *ErrorBody `json:"error,omitempty"`
	Missing []string   `json:"missing,omitempty"`
}

RenewResponse lists the sids the node no longer relays.

type Server 

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

Server serves clients on one application node: it answers the NATS auth callout, handles client requests and relays private channels. Run one per node; all nodes share the same Config.

func NewServer 

func NewServer(nc *nats.Conn, opts ServerOptions) (*Server, error)

NewServer creates a server on nc. Register callbacks, then call Start.

func (*Server) Authenticate 

func (s *Server) Authenticate(f AuthenticateFunc)

Authenticate registers the connection authenticator. It is required and must be called before Start; it panics afterwards.

func (*Server) Channel 

func (s *Server) Channel(pattern string, f AuthorizeFunc) error

Channel registers the authorizer of private channels matching pattern. Pattern tokens are literals or "{name}" placeholders matching one token, like "orders.{id}". The first matching pattern decides. Channels must be registered before Start; Channel returns an error afterwards.

func (*Server) Close 

func (s *Server) Close() error

Close stops serving and releases resources after a successful or failed Start. Relayed clients are told to resubscribe elsewhere.

func (*Server) Disconnect 

func (s *Server) Disconnect(ctx context.Context, t Target) (DisconnectResult, error)

Disconnect revokes the target's connections: it marks the target revoked so that authentications in progress fail, marks every connection record revoked so their requests are denied, stops their relays, asks clients to close and kicks them through the ConnectionAdmin. Repeated calls retry enforcement for existing revoked records. Callers must retry errors to complete enforcement; a failure to write the revocation mark is reported after the existing connections were revoked. Invalidate the session in the application first, or clients reconnect with the same credentials.

func (*Server) Grants 

func (s *Server) Grants(f GrantsFunc)

Grants registers the function returning directly subscribable private channel patterns. Optional; it must be called before Start and panics afterwards.

func (*Server) Leave 

func (s *Server) Leave(ctx context.Context, c Channel, t Target) error

Leave removes the target's current subscriptions to a channel: relays stop immediately and directly subscribed clients are told to unsubscribe. It does not revoke authorization; clients may subscribe again while the authorizer allows it.

func (*Server) Node 

func (s *Server) Node() string

Node returns the ID of this server node, random per process.

func (*Server) Refresh 

func (s *Server) Refresh(ctx context.Context, t Target) error

Refresh asks the target's connections to switch to a fresh connection after a random delay, so they pick up newly granted channels. It cannot revoke grants; use Disconnect.

func (*Server) Start 

func (s *Server) Start(ctx context.Context) error

Start ensures the stream and registry, then serves the auth callout and client requests until Close.

func (*Server) Stats 

func (s *Server) Stats() Stats

Stats returns the node's counters.

type ServerOptions 

type ServerOptions struct {
	Config Config
	// Account is the NATS account the auth callout places clients in: in
	// server configuration mode, its name. The application's own connection
	// belongs to the account of the auth_callout block, which holds the event
	// stream and the registry; a separate client account (see
	// embedded.Accounts) exchanges jetcast's subjects with it and limits
	// clients. Clients may also share the application's account.
	Account string
	// CalloutSigner signs the user JWTs and callout responses. It is the
	// account key pair whose public key is the auth_callout issuer.
	CalloutSigner nkeys.KeyPair
	// CalloutXKey decrypts callout requests and encrypts responses when the
	// auth_callout block configures an xkey. Optional.
	CalloutXKey nkeys.KeyPair
	// Admin kicks connections on Disconnect. Optional: without it Disconnect
	// relies on clients to close and on JWT expiry.
	Admin ConnectionAdmin
	// History configures event retention.
	History History
	// ManageStreams creates or updates the event stream and the registry
	// bucket; otherwise Start only validates them.
	ManageStreams bool
	// MaxConnectionTTL caps connection lifetime, 1 hour by default. Clients
	// switch to a fresh connection before it ends.
	MaxConnectionTTL time.Duration
	// RenewInterval is the relay lease renewal period, 20 seconds by default.
	// Relays not renewed for three periods are removed.
	RenewInterval time.Duration
	// ReauthorizeInterval is how often relayed subscriptions are authorized
	// again on renewal, 1 minute by default.
	ReauthorizeInterval time.Duration
	Limits              Limits
	Logger              *slog.Logger
}

ServerOptions configure a Server.

type Stats 

type Stats struct {
	CalloutAccepted uint64
	CalloutRejected uint64
	// CalloutDropped counts callout requests dropped because the queue was
	// full or they waited too long.
	CalloutDropped    uint64
	CalloutLatencyAvg time.Duration
	Relays            int
	RelayedEvents     uint64
	Recoveries        uint64
	RecoveryFailures  uint64
	// Overloaded counts requests answered overloaded because the
	// connection or the node had too many requests in flight. Subscriptions
	// refused for the relay limits are not counted.
	Overloaded uint64
}

Stats are cumulative counters of a server node.

type SubRequest 

type SubRequest struct {
	Channel string `json:"channel"` // "<kind>.<name>"
	Sid     string `json:"sid"`
	Path    string `json:"path,omitempty"` // the client's preference
}

SubRequest subscribes to a channel.

type SubResponse 

type SubResponse struct {
	Error       *ErrorBody `json:"error,omitempty"`
	Path        string     `json:"path,omitempty"`
	Node        string     `json:"node,omitempty"`
	Epoch       string     `json:"epoch,omitempty"`
	Recoverable bool       `json:"recoverable"`
	Head        uint64     `json:"head"`     // latest event sequence of the channel, 0 if none
	Position    uint64     `json:"position"` // last sequence of the stream
}

SubResponse answers a sub request.

type Target 

type Target struct {
	User    string
	Session string
}

Target selects the connections of a user or a session.

func BySession 

func BySession(user, session string) Target

BySession targets the connections of one session of a user.

func ByUser 

func ByUser(user string) Target

ByUser targets every connection of a user.

type User 

type User struct {
	// ID identifies the user: 1 to 64 ASCII letters, digits, "-" or "_".
	// Prefix it to separate kinds of principals, such as "u_42" and "v_9f3a".
	ID string
	// Session identifies the login session, with the same syntax as ID.
	// Disconnect can revoke one session. Defaults to ID.
	Session string
	// ExpiresAt bounds the connection lifetime; zero means the server's
	// MaxConnectionTTL.
	ExpiresAt time.Time
	// Info is public data about the user, returned to the client by hello.
	// GrantsFunc receives it as returned by Authenticate; AuthorizeFunc
	// receives its JSON encoding as a json.RawMessage.
	Info any
	// ConnectionTypes lists the allowed connection types; WebSocket only by
	// default.
	ConnectionTypes []string
}

User is an authenticated principal.

Source Files

View all Source files

Directories

Path Synopsis
Package client is the Go client of jetcast, for executors, command-line tools and other non-browser clients.
 Package client is the Go client of jetcast, for executors, command-line tools and other non-browser clients.
cmd
jetcast-dev command
Command jetcast-dev runs an embedded NATS server and a jetcast server for development and for the TypeScript SDK's tests.
 Command jetcast-dev runs an embedded NATS server and a jetcast server for development and for the TypeScript SDK's tests.
Package embedded adapts jetcast to a NATS server embedded in the same process: it renders the accounts jetcast needs and disconnects clients of the embedded server.
 Package embedded adapts jetcast to a NATS server embedded in the same process: it renders the accounts jetcast needs and disconnects clients of the embedded server.
examples
embedded command
Command embedded is a single-binary application that embeds NATS, serves browsers on one port and reverse proxies /nats to the embedded WebSocket listener.
 Command embedded is a single-binary application that embeds NATS, serves browsers on one port and reverse proxies /nats to the embedded WebSocket listener.
internal
devenv
Package devenv starts an embedded NATS server configured for jetcast: JetStream, WebSocket, a system account and an auth callout.
 Package devenv starts an embedded NATS server configured for jetcast: JetStream, WebSocket, a system account and an auth callout.

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.