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

realtime

package
v0.56.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: 12 Imported by: 0

Documentation

Overview

Package realtime implements the WebSocket hub behind filex's live collaboration surface: per-folder "rooms" that broadcast (1) file-change events so every viewer's listing updates without a manual refresh and (2) presence so viewers can see who else is in the same folder and which file each is focused on.

The package is transport-agnostic: it knows nothing about net/http or coder/websocket. The HTTP handler (internal/api/handlers/ws.go) owns the socket lifecycle and feeds parsed client intents (subscribe/focus) into the Hub, while the Hub pushes JSON frames back out through each Client's Send channel. That split keeps the Hub unit-testable with fake clients and no network — see hub_test.go.

Index

Constants

View Source 
const (
	VaultGenerationEvent = "vault.generation"
	VaultLockEvent       = "vault.lock"
)

The two frame types.

View Source 
const DefaultChangeLogCapacity = 4096

DefaultChangeLogCapacity is how many changes per storage the log keeps. A cursor older than that reads as "changed", which only costs one walk.

Variables

This section is empty.

Functions

func RoomKey 

func RoomKey(storageID int64, dir string) string

RoomKey is the canonical room identity for a (storage, dir) pair. dir is normalized to the node-table form (leading slash, no trailing slash, "" for the storage root) so that a subscriber's path and a mutation handler's relative dir resolve to the same room regardless of how each was spelled.

Types

type ChangeEvent 

type ChangeEvent struct {
	Action  string `json:"action"`             // create|delete|rename|move|upload|modify
	Name    string `json:"name,omitempty"`     // affected item's basename
	NewName string `json:"new_name,omitempty"` // rename target basename

	// Derived marks an event that reports a recomputed AGGREGATE (folder
	// sizes, internal/sync.SizeRefresher) rather than a change to the folder's
	// entries. Explorers still get it — their size column is what it repaints —
	// but recursive watchers (the desktop sync engine) do not: the real change
	// already reached them, and obeying these would re-list every ancestor
	// folder after every save. Not on the wire; browsers see exactly the frame
	// they always saw.
	Derived bool `json:"-"`
}

ChangeEvent describes a mutation that happened inside a folder. It is the value the mutation handlers hand to Hub.EmitChange; the hub wraps it in a {type:"change", …} frame addressed to the affected room. Frontends treat it as a signal to re-fetch the listing (Action/Name are advisory, for future incremental patching / toasts).

type ChangeLog 

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

ChangeLog remembers which folders of each storage changed recently, so a sync client can ask "has anything under this folder changed since I last looked?" with ONE request instead of listing the whole tree again.

The desktop app re-walked every paired tree every 30 seconds, one listing per folder: a single Mac with 7,048 synced folders sent 100–150 thousand listings an hour, around the clock, and two clients together kept a small server at ~60 requests a second doing nothing but answering "still the same".

It is fed by the same emitter chain that already refreshes open explorers and folder sizes after every change, from every surface (explorer, WebDAV, S3, SFTP, FTPS, NFS, trash and version restores, the ops queue), so it sees what they see. It lives in memory: a restart starts a new epoch, every old cursor then reads as "changed", and each client walks once. Changes the catalogue learns from a storage scan (bytes written straight into the backend) do not pass through here — clients keep a slower full walk as the safety net for those.

func NewChangeLog 

func NewChangeLog(capacity int) *ChangeLog

NewChangeLog returns an empty log with a fresh epoch.

func (*ChangeLog) ChangedSince 

func (l *ChangeLog) ChangedSince(storageID int64, rel, cursor string) (bool, string)

ChangedSince answers whether anything under rel (a folder of storageID) may have changed after cursor, and returns the cursor to ask with next time.

⚠ Every doubt is "changed": no cursor, a cursor from another boot or garbled, a cursor older than the ring still holds, an event on the way to rel that does not say which child it touched. A false "changed" costs one walk; a false "unchanged" leaves a file unsynced until the next full walk.

func (*ChangeLog) ChangedSinceFor 

func (l *ChangeLog) ChangedSinceFor(storageID int64, rel, cursor string, visible func(path string) bool) (bool, string)

ChangedSinceFor is ChangedSince for a caller who may not see everything: a change counts only when visible reports that the caller can see the item it touched (the folder itself when the event names no item). Otherwise a change inside a subfolder somebody has no grant on would still tell them "something happened there". A nil visible sees everything.

func (*ChangeLog) EmitChange 

func (l *ChangeLog) EmitChange(storageID int64, dir string, ev ChangeEvent)

EmitChange records one change. It is the ChangeEmitter method, so the log can also be used as the last link of a chain.

⚠ A change inside filex's own directories is not recorded — the same rule the hub applies to what it announces (internalChange): a version snapshot, a trash move's bookkeeping or the desktop's open-with working copy is nothing a sync client lists, and recording it answered "changed" to every client paired at the storage root, each of which then walked its tree for nothing. Only when EVERY name the event carries is filex's own: a user file renamed onto such a name still reads as changed (a false "changed" costs one walk, a false "unchanged" a missed edit).

func (*ChangeLog) String 

func (l *ChangeLog) String() string

String is for logs.

func (*ChangeLog) Wrap 

func (l *ChangeLog) Wrap(inner Emitter) Emitter

Wrap puts the log in front of inner: every event is recorded, then passed on.

type Client 

type Client struct {
	ID     int64  // process-unique connection id
	UserID int64  // authenticated user id (0 = anonymous, unused today)
	Name   string // display name shown in presence
	Avatar string // profile picture shown in presence ("" = draw initials)
	Send   chan []byte

	// PresenceKey (from the ticket) distinguishes end users behind a shared
	// proxy token; see realtime.Ticket.PresenceKey. Set once before any hub
	// interaction. Empty for native (cookie) sessions.
	PresenceKey string

	// Confinement carried by a ticket-authenticated connection (embedded
	// contexts like work.example.com proxy a root-confined token). When Confined is
	// true the client may only subscribe to rooms within (ConfineAdapter,
	// ConfineRel). Cookie-authenticated connections leave Confined=false and
	// rely on RBAC alone. Set once before any hub interaction.
	Confined       bool
	ConfineAdapter string
	ConfineRel     string

	// Tenant is the provider id of the tenant this connection's account
	// belongs to on a multi-tenant install, 0 for the platform's own accounts
	// and on a single-tenant install. A tenant's `access.changed` news reaches
	// only its own sockets (access.go). Set once before any hub interaction.
	Tenant int64
	// contains filtered or unexported fields
}

Client is one live WebSocket connection's presence in the hub. It carries the identity used for presence plus a buffered Send channel the hub writes outbound JSON frames to; the handler's write pump drains it to the socket.

The room/file/path fields are mutated only under Hub.mu, so a Client must not be shared across hubs. Send is created buffered — the hub uses a non-blocking send so one slow/stalled reader can never wedge a broadcast (a dropped frame degrades to a missed live update, which the next event or a manual refresh corrects).

func NewClient 

func NewClient(userID int64, name string, buffer int) *Client

NewClient builds a Client with a unique id and a buffered send channel. buffer sizes the outbound queue; a small value (e.g. 16) is plenty for presence + change frames.

func (*Client) AllowsPath 

func (c *Client) AllowsPath(adapter, rel string) bool

AllowsPath reports whether this client may subscribe to (adapter, rel). An unconfined client may go anywhere (RBAC still applies); a confined client is restricted to its ticket's root and everything under it.

func (*Client) Identity 

func (c *Client) Identity() string

Identity is the presence identity this client is de-duplicated and self-filtered by: the filex user id, refined by the per-end-user PresenceKey when an embedded host supplied one. Two tabs of the same person share an identity (one roster entry); two work users behind one proxy token do not.

type Emitter 

type Emitter interface {
	EmitChange(storageID int64, dir string, ev ChangeEvent)
}

Emitter is what the log wraps: the hub, or the next link of the chain.

type Hub 

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

Hub is the process-wide registry of rooms and their subscribers. All state is guarded by a single mutex; broadcasts build the frame under the lock and enqueue it with a non-blocking send, so the critical section is short and a stalled client never blocks others.

func NewHub 

func NewHub() *Hub

NewHub returns an empty, ready-to-use Hub.

func (*Hub) AccessChanged added in v0.54.0 

func (h *Hub) AccessChanged(userIDs ...int64)

AccessChanged tells the named accounts that what they may do may have changed. Naming nobody (or only ids <= 0) tells nobody. Never blocks.

func (*Hub) AccessChangedEverywhere added in v0.54.0 

func (h *Hub) AccessChangedEverywhere()

AccessChangedEverywhere tells every open socket: a change at the platform level, or any change on a single-tenant install.

func (*Hub) AccessChangedInTenant added in v0.54.0 

func (h *Hub) AccessChangedInTenant(providerID int64)

AccessChangedInTenant tells every account of one tenant (a provider id > 0). Sockets of other tenants and of the platform's own accounts hear nothing.

func (*Hub) AppUpdated added in v0.48.0 

func (h *Hub) AppUpdated(app, version string)

AppUpdated tells every connected client that app now runs at version. Non-blocking per client, like every other frame: a stalled reader misses it and loads the new version the next time it opens the app.

func (*Hub) Connect added in v0.48.0 

func (h *Hub) Connect(c *Client)

Connect registers c for instance-wide notices for as long as its socket is open. Unsubscribe (on disconnect) forgets it.

func (*Hub) EmitChange 

func (h *Hub) EmitChange(storageID int64, dir string, ev ChangeEvent)

EmitChange tells every client viewing (storageID, dir) that the folder changed. Safe to call for a room with no subscribers (no-op). This is the method the mutation handlers reach through the handlers.ChangeEmitter interface.

The first change in a quiet room is broadcast synchronously, exactly as it always was. Changes arriving on top of a recent one are merged and sent as a single frame when the room's window elapses — see room for why, and for the promise that the last event of a burst is the one that lands.

func (*Hub) EmitVault added in v0.54.0 

func (h *Hub) EmitVault(storageID int64, dir string, ev VaultEvent)

EmitVault sends ev to every client viewing the vault folder (storageID, dir). A room nobody is in costs nothing.

func (*Hub) Presence 

func (h *Hub) Presence(storageID int64, dir string) []PresenceUser

Presence returns the current roster for (storageID, dir). Exposed for tests and diagnostics; the live path broadcasts instead.

func (*Hub) SetFocus 

func (h *Hub) SetFocus(c *Client, file string)

SetFocus updates the file c is focused on within its current room and re-broadcasts presence. file "" clears the focus (e.g. preview closed).

func (*Hub) Subscribe 

func (h *Hub) Subscribe(c *Client, storageID int64, dir, displayPath string)

Subscribe moves c into the room for (storageID, dir), leaving whatever room it was in before. displayPath is the "<adapter>://<dir>" the client asked for; it's echoed in outbound frames. Presence is re-broadcast to both the old and new rooms so everyone (including c) gets the fresh roster.

func (*Hub) Unsubscribe 

func (h *Hub) Unsubscribe(c *Client)

Unsubscribe removes c from its room entirely (called on disconnect) and refreshes presence for the room it left.

func (*Hub) Watch 

func (h *Hub) Watch(c *Client, roots []WatchRoot)

Watch replaces the set of roots c watches. An empty roots slice removes every watch. Nothing is sent from here; the caller acknowledges.

func (*Hub) WatchedRoots added in v0.44.2 

func (h *Hub) WatchedRoots(storageID int64) []string

WatchedRoots lists the folders of storageID that some client watches recursively right now, each once, in normalizeDir form ("" is the storage root, otherwise "/a/b"). A lazily catalogued storage keeps these subtrees current, because a desktop sync pair mirrors every folder under its root whether or not anybody opens it (internal/sync lazy.go).

func (*Hub) Watching 

func (h *Hub) Watching(c *Client) int

Watching reports how many roots c currently watches. For tests and diagnostics.

type OneShot added in v0.56.0 

type OneShot[T any] struct {
	// contains filtered or unexported fields
}

OneShot is an in-memory store of short-lived, single-use tickets that carry a T. A ticket is consumed on first use and expires after its ttl, so the map stays tiny; a lazy sweep on mint drops anything stale. The zero value is ready to use. TicketStore (this package's socket) and the editing-together socket's tickets (handlers/coedit.go) are both one.

func (*OneShot[T]) Consume added in v0.56.0 

func (s *OneShot[T]) Consume(tok string) (T, bool)

Consume removes and returns what tok carries. ok is false if the token is unknown or expired (single-use: a second Consume of the same token fails).

func (*OneShot[T]) Mint added in v0.56.0 

func (s *OneShot[T]) Mint(v T, ttl time.Duration) (string, error)

Mint stores v for ttl and returns its opaque token. A best-effort sweep of expired tickets runs first so the map can't grow without bound.

type PresenceUser 

type PresenceUser struct {
	ID   int64  `json:"id"`
	UID  string `json:"uid"` // stable identity key (Client.Identity) for client-side keying/colours
	Name string `json:"name"`
	File string `json:"file,omitempty"`
	// Avatar is the profile picture to draw instead of initials — omitted when
	// the person has none, which is the fallback the strip has always used.
	Avatar string `json:"avatar,omitempty"`
}

PresenceUser is one person currently in a room and the file they're focused on (empty = just browsing the folder). Presence is de-duplicated per identity (user id + optional per-end-user key), so two tabs from the same person collapse to a single entry while two end users behind one shared proxy token stay distinct.

type Ticket 

type Ticket struct {
	UserID int64
	Name   string
	// Avatar is the presence picture for this connection (a small data: URI or
	// a URL). Resolved at mint time from the account the connection
	// authenticates as, so every client of that account — browser session,
	// desktop app, any API key minted under it — shows the same face. Empty
	// when the account has no picture, or when a proxy re-identified the
	// connection as a different end user without supplying one (drawing the
	// account owner's face for somebody else would be worse than initials).
	Avatar string
	// PresenceKey distinguishes different END users sharing one filex account.
	// Embedded hosts (work/fishapp) authenticate every end user with a single
	// proxy token — the same filex UserID — so presence would collapse them all
	// into one entry. The host proxy stamps a per-end-user key (e.g. "work-7")
	// via X-Filex-Presence-Key and presence identity becomes (UserID, key).
	// Empty = identity is the filex user alone (native sessions).
	PresenceKey string
	// Confinement captured from the minting request (a root-confined token or
	// X-Filex-Root header). Empty adapter = unconfined.
	ConfineAdapter string
	ConfineRel     string
}

Ticket is a short-lived, single-use grant that authenticates one WebSocket upgrade. Embedded consumers (the vendored webcomponent inside work.example.com / fishapp) can't set an Authorization header on a native WebSocket and connect cross-origin to fm.example.com, so instead they fetch a ticket through the host's existing HTTP proxy (which injects the real token server-side) and open `wss://fm.example.com/api/ws?ticket=<t>`. The durable token never reaches the browser; only this 60-second, one-shot ticket does.

type TicketStore 

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

TicketStore is the realtime socket's store of live tickets (OneShot).

func NewTicketStore 

func NewTicketStore() *TicketStore

NewTicketStore builds an empty store.

func (*TicketStore) Consume 

func (s *TicketStore) Consume(tok string) (Ticket, bool)

Consume removes and returns the ticket for tok. ok is false if the token is unknown or expired (single-use: a second Consume of the same token fails).

func (*TicketStore) Mint 

func (s *TicketStore) Mint(t Ticket, ttl time.Duration) (string, error)

Mint stores t for ttl and returns its opaque token.

type VaultEvent added in v0.54.0 

type VaultEvent struct {
	Type       string       `json:"type"` // "vault.generation" | "vault.lock"
	Generation uint64       `json:"generation,omitempty"`
	Held       bool         `json:"held"`
	Holder     *VaultHolder `json:"holder,omitempty"`
}

VaultEvent is one vault frame: Generation set for `vault.generation`, Held (and Holder while held) for `vault.lock`.

type VaultHolder added in v0.54.0 

type VaultHolder struct {
	Name   string `json:"name"`
	Client string `json:"client"`
	Label  string `json:"label,omitempty"`
}

VaultHolder is who holds a vault's write lock, as a frame names it.

type WatchRoot 

type WatchRoot struct {
	StorageID int64
	Dir       string // any spelling; normalised like a room key
	Display   string // the client's own spelling, echoed in frames
}

WatchRoot is one validated root a client asked to watch.

Source Files

View all Source files

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.