notify

Documentation

Overview

Package notify implements hotline's third ingress leg: event-driven notifies from local scripts and daemons (backup jobs, the email sentry, CI, watchers). A script calls `hotline notify --source <key>`; the CLI runs the full gate (capability-key lookup, level clamp, sanitize, dedup, per-source rate limit, quiet hours) and durably enqueues an accepted event into spool.json. A Dispatcher on the daemon side injects enqueued events as synthetic inbound turns (kind="notify") through the same sink real messages and schedules use.

State lives under <box root>/notify/: sources.json (the capability-key registry, operator-owned) and spool.json (pending entries plus per-source gate state), both guarded by the same flock/atomic-write pattern as schedules.json. The design is deliberately the existing house patterns applied to a new noun.

Index

• Variables
• func Dir(stateRoot string) string
• func MaxStdinBytes() int64
• func MutateRegistry(path string, fn func(*Registry) error) error
• func MutateSpool(path string, fn func(*SpoolDoc) error) error
• func RejectsPath(stateRoot string) string
• func Sanitize(s string) string
• func SaveRegistry(r *Registry, path string) error
• func SaveSpool(d *SpoolDoc, path string) error
• func SourcesPath(stateRoot string) string
• func SpoolPath(stateRoot string) string
• type Dispatcher
  • func NewDispatcher(spoolPath, sourcesPath, accessFile string, sources []string, ...) *Dispatcher
  • func (d *Dispatcher) Run(ctx context.Context, sink Sink) error
• type Entry
• type Level
  • func ClampLevel(requested, cap Level) (Level, bool)
  • func ParseLevel(s string) (Level, error)
• type Outcome
  • func Enqueue(spoolPath, rejectsPath string, reg *Registry, key string, level Level, ...) (Outcome, error)
• type OutcomeStatus
• type Rate
• type Registry
  • func LoadRegistry(path string) (*Registry, error)
  • func (r *Registry) FindByKey(key string) (Source, bool)
  • func (r *Registry) FindByLabel(label string) (Source, bool)
• type Sink
• type Source
  • func AddSource(path, label string, cap Level, rate Rate, chatID string, now time.Time) (Source, error)
  • func RevokeSource(path, label string) (Source, error)
• type SourceState
• type SpoolDoc
  • func LoadSpool(path string) (*SpoolDoc, error)

Variables

var ErrSourceNotFound = errors.New("no source with that label")

ErrSourceNotFound is returned when no source matches a label.

Functions

func Dir

func Dir(stateRoot string) string

Path helpers: everything notify owns lives under <box root>/notify/.

func MaxStdinBytes

func MaxStdinBytes() int64

MaxStdinBytes is the hard bound the CLI applies to a piped message read; Sanitize then truncates to maxMessageBytes.

func MutateRegistry

func MutateRegistry(path string, fn func(*Registry) error) error

MutateRegistry is a flock(LOCK_EX)-guarded read-modify-write on sources.json via path+".lock", so concurrent source add/revoke never race.

func MutateSpool

func MutateSpool(path string, fn func(*SpoolDoc) error) error

MutateSpool is a flock(LOCK_EX)-guarded read-modify-write on spool.json via path+".lock", so the CLI gate and the daemon's dispatcher never race — the same CLI-mutates-while-daemon-ticks concurrency schedules already live with.

func RejectsPath

func RejectsPath(stateRoot string) string

func Sanitize

func Sanitize(s string) string

Sanitize cleans a script-authored message once, at enqueue, so the spool only ever holds clean payloads: it neutralizes envelope-close forgery, strips control characters (ANSI escapes) except \n and \t, and truncates to maxMessageBytes at a UTF-8 boundary.

func SaveRegistry

func SaveRegistry(r *Registry, path string) error

SaveRegistry atomically writes sources.json (tmp file 0600 + rename).

func SaveSpool

func SaveSpool(d *SpoolDoc, path string) error

SaveSpool atomically writes spool.json (tmp file 0600 + rename).

func SourcesPath

func SourcesPath(stateRoot string) string

func SpoolPath

func SpoolPath(stateRoot string) string

Types

type Dispatcher

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

Dispatcher consumes the spool and injects enqueued events. now/loc/tick are fields (defaulted in NewDispatcher) so tests inject a fixed clock and never sleep.

func NewDispatcher

func NewDispatcher(spoolPath, sourcesPath, accessFile string, sources []string, log *transcript.Logger) *Dispatcher

NewDispatcher builds a Dispatcher over spool.json/sources.json at the notify paths. sources is router.Sources(); accessFile is the primary provider's access.json path (may be empty); log may be nil.

func (*Dispatcher) Run

func (d *Dispatcher) Run(ctx context.Context, sink Sink) error

Run injects enqueued notifies until ctx is cancelled: one eager catch-up scan immediately (the restart catch-up, zero special code), then a scan per tick. It returns nil on cancellation and never otherwise exits — store and injection failures are logged to stderr and retried on the next tick, never fatal.

type Entry

type Entry struct {
	ID      string `json:"id"`
	Label   string `json:"label"`
	Level   Level  `json:"level"`
	Message string `json:"message"`
	Hash    string `json:"hash"`
	Status  string `json:"status"` // statusReady | statusQueued
	Count   int    `json:"count"`  // dedup coalesce counter
	Clamped bool   `json:"clamped,omitempty"`
	FirstAt string `json:"firstAt"`
	LastAt  string `json:"lastAt"`
}

Entry is one pending spool item awaiting injection.

type Level

type Level string

Level is a notify urgency. urgent > normal > low; only urgent bypasses quiet hours (nothing bypasses the rate limit).

const (
	LevelUrgent Level = "urgent"
	LevelNormal Level = "normal"
	LevelLow    Level = "low"
)

func ClampLevel

func ClampLevel(requested, cap Level) (Level, bool)

ClampLevel returns min(requested, cap) and whether the request was clamped. A level above the source's cap is clamped, not rejected — a misconfigured script still gets its event through, just without the escalation it isn't entitled to.

func ParseLevel

func ParseLevel(s string) (Level, error)

ParseLevel normalizes a CLI --level value. Empty defaults to normal; anything other than urgent/normal/low is a usage error.

type Outcome

type Outcome struct {
	Status          OutcomeStatus
	Label           string
	Level           Level
	Clamped         bool
	ClampedTo       Level
	Count           int    // for Duplicate
	QueuedUntil     string // "HH:MM", for Queued
	Suppressed      int    // for RejectedRate
	SuppressedSince string // "HH:MM", for RejectedRate
}

Outcome carries everything the CLI needs to print the right line and exit code.

func Enqueue

func Enqueue(spoolPath, rejectsPath string, reg *Registry, key string, level Level, rawMessage string, now time.Time) (Outcome, error)

Enqueue runs the full gate for one notify and durably records the result. The registry is read (fresh, unlocked — atomic rename makes that safe) by the caller and passed in; the whole check-and-record runs inside the spool's flock critical section so a crashlooping caller cannot race the bucket. Returns an error only for internal failures (I/O, lock, quiet-hours parse) → exit 1; the gate decision rides Outcome.

type OutcomeStatus

type OutcomeStatus int

OutcomeStatus is the gate's decision, mapped by the CLI to an exit code.

const (
	Accepted          OutcomeStatus = iota // durably enqueued as ready
	Duplicate                              // coalesced into an existing/recent identical event
	Queued                                 // valid but held for quiet hours
	RejectedUnknown                        // unknown or revoked source key
	RejectedRate                           // rate-limit suppressed
	RejectedSpoolFull                      // spool at capacity
)

type Rate

type Rate struct {
	Burst      int `json:"burst,omitempty"`
	RefillMins int `json:"refillMins,omitempty"`
}

Rate is a per-source token-bucket override. Zero fields mean the defaults.

type Registry

type Registry struct {
	QuietHours    string   `json:"quietHours"`
	DefaultChatID string   `json:"defaultChatId"`
	Sources       []Source `json:"sources"`
}

Registry is the full persisted sources.json document. quietHours and defaultChatId are subsystem-level settings riding the same operator-owned file.

func LoadRegistry

func LoadRegistry(path string) (*Registry, error)

LoadRegistry reads sources.json. Missing file → empty Registry. Corrupt file → moved aside to path+".corrupt", empty Registry returned. (schedule.Load pattern.)

func (*Registry) FindByKey

func (r *Registry) FindByKey(key string) (Source, bool)

FindByKey resolves a bearer key to its source with a constant-time compare over every registered key (the property a bearer token needs; cheap locally, free forward-compatibility for a future webhook ingress).

func (*Registry) FindByLabel

func (r *Registry) FindByLabel(label string) (Source, bool)

FindByLabel resolves a source by its human label (provenance / chat routing).

type Sink

type Sink interface {
	SendChannel(ctx context.Context, content string, meta map[string]string) error
}

Sink is the inbound-injection seam: the one method the dispatcher needs from provider.InboundSink. Declared locally so this package never imports internal/provider (the same cycle the scheduler dodges). *mcpchan.Notifier and cmd/hotline's opencodeSink satisfy it structurally.

type Source

type Source struct {
	Label     string `json:"label"`
	Key       string `json:"key"`
	LevelCap  Level  `json:"levelCap"`
	Rate      Rate   `json:"rate,omitempty"`
	ChatID    string `json:"chatId,omitempty"`
	CreatedAt string `json:"createdAt"`
}

Source is one registered capability key. The key is a bearer credential; every human-facing surface shows Label, never Key.

func AddSource

func AddSource(path, label string, cap Level, rate Rate, chatID string, now time.Time) (Source, error)

AddSource mints a fresh capability key for a new label and appends it. Labels are unique (they are the provenance handle); cap defaults to normal so urgent must be granted deliberately. Runs entirely under MutateRegistry.

func RevokeSource

func RevokeSource(path, label string) (Source, error)

RevokeSource removes the source matching label (exact) and returns it. There are no tombstones — the audit trail is the transcript plus rejects.log. A revoked key immediately fails the gate because every CLI call reads the registry fresh.

type SourceState

type SourceState struct {
	Tokens          float64 `json:"tokens"`
	TokensAt        string  `json:"tokensAt,omitempty"`
	LastHash        string  `json:"lastHash,omitempty"`
	LastHashAt      string  `json:"lastHashAt,omitempty"`
	Suppressed      int     `json:"suppressed,omitempty"`
	SuppressedSince string  `json:"suppressedSince,omitempty"`
	Delivered       int     `json:"delivered,omitempty"`
	LastSeen        string  `json:"lastSeen,omitempty"`
}

SourceState is the persisted per-source gate state: token bucket, dedup fingerprint, suppression counters, and lifetime counters.

type SpoolDoc

type SpoolDoc struct {
	Pending []Entry                 `json:"pending"`
	State   map[string]*SourceState `json:"state"`
}

SpoolDoc is the full persisted spool.json document.

func LoadSpool

func LoadSpool(path string) (*SpoolDoc, error)

LoadSpool reads spool.json. Missing file → empty doc. Corrupt file → moved aside to path+".corrupt", empty doc returned. (schedule.Load pattern.)