notifier

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 28, 2026 License: MIT Imports: 8 Imported by: 0

README

notifier

CI Go version coverage tag

Notifier is a lightweight library that delivers queued items to destinations by transport (Telegram and Email at the moment), retrying until they land using go-retry. Message batches claimed from storage, sent, and resolved by pool of workers concurrently. Queued items and delivery state persisted in a pluggable store (driver-agnostic SQLite and in-memory at the moment). Stale deliveries are handled too. It also automatically probes destinations concurrently before sending to them.

Table of Contents

Usage

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"time"
	// or any other driver of your choice which implements 'database/sql'
	_ "modernc.org/sqlite"

	"github.com/arkhemlol/notifier"
	// import only what you actually need
	"github.com/arkhemlol/notifier/telegram"
	"github.com/arkhemlol/notifier/email"
	"github.com/arkhemlol/notifier/store/sqlite"
)

type Alert struct {
	Text string
}

type jsonCodec struct{}

func (jsonCodec) Encode(value Alert) ([]byte, error) {
	return json.Marshal(value)
}

func (jsonCodec) Decode(data []byte) (Alert, error) {
	var value Alert
	err := json.Unmarshal(data, &value)

	return value, err
}

// error handling ommited for clarity
func main() {
	ctx := context.Background()

	client, err := telegram.NewClient[Alert](telegram.Config{Token: "bot-token"})

	alerts, err := client.Chat("telegram:alerts", "chat_id", func(batch []Alert) string {
		return batch[0].Text
	})

	transport, err := email.NewTransport[Alert](email.Config{
		Host: "smtp.example.com", Port: "465",
		Username: "alerts@example.com", Password: "my_password", From: "alerts@example.com",
	})

	onCall, err := transport.Recipient("email:on-call", "on-call@example.com", func(batch []Alert) email.Message {
		return email.Message{Subject: "Alert", Text: batch[0].Text}
	})

	db, err := sql.Open("sqlite", "file:notifier.db?_fk=1&_busy_timeout=5000")

	store, err := sqlite.New[Alert](db, jsonCodec{})

	dispatcher, err := notifier.NewDispatcher(
		store,
		notifier.DispatcherConfig{
			Workers:      8,                // batches claimed and sent at once; defaults to GOMAXPROCS
			FirstSuccess: false,            // false: every destination must succeed; true: delivery success for first transport is enough (others not used)
			ProbeWorkers: 4,                // check destination availability before the first delivery, this many at once
			ProbeTimeout: 10 * time.Second, // time budget for one probe check
			SkipProbing:  false,            // true skips that check; delivery to an unreachable destination is then less reliable
		},
		// Destinations. For a custom probe check per destination, implement notifier.Prober, 
		// or wrap one in a type that embeds it and adds its own Probe.
		alerts, // telegram, tried first if FirstSuccess is set
		onCall, // email
		// .. more
	)

	wait := dispatcher.Start(ctx, time.Minute, func(_ notifier.Report, err error) {
		// handle err, e.g. log it
	})
	defer wait(ctx)

	// One-shot delivery cycle instead of a recurring schedule:
	report, err := dispatcher.Run(ctx)
}

Examples

More examples here: examples folder.

Structure

  • telegram sends notification batches through the Telegram Bot API.
  • email sends notification batches over implicit-TLS SMTP.
  • store/sqlite is a driver-agnostic adapter for SQLite (uses std's database/sql interface).
  • store/memory is an in-memory store, doesn't survive process restart.

See API description for the full public API.

TODO

  • add statistics
  • more transports (Discord, Slack, grpc, etc.)
  • centralized caching
  • PostgreSQL adapter
  • Redis adapter

Documentation

Overview

Package notifier provides stored, at-least-once notification delivery.

Create destinations and a dispatcher, enqueue items, then run a delivery cycle: 

transport, err := email.NewTransport[Alert](email.Config{
	Host: "smtp.example.com", Port: "465",
	Username: user, Password: pass, From: "alerts@example.com",
})
onCall, err := transport.Recipient("email:on-call", "on-call@example.com", render)
dispatcher, err := notifier.NewDispatcher(
	memory.New[Alert](), notifier.DispatcherConfig{}, onCall)
err = dispatcher.Enqueue(ctx, notifier.Item[Alert]{
	ID: 1, Payload: Alert{Text: "disk almost full"},
})
report, err := dispatcher.Run(ctx)

Run claims queued work, sends it, and records each outcome. Retryable failures remain queued. Permanent destination failures quarantine that destination. 

for _, result := range report.Results {
	if errors.Is(result.SendErr, notifier.ErrQuarantine) {
		log.Printf("destination %s quarantined", result.Destination)
	}
}

Multiple destinations must all succeed unless DispatcherConfig.FirstSuccess is set, in which case their order defines the fallback chain. The Store holds unfinished work; memory.New is process-local and sqlite.New survives restarts.

The first Run registers the plan and probes destinations that support checks. If delivery succeeds but recording fails, Run reports OutcomeDeliveredUnrecorded and the item may be delivered again.

Index

Constants

View Source 
const (
	// OutcomeUnknown is not a valid delivery resolution.
	OutcomeUnknown = core.OutcomeUnknown
	// OutcomeDelivered records successful provider acceptance.
	OutcomeDelivered = core.OutcomeDelivered
	// OutcomeRetryableFailure keeps the batch eligible for later delivery.
	OutcomeRetryableFailure = core.OutcomeRetryableFailure
	// OutcomeFailedPermanent terminalizes the affected failure scope.
	OutcomeFailedPermanent = core.OutcomeFailedPermanent
	// OutcomeDeliveredUnrecorded means delivery succeeded but recording failed; it may repeat.
	OutcomeDeliveredUnrecorded = core.OutcomeDeliveredUnrecorded
)

Variables

View Source 
var (
	// ErrInvalidDispatcherConfig marks invalid dispatcher configuration.
	ErrInvalidDispatcherConfig = errors.New("invalid dispatcher configuration")
	// ErrInvalidDestinationBinding marks an invalid destination list.
	ErrInvalidDestinationBinding = errors.New("invalid destination binding")
)
View Source 
var (
	// ErrStoreStaleLeaseToken marks an outcome written after losing its lease.
	ErrStoreStaleLeaseToken = core.ErrStoreStaleLeaseToken
	// ErrStorePayloadConflict marks reuse of an Item.ID with a different payload.
	ErrStorePayloadConflict = core.ErrStorePayloadConflict
	// ErrStoreWorkDoesntExist marks an outcome for an unknown batch.
	ErrStoreWorkDoesntExist = core.ErrStoreWorkDoesntExist
	// ErrStoreInvalidTransition marks a write that contradicts stored state.
	ErrStoreInvalidTransition = core.ErrStoreInvalidTransition
	// ErrStoreBusy marks transient backend contention, which is retried.
	ErrStoreBusy = core.ErrStoreBusy
	// ErrStoreUnavailable marks unavailable persistence infrastructure.
	ErrStoreUnavailable = core.ErrStoreUnavailable
)
View Source 
var (
	// ErrRetryable marks a failure that may succeed on a later attempt.
	ErrRetryable = core.ErrRetryable
	// ErrPermanent marks a failure that will not succeed on a later attempt.
	ErrPermanent = core.ErrPermanent
	// ErrQuarantine marks a permanent failure that disables the destination.
	ErrQuarantine = core.ErrQuarantine
)

Delivery errors retain the provider cause for errors.Is and errors.As. ErrQuarantine also matches ErrPermanent.

View Source 
var ErrDestinationImplementationMissing = errors.New(
	"dispatcher destination implementation missing",
)

ErrDestinationImplementationMissing marks work with no bound destination.

Functions

This section is empty.

Types

type Destination 

type Destination[T any] interface {
	// ID returns the stable identifier used in delivery plans.
	ID() core.DestinationID

	// Send delivers one batch and returns only once the provider has accepted or rejected it.
	Send(ctx context.Context, batch []T) error
}

Destination sends batches to one addressable endpoint. Use email.Transport.Recipient, telegram.Client.Chat, or another transport in this module.

type Dispatcher 

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

Dispatcher delivers queued items for one destination plan. Run calls are serialized; sends are concurrent and outcomes are persisted serially. Durable delivery state lives in the Store.

func NewDispatcher 

func NewDispatcher[T any](
	store Store[T], config DispatcherConfig, destinations ...Destination[T],
) (*Dispatcher[T], error)

NewDispatcher validates and binds destinations in order. Their IDs and order define the delivery plan and FirstSuccess fallback order.

func (*Dispatcher[T]) Enqueue 

func (d *Dispatcher[T]) Enqueue(ctx context.Context, items ...Item[T]) error

Enqueue persists items for later delivery; it does not send them. Reusing an Item.ID with the same payload is a no-op; a different payload returns ErrStorePayloadConflict.

func (*Dispatcher[T]) Run 

func (d *Dispatcher[T]) Run(ctx context.Context) (Report, error)

Run performs one delivery cycle. An empty queue is not an error. Concurrent calls wait for the active cycle or return when their context is canceled. The first call registers the plan and probes supported destinations before delivery. Before claiming, it retries still-leased writes for successful deliveries. Once its plan is empty, Run may drain another pending plan; unbound work fails permanently.

func (*Dispatcher[T]) Start 

func (d *Dispatcher[T]) Start(
	ctx context.Context,
	interval time.Duration,
	onCycle func(Report, error),
) (wait func(context.Context) error)

Start runs the dispatcher on a schedule until ctx is cancelled, calling onCycle on every cycle and returns the function that can be called to cancel the loop. 

wait := dispatcher.Start(ctx, time.Minute, func(_ notifier.Report, err error) {
	if err != nil {
		slog.Error("delivery cycle failed", "error", err)
	}
})
defer wait(shutdownCtx)

type DispatcherConfig 

type DispatcherConfig struct {
	// Workers limits batches claimed and sent concurrently. Defaults to GOMAXPROCS.
	Workers int

	// FirstSuccess stops at the first accepting destination. By default, all must succeed.
	FirstSuccess bool

	// MaxItemsPerWork limits one Destination.Send batch. Defaults to 100.
	MaxItemsPerWork int

	// AttemptLimit includes the initial send. Defaults to 5; the maximum is 20.
	AttemptLimit int

	// AttemptTimeout bounds each send. Timeouts are retryable. Defaults to 30s.
	AttemptTimeout time.Duration

	// InitialBackoff is doubled after each retry. Defaults to 1s; the maximum is 1h.
	InitialBackoff time.Duration

	// JitterPercent varies backoff by ±N percent. Defaults to 20; the maximum is 100.
	JitterPercent int

	// DisableJitter uses the nominal backoff without variation.
	DisableJitter bool

	// PersistFailureDetail stores provider error text for permanent failures.
	// It defaults off because provider errors may contain recipient data.
	PersistFailureDetail bool

	// ResolveTimeout bounds outcome persistence, including retries. Defaults to 5s.
	ResolveTimeout time.Duration

	// ProbeWorkers limits concurrent checks before the first delivery. Defaults to 4.
	ProbeWorkers int

	// ProbeTimeout bounds each check. A timeout leaves stored state unchanged. Defaults to 10s.
	ProbeTimeout time.Duration

	// SkipProbing disables the automatic probe before the first Run. Defaults to false;
	// skipping makes delivery to an unreachable destination less reliable.
	SkipProbing bool
}

DispatcherConfig controls dispatch concurrency, retries, and destination probes. Zero values select the defaults below; invalid numeric values return ErrInvalidDispatcherConfig.

type Item 

type Item[T any] = core.Item[T]

Item pairs a storage-level identifier with a delivery payload.

type Outcome 

type Outcome = core.Outcome

Outcome describes a delivery resolution reported by Run.

type Prober 

type Prober = core.Prober

Prober is a Destination that can check reachability before delivery. Dispatcher probes any bound destination implementing it, including a wrapper embedding a plain Destination to add a custom Probe.

type Report 

type Report struct {
	Results []result
}

Report contains results from one bounded dispatch wave.

type Store 

type Store[T any] = core.Store[T]

Store persists queued items, leases, and outcomes.

Source Files

View all Source files

Directories

Path Synopsis
Package email sends notification batches over implicit-TLS SMTP.
 Package email sends notification batches over implicit-TLS SMTP.
internal
core
Package core defines notifier's internal persistence model.
 Package core defines notifier's internal persistence model.
store
memory
Package memory provides a process-local implementation of core.Store.
 Package memory provides a process-local implementation of core.Store.
sqlite
Package sqlite implements core.Store on caller-owned, driver-neutral SQLite.
 Package sqlite implements core.Store on caller-owned, driver-neutral SQLite.
Package telegram sends notification batches through the Telegram Bot API.
 Package telegram sends notification batches through the Telegram Bot API.

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.