notifications

README

schoolaid-go-notifications

Thin Go wrapper around a Kafka producer for the SchoolAid notifications pipeline. Mirrors the schemas, topic names, and header conventions used by schoolaid-admin's Laravel KafkaChannel so a Go service can publish identical messages without re-deriving the wire format.

The package is intentionally a wrapper. There are no service-specific defaults, secrets, or business rules — brokers, topics, and client identity come from the caller (typically the same KAFKA_* environment variables Laravel reads).

Install

go get github.com/schoolaid/schoolaid-go-notifications

Usage

import notifications "github.com/schoolaid/schoolaid-go-notifications"

cfg := notifications.ConfigFromEnv()
client, err := notifications.NewClient(cfg)
if err != nil { return err }
defer client.Close()

err = client.PublishEmail(ctx, notifications.EmailMessage{
    SchoolID: 42,
    UserID:   99,
    Email:    "foo@example.com",
    Subject:  "Hello",
    Content:  "<p>hi</p>",
})

Environment

Reads the same variables schoolaid-admin/config/kafka.php reads:

Var	Default
KAFKA_ENABLED	false
KAFKA_BROKERS	localhost:9092
KAFKA_CLIENT_ID	schoolaid-go
KAFKA_TOPIC_NOTE_CREATED	notifications.note.created
KAFKA_TOPIC_NOTE_CREATED_PRIORITY	notifications.note.created.priority
KAFKA_TOPIC_EMAIL_BATCH	notifications.email.batch
KAFKA_TOPIC_PUSH_BATCH	notifications.push.batch
KAFKA_TOPIC_SMS_BATCH	notifications.sms.batch
KAFKA_TOPIC_WHATSAPP_BATCH	notifications.whatsapp.batch
KAFKA_TOPIC_DLQ	notifications.dlq
KAFKA_TOPIC_RETRY	notifications.retry

Helpers

Method	Topic	Partition key
PublishNoteCreated	note.created (or .priority)	recipient.student_id → note_id
PublishEmail	email.batch	user_id
PublishPush	push.batch	user_id
PublishSMS	sms.batch	user_id
PublishWhatsApp	whatsapp.batch	user_id
PublishRaw	caller-supplied	caller-supplied

All helpers add three Kafka headers: trace-id, school-id, priority.

Testing

NewClientWithProducer accepts any Producer implementation, so unit tests can stub the broker entirely. See client_test.go.

Schema source of truth

Field names and JSON tags mirror schoolaid-admin/app/Notifications/Messages/*. When the Laravel side changes, this package follows.

Documentation

Overview

Package notifications is a thin wrapper around a Kafka producer for the SchoolAid notifications pipeline. It mirrors the schemas, topics, and header conventions used by the schoolaid-admin Laravel KafkaChannel so Go services can publish identical messages without re-deriving the wire format.

The package intentionally carries no secrets or service-specific defaults. Brokers, topics, and client identity are supplied by the caller (typically from environment variables matching the Laravel KAFKA_* names).

Quick start:

c, err := notifications.NewClient(notifications.ConfigFromEnv())
if err != nil { return err }
defer c.Close()

err = c.PublishEmail(ctx, notifications.EmailMessage{
    EventID:   uuid.NewString(),
    TraceID:   traceID,
    SchoolID:  42,
    UserID:    99,
    Email:     "foo@example.com",
    Subject:   "Hello",
    Content:   "<p>hi</p>",
    Priority:  notifications.PriorityNormal,
})

Index

• Constants
• type Attachment
• type Client
  • func NewClient(cfg Config) (*Client, error)
  • func NewClientWithProducer(p Producer, topics Topics) *Client
  • func (c *Client) Close() error
  • func (c *Client) PublishEmail(ctx context.Context, m EmailMessage) error
  • func (c *Client) PublishNoteCreated(ctx context.Context, n NoteCreated) error
  • func (c *Client) PublishPush(ctx context.Context, m PushMessage) error
  • func (c *Client) PublishRaw(ctx context.Context, topic, key string, payload any, headers map[string]string) error
  • func (c *Client) PublishSMS(ctx context.Context, m SMSMessage) error
  • func (c *Client) PublishWhatsApp(ctx context.Context, m WhatsAppMessage) error
  • func (c *Client) Topics() Topics
• type Config
  • func ConfigFromEnv() Config
• type EmailAttachment
• type EmailMessage
• type Metadata
• type Note
• type NoteCreated
• type Priority
• type Producer
• type PushMessage
• type PushTitle
• type Recipient
• type SMSMessage
• type Topics
• type WhatsAppMessage

Constants

const (
	DefaultTopicNoteCreated         = "notifications.note.created"
	DefaultTopicNoteCreatedPriority = "notifications.note.created.priority"
	DefaultTopicEmailBatch          = "notifications.email.batch"
	DefaultTopicPushBatch           = "notifications.push.batch"
	DefaultTopicSMSBatch            = "notifications.sms.batch"
	DefaultTopicWhatsAppBatch       = "notifications.whatsapp.batch"
	DefaultTopicDLQ                 = "notifications.dlq"
	DefaultTopicRetry               = "notifications.retry"
)

Default topic names, kept in sync with schoolaid-admin/config/kafka.php.

const (
	EventNoteCreated = "note.created"
)

EventType values used in NoteCreated.EventType.

const SchemaVersion = "1.0"

Types

type Attachment

type Attachment struct {
	Filename string  `json:"filename"`
	Path     string  `json:"path,omitempty"`
	URL      *string `json:"url"`
}

Attachment represents a single signed attachment URL. Mirrors NoteMessage::buildCachedAttachments().

type Client

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

Client is the high-level entry point. It owns a Producer plus the resolved topic configuration and exposes one Publish helper per logical event.

func NewClient

func NewClient(cfg Config) (*Client, error)

NewClient wires up the default segmentio-kafka-go-backed producer. Use NewClientWithProducer to inject a stub or alternative implementation.

func NewClientWithProducer

func NewClientWithProducer(p Producer, topics Topics) *Client

NewClientWithProducer skips Kafka setup. Useful for tests or for plugging in a different transport (e.g. an in-memory ring buffer).

func (*Client) Close

func (c *Client) Close() error

Close flushes and shuts down the underlying producer.

func (*Client) PublishEmail

func (c *Client) PublishEmail(ctx context.Context, m EmailMessage) error

PublishEmail puts one entry on the email batch topic. Key is the recipient user id, matching the Laravel side.

func (*Client) PublishNoteCreated

func (c *Client) PublishNoteCreated(ctx context.Context, n NoteCreated) error

PublishNoteCreated routes to the priority topic when n.Note.Important is true and to the regular topic otherwise. The partition key is the recipient's student_id when present, falling back to the note id — matching NoteMessage::kafkaKey().

func (*Client) PublishPush

func (c *Client) PublishPush(ctx context.Context, m PushMessage) error

PublishPush puts one entry on the push batch topic.

func (*Client) PublishRaw

func (c *Client) PublishRaw(ctx context.Context, topic, key string, payload any, headers map[string]string) error

PublishRaw is the escape hatch for anything that does not yet have a first-class helper. The caller owns topic, key, and headers.

func (*Client) PublishSMS

func (c *Client) PublishSMS(ctx context.Context, m SMSMessage) error

PublishSMS puts one entry on the SMS batch topic.

func (*Client) PublishWhatsApp

func (c *Client) PublishWhatsApp(ctx context.Context, m WhatsAppMessage) error

PublishWhatsApp puts one entry on the WhatsApp batch topic.

func (*Client) Topics

func (c *Client) Topics() Topics

Topics returns the resolved topic set, useful for diagnostics.

type Config

type Config struct {
	// Brokers is a comma-separated list of host:port pairs. Required.
	Brokers string

	// ClientID identifies the producer to the broker. Optional; defaults to
	// "schoolaid-go".
	ClientID string

	// Topics override the default Laravel topic names. Empty values fall
	// back to defaults defined in topics.go.
	Topics Topics

	// Async controls whether writes are fire-and-forget. Mirrors the
	// Laravel side which dispatches a queued job. Defaults to true.
	Async bool

	// BatchTimeout maps to Kafka's linger.ms. Defaults to 20ms to match
	// the Laravel KAFKA_LINGER_MS default.
	BatchTimeout time.Duration

	// BatchSize maps to batch.num.messages. Defaults to 10000 to match
	// the Laravel KAFKA_BATCH_NUM_MESSAGES default.
	BatchSize int

	// RequireAll, when true, sets acks=all. Defaults to true.
	RequireAll bool

	// WriteTimeout caps how long a synchronous write may block.
	// Defaults to 10s.
	WriteTimeout time.Duration

	// Enabled lets a service ship the wrapper but keep production traffic
	// on the legacy path until a flag flip. When false, all Publish calls
	// return nil without contacting Kafka.
	Enabled bool
}

Config controls how the producer connects to Kafka and which topics it resolves for each high-level Publish helper. All fields are optional except Brokers; topic fields fall back to the Laravel defaults so a Go service can publish to the same destinations a Laravel producer would.

func ConfigFromEnv

func ConfigFromEnv() Config

ConfigFromEnv reads the same KAFKA_* environment variables Laravel uses, so the same deployment manifest can drive both sides of the cutover.

KAFKA_ENABLED                    -> Enabled (bool, default false)
KAFKA_BROKERS                    -> Brokers (default "localhost:9092")
KAFKA_CLIENT_ID                  -> ClientID (default "schoolaid-go")
KAFKA_TOPIC_NOTE_CREATED         -> Topics.NoteCreated
KAFKA_TOPIC_NOTE_CREATED_PRIORITY-> Topics.NoteCreatedPriority
KAFKA_TOPIC_EMAIL_BATCH          -> Topics.EmailBatch
KAFKA_TOPIC_PUSH_BATCH           -> Topics.PushBatch
KAFKA_TOPIC_SMS_BATCH            -> Topics.SMSBatch
KAFKA_TOPIC_WHATSAPP_BATCH       -> Topics.WhatsAppBatch
KAFKA_TOPIC_DLQ                  -> Topics.DLQ
KAFKA_TOPIC_RETRY                -> Topics.Retry

type EmailAttachment

type EmailAttachment struct {
	Filename string  `json:"filename"`
	URL      *string `json:"url"`
}

EmailAttachment is the trimmed shape attached to email batch payloads.

type EmailMessage

type EmailMessage struct {
	EventID       string            `json:"event_id"`
	TraceID       string            `json:"trace_id"`
	NoteID        *int64            `json:"note_id"`
	SchoolID      int64             `json:"school_id"`
	StudentID     *int64            `json:"student_id"`
	UserID        int64             `json:"user_id"`
	Email         string            `json:"email"`
	FromName      *string           `json:"from_name"`
	Subject       string            `json:"subject"`
	Content       string            `json:"content"`
	FeaturedImage *string           `json:"featured_image"`
	Signature     *string           `json:"signature"`
	Attachments   []EmailAttachment `json:"attachments"`
	Priority      Priority          `json:"priority"`
}

EmailMessage is the wire format of one notifications.email.batch entry. Matches NoteMessage::additionalKafkaMessages() payload shape.

type Metadata

type Metadata struct {
	TraceID     string  `json:"trace_id"`
	UserID      *int64  `json:"user_id"`
	CreatedAt   string  `json:"created_at"`
	ScheduledAt *string `json:"scheduled_at"`
}

Metadata mirrors NoteMessage::toKafka()['metadata'].

type Note

type Note struct {
	NoteID         int64     `json:"note_id"`
	SchoolID       int64     `json:"school_id"`
	Title          string    `json:"title"`
	Subject        string    `json:"subject"`
	FromName       *string   `json:"from_name"`
	PushTitle      PushTitle `json:"push_title"`
	Content        string    `json:"content"`
	Important      bool      `json:"important"`
	FeaturedImage  *string   `json:"featured_image"`
	Signature      *string   `json:"signature"`
	Channels       []string  `json:"channels"`
	RelativeTypeID *int64    `json:"relative_type_id"`
}

Note is the note-level body shared across recipients in a NoteCreated event. Matches NoteMessage::buildCachedNotePayload().

type NoteCreated

type NoteCreated struct {
	Version     string       `json:"version"`
	EventType   string       `json:"event_type"`
	EventID     string       `json:"event_id"`
	Timestamp   string       `json:"timestamp"`
	Note        Note         `json:"note"`
	Recipient   Recipient    `json:"recipient"`
	Attachments []Attachment `json:"attachments"`
	Metadata    Metadata     `json:"metadata"`
}

NoteCreated is the full notifications.note.created[.priority] payload.

type Priority

type Priority string

Priority is the value carried by the "priority" Kafka header and by the priority field embedded in payloads.

const (
	PriorityNormal Priority = "normal"
	PriorityHigh   Priority = "high"
)

type Producer

type Producer interface {
	Produce(ctx context.Context, topic string, key string, payload any, headers map[string]string) error
	Close() error
}

Producer is the minimum surface needed to publish a raw Kafka message. Defining it as an interface lets callers stub the wrapper in tests without spinning up a broker.

type PushMessage

type PushMessage struct {
	EventID   string         `json:"event_id"`
	TraceID   string         `json:"trace_id"`
	SchoolID  int64          `json:"school_id"`
	StudentID *int64         `json:"student_id"`
	UserID    int64          `json:"user_id"`
	DeviceIDs []string       `json:"device_ids"`
	TitleES   string         `json:"title_es"`
	TitleEN   string         `json:"title_en"`
	BodyES    string         `json:"body_es"`
	BodyEN    string         `json:"body_en"`
	ImageURL  *string        `json:"image_url"`
	Data      map[string]any `json:"data"`
	Priority  Priority       `json:"priority"`
}

PushMessage is one entry on notifications.push.batch.

type PushTitle

type PushTitle struct {
	ES *string `json:"es"`
	EN *string `json:"en"`
}

PushTitle holds localized push notification titles.

type Recipient

type Recipient map[string]any

Recipient is the per-student block produced by the Laravel kafka resolver. Carried verbatim through the wrapper; consumers know the shape.

type SMSMessage

type SMSMessage struct {
	EventID   string   `json:"event_id"`
	TraceID   string   `json:"trace_id"`
	SchoolID  int64    `json:"school_id"`
	StudentID *int64   `json:"student_id"`
	UserID    int64    `json:"user_id"`
	Phone     string   `json:"phone"`
	Body      string   `json:"body"`
	Priority  Priority `json:"priority"`
}

SMSMessage is one entry on notifications.sms.batch.

type Topics

type Topics struct {
	NoteCreated         string
	NoteCreatedPriority string
	EmailBatch          string
	PushBatch           string
	SMSBatch            string
	WhatsAppBatch       string
	DLQ                 string
	Retry               string
}

Topics is the set of Kafka topics the wrapper knows how to publish to. Defaults match the values in schoolaid-admin's config/kafka.php so a Go service can drop in alongside Laravel without re-wiring topic names.

type WhatsAppMessage

type WhatsAppMessage struct {
	EventID   string   `json:"event_id"`
	TraceID   string   `json:"trace_id"`
	SchoolID  int64    `json:"school_id"`
	StudentID *int64   `json:"student_id"`
	UserID    int64    `json:"user_id"`
	Phone     string   `json:"phone"`
	Text      string   `json:"text"`
	Priority  Priority `json:"priority"`
}

WhatsAppMessage is one entry on notifications.whatsapp.batch.