diff --git a/admin.go b/admin.go index 9923c73..5d19290 100644 --- a/admin.go +++ b/admin.go @@ -30,7 +30,7 @@ func handleGetConfigScript(c echo.Context) error { RootURL: app.Constants.RootURL, UploadURI: app.Constants.UploadURI, FromEmail: app.Constants.FromEmail, - Messengers: app.Runner.GetMessengerNames(), + Messengers: app.Manager.GetMessengerNames(), } b = bytes.Buffer{} diff --git a/campaigns.go b/campaigns.go index 980716f..be4f7dc 100644 --- a/campaigns.go +++ b/campaigns.go @@ -127,13 +127,13 @@ func handlePreviewCampaign(c echo.Context) error { camp.Body = body } - if err := camp.CompileTemplate(app.Runner.TemplateFuncs(camp)); err != nil { + if err := camp.CompileTemplate(app.Manager.TemplateFuncs(camp)); err != nil { return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Error compiling template: %v", err)) } // Render the message body. - m := app.Runner.NewMessage(camp, &sub) + m := app.Manager.NewMessage(camp, &sub) if err := m.Render(); err != nil { return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Error rendering message: %v", err)) @@ -159,7 +159,7 @@ func handleCreateCampaign(c echo.Context) error { return echo.NewHTTPError(http.StatusBadRequest, err.Error()) } - if !app.Runner.HasMessenger(o.MessengerID) { + if !app.Manager.HasMessenger(o.MessengerID) { return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Unknown messenger %s", o.MessengerID)) } @@ -466,12 +466,12 @@ func handleTestCampaign(c echo.Context) error { // sendTestMessage takes a campaign and a subsriber and sends out a sample campain message. func sendTestMessage(sub *models.Subscriber, camp *models.Campaign, app *App) error { - if err := camp.CompileTemplate(app.Runner.TemplateFuncs(camp)); err != nil { + if err := camp.CompileTemplate(app.Manager.TemplateFuncs(camp)); err != nil { return fmt.Errorf("Error compiling template: %v", err) } // Render the message body. - m := app.Runner.NewMessage(camp, sub) + m := app.Manager.NewMessage(camp, sub) if err := m.Render(); err != nil { return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Error rendering message: %v", err)) @@ -511,7 +511,7 @@ func validateCampaignFields(c campaignReq, app *App) error { } camp := models.Campaign{Body: c.Body, TemplateBody: tplTag} - if err := c.CompileTemplate(app.Runner.TemplateFuncs(&camp)); err != nil { + if err := c.CompileTemplate(app.Manager.TemplateFuncs(&camp)); err != nil { return fmt.Errorf("Error compiling campaign body: %v", err) } diff --git a/main.go b/main.go index 14cb428..719b751 100644 --- a/main.go +++ b/main.go @@ -12,8 +12,8 @@ import ( _ "github.com/jinzhu/gorm/dialects/postgres" "github.com/jmoiron/sqlx" "github.com/knadh/goyesql" + "github.com/knadh/listmonk/manager" "github.com/knadh/listmonk/messenger" - "github.com/knadh/listmonk/runner" "github.com/knadh/listmonk/subimporter" "github.com/labstack/echo" flag "github.com/spf13/pflag" @@ -36,7 +36,7 @@ type App struct { DB *sqlx.DB Queries *Queries Importer *subimporter.Importer - Runner *runner.Runner + Manager *manager.Manager Logger *log.Logger NotifTpls *template.Template Messenger messenger.Messenger @@ -158,7 +158,7 @@ func registerHandlers(e *echo.Echo) { } // initMessengers initializes various messaging backends. -func initMessengers(r *runner.Runner) messenger.Messenger { +func initMessengers(r *manager.Manager) messenger.Messenger { // Load SMTP configurations for the default e-mail Messenger. var srv []messenger.Server for name := range viper.GetStringMapString("smtp") { @@ -251,7 +251,7 @@ func main() { campNotifCB := func(subject string, data map[string]interface{}) error { return sendNotification(notifTplCampaign, subject, data, app) } - r := runner.New(runner.Config{ + m := manager.New(manager.Config{ Concurrency: viper.GetInt("app.concurrency"), MaxSendErrors: viper.GetInt("app.max_send_errors"), FromEmail: app.Constants.FromEmail, @@ -264,14 +264,14 @@ func main() { // url.com/campaign/{campaign_uuid}/{subscriber_uuid}/px.png ViewTrackURL: fmt.Sprintf("%s/campaign/%%s/%%s/px.png", app.Constants.RootURL), - }, newRunnerDB(q), campNotifCB, logger) - app.Runner = r + }, newManagerDB(q), campNotifCB, logger) + app.Manager = m // Add messengers. - app.Messenger = initMessengers(app.Runner) + app.Messenger = initMessengers(app.Manager) - go r.Run(time.Duration(time.Second * 5)) - r.SpawnWorkers() + go m.Run(time.Duration(time.Second * 5)) + m.SpawnWorkers() // Initialize the server. var srv = echo.New() diff --git a/runner/runner.go b/manager/manager.go similarity index 64% rename from runner/runner.go rename to manager/manager.go index e88b0c8..6943ee2 100644 --- a/runner/runner.go +++ b/manager/manager.go @@ -1,4 +1,4 @@ -package runner +package manager import ( "bytes" @@ -33,9 +33,9 @@ type DataSource interface { CreateLink(url string) (string, error) } -// Runner handles the scheduling, processing, and queuing of campaigns +// Manager handles the scheduling, processing, and queuing of campaigns // and message pushes. -type Runner struct { +type Manager struct { cfg Config src DataSource messengers map[string]messenger.Messenger @@ -67,7 +67,7 @@ type Message struct { to string } -// Config has parameters for configuring the runner. +// Config has parameters for configuring the manager. type Config struct { Concurrency int MaxSendErrors int @@ -84,8 +84,8 @@ type msgError struct { } // New returns a new instance of Mailer. -func New(cfg Config, src DataSource, notifCB models.AdminNotifCallback, l *log.Logger) *Runner { - r := Runner{ +func New(cfg Config, src DataSource, notifCB models.AdminNotifCallback, l *log.Logger) *Manager { + return &Manager{ cfg: cfg, src: src, notifCB: notifCB, @@ -98,37 +98,35 @@ func New(cfg Config, src DataSource, notifCB models.AdminNotifCallback, l *log.L msgErrorQueue: make(chan msgError, cfg.MaxSendErrors), msgErrorCounts: make(map[int]int), } - - return &r } // NewMessage creates and returns a Message that is made available // to message templates while they're compiled. -func (r *Runner) NewMessage(c *models.Campaign, s *models.Subscriber) *Message { +func (m *Manager) NewMessage(c *models.Campaign, s *models.Subscriber) *Message { return &Message{ from: c.FromEmail, to: s.Email, Campaign: c, Subscriber: s, - UnsubscribeURL: fmt.Sprintf(r.cfg.UnsubscribeURL, c.UUID, s.UUID), + UnsubscribeURL: fmt.Sprintf(m.cfg.UnsubscribeURL, c.UUID, s.UUID), } } -// AddMessenger adds a Messenger messaging backend to the runner process. -func (r *Runner) AddMessenger(msg messenger.Messenger) error { +// AddMessenger adds a Messenger messaging backend to the manager. +func (m *Manager) AddMessenger(msg messenger.Messenger) error { id := msg.Name() - if _, ok := r.messengers[id]; ok { + if _, ok := m.messengers[id]; ok { return fmt.Errorf("messenger '%s' is already loaded", id) } - r.messengers[id] = msg + m.messengers[id] = msg return nil } // GetMessengerNames returns the list of registered messengers. -func (r *Runner) GetMessengerNames() []string { +func (m *Manager) GetMessengerNames() []string { var names []string - for n := range r.messengers { + for n := range m.messengers { names = append(names, n) } @@ -136,8 +134,8 @@ func (r *Runner) GetMessengerNames() []string { } // HasMessenger checks if a given messenger is registered. -func (r *Runner) HasMessenger(id string) bool { - _, ok := r.messengers[id] +func (m *Manager) HasMessenger(id string) bool { + _, ok := m.messengers[id] return ok } @@ -147,55 +145,55 @@ func (r *Runner) HasMessenger(id string) bool { // subscribers and pushes messages to them for each queued campaign // until all subscribers are exhausted, at which point, a campaign is marked // as "finished". -func (r *Runner) Run(tick time.Duration) { +func (m *Manager) Run(tick time.Duration) { go func() { t := time.NewTicker(tick) for { select { // Periodically scan the data source for campaigns to process. case <-t.C: - campaigns, err := r.src.NextCampaigns(r.getPendingCampaignIDs()) + campaigns, err := m.src.NextCampaigns(m.getPendingCampaignIDs()) if err != nil { - r.logger.Printf("error fetching campaigns: %v", err) + m.logger.Printf("error fetching campaigns: %v", err) continue } for _, c := range campaigns { - if err := r.addCampaign(c); err != nil { - r.logger.Printf("error processing campaign (%s): %v", c.Name, err) + if err := m.addCampaign(c); err != nil { + m.logger.Printf("error processing campaign (%s): %v", c.Name, err) continue } - r.logger.Printf("start processing campaign (%s)", c.Name) + m.logger.Printf("start processing campaign (%s)", c.Name) // If subscriber processing is busy, move on. Blocking and waiting // can end up in a race condition where the waiting campaign's // state in the data source has changed. select { - case r.subFetchQueue <- c: + case m.subFetchQueue <- c: default: } } // Aggregate errors from sending messages to check against the error threshold // after which a campaign is paused. - case e := <-r.msgErrorQueue: - if r.cfg.MaxSendErrors < 1 { + case e := <-m.msgErrorQueue: + if m.cfg.MaxSendErrors < 1 { continue } // If the error threshold is met, pause the campaign. - r.msgErrorCounts[e.camp.ID]++ - if r.msgErrorCounts[e.camp.ID] >= r.cfg.MaxSendErrors { - r.logger.Printf("error counted exceeded %d. pausing campaign %s", - r.cfg.MaxSendErrors, e.camp.Name) + m.msgErrorCounts[e.camp.ID]++ + if m.msgErrorCounts[e.camp.ID] >= m.cfg.MaxSendErrors { + m.logger.Printf("error counted exceeded %d. pausing campaign %s", + m.cfg.MaxSendErrors, e.camp.Name) - if r.isCampaignProcessing(e.camp.ID) { - r.exhaustCampaign(e.camp, models.CampaignStatusPaused) + if m.isCampaignProcessing(e.camp.ID) { + m.exhaustCampaign(e.camp, models.CampaignStatusPaused) } - delete(r.msgErrorCounts, e.camp.ID) + delete(m.msgErrorCounts, e.camp.ID) // Notify admins. - r.sendNotif(e.camp, + m.sendNotif(e.camp, models.CampaignStatusPaused, "Too many errors") } @@ -204,50 +202,50 @@ func (r *Runner) Run(tick time.Duration) { }() // Fetch the next set of subscribers for a campaign and process them. - for c := range r.subFetchQueue { - has, err := r.nextSubscribers(c, batchSize) + for c := range m.subFetchQueue { + has, err := m.nextSubscribers(c, batchSize) if err != nil { - r.logger.Printf("error processing campaign batch (%s): %v", c.Name, err) + m.logger.Printf("error processing campaign batch (%s): %v", c.Name, err) continue } if has { // There are more subscribers to fetch. - r.subFetchQueue <- c - } else if r.isCampaignProcessing(c.ID) { + m.subFetchQueue <- c + } else if m.isCampaignProcessing(c.ID) { // There are no more subscribers. Either the campaign status // has changed or all subscribers have been processed. - newC, err := r.exhaustCampaign(c, "") + newC, err := m.exhaustCampaign(c, "") if err != nil { - r.logger.Printf("error exhausting campaign (%s): %v", c.Name, err) + m.logger.Printf("error exhausting campaign (%s): %v", c.Name, err) continue } - r.sendNotif(newC, newC.Status, "") + m.sendNotif(newC, newC.Status, "") } } } // SpawnWorkers spawns workers goroutines that push out messages. -func (r *Runner) SpawnWorkers() { - for i := 0; i < r.cfg.Concurrency; i++ { +func (m *Manager) SpawnWorkers() { + for i := 0; i < m.cfg.Concurrency; i++ { go func() { - for m := range r.msgQueue { - if !r.isCampaignProcessing(m.Campaign.ID) { + for msg := range m.msgQueue { + if !m.isCampaignProcessing(msg.Campaign.ID) { continue } - err := r.messengers[m.Campaign.MessengerID].Push( - m.from, - []string{m.to}, - m.Campaign.Subject, - m.Body) + err := m.messengers[msg.Campaign.MessengerID].Push( + msg.from, + []string{msg.to}, + msg.Campaign.Subject, + msg.Body) if err != nil { - r.logger.Printf("error sending message in campaign %s: %v", - m.Campaign.Name, err) + m.logger.Printf("error sending message in campaign %s: %v", + msg.Campaign.Name, err) select { - case r.msgErrorQueue <- msgError{camp: m.Campaign, err: err}: + case m.msgErrorQueue <- msgError{camp: msg.Campaign, err: err}: default: } } @@ -257,28 +255,28 @@ func (r *Runner) SpawnWorkers() { } // addCampaign adds a campaign to the process queue. -func (r *Runner) addCampaign(c *models.Campaign) error { +func (m *Manager) addCampaign(c *models.Campaign) error { // Validate messenger. - if _, ok := r.messengers[c.MessengerID]; !ok { - r.src.UpdateCampaignStatus(c.ID, models.CampaignStatusCancelled) + if _, ok := m.messengers[c.MessengerID]; !ok { + m.src.UpdateCampaignStatus(c.ID, models.CampaignStatusCancelled) return fmt.Errorf("unknown messenger %s on campaign %s", c.MessengerID, c.Name) } // Load the template. - if err := c.CompileTemplate(r.TemplateFuncs(c)); err != nil { + if err := c.CompileTemplate(m.TemplateFuncs(c)); err != nil { return err } // Add the campaign to the active map. - r.camps[c.ID] = c + m.camps[c.ID] = c return nil } // getPendingCampaignIDs returns the IDs of campaigns currently being processed. -func (r *Runner) getPendingCampaignIDs() []int64 { +func (m *Manager) getPendingCampaignIDs() []int64 { // Needs to return an empty slice in case there are no campaigns. ids := make([]int64, 0) - for _, c := range r.camps { + for _, c := range m.camps { ids = append(ids, int64(c.ID)) } @@ -289,9 +287,9 @@ func (r *Runner) getPendingCampaignIDs() []int64 { // If returns a bool indicating whether there any subscribers were processed // in the current batch or not. This can happen when all the subscribers // have been processed, or if a campaign has been paused or cancelled abruptly. -func (r *Runner) nextSubscribers(c *models.Campaign, batchSize int) (bool, error) { +func (m *Manager) nextSubscribers(c *models.Campaign, batchSize int) (bool, error) { // Fetch a batch of subscribers. - subs, err := r.src.NextSubscribers(c.ID, batchSize) + subs, err := m.src.NextSubscribers(c.ID, batchSize) if err != nil { return false, fmt.Errorf("error fetching campaign subscribers (%s): %v", c.Name, err) } @@ -303,42 +301,42 @@ func (r *Runner) nextSubscribers(c *models.Campaign, batchSize int) (bool, error // Push messages. for _, s := range subs { - m := r.NewMessage(c, s) - if err := m.Render(); err != nil { - r.logger.Printf("error rendering message (%s) (%s): %v", c.Name, s.Email, err) + msg := m.NewMessage(c, s) + if err := msg.Render(); err != nil { + m.logger.Printf("error rendering message (%s) (%s): %v", c.Name, s.Email, err) continue } // Push the message to the queue while blocking and waiting until // the queue is drained. - r.msgQueue <- m + m.msgQueue <- msg } return true, nil } // isCampaignProcessing checks if the campaign is bing processed. -func (r *Runner) isCampaignProcessing(id int) bool { - _, ok := r.camps[id] +func (m *Manager) isCampaignProcessing(id int) bool { + _, ok := m.camps[id] return ok } -func (r *Runner) exhaustCampaign(c *models.Campaign, status string) (*models.Campaign, error) { - delete(r.camps, c.ID) +func (m *Manager) exhaustCampaign(c *models.Campaign, status string) (*models.Campaign, error) { + delete(m.camps, c.ID) // A status has been passed. Change the campaign's status // without further checks. if status != "" { - if err := r.src.UpdateCampaignStatus(c.ID, status); err != nil { - r.logger.Printf("error updating campaign (%s) status to %s: %v", c.Name, status, err) + if err := m.src.UpdateCampaignStatus(c.ID, status); err != nil { + m.logger.Printf("error updating campaign (%s) status to %s: %v", c.Name, status, err) } else { - r.logger.Printf("set campaign (%s) to %s", c.Name, status) + m.logger.Printf("set campaign (%s) to %s", c.Name, status) } return c, nil } // Fetch the up-to-date campaign status from the source. - cm, err := r.src.GetCampaign(c.ID) + cm, err := m.src.GetCampaign(c.ID) if err != nil { return nil, err } @@ -346,13 +344,13 @@ func (r *Runner) exhaustCampaign(c *models.Campaign, status string) (*models.Cam // If a running campaign has exhausted subscribers, it's finished. if cm.Status == models.CampaignStatusRunning { cm.Status = models.CampaignStatusFinished - if err := r.src.UpdateCampaignStatus(c.ID, models.CampaignStatusFinished); err != nil { - r.logger.Printf("error finishing campaign (%s): %v", c.Name, err) + if err := m.src.UpdateCampaignStatus(c.ID, models.CampaignStatusFinished); err != nil { + m.logger.Printf("error finishing campaign (%s): %v", c.Name, err) } else { - r.logger.Printf("campaign (%s) finished", c.Name) + m.logger.Printf("campaign (%s) finished", c.Name) } } else { - r.logger.Printf("stop processing campaign (%s)", c.Name) + m.logger.Printf("stop processing campaign (%s)", c.Name) } return cm, nil @@ -371,32 +369,32 @@ func (m *Message) Render() error { // trackLink register a URL and return its UUID to be used in message templates // for tracking links. -func (r *Runner) trackLink(url, campUUID, subUUID string) string { - r.linksMutex.RLock() - if uu, ok := r.links[url]; ok { - r.linksMutex.RUnlock() - return fmt.Sprintf(r.cfg.LinkTrackURL, uu, campUUID, subUUID) +func (m *Manager) trackLink(url, campUUID, subUUID string) string { + m.linksMutex.RLock() + if uu, ok := m.links[url]; ok { + m.linksMutex.RUnlock() + return fmt.Sprintf(m.cfg.LinkTrackURL, uu, campUUID, subUUID) } - r.linksMutex.RUnlock() + m.linksMutex.RUnlock() // Register link. - uu, err := r.src.CreateLink(url) + uu, err := m.src.CreateLink(url) if err != nil { - r.logger.Printf("error registering tracking for link '%s': %v", url, err) + m.logger.Printf("error registering tracking for link '%s': %v", url, err) // If the registration fails, fail over to the original URL. return url } - r.linksMutex.Lock() - r.links[url] = uu - r.linksMutex.Unlock() + m.linksMutex.Lock() + m.links[url] = uu + m.linksMutex.Unlock() - return fmt.Sprintf(r.cfg.LinkTrackURL, uu, campUUID, subUUID) + return fmt.Sprintf(m.cfg.LinkTrackURL, uu, campUUID, subUUID) } // sendNotif sends a notification to registered admin e-mails. -func (r *Runner) sendNotif(c *models.Campaign, status, reason string) error { +func (m *Manager) sendNotif(c *models.Campaign, status, reason string) error { var ( subject = fmt.Sprintf("%s: %s", strings.Title(status), c.Name) data = map[string]interface{}{ @@ -409,19 +407,19 @@ func (r *Runner) sendNotif(c *models.Campaign, status, reason string) error { } ) - return r.notifCB(subject, data) + return m.notifCB(subject, data) } // TemplateFuncs returns the template functions to be applied into // compiled campaign templates. -func (r *Runner) TemplateFuncs(c *models.Campaign) template.FuncMap { +func (m *Manager) TemplateFuncs(c *models.Campaign) template.FuncMap { return template.FuncMap{ "TrackLink": func(url, campUUID, subUUID string) string { - return r.trackLink(url, campUUID, subUUID) + return m.trackLink(url, campUUID, subUUID) }, "TrackView": func(campUUID, subUUID string) template.HTML { return template.HTML(fmt.Sprintf(``, - fmt.Sprintf(r.cfg.ViewTrackURL, campUUID, subUUID))) + fmt.Sprintf(m.cfg.ViewTrackURL, campUUID, subUUID))) }, "Date": func(layout string) string { if layout == "" { diff --git a/runner_db.go b/manager_db.go similarity index 97% rename from runner_db.go rename to manager_db.go index 1b5aa82..bff11b2 100644 --- a/runner_db.go +++ b/manager_db.go @@ -12,7 +12,7 @@ type runnerDB struct { queries *Queries } -func newRunnerDB(q *Queries) *runnerDB { +func newManagerDB(q *Queries) *runnerDB { return &runnerDB{ queries: q, } diff --git a/templates.go b/templates.go index a1c2555..0806eef 100644 --- a/templates.go +++ b/templates.go @@ -111,12 +111,12 @@ func handlePreviewTemplate(c echo.Context) error { Body: dummyTpl, } - if err := camp.CompileTemplate(app.Runner.TemplateFuncs(&camp)); err != nil { + if err := camp.CompileTemplate(app.Manager.TemplateFuncs(&camp)); err != nil { return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Error compiling template: %v", err)) } // Render the message body. - m := app.Runner.NewMessage(&camp, &dummySubscriber) + m := app.Manager.NewMessage(&camp, &dummySubscriber) if err := m.Render(); err != nil { return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Error rendering message: %v", err))