Custom MySQL Binlog Parsers: Building a Real-Time Outbox Pattern Processor in Go — code snippets

snippet-1.sql

CREATE TABLE IF NOT EXISTS outbox_events (
    id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSON NOT NULL,
    created_at TIMESTAMP(6) DEFAULT CURRENT_TIMESTAMP(6),
    INDEX idx_created_at (created_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

snippet-2.go

package main

import (
	"context"
	"fmt"
	"os"

	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/mysql"
)

type OutboxSubscriber struct {
	canalInstance *canal.Canal
	dbName        string
	tableName     string
}

func NewOutboxSubscriber(cfg *canal.Config, dbName, tableName string) (*OutboxSubscriber, error) {
	c, err := canal.NewCanal(cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to initialize canal: %w", err)
	}

	sub := &OutboxSubscriber{
		canalInstance: c,
		dbName:        dbName,
		tableName:     tableName,
	}

	// Register the custom rows event handler
	c.SetEventHandler(&OutboxHandler{
		sub: sub,
	})

	return sub, nil
}

func (s *OutboxSubscriber) Start(ctx context.Context, gtidSet mysql.GTIDSet) error {
	if gtidSet != nil {
		return s.canalInstance.StartFromGTID(gtidSet)
	}
	return s.canalInstance.Start()
}

func (s *OutboxSubscriber) Close() {
	s.canalInstance.Close()
}

snippet-3.go

package main

import (
	"encoding/json"
	"fmt"

	"github.com/go-mysql-org/go-mysql/canal"
	"github.com/go-mysql-org/go-mysql/schema"
)

type OutboxEvent struct {
	ID            uint64          `json:"id"`
	AggregateType string          `json:"aggregate_type"`
	AggregateId   string          `json:"aggregate_id"`
	EventType     string          `json:"event_type"`
	Payload       json.RawMessage `json:"payload"`
	CreatedAt     string          `json:"created_at"`
}

type OutboxHandler struct {
	canal.DummyEventHandler
	sub *OutboxSubscriber
}

func (h *OutboxHandler) OnRow(e *canal.RowsEvent) error {
	// Only process inserts in our target database and table
	if e.Table.Schema != h.sub.dbName || e.Table.Name != h.sub.tableName {
		return nil
	}

	// We only care about INSERT events (WRITE_ROWS_EVENT)
	if e.Action != canal.InsertAction {
		return nil
	}

	for _, row := range e.Rows {
		event, err := h.parseRow(e.Table, row)
		if err != nil {
			return fmt.Errorf("error parsing row event: %w", err)
		}
		
		h.dispatch(event)
	}

	return nil
}

func (h *OutboxHandler) parseRow(table *schema.Table, row []interface{}) (*OutboxEvent, error) {
	event := &OutboxEvent{}
	
	for colIndex, value := range row {
		if value == nil {
			continue
		}

		columnName := table.Columns[colIndex].Name
		switch columnName {
		case "id":
			val, ok := value.(int64)
			if !ok {
				// Fallback for unsigned bigint conversion variations
				uVal, ok := value.(uint64)
				if !ok {
					return nil, fmt.Errorf("failed to cast id column, type is %T", value)
				}
				event.ID = uVal
			} else {
				event.ID = uint64(val)
			}
		case "aggregate_type":
			val, ok := value.(string)
			if !ok {
				valBytes, ok := value.([]byte)
				if !ok {
					return nil, fmt.Errorf("failed to cast aggregate_type column")
				}
				event.AggregateType = string(valBytes)
			} else {
				event.AggregateType = val
			}
		case "aggregate_id":
			val, ok := value.(string)
			if !ok {
				valBytes, ok := value.([]byte)
				if !ok {
					return nil, fmt.Errorf("failed to cast aggregate_id column")
				}
				event.AggregateId = string(valBytes)
			} else {
				event.AggregateId = val
			}
		case "event_type":
			val, ok := value.(string)
			if !ok {
				valBytes, ok := value.([]byte)
				if !ok {
					return nil, fmt.Errorf("failed to cast event_type column")
				}
				event.EventType = string(valBytes)
			} else {
				event.EventType = val
			}
		case "payload":
			// MySQL JSON columns are returned as []byte or strings
			switch v := value.(type) {
			case []byte:
				event.Payload = json.RawMessage(v)
			case string:
				event.Payload = json.RawMessage(v)
			default:
				return nil, fmt.Errorf("unexpected type for payload column: %T", value)
			}
		case "created_at":
			event.CreatedAt = fmt.Sprintf("%v", value)
		}
	}

	return event, nil
}

func (h *OutboxHandler) dispatch(event *OutboxEvent) {
	// Processing placeholder: in a production pipeline, this hands off to a worker pool
	fmt.Printf("Parsed Event ID %d for aggregate %s/%s\n", event.ID, event.AggregateType, event.AggregateId)
}

snippet-4.go

package main

import (
	"context"
	"hash/fnv"
	"sync"
)

type Dispatcher struct {
	workers    []chan *OutboxEvent
	numWorkers int
	wg         sync.WaitGroup
}

func NewDispatcher(numWorkers int, queueSize int) *Dispatcher {
	workers := make([]chan *OutboxEvent, numWorkers)
	for i := 0; i < numWorkers; i++ {
		workers[i] = make(chan *OutboxEvent, queueSize)
	}
	return &Dispatcher{
		workers:    workers,
		numWorkers: numWorkers,
	}
}

func (d *Dispatcher) Start(ctx context.Context, handler func(ctx context.Context, event *OutboxEvent)) {
	for i := 0; i < d.numWorkers; i++ {
		d.wg.Add(1)
		go func(workerId int, ch chan *OutboxEvent) {
			defer d.wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case event, ok := <-ch:
					if !ok {
						return
					}
					handler(ctx, event)
				}
			}
		}(i, d.workers[i])
	}
}

func (d *Dispatcher) Submit(event *OutboxEvent) {
	// Hash the aggregate ID to pin events for a single aggregate to a single worker channel
	h := fnv.New32a()
	h.Write([]byte(event.AggregateId))
	workerIndex := int(h.Sum32() % uint32(d.numWorkers))
	
	d.workers[workerIndex] <- event
}

func (d *Dispatcher) Stop() {
	for _, ch := range d.workers {
		close(ch)
	}
	d.wg.Wait()
}

snippet-5.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/go-mysql-org/go-mysql/mysql"
)

type CheckpointStore interface {
	Save(ctx context.Context, key string, gtidSet string) error
	Load(ctx context.Context, key string) (string, error)
}

type Checkpointer struct {
	store CheckpointStore
	key   string
}

func NewCheckpointer(store CheckpointStore, key string) *Checkpointer {
	return &Checkpointer{
		store: store,
		key:   key,
	}
}

func (c *Checkpointer) SaveCheckpoint(ctx context.Context, gtidSet mysql.GTIDSet) error {
	return c.store.Save(ctx, c.key, gtidSet.String())
}

func (c *Checkpointer) LoadCheckpoint(ctx context.Context) (mysql.GTIDSet, error) {
	raw, err := c.store.Load(ctx, c.key)
	if err != nil {
		return nil, fmt.Errorf("failed to load checkpoint from store: %w", err)
	}
	if raw == "" {
		return nil, nil
	}

	gtidSet, err := mysql.ParseGTIDSet("mysql", raw)
	if err != nil {
		return nil, fmt.Errorf("failed to parse loaded GTID set: %w", err)
	}

	return gtidSet, nil
}

// Example usage hook for OnGTID event inside canal handler
type AdvancedHandler struct {
	canal.DummyEventHandler
	checkpointer *Checkpointer
	ctx          context.Context
}

func (h *AdvancedHandler) OnGTID(gtidSet mysql.GTIDSet) error {
	// Call this periodically or on every GTID block to persist progress
	ctx, cancel := context.WithTimeout(h.ctx, 2*time.Second)
	defer cancel()
	
	if err := h.checkpointer.SaveCheckpoint(ctx, gtidSet); err != nil {
		// Log error but do not crash; replication can continue unless safety rules mandate hard stop
		fmt.Fprintf(os.Stderr, "failed to save checkpoint: %v\n", err)
	}
	return nil
}

snippet-6.go

func (h *OutboxHandler) OnRowWithSafeValidation(e *canal.RowsEvent) error {
	// Inspect the schema to guarantee expected columns exist
	requiredColumns := map[string]bool{
		"id":             false,
		"aggregate_type": false,
		"aggregate_id":   false,
		"event_type":     false,
		"payload":        false,
	}

	for _, col := range e.Table.Columns {
		if _, ok := requiredColumns[col.Name]; ok {
			requiredColumns[col.Name] = true
		}
	}

	for colName, found := range requiredColumns {
		if !found {
			// Trigger a critical alert - schema drift has broken the contract
			return fmt.Errorf("schema violation: required column %s not found in current table structure", colName)
		}
	}

	return h.OnRow(e)
}

snippet-7.go

func ConfigureHeartbeats(cfg *canal.Config) {
	// Request replication stream heartbeats every 10 seconds during idle periods
	cfg.HeartbeatPeriod = 10 * time.Second
	// Ensure ReadTimeout is set higher than the heartbeat period to prevent false drops
	cfg.ReadTimeout = 30 * time.Second
}