EagleCast/internal/bounce/bounce.go

package bounce

import (
	"errors"
	"log"
	"time"

	"github.com/jmoiron/sqlx"
	"source.offmarket.win/aleagle/eaglecast/internal/bounce/mailbox"
	"source.offmarket.win/aleagle/eaglecast/internal/bounce/webhooks"
	"source.offmarket.win/aleagle/eaglecast/models"
)

// Mailbox represents a POP/IMAP mailbox client that can scan messages and pass
// them to a given channel.
type Mailbox interface {
	Scan(limit int, ch chan models.Bounce) error
}

// Opt represents bounce processing options.
type Opt struct {
	MailboxEnabled          bool        `json:"mailbox_enabled"`
	MailboxType             string      `json:"mailbox_type"`
	Mailbox                 mailbox.Opt `json:"mailbox"`
	WebhooksEnabled         bool        `json:"webhooks_enabled"`
	SESEnabled              bool        `json:"ses_enabled"`
	AzureEnabled            bool        `json:"azure_enabled"`
	AzureSharedSecret       string      `json:"azure_shared_secret"`
	AzureSharedSecretHeader string      `json:"azure_shared_secret_header"`
	SendgridEnabled         bool        `json:"sendgrid_enabled"`
	SendgridKey             string      `json:"sendgrid_key"`
	Postmark                struct {
		Enabled  bool
		Username string
		Password string
	}
	ForwardEmail struct {
		Enabled bool
		Key     string
	}
	Lettermint struct {
		Enabled bool
		Key     string
	}

	RecordBounceCB func(models.Bounce) error
}

// Manager handles e-mail bounces.
type Manager struct {
	queue        chan models.Bounce
	mailbox      Mailbox
	SES          *webhooks.SES
	Azure        *webhooks.Azure
	Sendgrid     *webhooks.Sendgrid
	Postmark     *webhooks.Postmark
	Forwardemail *webhooks.Forwardemail
	Lettermint   *webhooks.Lettermint
	queries      *Queries
	opt          Opt
	log          *log.Logger
}

// Queries contains the queries.
type Queries struct {
	DB          *sqlx.DB
	RecordQuery *sqlx.Stmt
}

// New returns a new instance of the bounce manager.
func New(opt Opt, q *Queries, lo *log.Logger) (*Manager, error) {
	m := &Manager{
		opt:     opt,
		queries: q,
		queue:   make(chan models.Bounce, 1000),
		log:     lo,
	}

	// Is there a mailbox?
	if opt.MailboxEnabled {
		switch opt.MailboxType {
		case "pop":
			m.mailbox = mailbox.NewPOP(opt.Mailbox, lo)
		default:
			return nil, errors.New("unknown bounce mailbox type")
		}
	}

	if opt.WebhooksEnabled {
		if opt.SESEnabled {
			m.SES = webhooks.NewSES()
		}
		if opt.AzureEnabled {
			m.Azure = webhooks.NewAzure(opt.AzureSharedSecret, opt.AzureSharedSecretHeader)
		}

		if opt.SendgridEnabled {
			sg, err := webhooks.NewSendgrid(opt.SendgridKey)
			if err != nil {
				lo.Printf("error initializing sendgrid webhooks: %v", err)
			} else {
				m.Sendgrid = sg
			}
		}

		if opt.Postmark.Enabled {
			m.Postmark = webhooks.NewPostmark(opt.Postmark.Username, opt.Postmark.Password)
		}

		if opt.ForwardEmail.Enabled {
			fe := webhooks.NewForwardemail([]byte(opt.ForwardEmail.Key))
			m.Forwardemail = fe
		}

		if opt.Lettermint.Enabled {
			m.Lettermint = webhooks.NewLettermint([]byte(opt.Lettermint.Key))
		}
	}

	return m, nil
}

// Run is a blocking function that listens for bounce events from webhooks and or mailboxes
// and executes them on the DB.
func (m *Manager) Run() {
	if m.opt.MailboxEnabled {
		go m.runMailboxScanner()
	}

	for b := range m.queue {
		if b.CreatedAt.IsZero() {
			b.CreatedAt = time.Now()
		}

		if err := m.opt.RecordBounceCB(b); err != nil {
			continue
		}
	}
}

// runMailboxScanner runs a blocking loop that scans the mailbox at given intervals.
func (m *Manager) runMailboxScanner() {
	for {
		m.log.Printf("scanning bounce mailbox %s", m.opt.Mailbox.Host)
		if err := m.mailbox.Scan(1000, m.queue); err != nil {
			m.log.Printf("error scanning bounce mailbox: %v", err)
		}

		time.Sleep(m.opt.Mailbox.ScanInterval)
	}
}

// Record records a new bounce event given the subscriber's email or UUID.
func (m *Manager) Record(b models.Bounce) error {
	m.queue <- b
	return nil
}