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.PublishPush(ctx, notifications.PushMessage{
    NoteID:   1001,
    SchoolID: 42,
    UserID:   99,
    Title:    "Nueva circular: Example",
    Devices:  []string{"fcm-token-a", "fcm-token-b"},
    Data:     map[string]string{"item_id": "1001", "notification_type": "note"},
})

Schemas mirror the consumer at github.com/schoolaid/notifications (internal/models/event.go + batch.go). Producers MUST match — mismatched fields land in the DLQ.

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 EmailUser
• type EventMetadata
• type Note
• type NoteCreated
• type Priority
• type Producer
• type PushMessage
• type PushUser
• 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"`
	URL      string `json:"url"`
}

Attachment is the note-level attachment shape carried inside NoteCreated. Consumer: notifications/internal/models/event.go

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 the Laravel 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. Consumer: notifications/internal/models/batch.go

type EmailMessage

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

EmailMessage is one entry on notifications.email.batch. Consumer model: EmailBatchMessage in batch.go.

type EmailUser

type EmailUser struct {
	UserID int    `json:"user_id"`
	Email  string `json:"email"`
}

EmailUser is one recipient on the NoteCreated.Recipient.EmailUsers list.

type EventMetadata

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

EventMetadata carries tracing info on every NoteCreated event.

type Note

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

Note holds the note-level body for a NoteCreated event. PushTitle and PushBody keys are language codes ("es", "en"); the consumer's Note.PushTitleFor / PushBodyFor fall back to "es" then to Title (for PushTitle) or empty string (for PushBody).

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    EventMetadata `json:"metadata"`
}

NoteCreated is the full notifications.note.created[.priority] payload. The fan-out consumer reads this and emits PushBatch + EmailBatch messages.

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"`
	NoteID    int               `json:"note_id"`
	SchoolID  int               `json:"school_id"`
	StudentID int               `json:"student_id"`
	UserID    int               `json:"user_id"`
	Title     string            `json:"title"`
	Body      string            `json:"body,omitempty"`
	ImageURL  *string           `json:"image_url,omitempty"`
	Priority  Priority          `json:"priority"`
	Devices   []string          `json:"devices"`
	Data      map[string]string `json:"data"`
}

PushMessage is one entry on notifications.push.batch. Consumer model: PushBatchMessage in batch.go.

type PushUser

type PushUser struct {
	UserID   int      `json:"user_id"`
	Language string   `json:"language"`
	Devices  []string `json:"devices"`
}

PushUser is one recipient on the NoteCreated.Recipient.Users list.

type Recipient

type Recipient struct {
	StudentID  int         `json:"student_id"`
	Users      []PushUser  `json:"users"`
	EmailUsers []EmailUser `json:"email_users,omitempty"`
}

Recipient bundles the per-student recipient set for a NoteCreated event.

type SMSMessage

type SMSMessage struct {
	EventID   string   `json:"event_id"`
	TraceID   string   `json:"trace_id"`
	NoteID    int      `json:"note_id,omitempty"`
	SchoolID  int      `json:"school_id"`
	StudentID int      `json:"student_id,omitempty"`
	UserID    int      `json:"user_id"`
	Phone     string   `json:"phone"`
	Body      string   `json:"body"`
	Priority  Priority `json:"priority"`
}

SMSMessage is one entry on notifications.sms.batch. The consumer does not yet handle this topic; field shape mirrors the other batch types so it can be added later without a wire break.

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"`
	NoteID    int      `json:"note_id,omitempty"`
	SchoolID  int      `json:"school_id"`
	StudentID int      `json:"student_id,omitempty"`
	UserID    int      `json:"user_id"`
	Phone     string   `json:"phone"`
	Text      string   `json:"text"`
	Priority  Priority `json:"priority"`
}

WhatsAppMessage is one entry on notifications.whatsapp.batch. As with SMSMessage, the consumer does not yet handle this topic.