Documentation
¶
Overview ¶
Package notify implements outbound transports for newly-detected signals. Notification is decoupled from detection: any code path that calls SignalRepository.Save through a NotifyingSignalRepository triggers the configured ports.Notifier without the engine, the detectors, or the pipeline knowing or caring how delivery happens.
Available transports:
- WebhookNotifier — HMAC-signed POST per signal to one or more URLs
- SSE — bounded in-memory broadcast for /v1/signals/stream clients
- Multi — composite that fans out to several Notifiers
Delivery is at-most-once per consumer; consumers de-duplicate by Signal.ID. Persistence is the source of truth — push is a courtesy.
Index ¶
- type AckingNotifier
- type Multi
- type NotifyingSignalRepository
- func (n *NotifyingSignalRepository) Count(ctx context.Context, filter ports.SignalFilter) (int64, error)
- func (n *NotifyingSignalRepository) Get(ctx context.Context, id uuid.UUID) (domain.Signal, error)
- func (n *NotifyingSignalRepository) List(ctx context.Context, filter ports.SignalFilter) ([]domain.Signal, error)
- func (n *NotifyingSignalRepository) Save(ctx context.Context, sig domain.Signal) error
- type Outbox
- type OutboxConfig
- type SSE
- type Webhook
- type WebhookConfig
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AckingNotifier ¶ added in v0.4.0
AckingNotifier is the contract for downstream notifiers that can report whether a delivery succeeded. The standard webhook is best-effort; AckingWebhook (next sibling) reports per-call success so the outbox can decide whether to retry.
type Multi ¶
Multi composes several Notifiers into one. Failures in any single notifier do not block the others; each is invoked synchronously in slice order (transports are responsible for their own concurrency).
type NotifyingSignalRepository ¶
type NotifyingSignalRepository struct {
// contains filtered or unexported fields
}
NotifyingSignalRepository wraps a ports.SignalRepository and calls the configured Notifier *after* a successful Save. Failures from the inner repository propagate normally; the notifier is only invoked on the persistence happy path. A nil notifier disables notification (the wrapper still delegates Save/List/Get/Count, so it can be installed unconditionally and the configuration decides whether anything actually fires).
func WrapSignals ¶
func WrapSignals(inner ports.SignalRepository, notifier ports.Notifier) *NotifyingSignalRepository
WrapSignals returns a SignalRepository that calls notifier on each successful Save. Pass a nil notifier to disable push transparently.
func (*NotifyingSignalRepository) Count ¶
func (n *NotifyingSignalRepository) Count(ctx context.Context, filter ports.SignalFilter) (int64, error)
Count delegates to the inner repository.
func (*NotifyingSignalRepository) List ¶
func (n *NotifyingSignalRepository) List(ctx context.Context, filter ports.SignalFilter) ([]domain.Signal, error)
List delegates to the inner repository.
type Outbox ¶ added in v0.4.0
type Outbox struct {
// contains filtered or unexported fields
}
Outbox is a Notifier wrapper that retries failed deliveries on a background sweeper, providing at-least-once semantics within a process lifetime. Pending deliveries do NOT survive process restart — operators wanting durable at-least-once delivery wire their own persistent Notifier.
Retry policy: exponential backoff starting at MinBackoff, doubling up to MaxBackoff, with up to MaxAttempts before the delivery is dropped and metrics increment a "give-up" counter.
func NewOutbox ¶ added in v0.4.0
func NewOutbox(inner AckingNotifier, cfg OutboxConfig) (*Outbox, error)
NewOutbox constructs an Outbox. Callers must invoke Start to activate the sweeper goroutine; Stop drains in-flight retries.
func (*Outbox) Notify ¶ added in v0.4.0
Notify is the ports.Notifier entry point: fire-and-forget. The underlying delivery is enqueued in the outbox; immediate delivery is attempted once, and failed deliveries are retried by the sweeper.
func (*Outbox) PendingCount ¶ added in v0.4.0
PendingCount returns the number of deliveries waiting to retry. Useful for tests and metrics surface.
type OutboxConfig ¶ added in v0.4.0
type OutboxConfig struct {
MaxAttempts int
MinBackoff time.Duration
MaxBackoff time.Duration
PersistencePath string
}
OutboxConfig tunes the retrier. All fields default to sensible values when zero:
MaxAttempts default 5 MinBackoff default 1s MaxBackoff default 30s
PersistencePath, when set, makes the outbox durable across process restarts: pending deliveries are snapshotted to the path after each enqueue/flush, and re-loaded at Start. The file format is JSON and the parent directory is created with 0o700 permissions.
type SSE ¶
type SSE struct {
// contains filtered or unexported fields
}
SSE is an in-memory broadcaster for Server-Sent Events. It implements ports.Notifier (push side) and exposes Subscribe / Unsubscribe so an HTTP handler can register browser/runtime clients for /v1/signals/stream. Each subscriber owns a buffered channel; a signal is delivered with a non-blocking send, so a slow client is silently dropped rather than back-pressuring the entire bus. SSE is therefore a best-effort live feed; consumers needing replay must combine the stream with a /v1/signals query keyed on the last-seen DetectedAt.
func NewSSE ¶
NewSSE returns a broadcaster with the given per-client buffer capacity. A buffer of 8-32 is typically right: enough to absorb a burst, small enough that genuinely stuck clients are dropped fast. A bufferSize of 0 forces every send to block until the receiver reads — so we floor at 1 silently to keep the broadcaster healthy.
func (*SSE) Notify ¶
Notify fans the signal out to every matching subscriber via a non-blocking send. Slow consumers are dropped silently. Honours ctx cancellation to skip iteration when the broadcaster is being shut down.
func (*SSE) Subscribe ¶
Subscribe registers a new client filtered by an allowlist of scope ids (nil = any scope, single-element slice = legacy per-scope stream) and pattern (empty = any pattern). It returns:
- id used to unregister
- read-only channel of domain.Signal matching the filter
The returned channel is closed by Unsubscribe; receivers should detect the close and exit their read loop. The signature is deliberately bare types so internal/api can declare a matching interface without importing this package — that breaks what would otherwise be a cycle (api -> notify -> api via the wire shape).
func (*SSE) SubscriberCount ¶
SubscriberCount returns the number of currently-attached clients. Useful for /metrics and tests.
func (*SSE) Unsubscribe ¶
Unsubscribe removes the client and closes its channel. Safe to call with an unknown id (no-op).
type Webhook ¶
type Webhook struct {
// contains filtered or unexported fields
}
Webhook posts each newly-persisted signal as JSON to a configured set of URLs. It is a transport: it does not interpret the signal, does not decide whether to deliver, and never blocks the persistence path. Delivery is at-most-once per URL with a small best-effort retry on 5xx responses; consumers de-duplicate by Signal.ID and treat headers as authoritative.
Headers per request:
X-Chronos-Signature: sha256=<hex hmac-sha256 of body> // when Secret is set X-Chronos-Delivery: <uuid v4> // unique per send attempt X-Chronos-Event: signal.detected
The signed payload is the same JSON shape as /v1/signals returns (api.SignalDTO), so a consumer can treat webhook bodies and pulled list responses identically.
func NewWebhook ¶
func NewWebhook(cfg WebhookConfig, metrics *observability.Metrics, logger *slog.Logger) *Webhook
NewWebhook constructs a Webhook from configuration. The returned notifier is safe for concurrent use; pass it through ports.Notifier.
type WebhookConfig ¶
type WebhookConfig struct {
URLs []string
Secret string
Timeout time.Duration // default 5s when zero
Retries int // default 1 when negative
}
WebhookConfig describes a webhook fan-out target set. URLs is the only required field; an empty slice disables the transport. Secret is optional — when empty, no X-Chronos-Signature header is sent and receivers cannot authenticate the payload (acceptable for trusted network paths but not over the public internet).
Source Files