livewire

README

livewire (Go)

Live query synchronisation: a client subscribes to a question, receives its answer, then every answer after it.

go get github.com/softwarity/livewire/go

There is no registry to publish to — the import path is the repository, and a version is a git tag. go/v0.1.0 is what this module is fetched under; the npm packages carry v0.1.0 for the same release.

A server

registry := livewire.NewRegistry(0) // 0 = default coalescing window
registry.Register("messages", &MessagesSource{db: db})

mux.Handle("/my-service/ws", livewire.NewServer(registry, livewire.Options{
    Authorize: func(r *http.Request) bool { return rolesOf(r).Any(known) },
    Refusal:   func(r *http.Request) string { return "no role reached this service" },
}))

Registration is explicit. Go has no annotations to discover, and a list of what a server serves is worth reading anyway.

A source

type MessagesSource struct{ db *sql.DB; changes chan struct{} }

func (s *MessagesSource) ReadQuery(raw json.RawMessage) (any, error) {
    asked := map[string]any{}
    _ = json.Unmarshal(raw, &asked)
    return query{
        search: livewire.Text(asked, "search"),
        offset: livewire.Whole(asked, "offset", 0),
        limit:  livewire.LimitOf(asked, "limit", 50),
    }, nil
}

func (s *MessagesSource) Key(q any) string { … }        // two identical questions
func (s *MessagesSource) Wake() <-chan struct{} { … }   // what makes it read again
func (s *MessagesSource) Read(ctx context.Context, q any) (livewire.Window, error) { … }

What the registry does for you

• One read per question. Ten clients asking the same thing share one query.
• Silence on an unchanged read. A busy feed does not repaint a screen it did not move.
• Bursts gathered. A salvo of wakes becomes one read.
• The diff, per subscription. A client that joins mid-stream gets a snapshot, and its patches are computed against the rows it actually holds.
• Cleanup. The last watcher leaves, the read stops and the entry is dropped.

Commands and notifications

Level 2, and optional.

registry.Handle("flight.acknowledge", func(ctx context.Context, payload json.RawMessage) (any, error) {
    asked := struct{ ID string }{}
    _ = json.Unmarshal(payload, &asked)
    return nil, flights.Acknowledge(ctx, asked.ID)
})

// Something that happened, told once, outside any window.
server.Notify(ctx, "import.finished", map[string]any{"count": 412})

Returning an error refuses the command, and its message becomes the reason the client is given. Every command is answered exactly once, whatever happens - including one naming something nothing handles.

What a command changed does not go in its answer. A list it touched is republished by its own subscription. Two answers to one question is what this protocol exists to avoid.

The one rule to remember

UpdatedAt is the version of a row, and everything the row shows has to be in it — not only what a write touched. A value read from the clock changes with no write behind it, and a version that ignores it makes the server believe the row unchanged: nothing is published and the client keeps a value that stopped being true.

See SPEC.md for the contract in full. Where this implementation and the specification disagree, the specification is right.

Documentation

Overview

Package livewire synchronises the result of a query over one WebSocket: a client subscribes to a question, receives its answer, then every answer after it.

This is the Go implementation of the contract in packages/protocol/SPEC.md. That document is normative — where this code and the specification disagree, the specification is right and this is a bug.

Index

• Constants
• func LimitOf(raw map[string]any, field string, fallback int) int
• func Text(raw map[string]any, field string) string
• func Whole(raw map[string]any, field string, fallback int) int
• type Command
• type Envelope
• type Options
• type Registry
  • func NewRegistry(coalesce time.Duration) *Registry
  • func (r *Registry) Command(name string) Command
  • func (r *Registry) Find(topic string) Source
  • func (r *Registry) Handle(name string, command Command)
  • func (r *Registry) Register(topic string, source Source)
  • func (r *Registry) Topics() []string
  • func (r *Registry) Watch(topic string, source Source, query any) (<-chan Window, func())
• type Row
  • func (r Row) MarshalJSON() ([]byte, error)
• type Server
  • func NewServer(registry *Registry, options Options) *Server
  • func (s *Server) Notify(ctx context.Context, topic string, payload any)
  • func (s *Server) ServeHTTP(writer http.ResponseWriter, request *http.Request)
• type Source
• type Window

Constants

const (
	SubscribeEvent   = "subscribe"
	UnsubscribeEvent = "unsubscribe"
	UpdateEvent      = "update"

	// Level 2 — SPEC §6. A server may implement neither, either or both.
	CommandEvent = "command"
	AckEvent     = "ack"
	NotifyEvent  = "notify"
)

Frame names are the vocabulary of the protocol. There is no other.

const CoalesceDefault = 300 * time.Millisecond

CoalesceDefault is how long a burst of changes gathers before a window is read again.

A feed that fires several times a second would otherwise spend itself re-running the same query. Long enough to turn a salvo into one read, short enough that nobody notices the wait.

const MaxLimit = 200

MaxLimit is the widest window a client may ask for.

Wider than a screen, narrower than a scan. The real ceiling is the wire, and it is not this: some proxies silently drop frames over ~64 kB, so what fits depends on the size of a row and is the source's business.

const NotAuthorised = 1008

NotAuthorised is RFC 6455 policy violation: the socket opened, the caller may not use it.

Functions

func LimitOf

func LimitOf(raw map[string]any, field string, fallback int) int

LimitOf keeps whatever was asked for inside what this server will send.

func Text

func Text(raw map[string]any, field string) string

Text answers a non-empty trimmed string, or "".

func Whole

func Whole(raw map[string]any, field string, fallback int) int

Whole answers a whole number, at least zero, or the fallback.

Types

type Command

type Command func(ctx context.Context, payload json.RawMessage) (any, error)

Registry holds the sources this server publishes, and the reads they share.

Explicit registration rather than discovery: Go has no annotations, and a list of what a server serves is worth reading anyway. Command is something a client can ask the server to do — SPEC §6.1.

Answer what the caller should get back, or nil. Return an error to refuse: its message becomes the reason the client is given, rather than silence.

What the command changed is not returned here. It reaches the screens through whatever subscriptions were watching it, on their own schedule.

type Envelope

type Envelope struct {
	Event string          `json:"event"`
	Data  json.RawMessage `json:"data"`
}

Envelope is how every frame travels, both ways.

type Options

type Options struct {
	// Authorize answers whether this caller may use the socket at all.
	//
	// The only place the library touches your application's idea of identity.
	// Nil accepts every upgrade, which is right behind a gateway that has
	// already authenticated and wrong on the open internet.
	Authorize func(request *http.Request) bool

	// Refusal is what to say before closing a socket that was refused. Said on
	// the socket and not only in a close code: a refusal arriving as a bare
	// disconnection is indistinguishable from a network fault.
	Refusal func(request *http.Request) string

	// Origins allowed to open a socket. Empty means same-origin only.
	Origins []string

	// Logger. Nil uses the default.
	Logger *slog.Logger
}

Options is how a server is configured.

type Registry

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

func NewRegistry

func NewRegistry(coalesce time.Duration) *Registry

NewRegistry builds an empty registry. `coalesce` is how long a burst gathers before a read; zero means CoalesceDefault.

func (*Registry) Command

func (r *Registry) Command(name string) Command

Command answers the handler behind a name, or nil — the caller says so on the socket rather than staying quiet.

func (*Registry) Find

func (r *Registry) Find(topic string) Source

Find answers the source behind a topic, or nil.

func (*Registry) Handle

func (r *Registry) Handle(name string, command Command)

Handle adds something the server can be asked to do. Registering the same name twice replaces it.

func (*Registry) Register

func (r *Registry) Register(topic string, source Source)

Register adds a source under a topic. Registering twice replaces.

func (*Registry) Topics

func (r *Registry) Topics() []string

Topics answers what this server publishes, in no particular order.

func (*Registry) Watch

func (r *Registry) Watch(topic string, source Source, query any) (<-chan Window, func())

Watch answers a channel of windows, and a function to stop watching.

Two callers asking the same question share one read: the second is handed what the window already holds, and no query is run. The read stops and the entry is dropped when the last of them leaves — otherwise the map is a leak the size of every filter ever typed.

type Row

type Row struct {
	ID string `json:"id"`

	// UpdatedAt is the version of this row. It changes whenever anything the
	// row shows changes.
	//
	// Not necessarily a timestamp: a filter entry whose only content is its
	// label uses the label, and a row carrying a value derived from the clock
	// has to fold that value in — otherwise the server believes the row
	// unchanged and never sends it again. See SPEC.md, "Versions".
	UpdatedAt string `json:"updatedAt"`

	// Data is what the row actually shows. Marshalled flat beside ID and
	// UpdatedAt, so the wire carries one object per row rather than a nested
	// one — see MarshalJSON.
	Data map[string]any `json:"-"`
}

Row is what every row a source publishes must carry.

func (Row) MarshalJSON

func (r Row) MarshalJSON() ([]byte, error)

MarshalJSON writes id, updatedAt and the row's own fields as one object.

type Server

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

Server is one endpoint: every subscription of every client passes through it.

It implements http.Handler, so it is mounted wherever the application wants:

mux.Handle("/my-service/ws", livewire.NewServer(registry, livewire.Options{...}))

func NewServer

func NewServer(registry *Registry, options Options) *Server

func (*Server) Notify

func (s *Server) Notify(ctx context.Context, topic string, payload any)

Notify tells every open connection that something happened — SPEC §6.2.

An event, not a window: nothing here is applied to a list, and a client that does not know the topic ignores it. Who receives one is the server's business, which here means everybody it is talking to.

func (*Server) ServeHTTP

func (s *Server) ServeHTTP(writer http.ResponseWriter, request *http.Request)

type Source

type Source interface {
	// ReadQuery is the trust boundary: what arrives is JSON off a socket.
	// Clamp it, whitelist it, default it, and hand back something Read can act
	// on without checking again.
	ReadQuery(raw json.RawMessage) (any, error)

	// Key answers what two identical questions share. Two queries with the
	// same key share one read.
	Key(query any) string

	// Wake fires whenever this source may have something new to say. Only the
	// fact of a send is read, never its value.
	//
	// It must not be closed while the source is in use; a source with nothing
	// to follow returns a channel that never sends.
	Wake() <-chan struct{}

	// Read is the window as it stands.
	Read(ctx context.Context, query any) (Window, error)
}

Source is one live list.

Read returns the whole window, never a delta. Turning it into a patch is the server's business, per subscription, because only it knows what that client actually received.

type Window

type Window struct {
	Rows []Row

	// Total is the length the window is a page of. Nil when the source does
	// not page.
	Total *int

	// Pivot is an index in the whole list the source points the client at.
	//
	// A number and nothing more — neither side interprets it. A departure
	// board uses it for the boundary between what has left and what has not: a
	// position in the list that a client holding one page of six hundred
	// cannot work out from the rows it happens to have.
	Pivot *int
}

Window is what a source answers with: a window of rows, and what it is a window of.