Internal Structure of Listmonk's Bounce Manager: Architecture and Data Flow

Listmonk's bounce manager uses a modular, queue-based architecture to ingest delivery failure events from mailboxes (via POP) or webhook providers (SES, SendGrid, Postmark, etc.), processing them through a buffered channel and persisting to the database via a configurable callback.

The bounce manager in the knadh/listmonk repository orchestrates the collection and storage of email bounce events. Designed with a pluggable pipeline, it separates ingestion sources from persistence logic, allowing operators to mix mailbox scanning and webhook endpoints without modifying core processing code.

Core Architectural Components

The Manager Struct

The central orchestrator is the Manager struct defined in internal/bounce/bounce.go (lines 46–58). It holds:

• A bounded channel (queue chan models.Bounce) that buffers bounce events before processing
• A concrete mailbox client implementing the Mailbox interface
• Pointers to webhook clients for each enabled provider
• The configuration (Opt)
• Database query helpers (Queries)
• A logger instance

This struct coordinates the flow between external data sources and the database.

Configuration Options

The Opt struct (lines 20–44) controls which ingestion methods are active:

• MailboxEnabled and MailboxType toggle POP/IMAP scanning
• WebhooksEnabled activates HTTP endpoints
• RecordBounceCB is a critical function field that receives a models.Bounce and returns error, acting as the bridge to persistence logic

The Mailbox Interface

Bounce sources implement the Mailbox interface (lines 14–18):

type Mailbox interface {
    Scan(limit int, ch chan models.Bounce) error
}

Currently, only POP is implemented in internal/bounce/mailbox/pop.go. The Scan method reads messages, parses them into models.Bounce structs, and pushes them onto the supplied channel.

Data Ingestion Sources

POP/IMAP Mailbox Scanning

When MailboxEnabled is true, the manager spawns a background goroutine running runMailboxScanner (lines 134–144). This loop:

1. Calls m.mailbox.Scan(1000, m.queue) to fetch up to 1000 messages
2. Pushes parsed bounces onto the manager's internal queue
3. Sleeps for the configured ScanInterval before repeating

Errors during scanning are logged, but the loop continues indefinitely.

Webhook Providers

Provider-specific logic resides in internal/bounce/webhooks/. Each implements an HTTP handler that converts proprietary JSON payloads into models.Bounce objects:

• SES: webhooks.NewSES() parses Amazon SNS notifications
• SendGrid: webhooks.NewSendgrid(key) validates secrets and unmarshals JSON
• Postmark: webhooks.NewPostmark(username, password) handles basic authentication
• ForwardEmail: webhooks.NewForwardemail(key) verifies signatures
• Lettermint: webhooks.NewLettermint(key) validates provider-specific tokens

These handlers write directly to the manager's queue channel, unifying webhook and mailbox data streams.

Initialization and Runtime Flow

The Constructor

The New function (lines 66–113) wires the system:

1. Creates a buffered channel for the bounce queue
2. Instantiates pop.NewPOP if mailbox scanning is enabled
3. Initializes webhook clients only for configured providers
4. Returns the fully wired Manager or an error for unknown mailbox types

The Main Processing Loop

The Run method (lines 115–132) starts the system:

func (m *Manager) Run() {
    if m.opt.MailboxEnabled {
        go m.runMailboxScanner()
    }
    for b := range m.queue {
        if b.CreatedAt.IsZero() {
            b.CreatedAt = time.Now()
        }
        _ = m.opt.RecordBounceCB(b)
    }
}

If mailbox scanning is enabled, it launches a goroutine. The main goroutine then blocks on m.queue, applying default timestamps and invoking RecordBounceCB for each bounce event. This design ensures that slow database writes do not block fast webhook deliveries or mailbox scanning.

Database Persistence

The actual SQL insertion is decoupled from the bounce manager. The RecordBounceCB callback is typically implemented in internal/core/bounces.go, which uses the prepared statement queries.RecordQuery to insert into the bounces table.

The Queries struct (lines 60–64) wraps a *sqlx.DB connection:

type Queries struct {
    *sqlx.DB
    RecordQuery *sqlx.Stmt
}

This separation allows the bounce manager to be tested without a database, and lets persistence logic evolve independently.

Practical Implementation Examples

Creating a Manager with POP Mailbox

import (
    "log"
    "os"
    "time"

    "github.com/jmoiron/sqlx"
    "github.com/knadh/listmonk/internal/bounce"
    "github.com/knadh/listmonk/models"
)

// Prepare database connection and statement
queries := &bounce.Queries{
    DB:          db,
    RecordQuery: db.MustPreparex(`INSERT INTO bounces 
        (subscriber_id, email, reason, created_at) 
        VALUES (?, ?, ?, ?)`),
}

// Define persistence callback
recordCB := func(b models.Bounce) error {
    _, err := queries.RecordQuery.Exec(
        b.SubscriberID, b.Email, b.Reason, b.CreatedAt,
    )
    return err
}

// Configure options
opt := bounce.Opt{
    MailboxEnabled: true,
    MailboxType:    "pop",
    Mailbox:        mailbox.Opt{
        Host:         "mail.example.com",
        Username:     "bounceuser",
        Password:     "secret",
        ScanInterval: time.Minute * 5,
    },
    RecordBounceCB: recordCB,
}

// Initialize and start
mgr, err := bounce.New(opt, queries, log.New(os.Stderr, "bounce: ", log.LstdFlags))
if err != nil {
    log.Fatal(err)
}
go mgr.Run()

Recording a Bounce Manually

For testing or manual insertion, push directly to the queue:

b := models.Bounce{
    SubscriberID: 42,
    Email:        "bounced@example.com",
    Reason:       "hard-bounce",
    CreatedAt:    time.Now(),
}
_ = mgr.Record(b) // Enqueues for background processing

Registering a SendGrid Webhook Handler

http.HandleFunc("/webhooks/sendgrid", func(w http.ResponseWriter, r *http.Request) {
    if err := mgr.Sendgrid.Handle(r, mgr.queue); err != nil {
        http.Error(w, "invalid payload", http.StatusBadRequest)
        return
    }
    w.WriteHeader(http.StatusOK)
})

Summary

• The Manager struct in internal/bounce/bounce.go orchestrates ingestion via a buffered channel that decouples receiving from persistence.
• Mailbox implementations (currently POP) poll mailboxes continuously in a background goroutine, pushing parsed bounces to the queue.
• Webhook clients for SES, SendGrid, Postmark, ForwardEmail, and Lettermint translate provider-specific payloads into standard models.Bounce structs.
• The RecordBounceCB callback decouples the manager from database logic, allowing the core bounce processing to remain provider-agnostic.
• Initialization happens through bounce.New(), which wires enabled sources based on the Opt configuration struct.

Frequently Asked Questions

How does the bounce manager handle concurrent inputs from webhooks and mailboxes?

The manager uses a single buffered channel (m.queue) as the concurrency control point. Both the mailbox scanner goroutine and multiple webhook HTTP handlers write to this channel concurrently. The main Run() loop reads from the channel sequentially, ensuring that database writes happen in a controlled manner while ingress operations remain non-blocking up to the channel buffer size.

Can I extend the manager to support IMAP mailboxes?

Yes. Create a new file in internal/bounce/mailbox/ implementing the Mailbox interface with a Scan(limit int, ch chan models.Bounce) error method. Update bounce.New() to instantiate your IMAP client when opt.MailboxType == "imap". The interface abstracts all connection details, so the manager requires no other changes.

Where is the bounce data actually stored?

The manager delegates persistence to the RecordBounceCB callback defined in Opt. By default, this callback is implemented in internal/core/bounces.go, which executes a prepared SQL statement against the bounces table using the Queries.RecordQuery statement. This design keeps storage logic separate from the ingestion pipeline.

What happens if the database is slow or temporarily unavailable?

The buffered m.queue channel provides backpressure. If the database slows down, the channel fills up. Once the buffer is full, webhook handlers and the mailbox scanner will block on send operations until the Run() loop drains the queue. This prevents memory exhaustion during database slowdowns while maintaining data integrity.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}