feat: add webhook to settings for custom event types

cmd/handlers.go

@@ -217,6 +217,9 @@ func initHTTPHandlers(e *echo.Echo, a *App) {
 		g.PUT("/api/roles/lists/:id", pm(hasID(a.UpdateListRole), "roles:manage"))
 		g.DELETE("/api/roles/:id", pm(hasID(a.DeleteRole), "roles:manage"))
 
+		// Webhook events list (webhooks are configured via settings).
+		g.GET("/api/settings/webhooks/events", pm(a.GetWebhookEvents, "settings:get"))
+
 		if a.cfg.BounceWebhooksEnabled {
 			// Private authenticated bounce endpoint.
 			g.POST("/webhooks/bounce", pm(a.BounceWebhook, "webhooks:post_bounce"))

cmd/init.go

@@ -47,6 +47,7 @@ import (
 	"github.com/knadh/listmonk/internal/messenger/postback"
 	"github.com/knadh/listmonk/internal/notifs"
 	"github.com/knadh/listmonk/internal/subimporter"
+	"github.com/knadh/listmonk/internal/webhooks"
 	"github.com/knadh/listmonk/models"
 	"github.com/knadh/stuffbin"
 	"github.com/labstack/echo/v4"
@@ -531,7 +532,7 @@ func initI18n(lang string, fs stuffbin.FileSystem) *i18n.I18n {
 }
 
 // initCore initializes the CRUD DB core .
-func initCore(fnNotify func(sub models.Subscriber, listIDs []int) (int, error), queries *models.Queries, db *sqlx.DB, i *i18n.I18n, ko *koanf.Koanf) *core.Core {
+func initCore(fnNotify func(sub models.Subscriber, listIDs []int) (int, error), queries *models.Queries, db *sqlx.DB, i *i18n.I18n, ko *koanf.Koanf, whMgr *webhooks.Manager) *core.Core {
 	opt := &core.Opt{
 		Constants: core.Constants{
 			SendOptinConfirmation: ko.Bool("app.send_optin_confirmation"),
@@ -551,6 +552,7 @@ func initCore(fnNotify func(sub models.Subscriber, listIDs []int) (int, error),
 	// Initialize the CRUD core.
 	return core.New(opt, &core.Hooks{
 		SendOptinConfirmation: fnNotify,
+		TriggerWebhook:        whMgr.Trigger,
 	})
 }
 
@@ -617,6 +619,8 @@ func initImporter(q *models.Queries, db *sqlx.DB, core *core.Core, i *i18n.I18n,
 			BlocklistStmt:      q.UpsertBlocklistSubscriber.Stmt,
 			UpdateListDateStmt: q.UpdateListsDate.Stmt,
 
+			TriggerWebhook: core.TriggerWebhook,
+
 			// Hook for triggering admin notifications and refreshing stats materialized
 			// views after a successful import.
 			PostCB: func(subject string, data any) error {

cmd/main.go

@@ -26,6 +26,7 @@ import (
 	"github.com/knadh/listmonk/internal/media"
 	"github.com/knadh/listmonk/internal/messenger/email"
 	"github.com/knadh/listmonk/internal/subimporter"
+	"github.com/knadh/listmonk/internal/webhooks"
 	"github.com/knadh/listmonk/models"
 	"github.com/knadh/paginator"
 	"github.com/knadh/stuffbin"
@@ -46,6 +47,7 @@ type App struct {
 	auth       *auth.Auth
 	media      media.Store
 	bounce     *bounce.Manager
+	webhooks   *webhooks.Manager
 	captcha    *captcha.Captcha
 	i18n       *i18n.I18n
 	pg         *paginator.Paginator
@@ -82,6 +84,8 @@ var (
 	db      *sqlx.DB
 	queries *models.Queries
 
+	webhookMgr *webhooks.Manager
+
 	// Compile-time variables.
 	buildString   string
 	versionString string
@@ -174,6 +178,9 @@ func init() {
 
 	// Prepare queries.
 	queries = prepareQueries(qMap, db, ko)
+
+	// Initialize the webhook manager for outgoing event webhooks.
+	webhookMgr = webhooks.New(lo, versionString, queries)
 }
 
 func main() {
@@ -193,7 +200,7 @@ func main() {
 		fbOptinNotify = makeOptinNotifyHook(ko.Bool("privacy.unsubscribe_header"), urlCfg, queries, i18n)
 
 		// Crud core.
-		core = initCore(fbOptinNotify, queries, db, i18n, ko)
+		core = initCore(fbOptinNotify, queries, db, i18n, ko, webhookMgr)
 
 		// Initialize all messengers, SMTP and postback.
 		msgrs = append(initSMTPMessengers(), initPostbackMessengers(ko)...)
@@ -240,6 +247,38 @@ func main() {
 		go bounce.Run()
 	}
 
+	// Load webhooks from settings.
+	var settings models.Settings
+	var settingsLoaded bool
+	if s, err := core.GetSettings(); err == nil {
+		settings = s
+		settingsLoaded = true
+		webhookMgr.Load(settings.Webhooks)
+	}
+
+	// Initialize and start the webhook worker pool.
+	webhookWorkerCfg := webhooks.WorkerConfig{
+		NumWorkers: ko.Int("app.webhook_workers"),
+		BatchSize:  ko.Int("app.webhook_batch_size"),
+	}
+	if webhookWorkerCfg.NumWorkers < 1 {
+		webhookWorkerCfg.NumWorkers = 2
+	}
+	if webhookWorkerCfg.BatchSize < 1 {
+		webhookWorkerCfg.BatchSize = 50
+	}
+	webhookWorkerPool := webhooks.NewWorkerPool(webhookWorkerCfg, db, queries, lo, versionString)
+	if settingsLoaded {
+		webhookWorkerPool.LoadWebhooks(settings.Webhooks)
+	}
+	go webhookWorkerPool.Run()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// write each webhook to db for the worker pool to pick up
+	go core.PersistWebhookLogs(ctx)
+
 	// Start cronjobs.
 	initCron(core, db)
 
@@ -263,6 +302,7 @@ func main() {
 		auth:       auth,
 		media:      media,
 		bounce:     bounce,
+		webhooks:   webhookMgr,
 		captcha:    initCaptcha(),
 		i18n:       i18n,
 		log:        lo,
@@ -310,6 +350,15 @@ func main() {
 		// Close the campaign manager.
 		mgr.Close()
 
+		// Close the webhook worker pool.
+		webhookWorkerPool.Close()
+
+		// Close the webhook manager.
+		webhookMgr.Close()
+
+		// close persist webhook log goroutine
+		cancel()
+
 		// Close the DB pool.
 		db.Close()
 

cmd/manager_store.go

@@ -74,8 +74,9 @@ func (s *store) GetCampaign(campID int) (*models.Campaign, error) {
 }
 
 // UpdateCampaignStatus updates a campaign's status.
+// Uses Core to ensure webhooks are triggered for status changes.
 func (s *store) UpdateCampaignStatus(campID int, status string) error {
-	_, err := s.queries.UpdateCampaignStatus.Exec(campID, status)
+	_, err := s.core.UpdateCampaignStatus(campID, status)
 	return err
 }
 

cmd/settings.go

@@ -3,6 +3,7 @@ package main
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -71,6 +72,10 @@ func (a *App) GetSettings(c echo.Context) error {
 	for i := range s.Messengers {
 		s.Messengers[i].Password = strings.Repeat(pwdMask, utf8.RuneCountInString(s.Messengers[i].Password))
 	}
+	for i := range s.Webhooks {
+		s.Webhooks[i].AuthBasicPass = strings.Repeat(pwdMask, utf8.RuneCountInString(s.Webhooks[i].AuthBasicPass))
+		s.Webhooks[i].AuthToken = strings.Repeat(pwdMask, utf8.RuneCountInString(s.Webhooks[i].AuthToken))
+	}
 
 	s.UploadS3AwsSecretAccessKey = strings.Repeat(pwdMask, utf8.RuneCountInString(s.UploadS3AwsSecretAccessKey))
 	s.SendgridKey = strings.Repeat(pwdMask, utf8.RuneCountInString(s.SendgridKey))
@@ -209,6 +214,44 @@ func (a *App) UpdateSettings(c echo.Context) error {
 		names[name] = true
 	}
 
+	// Webhooks password/secret handling.
+	for i, w := range set.Webhooks {
+		u, err := url.Parse(w.URL)
+		if err != nil {
+			return err
+		}
+
+		if u.Scheme != "http" && u.Scheme != "https" {
+			return fmt.Errorf("invalid scheme in the url provided for webhook: %s", w.URL)
+		}
+
+		if u.Host == "" {
+			return fmt.Errorf("invalid host in the url provided for webhook: %s", w.URL)
+		}
+
+		// UUID to keep track of password changes similar to the SMTP logic above.
+		if w.UUID == "" {
+			set.Webhooks[i].UUID = uuid.Must(uuid.NewV4()).String()
+		}
+
+		// If there's no password/token coming in from the frontend, copy the existing
+		// values by matching the UUID.
+		if w.AuthBasicPass == "" {
+			for _, c := range cur.Webhooks {
+				if w.UUID == c.UUID {
+					set.Webhooks[i].AuthBasicPass = c.AuthBasicPass
+				}
+			}
+		}
+		if w.AuthToken == "" {
+			for _, c := range cur.Webhooks {
+				if w.UUID == c.UUID {
+					set.Webhooks[i].AuthToken = c.AuthToken
+				}
+			}
+		}
+	}
+
 	// S3 password?
 	if set.UploadS3AwsSecretAccessKey == "" {
 		set.UploadS3AwsSecretAccessKey = cur.UploadS3AwsSecretAccessKey

cmd/upgrade.go

@@ -43,6 +43,7 @@ var migList = []migFunc{
 	{"v5.0.0", migrations.V5_0_0},
 	{"v5.1.0", migrations.V5_1_0},
 	{"v6.0.0", migrations.V6_0_0},
+	{"v6.1.0", migrations.V6_1_0},
 }
 
 // upgrade upgrades the database to the current version by running SQL migration files

cmd/webhooks.go

@@ -0,0 +1,13 @@
+package main
+
+import (
+	"net/http"
+
+	"github.com/knadh/listmonk/models"
+	"github.com/labstack/echo/v4"
+)
+
+// GetWebhookEvents returns the list of available webhook events.
+func (a *App) GetWebhookEvents(c echo.Context) error {
+	return c.JSON(http.StatusOK, okResp{models.AllWebhookEvents()})
+}

frontend/src/api/index.js

@@ -574,3 +574,9 @@ export const disableTOTP = (id, data) => http.delete(
   `/api/users/${id}/twofa`,
   { data },
 );
+
+// Webhooks.
+export const getWebhookEvents = async () => http.get(
+  '/api/settings/webhooks/events',
+  { camelCase: false },
+);

frontend/src/views/Settings.vue

@@ -54,6 +54,10 @@
             <messenger-settings :form="form" :key="key" />
           </b-tab-item><!-- messengers -->
 
+          <b-tab-item :label="$t('settings.webhooks.name')">
+            <webhook-settings :form="form" :events="webhookEvents" :key="key" />
+          </b-tab-item><!-- webhooks -->
+
           <b-tab-item :label="$t('settings.appearance.name')">
             <appearance-settings :form="form" :key="key" />
           </b-tab-item><!-- appearance -->
@@ -75,6 +79,7 @@ import PerformanceSettings from './settings/performance.vue';
 import PrivacySettings from './settings/privacy.vue';
 import SecuritySettings from './settings/security.vue';
 import SmtpSettings from './settings/smtp.vue';
+import WebhookSettings from './settings/webhooks.vue';
 
 export default Vue.extend({
   components: {
@@ -86,6 +91,7 @@ export default Vue.extend({
     SmtpSettings,
     BounceSettings,
     MessengerSettings,
+    WebhookSettings,
     AppearanceSettings,
   },
 
@@ -102,6 +108,7 @@ export default Vue.extend({
       formCopy: '',
       form: null,
       tab: 0,
+      webhookEvents: [],
     };
   },
 
@@ -187,6 +194,22 @@ export default Vue.extend({
         }
       }
 
+      // Webhook secrets.
+      for (let i = 0; i < form.webhooks.length; i += 1) {
+        // If it's the dummy UI password placeholder, ignore it.
+        if (this.isDummy(form.webhooks[i].auth_basic_pass)) {
+          form.webhooks[i].auth_basic_pass = '';
+        } else if (this.hasDummy(form.webhooks[i].auth_basic_pass)) {
+          hasDummy = `webhook #${i + 1} password`;
+        }
+
+        if (this.isDummy(form.webhooks[i].auth_hmac_secret)) {
+          form.webhooks[i].auth_hmac_secret = '';
+        } else if (this.hasDummy(form.webhooks[i].auth_hmac_secret)) {
+          hasDummy = `webhook #${i + 1} HMAC secret`;
+        }
+      }
+
       if (hasDummy) {
         this.$utils.toast(this.$t('globals.messages.passwordChangeFull', { name: hasDummy }), 'is-danger');
         return false;
@@ -245,6 +268,12 @@ export default Vue.extend({
     hasDummy(pwd) {
       return pwd.includes('•');
     },
+
+    getWebhookEvents() {
+      this.$api.getWebhookEvents().then((data) => {
+        this.webhookEvents = data;
+      });
+    },
   },
 
   computed: {
@@ -269,6 +298,7 @@ export default Vue.extend({
   mounted() {
     this.tab = this.$utils.getPref('settings.tab') || 0;
     this.getSettings();
+    this.getWebhookEvents();
   },
 
   watch: {

frontend/src/views/settings/performance.vue

@@ -24,6 +24,24 @@
         min="0" max="100000" />
     </b-field>
 
+    <hr />
+    <div class="columns">
+      <div class="column is-6">
+        <b-field :label="$t('settings.performance.webhookWorkers')" label-position="on-border"
+          :message="$t('settings.performance.webhookWorkersHelp')">
+          <b-numberinput v-model="data['app.webhook_workers']" name="app.webhook_workers" type="is-light" placeholder="2"
+            min="1" max="100" />
+        </b-field>
+      </div>
+      <div class="column is-6">
+        <b-field :label="$t('settings.performance.webhookBatchSize')" label-position="on-border"
+          :message="$t('settings.performance.webhookBatchSizeHelp')">
+          <b-numberinput v-model="data['app.webhook_batch_size']" name="app.webhook_batch_size" type="is-light"
+            placeholder="50" min="1" max="1000" />
+        </b-field>
+      </div>
+    </div>
+
     <div>
       <div class="columns">
         <div class="column is-6">