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

daemon

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 29, 2026 License: MIT Imports: 32 Imported by: 0

Documentation

Overview

Package daemon holds both sides of the kithd wire: Serve hosts an api.Backend over a unix socket via Connect, and Remote is the api.Backend the TUI runs against. Schema: api/proto/backend/v1; mapping: protoconv.

Index

Constants

View Source 
const LogFlag = "--log-file"

LogFlag is the kithd flag naming a log file instead of stderr.

Variables

View Source 
var ErrHandlersRunning = errors.New("handlers still running")

ErrHandlersRunning is Serve giving up on handlers that did not finish. They may be mid-write, so the caller must leave the stores open and let the process exit release them: closing the crypto store under a write can corrupt it.

Functions

func DaemonLogPath 

func DaemonLogPath(profile string) string

DaemonLogPath is where a daemon started without systemd logs when there is no journal to log to (or `[log] target` is "file"): $XDG_STATE_HOME/kith/kithd.log, or kithd-<profile>.log for a profile. A systemd unit logs to the journal.

func Dial 

func Dial(path string) *http.Client

Dial returns an HTTP client that reaches the daemon's socket at path, whatever host a request names.

func Listen 

func Listen(ctx context.Context, path string) (net.Listener, error)

Listen opens the daemon's listening socket at path, creating its parent directory owner-only. It never removes an existing socket — only the lock holder may (see Lock.Listen). ctx bounds opening only.

func LockPath 

func LockPath(user string) (string, error)

LockPath returns the path of the daemon's lock file for user (an MXID), beside the socket and named from the same hash.

func Serve 

func Serve(ctx context.Context, ln net.Listener, d *Daemon) error

Serve hosts d on ln until ctx is canceled, then shuts down gracefully. Feeding Streams is the caller's job (Streams.Run).

There is no authentication: the socket is 0600, so the kernel already decided who may connect.

func SocketPath 

func SocketPath(user string) (string, error)

SocketPath returns the path of the daemon's unix socket for user (an MXID), named by a hash to keep the MXID out of directory listings. It is always under the user's runtime directory: never the shared temp directory xdg.RuntimeFile falls back to, where another local user could create the directory first.

func SyncFault 

func SyncFault(ctx context.Context, err error) error

SyncFault returns nil when err (from api.Backend.Start) is just our own shutdown, so a clean stop does not exit non-zero or clear readiness.

Types

type Daemon 

type Daemon struct {
	Backend api.Backend
	// Streams fans each backend channel out to every subscriber.
	Streams *Streams
	// State is readiness: whether a sync has landed, and what broke if not.
	State *State
	// Notifications may be nil in a test harness; its handlers then refuse.
	Notifications *Notifications
	// Reload re-reads the config file; nil when the daemon was given none.
	Reload Reload
	// Scheduler is nil when the queue path could not be resolved; its handlers
	// then refuse.
	Scheduler *Scheduler
	// Log receives every failed call, with the procedure and the reason, so a
	// failure a client only counted still reaches the journal. nil logs nothing.
	Log *slog.Logger
}

Daemon is everything the socket serves: the Matrix backend plus facts about this process that api.Backend cannot answer.

type KeyBackup 

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

KeyBackup uploads room keys this device holds into the account's server-side backup, on a timer, for as long as the daemon runs.

func NewKeyBackup 

func NewKeyBackup(b backupBackend, log func(level slog.Level, line string)) *KeyBackup

NewKeyBackup returns a sweeper over b, reporting through log: problems at warn, progress at info.

func (*KeyBackup) Run 

func (k *KeyBackup) Run(ctx context.Context)

Run sweeps once at startup (keys received while stopped are in no backup) and then every interval, until ctx ends. It blocks.

type Lock 

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

Lock is the daemon's single-instance guard: an exclusive advisory lock on a file beside the socket, taken before any store is opened. Two processes sharing one device's olm/megolm state corrupt it irreparably.

func Acquire 

func Acquire(user string) (lock *Lock, held bool, err error)

Acquire takes the exclusive lock for user's daemon.

held is false, with no error, when another daemon already holds it: two clients racing to auto-spawn is expected, and the caller exits silently.

func (*Lock) Listen 

func (l *Lock) Listen(ctx context.Context) (net.Listener, error)

Listen opens the daemon's listening socket, removing a stale one first. Safe only because we hold the lock; the package-level Listen never unlinks.

func (*Lock) Release 

func (l *Lock) Release() error

Release removes the socket (while still holding the lock) and drops the lock. The lock file itself stays: unlinking it is the classic flock race that lets two daemons each hold "the" lock.

func (*Lock) Socket 

func (l *Lock) Socket() string

Socket returns the path of the socket this lock governs.

type Notifications 

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

Notifications decides and delivers notifications — the daemon sees every message whether or not a TUI is attached. It holds no per-client state on purpose. Config-derived state behind mu is replaced wholesale by Reload.

func NewNotifications 

func NewNotifications(cfg config.Config, src scopeSource, me string, sinks Sinks) (*Notifications, error)

NewNotifications builds the notifier from cfg. me is the logged-in MXID, so our own messages never notify. A config that will not parse is refused.

func (*Notifications) ClearAllDND 

func (n *Notifications) ClearAllDND() notify.Temps

ClearAllDND lifts every temporary rule.

func (*Notifications) ClearDND 

func (n *Notifications) ClearDND(match, sender string) notify.Temps

ClearDND lifts the rule naming this place and person, and returns what is left.

func (*Notifications) DND 

func (n *Notifications) DND() notify.Temps

DND reports the temporary rules in force, lapsed ones dropped.

func (*Notifications) Deliver 

func (n *Notifications) Deliver(ctx context.Context, msg domain.Message) (notify.Notification, bool)

Deliver acts on one incoming message: it captures a verification code if configured, then raises a notification if one is earned, and reports it.

It runs synchronously on the stream pump, which serve joins on shutdown, so every sink must be bounded (command hooks are not waited on, D-Bus has a deadline, the clipboard has copyTimeout) — a wedged notification daemon once held SIGTERM.

func (*Notifications) InvalidateScope 

func (n *Notifications) InvalidateScope()

InvalidateScope drops the room/space index, e.g. after a refresh has landed.

func (*Notifications) Reload 

func (n *Notifications) Reload(cfg config.Config) error

Reload replaces the config-derived state — rules, sinks, room aliases — or returns the error that stopped it, changing nothing. Temporary (DND) rules are runtime state and are left untouched.

func (*Notifications) Run 

func (n *Notifications) Run(ctx context.Context, s *Streams)

Run delivers a notification for every message on the fan-out, in order, until ctx is canceled or the stream ends. It blocks.

func (*Notifications) SetDND 

func (n *Notifications) SetDND(r notify.Rule) notify.Temps

SetDND puts one temporary rule in force, replacing one naming the same thing, and returns the resulting set.

func (*Notifications) Synced 

func (n *Notifications) Synced(time.Time)

Synced records a sync response arriving. mautrix runs it before that response's events, so messages handled while the count is one are the catch-up batch.

func (*Notifications) UseLogger 

func (n *Notifications) UseLogger(log *slog.Logger)

UseLogger sets where failed deliveries, copies and cache reads are logged. Call before Run; nil keeps the silent default.

type Queue 

type Queue interface {
	Load() ([]domain.ScheduledMessage, error)
	Save(queue []domain.ScheduledMessage) error
}

Queue is the durable half, declared here because its store (internal/schedule) reads and writes files, and the wire package every client links must not.

type Refresher 

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

Refresher re-reads the room list and space hierarchy when the sync loop reports they changed (/sync never writes those caches), so a daemon running alone does not drift. after runs once a refresh has landed.

func NewRefresher 

func NewRefresher(b refreshBackend, after func()) *Refresher

NewRefresher returns a Refresher over b.

func (*Refresher) Changed 

func (r *Refresher) Changed()

Changed flags that rooms or spaces changed. It never blocks.

func (*Refresher) Run 

func (r *Refresher) Run(ctx context.Context)

Run refreshes once at startup (a cold cache would otherwise serve an empty room list) and then on each debounced signal, until ctx ends. It blocks.

func (*Refresher) UseLogger 

func (r *Refresher) UseLogger(log *slog.Logger)

UseLogger sets where a failed refresh is logged (it is retried on the next change, so a warning is all it costs). Call before Run.

type Reload 

type Reload func(ctx context.Context) error

Reload re-reads the daemon's configuration and applies it, or returns the error that stopped it having changed nothing.

type Remote 

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

Remote is the api.Backend the TUI runs against: every method is a request over the daemon's unix socket, and every stream is a subscription re-exposed as a Go channel. Its methods are documented where they are declared, on api.Backend.

func Attach 

func Attach(ctx context.Context, socket string, timeout time.Duration) (*Remote, error)

Attach returns a client attached to the daemon at socket once it reports ready (not merely listening: that precedes the first sync). It never starts a daemon.

func Ensure 

func Ensure(ctx context.Context, user, profile string, timeout time.Duration) (r *Remote, note string, err error)

Ensure returns a client attached to user's daemon, starting one if none is running. note is non-empty when one had to be started and says what that cost (nothing was notified while it was down).

func NewRemote 

func NewRemote(path string) *Remote

NewRemote returns a Remote for the daemon socket at path. It performs no I/O.

func (*Remote) AcceptVerification 

func (r *Remote) AcceptVerification(ctx context.Context, txnID string) error

func (*Remote) Activity 

func (r *Remote) Activity() <-chan domain.Activity

func (*Remote) AddToSpace 

func (r *Remote) AddToSpace(ctx context.Context, spaceID domain.SpaceID, roomID domain.RoomID) error

func (*Remote) AllowRareWord 

func (r *Remote) AllowRareWord(ctx context.Context, word string) error

func (*Remote) Attached 

func (r *Remote) Attached() <-chan bool

Attached reports losing the daemon (false) and getting it back (true); see api.Backend. true is sent just before re-subscribing, so a message in that window reaches neither the catch-up read nor the stream; the unread stream still marks its room.

func (*Remote) BanUser 

func (r *Remote) BanUser(ctx context.Context, roomID domain.RoomID, userID, reason string) error

func (*Remote) BootstrapKeyBackup 

func (r *Remote) BootstrapKeyBackup(ctx context.Context, password string) (domain.KeyBackup, error)

func (*Remote) CachedInvites 

func (r *Remote) CachedInvites(ctx context.Context) ([]domain.Room, error)

func (*Remote) CachedReactions 

func (r *Remote) CachedReactions(ctx context.Context, roomID domain.RoomID) ([]domain.Reaction, error)

func (*Remote) CachedTimeline 

func (r *Remote) CachedTimeline(ctx context.Context, roomID domain.RoomID) ([]domain.Message, error)

func (*Remote) CachedUnread 

func (r *Remote) CachedUnread(ctx context.Context) ([]domain.Unread, error)

func (*Remote) CancelScheduled 

func (r *Remote) CancelScheduled(ctx context.Context, id string) error

CancelScheduled drops one pending message.

func (*Remote) CancelVerification 

func (r *Remote) CancelVerification(ctx context.Context, txnID string) error

func (*Remote) CanonicalParent 

func (r *Remote) CanonicalParent(ctx context.Context, roomID domain.RoomID) (domain.SpaceID, error)

func (*Remote) CheckSpelling 

func (r *Remote) CheckSpelling(ctx context.Context, text string) ([]domain.Misspelling, error)

CheckSpelling carries api.ErrSpellUnavailable when there is no engine.

func (*Remote) ClearAllDND 

func (r *Remote) ClearAllDND(ctx context.Context) (notify.Temps, error)

ClearAllDND lifts every temporary rule.

func (*Remote) ClearCache 

func (r *Remote) ClearCache(ctx context.Context) error

func (*Remote) ClearDND 

func (r *Remote) ClearDND(ctx context.Context, match, sender string) (notify.Temps, error)

ClearDND lifts the rule for this place and person, returning what is left.

func (*Remote) CompleteWord 

func (r *Remote) CompleteWord(ctx context.Context, req domain.CompleteRequest) ([]domain.WordCandidate, error)

func (*Remote) ConfirmSAS 

func (r *Remote) ConfirmSAS(ctx context.Context, txnID string) error

func (*Remote) CreateRoom 

func (r *Remote) CreateRoom(ctx context.Context, spec domain.NewRoom) (domain.RoomID, error)

CreateRoom keeps the in-process contract: a partial failure returns the new room's ID together with an error.

func (*Remote) DND 

func (r *Remote) DND(ctx context.Context) (notify.Temps, error)

DND reports the temporary rules in force.

func (*Remote) DetectLanguages 

func (r *Remote) DetectLanguages(ctx context.Context) (domain.SpellSuggestion, error)

func (*Remote) DetectModel 

func (r *Remote) DetectModel(ctx context.Context) (domain.ModelSuggestion, error)

func (*Remote) DirectCandidates 

func (r *Remote) DirectCandidates(ctx context.Context, limit int) ([]domain.Member, error)

func (*Remote) Drafts 

func (r *Remote) Drafts(ctx context.Context) ([]domain.StoredDraft, error)

func (*Remote) EmojiScores 

func (r *Remote) EmojiScores(ctx context.Context, kind domain.EmojiKind, roomID domain.RoomID, spaceRooms []domain.RoomID, scope string) (map[string]int, error)

func (*Remote) ExportRoomKeys 

func (r *Remote) ExportRoomKeys(ctx context.Context, passphrase string) ([]byte, error)

func (*Remote) FetchEvent 

func (r *Remote) FetchEvent(ctx context.Context, roomID domain.RoomID, eventID domain.EventID) (domain.Message, error)

func (*Remote) Follow 

func (r *Remote) Follow(ctx context.Context, uri string) (bool, error)

Follow hands a URI to whichever client is attached, reporting whether one was.

func (*Remote) Follows 

func (r *Remote) Follows() <-chan string

Follows streams matrix URIs handed over by `kith --open`.

func (*Remote) ImportRoomKeys 

func (r *Remote) ImportRoomKeys(ctx context.Context, passphrase string, data []byte) (int, int, error)

func (*Remote) InstallDictionary 

func (r *Remote) InstallDictionary(ctx context.Context, tag string) error

func (*Remote) InstallFrequencies 

func (r *Remote) InstallFrequencies(ctx context.Context, tag string) error

func (*Remote) InstallModel 

func (r *Remote) InstallModel(ctx context.Context, tag string) error

func (*Remote) InviteUser 

func (r *Remote) InviteUser(ctx context.Context, roomID domain.RoomID, userID string) error

func (*Remote) Invites 

func (r *Remote) Invites() <-chan []domain.Room

func (*Remote) JoinRoom 

func (r *Remote) JoinRoom(ctx context.Context, roomIDOrAlias string, via []string) (domain.RoomID, error)

func (*Remote) KickUser 

func (r *Remote) KickUser(ctx context.Context, roomID domain.RoomID, userID, reason string) error

func (*Remote) LastMessages 

func (r *Remote) LastMessages(ctx context.Context) (map[domain.RoomID]time.Time, error)

func (*Remote) LearnWord 

func (r *Remote) LearnWord(ctx context.Context, word string, forever bool) error

func (*Remote) LeaveRoom 

func (r *Remote) LeaveRoom(ctx context.Context, roomID domain.RoomID) error

func (*Remote) ListThreads 

func (r *Remote) ListThreads(ctx context.Context, roomID domain.RoomID) ([]domain.Thread, error)

func (*Remote) LoadImage 

func (r *Remote) LoadImage(ctx context.Context, roomID domain.RoomID, eventID domain.EventID) ([]byte, error)

LoadImage returns an attachment's bytes, from the daemon's media cache or downloaded (and decrypted) by it.

func (*Remote) MarkRead 

func (r *Remote) MarkRead(ctx context.Context, roomID domain.RoomID, eventID domain.EventID, private bool) error

func (*Remote) MarkRoomUnread 

func (r *Remote) MarkRoomUnread(ctx context.Context, roomID domain.RoomID, unread bool) error

func (*Remote) MarkRoomsRead 

func (r *Remote) MarkRoomsRead(ctx context.Context, roomIDs []domain.RoomID, private bool) (domain.ReadResult, error)

MarkRoomsRead sends only room IDs: each room's newest event is resolved from the daemon's cache.

func (*Remote) MarkSpam 

func (r *Remote) MarkSpam(ctx context.Context, verdict domain.SpamVerdict) error

func (*Remote) MarkThreadRead 

func (r *Remote) MarkThreadRead(ctx context.Context, roomID domain.RoomID, root, eventID domain.EventID, private bool) error

func (*Remote) Members 

func (r *Remote) Members(ctx context.Context, roomID domain.RoomID, limit int) ([]domain.Member, error)

func (*Remote) MentionCandidates 

func (r *Remote) MentionCandidates(ctx context.Context, roomID domain.RoomID, limit int) ([]domain.Member, error)

func (*Remote) MessageHistory 

func (r *Remote) MessageHistory(ctx context.Context, roomID domain.RoomID, eventID domain.EventID) ([]domain.Revision, domain.Deletion, error)

func (*Remote) Messages 

func (r *Remote) Messages() <-chan domain.Message

func (*Remote) MessagesAround 

func (r *Remote) MessagesAround(ctx context.Context, roomID domain.RoomID, event domain.EventID, before, after int) ([]domain.Message, error)

func (*Remote) ModelTask 

func (r *Remote) ModelTask(ctx context.Context, req domain.ModelRequest) (domain.ModelResult, error)

func (*Remote) ReactionRefusals 

func (r *Remote) ReactionRefusals(ctx context.Context) ([]domain.ReactionRefusal, error)

func (*Remote) Reactions 

func (r *Remote) Reactions() <-chan domain.ReactionUpdate

func (*Remote) RecordEmoji 

func (r *Remote) RecordEmoji(ctx context.Context, kind domain.EmojiKind, roomID domain.RoomID, emoji string) error

func (*Remote) RecordReactionRefusal 

func (r *Remote) RecordReactionRefusal(ctx context.Context, protocol, emoji string) error

func (*Remote) Redact 

func (r *Remote) Redact(ctx context.Context, roomID domain.RoomID, eventID domain.EventID, reason string) error

func (*Remote) RefreshMembers 

func (r *Remote) RefreshMembers(ctx context.Context, roomID domain.RoomID) ([]domain.Member, error)

func (*Remote) RefreshRooms 

func (r *Remote) RefreshRooms(ctx context.Context) ([]domain.Room, error)

func (*Remote) RefreshSpaces 

func (r *Remote) RefreshSpaces(ctx context.Context) ([]domain.Space, error)

func (*Remote) ReloadConfig 

func (r *Remote) ReloadConfig(ctx context.Context) error

ReloadConfig asks the daemon to re-read its config, returning the parse error if it will not load.

func (*Remote) RemoveFromSpace 

func (r *Remote) RemoveFromSpace(ctx context.Context, spaceID domain.SpaceID, roomID domain.RoomID) error

func (*Remote) ReplaceDraft 

func (r *Remote) ReplaceDraft(ctx context.Context, draft, over domain.StoredDraft) (bool, error)

ReplaceDraft names this window once it has the seat: from then on its writes are refused if another window takes it.

func (*Remote) RestoreKeyBackup 

func (r *Remote) RestoreKeyBackup(ctx context.Context, secret string) (int, error)

RestoreKeyBackup sends the secret in the request body over the 0600 socket, never on argv.

func (*Remote) RoomEncryption 

func (r *Remote) RoomEncryption(ctx context.Context, roomIDs []domain.RoomID) (map[domain.RoomID]bool, error)

RoomEncryption reports rooms the daemon did not name as unencrypted: the daemon already reported its unknowns as encrypted.

func (*Remote) Rooms 

func (r *Remote) Rooms(ctx context.Context) ([]domain.Room, error)

func (*Remote) RoomsWith 

func (r *Remote) RoomsWith(ctx context.Context, userIDs []string, rooms domain.RoomSet, limit int) ([]domain.Room, error)

func (*Remote) SaveSenderSlots 

func (r *Remote) SaveSenderSlots(ctx context.Context, roomID domain.RoomID, slots map[string]int) error

func (*Remote) Schedule 

func (r *Remote) Schedule(ctx context.Context, msg domain.ScheduledMessage) (string, error)

Schedule queues a message for later, returning the handle that cancels it.

func (*Remote) ScheduledMessages 

func (r *Remote) ScheduledMessages(ctx context.Context) ([]domain.ScheduledMessage, error)

ScheduledMessages is the whole queue, soonest first.

func (*Remote) SearchMessages 

func (r *Remote) SearchMessages(ctx context.Context, search domain.SearchRequest) ([]domain.SearchHit, error)

func (*Remote) SearchSenders 

func (r *Remote) SearchSenders(ctx context.Context, rooms domain.RoomSet, limit int) ([]domain.Member, error)

func (*Remote) SeatLost 

func (r *Remote) SeatLost() <-chan struct{}

SeatLost is closed when another window takes the seat.

func (*Remote) Send 

func (r *Remote) Send(ctx context.Context, roomID domain.RoomID, draft domain.Draft) error

func (*Remote) SendFile 

func (r *Remote) SendFile(ctx context.Context, roomID domain.RoomID, path, caption string) error

SendFile passes the path, not the bytes: client and daemon are the same user on the same machine. The caller makes it absolute.

func (*Remote) SendReaction 

func (r *Remote) SendReaction(ctx context.Context, roomID domain.RoomID, target domain.EventID, key string) error

func (*Remote) SendTyping 

func (r *Remote) SendTyping(ctx context.Context, roomID domain.RoomID, typing bool, timeout time.Duration) error

func (*Remote) SenderSlots 

func (r *Remote) SenderSlots(ctx context.Context, roomID domain.RoomID) (map[string]int, error)

func (*Remote) SetDND 

func (r *Remote) SetDND(ctx context.Context, rule notify.Rule) (notify.Temps, error)

SetDND puts one temporary rule in force and returns every rule in force.

func (*Remote) Spaces 

func (r *Remote) Spaces(ctx context.Context) ([]domain.Space, error)

func (*Remote) SpamRooms 

func (r *Remote) SpamRooms(ctx context.Context) ([]domain.SpamVerdict, error)

func (*Remote) StarMessage 

func (r *Remote) StarMessage(ctx context.Context, roomID domain.RoomID, eventID domain.EventID, starred bool) error

func (*Remote) StarredIn 

func (r *Remote) StarredIn(ctx context.Context, roomID domain.RoomID) ([]domain.EventID, error)

func (*Remote) Start 

func (r *Remote) Start(ctx context.Context) error

Start attaches to every daemon stream and pumps each onto its channel, blocking until ctx is canceled or Stop is called. It starts no sync loop: the daemon syncs regardless, so "start" means "attach".

It survives the daemon restarting: losing the streams is a reconnect, not an ending (see hold and Attached). The channels close once, when Start returns.

func (*Remote) StartVerification 

func (r *Remote) StartVerification(ctx context.Context) (string, error)

func (*Remote) Status 

func (r *Remote) Status(ctx context.Context) (ready bool, syncedAt time.Time, lastErr string, err error)

Status reports what the daemon says about itself; a dial failure means no daemon is listening (see Ensure).

func (*Remote) Stop 

func (r *Remote) Stop()

Stop detaches from the daemon, unblocking Start. Safe before Start, more than once, and concurrently.

func (*Remote) TakeSeat 

func (r *Remote) TakeSeat(ctx context.Context, force bool, where domain.SeatHolder) error

TakeSeat asks for the seat and keeps it until ctx ends (api.Seat). It answers once the daemon has: granted, or refused with ErrSeatTaken naming the other window. The stream is held open behind it; when the daemon restarts, the seat is asked for again (without force), and lost if another window took it meanwhile.

func (*Remote) ThreadPage 

func (r *Remote) ThreadPage(ctx context.Context, roomID domain.RoomID, root domain.EventID, from string, limit int) (domain.TimelinePage, error)

func (*Remote) Timeline 

func (r *Remote) Timeline(ctx context.Context, roomID domain.RoomID, from string, limit int) (domain.TimelinePage, error)

func (*Remote) UnbanUser 

func (r *Remote) UnbanUser(ctx context.Context, roomID domain.RoomID, userID string) error

func (*Remote) Unread 

func (r *Remote) Unread() <-chan domain.Unread

func (*Remote) Verifications 

func (r *Remote) Verifications() <-chan domain.Verification

type Scheduler 

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

Scheduler owns the pending-send queue: the file, the timer and the sending.

func NewScheduler 

func NewScheduler(store Queue, sender Sender, cutoff func() time.Duration, report func(level slog.Level, line string)) *Scheduler

NewScheduler builds a scheduler over the queue at store.

func (*Scheduler) Add 

func (s *Scheduler) Add(msg domain.ScheduledMessage) error

Add queues a message; a past time means "send now". It takes no context on purpose: its caller's request context dies before the message is due.

func (*Scheduler) Cancel 

func (s *Scheduler) Cancel(id string) error

Cancel drops one pending message; canceling one already gone is not an error.

func (*Scheduler) List 

func (s *Scheduler) List() []domain.ScheduledMessage

List is the queue as it stands, soonest first.

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context)

Start loads the queue and starts the first pass, which sends what came due while the daemon was down. An unreadable queue is reported and treated as empty.

The load is synchronous, so no Add can save over a queue not yet read. The pass is not: an overdue send can take the homeserver's full timeout, and the socket must be serving meanwhile. It needs no join of its own: a pass registers under the lock Stop takes, so Stop either waits for it or it sees stopped and does nothing.

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

Stop cancels the pending wake-up and waits for any pass still running, so none reaches a closed SQLite handle. Nil-safe.

type Sender 

type Sender interface {
	Send(ctx context.Context, roomID domain.RoomID, draft domain.Draft) error
}

Sender is what the scheduler needs from the backend. Scheduling lives in the daemon because nothing else may be running when a message comes due; it also makes the queue single-writer.

type Sinks 

type Sinks func(config.Notifications) notify.Notifier

Sinks builds the delivery sinks for a configuration (setup.Notifier in the daemon). Reload rebuilds them, so the builder outlives any one set.

type State 

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

State is the daemon's view of itself: whether it is ready (a sync has landed — a listening socket over a cold cache is not ready), when it last synced, and the last error. Safe for concurrent use.

func NewState 

func NewState() *State

NewState returns a State that is not ready: nothing has synced yet.

func (*State) Failed 

func (s *State) Failed(err error)

Failed records why the sync loop stopped and clears readiness: a dead sync loop over a warm cache would otherwise look healthy while serving a frozen account.

func (*State) Snapshot 

func (s *State) Snapshot() (ready bool, syncedAt time.Time, lastErr string)

Snapshot reports readiness, the last sync time (zero before the first) and the last error (empty when there has been none).

func (*State) Synced 

func (s *State) Synced(t time.Time)

Synced records a sync response arriving at t, making the daemon ready and clearing the last error.

type Streams 

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

Streams owns the fan-out for every channel api.Backend exposes (see hub).

func NewStreams 

func NewStreams() *Streams

NewStreams returns Streams with nothing attached. Subscribing before Run is safe.

func (*Streams) Attached 

func (s *Streams) Attached() int

Attached reports how many clients are fully attached: the minimum subscriber count across the five streams, since a client subscribes to them one by one.

func (*Streams) Follow 

func (s *Streams) Follow(uri string) bool

Follow hands a URI to every attached client and reports whether there was one; `kith --open` starts a terminal when there was not.

func (*Streams) ReleaseMessages 

func (s *Streams) ReleaseMessages(id uint64)

ReleaseMessages detaches a subscriber taken out by SubscribeMessages.

func (*Streams) Run 

func (s *Streams) Run(ctx context.Context, b api.Backend)

Run pumps every backend channel into its hub until ctx is canceled or the backend closes them, then ends every stream. It blocks.

func (*Streams) SubscribeMessages 

func (s *Streams) SubscribeMessages() (uint64, <-chan domain.Message)

SubscribeMessages attaches an in-process consumer (the notifier) to the message stream, returning its id and channel. It does not affect Attached.

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.