channels

type Channel ¶

type Channel struct {
	Name       Name
	State      State
	TTL        time.Duration
	CreatedAt  time.Time
	LastActive time.Time
	// DeltaEnabled toggles centrifuge fossil-delta encoding for this
	// channel. Off by default; enable for high-frequency channels with
	// small mutations (scoreboards, tickers, partial state snapshots).
	DeltaEnabled bool
}

Channel is the persisted view of a managed channel. Token material is never stored here — the auth package owns secrets; the manager owns channel lifecycle.

type Event ¶

type Event struct {
	Kind EventKind
	Name Name
	At   time.Time
}

Event is the payload emitted on every lifecycle transition.

type EventBus ¶

type EventBus interface {
	// Publish broadcasts ev to remote nodes. The caller will already have
	// fanned it out locally; the bus must NOT re-emit on the originating
	// node.
	Publish(ctx context.Context, ev Event) error

	// Run subscribes to the bus. onRemote fires for every remote event
	// (own publishes are filtered by NodeID). Blocks until ctx is done or
	// the underlying transport errors fatally.
	Run(ctx context.Context, onRemote func(Event)) error
}

EventBus is the cross-node bridge for channel lifecycle events. A publish on node A is delivered to every other node's local Manager so subscribers everywhere see the same lifecycle transitions.

Implementations are responsible for tagging publications with the originating NodeID and skipping local re-delivery — the Manager fans out local events itself; the bus carries only the cross-node fan-in.

type EventKind ¶

type EventKind string

EventKind is the lifecycle transition a Channel just made.

const (
	// EventOpened fires on first open or re-open of a public channel.
	EventOpened EventKind = "opened"
	// EventClosed fires when a public channel transitions to StateClosed via
	// Sweep (TTL exceeded). Public-close semantics: no new subscribers, but
	// existing subscribers are not kicked.
	EventClosed EventKind = "closed"
	// EventDeleted fires when a channel is removed — either by an explicit
	// Delete call or because a private channel expired and was reaped from
	// the manager. Subscribers MUST be kicked.
	EventDeleted EventKind = "deleted"
)

type Manager ¶

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

Manager owns channel lifecycle, TTL enforcement, and the inactivity tick. State persistence is delegated to a Store; event fanout to subscribers stays in process. Safe for concurrent use.

func NewManager() *Manager

func NewManagerWithStore(s Store) *Manager

func (m *Manager) CreatePrivate(name Name, ttl time.Duration) (*Channel, error)

func (m *Manager) Delete(name Name) error

func (m *Manager) EventBus() EventBus

func (m *Manager) Get(name Name) (*Channel, error)

func (m *Manager) IsDelta(name Name) bool

func (m *Manager) IsOpen(name Name) bool

func (m *Manager) List() []Channel

func (m *Manager) OpenPublic(name Name, ttl time.Duration) (*Channel, error)

func (m *Manager) RunEventBus(ctx context.Context) error

func (m *Manager) RunSweeper(ctx context.Context, every time.Duration)

func (m *Manager) SetClock(c func() time.Time)

func (m *Manager) SetDelta(name Name, enabled bool) error

func (m *Manager) SetEventBus(b EventBus)

func (m *Manager) SetOnSweep(cb func([]Channel))

func (m *Manager) Subscribe(ch chan<- Event)

func (m *Manager) Sweep(now time.Time) int

func (m *Manager) Touch(name Name) error

type MemoryStore ¶

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

MemoryStore is the in-process Store. Single-node Parsec deployments use it by default; multi-node deployments swap it for a RedisStore.

func NewMemoryStore() *MemoryStore

func (s *MemoryStore) CreatePrivate(name Name, ttl time.Duration, now time.Time) (*Channel, Event, error)

func (s *MemoryStore) Delete(name Name, now time.Time) (Event, error)

func (s *MemoryStore) Get(name Name) (*Channel, error)

func (s *MemoryStore) IsOpen(name Name) bool

func (s *MemoryStore) List() ([]Channel, error)

func (s *MemoryStore) OpenPublic(name Name, ttl time.Duration, now time.Time) (*Channel, Event, error)

func (s *MemoryStore) SetDelta(name Name, enabled bool) error

func (s *MemoryStore) Sweep(now time.Time) ([]Event, error)

func (s *MemoryStore) Touch(name Name, now time.Time) error

type Name ¶

type Name struct {
	Visibility Visibility
	App        string
	Domain     string
	ID         string // optional for public, required for private
	Topic      string // optional
}

Name is a parsed, validated channel name. The zero value is invalid; build one with ParseName or BuildName.

func BuildName(vis Visibility, app, domain, id, topic string) (Name, error)

func ParseName(s string) (Name, error)

func (n Name) IsPrivate() bool

func (n Name) String() string

type Pattern ¶

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

Pattern is a parsed channel-name pattern. It carries a fixed visibility (cross-visibility patterns are rejected at parse time) and a slice of segments with optional `*` and trailing `**` wildcards.

The zero value is invalid; build one with ParsePattern.

func ParsePattern(s string) (Pattern, error)

func (p Pattern) IsValid() bool

func (p Pattern) Matches(n Name) bool

func (p Pattern) String() string

func (p Pattern) Visibility() Visibility

type RedisEventBus ¶

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

RedisEventBus fans channel lifecycle events across Parsec nodes via Redis pub/sub. Every locally emitted event is published with the originating NodeID; remote subscribers re-emit it locally, skipping echoes of their own publishes.

func NewRedisEventBus(client redis.UniversalClient, nodeID string) *RedisEventBus

func (b *RedisEventBus) NodeID() string

func (b *RedisEventBus) Publish(ctx context.Context, ev Event) error

func (b *RedisEventBus) Run(ctx context.Context, onRemote func(Event)) error

func (b *RedisEventBus) WithKeyPrefix(p string) *RedisEventBus

type RedisStore ¶

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

RedisStore persists channel records in a Redis hash so that every Parsec node sharing the same Redis sees the same channel registry. Sweep transitions run inside a Lua script for atomicity.

Storage layout:

parsec:channels        — HASH, field = channel name, value = JSON record

The store is safe for concurrent use across processes; Redis is the single source of truth.

func NewRedisStore(client redis.UniversalClient) *RedisStore

func (s *RedisStore) CreatePrivate(name Name, ttl time.Duration, now time.Time) (*Channel, Event, error)

func (s *RedisStore) Delete(name Name, now time.Time) (Event, error)

func (s *RedisStore) Get(name Name) (*Channel, error)

func (s *RedisStore) IsOpen(name Name) bool

func (s *RedisStore) List() ([]Channel, error)

func (s *RedisStore) OpenPublic(name Name, ttl time.Duration, now time.Time) (*Channel, Event, error)

func (s *RedisStore) SetDelta(name Name, enabled bool) error

func (s *RedisStore) Sweep(now time.Time) ([]Event, error)

func (s *RedisStore) Touch(name Name, now time.Time) error

func (s *RedisStore) WithKeyPrefix(p string) *RedisStore

type State ¶

type State string

State describes the lifecycle position of a managed channel.

const (
	StateOpen    State = "open"
	StateClosed  State = "closed" // public only — inactive past TTL, can reopen
	StateDeleted State = "deleted"
)

type Store ¶

type Store interface {
	// Get returns a snapshot. Deleted channels are reported as not-found
	// so callers do not need to redo the state check.
	Get(name Name) (*Channel, error)

	// List returns a snapshot of all non-deleted channel records.
	List() ([]Channel, error)

	// IsOpen reports whether the channel exists and is in StateOpen.
	IsOpen(name Name) bool

	// OpenPublic creates a new public channel, or re-opens a closed one.
	// Returns the channel snapshot and the event the caller should emit
	// (zero-value Event{} means no transition happened — e.g. already-open
	// channel had its TTL refreshed but no state change).
	OpenPublic(name Name, ttl time.Duration, now time.Time) (*Channel, Event, error)

	// CreatePrivate registers a new private channel. Fails if one already
	// exists with the same name.
	CreatePrivate(name Name, ttl time.Duration, now time.Time) (*Channel, Event, error)

	// Touch refreshes LastActive on the channel.
	Touch(name Name, now time.Time) error

	// Delete marks the channel deleted. The second return value is the
	// event the caller should emit (zero-value Event{} on a redundant
	// delete of an already-deleted channel).
	Delete(name Name, now time.Time) (Event, error)

	// Sweep runs one expiry pass and returns transition events. The Store
	// performs the writes; the Manager fans the events out.
	Sweep(now time.Time) ([]Event, error)

	// SetDelta toggles fossil-delta encoding for the channel. Used by
	// the Manager after open/create to enable the per-channel delta
	// flag without changing the OpenPublic/CreatePrivate signatures.
	SetDelta(name Name, enabled bool) error
}

Store is the persistence interface for channel records. Two implementations ship: an in-memory store (single-node) and a Redis- backed store that lets multiple Parsec nodes share the same channel registry.

Implementations MUST be safe for concurrent use. All mutating methods return the event(s) that the Manager should fan out to subscribers; the Store does NOT do its own emit — that's the Manager's job.

type Visibility ¶

type Visibility string

Visibility is the public/private discriminator that prefixes every channel name. It maps 1:1 to a Centrifuge namespace at broker boot.

const (
	VisibilityPublic  Visibility = "public"
	VisibilityPrivate Visibility = "private"
)

func (v Visibility) String() string