Merge 4e83073152 into 16f4dfd3e9

(cherry picked from commit e0473db7861b63d328be506542c340f2e95b6a03)

cmd/handlers.go

@@ -163,6 +163,7 @@ func initHTTPHandlers(e *echo.Echo, app *App) {
 	g.DELETE("/api/maintenance/subscriptions/unconfirmed", handleGCSubscriptions)
 
 	g.POST("/api/tx", handleSendTxMessage)
+	g.POST("/api/tx/external", handleSendExternalTxMessage)
 
 	g.GET("/api/events", handleEventStream)
 

cmd/subscribers.go

@@ -28,6 +28,7 @@ type subQueryReq struct {
 	SubscriberIDs      []int  `json:"ids"`
 	Action             string `json:"action"`
 	Status             string `json:"status"`
+	SubscriptionStatus string `json:"subscription_status"`
 }
 
 // subProfileData represents a subscriber's collated data in JSON
@@ -415,7 +416,7 @@ func handleDeleteSubscribersByQuery(c echo.Context) error {
 		return err
 	}
 
-	if err := app.core.DeleteSubscribersByQuery(req.Query, req.ListIDs); err != nil {
+	if err := app.core.DeleteSubscribersByQuery(req.Query, req.ListIDs, req.SubscriptionStatus); err != nil {
 		return err
 	}
 
@@ -434,7 +435,7 @@ func handleBlocklistSubscribersByQuery(c echo.Context) error {
 		return err
 	}
 
-	if err := app.core.BlocklistSubscribersByQuery(req.Query, req.ListIDs); err != nil {
+	if err := app.core.BlocklistSubscribersByQuery(req.Query, req.ListIDs, req.SubscriptionStatus); err != nil {
 		return err
 	}
 
@@ -461,11 +462,11 @@ func handleManageSubscriberListsByQuery(c echo.Context) error {
 	var err error
 	switch req.Action {
 	case "add":
-		err = app.core.AddSubscriptionsByQuery(req.Query, req.ListIDs, req.TargetListIDs, req.Status)
+		err = app.core.AddSubscriptionsByQuery(req.Query, req.ListIDs, req.TargetListIDs, req.Status, req.SubscriptionStatus)
 	case "remove":
-		err = app.core.DeleteSubscriptionsByQuery(req.Query, req.ListIDs, req.TargetListIDs)
+		err = app.core.DeleteSubscriptionsByQuery(req.Query, req.ListIDs, req.TargetListIDs, req.SubscriptionStatus)
 	case "unsubscribe":
-		err = app.core.UnsubscribeListsByQuery(req.Query, req.ListIDs, req.TargetListIDs)
+		err = app.core.UnsubscribeListsByQuery(req.Query, req.ListIDs, req.TargetListIDs, req.SubscriptionStatus)
 	default:
 		return echo.NewHTTPError(http.StatusBadRequest, app.i18n.T("subscribers.invalidAction"))
 	}

cmd/tx.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"io"
 	"net/http"
-	"net/textproto"
 	"strings"
 
 	"github.com/knadh/listmonk/internal/manager"
@@ -20,49 +19,8 @@ func handleSendTxMessage(c echo.Context) error {
 		m   models.TxMessage
 	)
 
-	// If it's a multipart form, there may be file attachments.
-	if strings.HasPrefix(c.Request().Header.Get("Content-Type"), "multipart/form-data") {
-		form, err := c.MultipartForm()
+	m, err := parseTxMessage(c, app)
 	if err != nil {
-			return echo.NewHTTPError(http.StatusBadRequest,
-				app.i18n.Ts("globals.messages.invalidFields", "name", err.Error()))
-		}
-
-		data, ok := form.Value["data"]
-		if !ok || len(data) != 1 {
-			return echo.NewHTTPError(http.StatusBadRequest,
-				app.i18n.Ts("globals.messages.invalidFields", "name", "data"))
-		}
-
-		// Parse the JSON data.
-		if err := json.Unmarshal([]byte(data[0]), &m); err != nil {
-			return echo.NewHTTPError(http.StatusBadRequest,
-				app.i18n.Ts("globals.messages.invalidFields", "name", fmt.Sprintf("data: %s", err.Error())))
-		}
-
-		// Attach files.
-		for _, f := range form.File["file"] {
-			file, err := f.Open()
-			if err != nil {
-				return echo.NewHTTPError(http.StatusInternalServerError,
-					app.i18n.Ts("globals.messages.invalidFields", "name", fmt.Sprintf("file: %s", err.Error())))
-			}
-			defer file.Close()
-
-			b, err := io.ReadAll(file)
-			if err != nil {
-				return echo.NewHTTPError(http.StatusInternalServerError,
-					app.i18n.Ts("globals.messages.invalidFields", "name", fmt.Sprintf("file: %s", err.Error())))
-			}
-
-			m.Attachments = append(m.Attachments, models.Attachment{
-				Name:    f.Filename,
-				Header:  manager.MakeAttachmentHeader(f.Filename, "base64", f.Header.Get("Content-Type")),
-				Content: b,
-			})
-		}
-
-	} else if err := c.Bind(&m); err != nil {
 		return err
 	}
 
@@ -73,11 +31,10 @@ func handleSendTxMessage(c echo.Context) error {
 		m = r
 	}
 
-	// Get the cached tx template.
-	tpl, err := app.manager.GetTpl(m.TemplateID)
+	// Get the template
+	tpl, err := getTemplate(app, m.TemplateID)
 	if err != nil {
-		return echo.NewHTTPError(http.StatusBadRequest,
-			app.i18n.Ts("globals.messages.notFound", "name", fmt.Sprintf("template %d", m.TemplateID)))
+		return err
 	}
 
 	var (
@@ -121,34 +78,9 @@ func handleSendTxMessage(c echo.Context) error {
 		}
 
 		// Prepare the final message.
-		msg := models.Message{}
-		msg.Subscriber = sub
-		msg.To = []string{sub.Email}
-		msg.From = m.FromEmail
-		msg.Subject = m.Subject
-		msg.ContentType = m.ContentType
-		msg.Messenger = m.Messenger
-		msg.Body = m.Body
-		for _, a := range m.Attachments {
-			msg.Attachments = append(msg.Attachments, models.Attachment{
-				Name:    a.Name,
-				Header:  a.Header,
-				Content: a.Content,
-			})
-		}
+		msg := models.CreateMailMessage(sub, m)
 
-		// Optional headers.
-		if len(m.Headers) != 0 {
-			msg.Headers = make(textproto.MIMEHeader, len(m.Headers))
-			for _, set := range m.Headers {
-				for hdr, val := range set {
-					msg.Headers.Add(hdr, val)
-				}
-			}
-		}
-
-		if err := app.manager.PushMessage(msg); err != nil {
-			app.log.Printf("error sending message (%s): %v", msg.Subject, err)
+		if err := sendEmail(app, msg); err != nil {
 			return err
 		}
 	}
@@ -160,6 +92,27 @@ func handleSendTxMessage(c echo.Context) error {
 	return c.JSON(http.StatusOK, okResp{true})
 }
 
+func parseTxMessage(c echo.Context, app *App) (models.TxMessage, error) {
+	m := models.TxMessage{}
+	// If it's a multipart form, there may be file attachments.
+	if strings.HasPrefix(c.Request().Header.Get("Content-Type"), "multipart/form-data") {
+		if data, attachments, err := parseMultiPartMessageDetails(c, app); err != nil {
+			return models.TxMessage{}, err
+		} else {
+			// Parse the JSON data.
+			if err := json.Unmarshal([]byte(data[0]), &m); err != nil {
+				return models.TxMessage{}, echo.NewHTTPError(http.StatusBadRequest,
+					app.i18n.Ts("globals.messages.invalidFields", "name", fmt.Sprintf("data: %s", err.Error())))
+			}
+
+			m.Attachments = append(m.Attachments, attachments...)
+		}
+	} else if err := c.Bind(&m); err != nil {
+		return models.TxMessage{}, err
+	}
+	return m, nil
+}
+
 func validateTxMessage(m models.TxMessage, app *App) (models.TxMessage, error) {
 	if len(m.SubscriberEmails) > 0 && m.SubscriberEmail != "" {
 		return m, echo.NewHTTPError(http.StatusBadRequest,
@@ -205,3 +158,163 @@ func validateTxMessage(m models.TxMessage, app *App) (models.TxMessage, error) {
 
 	return m, nil
 }
+
+// handleSendExternalTxMessage handles the sending of a transactional message to an external recipient.
+func handleSendExternalTxMessage(c echo.Context) error {
+	var (
+		app = c.Get("app").(*App)
+		m   models.ExternalTxMessage
+	)
+
+	m, err := parseExternalTxMessage(c, app)
+	if err != nil {
+		return err
+	}
+
+	// Validate input.
+	if r, err := validateExternalTxMessage(m, app); err != nil {
+		return err
+	} else {
+		m = r
+	}
+
+	// Get the template
+	tpl, err := getTemplate(app, m.TemplateID)
+	if err != nil {
+		return err
+	}
+
+	txMessage := m.MapToTxMessage()
+	notFound := []string{}
+	for n := 0; n < len(txMessage.SubscriberEmails); n++ {
+		// Render the message.
+		if err := txMessage.Render(models.Subscriber{}, tpl); err != nil {
+			return echo.NewHTTPError(http.StatusBadRequest,
+				app.i18n.Ts("globals.messages.errorFetching", "name"))
+		}
+
+		// Prepare the final message.
+		msg := models.CreateMailMessage(models.Subscriber{Email: txMessage.SubscriberEmails[n]}, txMessage)
+
+		if err := sendEmail(app, msg); err != nil {
+			return err
+		}
+	}
+
+	if len(notFound) > 0 {
+		return echo.NewHTTPError(http.StatusBadRequest, strings.Join(notFound, "; "))
+	}
+
+	return c.JSON(http.StatusOK, okResp{true})
+}
+
+func parseExternalTxMessage(c echo.Context, app *App) (models.ExternalTxMessage, error) {
+	m := models.ExternalTxMessage{}
+	// If it's a multipart form, there may be file attachments.
+	if strings.HasPrefix(c.Request().Header.Get("Content-Type"), "multipart/form-data") {
+		if data, attachments, err := parseMultiPartMessageDetails(c, app); err != nil {
+			return models.ExternalTxMessage{}, err
+		} else {
+			// Parse the JSON data.
+			if err := json.Unmarshal([]byte(data[0]), &m); err != nil {
+				return models.ExternalTxMessage{}, echo.NewHTTPError(http.StatusBadRequest,
+					app.i18n.Ts("globals.messages.invalidFields", "name", fmt.Sprintf("data: %s", err.Error())))
+			}
+
+			m.Attachments = append(m.Attachments, attachments...)
+		}
+	} else if err := c.Bind(&m); err != nil {
+		return models.ExternalTxMessage{}, err
+	}
+	return m, nil
+}
+
+func validateExternalTxMessage(m models.ExternalTxMessage, app *App) (models.ExternalTxMessage, error) {
+	if len(m.RecipientEmails) > 0 && m.RecipientEmail != "" {
+		return m, echo.NewHTTPError(http.StatusBadRequest,
+			app.i18n.Ts("globals.messages.invalidFields", "name", "do not send `subscriber_email`"))
+	}
+
+	if m.RecipientEmail != "" {
+		m.RecipientEmails = append(m.RecipientEmails, m.RecipientEmail)
+	}
+
+	for n, email := range m.RecipientEmails {
+		if m.RecipientEmail != "" {
+			em, err := app.importer.SanitizeEmail(email)
+			if err != nil {
+				return m, echo.NewHTTPError(http.StatusBadRequest, err.Error())
+			}
+			m.RecipientEmails[n] = em
+		}
+	}
+
+	if m.FromEmail == "" {
+		m.FromEmail = app.constants.FromEmail
+	}
+
+	if m.Messenger == "" {
+		m.Messenger = emailMsgr
+	} else if !app.manager.HasMessenger(m.Messenger) {
+		return m, echo.NewHTTPError(http.StatusBadRequest, app.i18n.Ts("campaigns.fieldInvalidMessenger", "name", m.Messenger))
+	}
+
+	return m, nil
+}
+
+func parseMultiPartMessageDetails(c echo.Context, app *App) ([]string, []models.Attachment, error) {
+	form, err := c.MultipartForm()
+	if err != nil {
+		return nil, nil, echo.NewHTTPError(http.StatusBadRequest,
+			app.i18n.Ts("globals.messages.invalidFields", "name", err.Error()))
+	}
+
+	data, ok := form.Value["data"]
+	if !ok || len(data) != 1 {
+		return nil, nil, echo.NewHTTPError(http.StatusBadRequest,
+			app.i18n.Ts("globals.messages.invalidFields", "name", "data"))
+	}
+
+	attachments := []models.Attachment{}
+	// Attach files.
+	for _, f := range form.File["file"] {
+		file, err := f.Open()
+		if err != nil {
+			return nil, nil, echo.NewHTTPError(http.StatusInternalServerError,
+				app.i18n.Ts("globals.messages.invalidFields", "name", fmt.Sprintf("file: %s", err.Error())))
+		}
+		defer file.Close()
+
+		b, err := io.ReadAll(file)
+		if err != nil {
+			return nil, nil, echo.NewHTTPError(http.StatusInternalServerError,
+				app.i18n.Ts("globals.messages.invalidFields", "name", fmt.Sprintf("file: %s", err.Error())))
+		}
+
+		attachments = append(attachments, models.Attachment{
+			Name:    f.Filename,
+			Header:  manager.MakeAttachmentHeader(f.Filename, "base64", f.Header.Get("Content-Type")),
+			Content: b,
+		})
+	}
+
+	return data, attachments, nil
+}
+
+func getTemplate(app *App, templateId int) (*models.Template, error) {
+	// Get the cached tx template.
+	tpl, err := app.manager.GetTpl(templateId)
+	if err != nil {
+		return nil, echo.NewHTTPError(http.StatusBadRequest,
+			app.i18n.Ts("globals.messages.notFound", "name", fmt.Sprintf("template %d", templateId)))
+	}
+	return tpl, nil
+}
+
+func sendEmail(app *App, msg models.Message) error {
+	if err := app.manager.PushMessage(msg); err != nil {
+		app.log.Printf("error sending message (%s): %v", msg.Subject, err)
+		return err
+	}
+	return nil
+}

docs/docs/content/apis/transactional.md

@@ -2,7 +2,8 @@
 
 | Method | Endpoint | Description                    |
 |:-------|:---------|:-------------------------------|
-| POST   | /api/tx  | Send transactional messages    |
+| POST   | /api/tx  | Send transactional messages to subscribers    |
+| POST   | /api/tx/external  | Send transactional messages to anyone    |
 
 ______________________________________________________________________
 
@@ -50,6 +51,49 @@ EOF
 
 ______________________________________________________________________
 
+#### POST /api/tx/external
+
+Allows sending transactional messages to one or more external recipients via a preconfigured transactional template.
+The recipients don't have to be subscribers.
+This means that the template will not have access to subscriber metadata.
+
+##### Parameters
+
+| Name              | Type      | Required | Description                                                                |
+|:------------------|:----------|:---------|:---------------------------------------------------------------------------|
+| recipient_email  | string    |          | Email of the recipient.              |
+| recipient_emails | string\[\]  |          | Multiple recipient emails as alternative to `recipient_email`.           |
+| template_id       | number    | Yes      | ID of the transactional template to be used for the message.               |
+| from_email        | string    |          | Optional sender email.                                                     |
+| data              | JSON      |          | Optional nested JSON map. Available in the template as `{{ .Tx.Data.* }}`. |
+| headers           | JSON\[\]    |          | Optional array of email headers.                                           |
+| messenger         | string    |          | Messenger to send the message. Default is `email`.                         |
+| content_type      | string    |          | Email format options include `html`, `markdown`, and `plain`.              |
+
+##### Example
+
+```shell
+curl -u "username:password" "http://localhost:9000/api/tx/external" -X POST \
+     -H 'Content-Type: application/json; charset=utf-8' \
+     --data-binary @- << EOF
+    {
+        "recipient_email": "user@test.com",
+        "template_id": 2,
+        "data": {"order_id": "1234", "date": "2022-07-30", "items": [1, 2, 3]},
+        "content_type": "html"
+    }
+EOF
+```
+
+##### Example response
+
+```json
+{
+    "data": true
+}
+```
+______________________________________________________________________
+
 #### File Attachments
 
 To include file attachments in a transactional message, use the `multipart/form-data` Content-Type. Use `data` param for the parameters described above as a JSON object. Include any number of attachments via the `file` param.

docs/site/layouts/index.html

@@ -110,7 +110,7 @@ sh -c "$(curl -fsSL https://raw.githubusercontent.com/knadh/listmonk/master/inst
       <img class="box" src="static/images/tx.png" alt="Screenshot of transactional API" />
     </div>
     <p>
-      Simple API to send arbitrary transactional messages to subscribers
+      Simple API to send arbitrary transactional messages to subscribers and external recipients
       using pre-defined templates. Send messages as e-mail, SMS, Whatsapp messages or any medium via Messenger interfaces.
     </p>
   </section>

docs/swagger/collections.yaml

@@ -1846,6 +1846,29 @@ paths:
                   data:
                     type: boolean
 
+  /tx/external:
+    post:
+      tags:
+        - Transactional
+      description: send message to anyone
+      operationId: transactWithAnyone
+      requestBody:
+        description: email message to recipient
+        content:
+          application/json:
+            schema:
+              $ref: "#/components/schemas/TransactionalMessage"
+      responses:
+        "200":
+          description: OK
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  data:
+                    type: boolean
+
   "/maintenance/subscribers/{type}":
     delete:
       description: garbage collects (deletes) orphaned or blocklisted subscribers.

frontend/src/views/Subscribers.vue

@@ -375,6 +375,7 @@ export default Vue.extend({
           this.$api.blocklistSubscribersByQuery({
             query: this.queryParams.queryExp,
             list_ids: this.queryParams.listID ? [this.queryParams.listID] : null,
+            subscription_status: this.queryParams.subStatus,
           }).then(() => this.querySubscribers());
         };
       }
@@ -426,6 +427,7 @@ export default Vue.extend({
           this.$api.deleteSubscribersByQuery({
             query: this.queryParams.queryExp,
             list_ids: this.queryParams.listID ? [this.queryParams.listID] : null,
+            subscription_status: this.queryParams.subStatus,
           }).then(() => {
             this.querySubscribers();
 
@@ -460,6 +462,7 @@ export default Vue.extend({
       } else {
         // 'All' is selected, perform by query.
         data.query = this.queryParams.queryExp;
+        data.subscription_status = this.queryParams.subStatus;
         fn = this.$api.addSubscribersToListsByQuery;
       }
 
@@ -503,7 +506,6 @@ export default Vue.extend({
       // Get subscribers on load.
       this.querySubscribers();
     }
-
     if (this.$route.query.subscription_status) {
       this.queryParams.subStatus = this.$route.query.subscription_status;
     }

internal/core/subscribers.go

@@ -394,8 +394,8 @@ func (c *Core) BlocklistSubscribers(subIDs []int) error {
 }
 
 // BlocklistSubscribersByQuery blocklists the given list of subscribers.
-func (c *Core) BlocklistSubscribersByQuery(query string, listIDs []int) error {
-	if err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.BlocklistSubscribersByQuery, listIDs, c.db); err != nil {
+func (c *Core) BlocklistSubscribersByQuery(query string, listIDs []int, subStatus string) error {
+	if err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.BlocklistSubscribersByQuery, listIDs, c.db, subStatus); err != nil {
 		c.log.Printf("error blocklisting subscribers: %v", err)
 		return echo.NewHTTPError(http.StatusInternalServerError,
 			c.i18n.Ts("subscribers.errorBlocklisting", "error", pqErrMsg(err)))
@@ -423,8 +423,8 @@ func (c *Core) DeleteSubscribers(subIDs []int, subUUIDs []string) error {
 }
 
 // DeleteSubscribersByQuery deletes subscribers by a given arbitrary query expression.
-func (c *Core) DeleteSubscribersByQuery(query string, listIDs []int) error {
-	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.DeleteSubscribersByQuery, listIDs, c.db)
+func (c *Core) DeleteSubscribersByQuery(query string, listIDs []int, subStatus string) error {
+	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.DeleteSubscribersByQuery, listIDs, c.db, subStatus)
 	if err != nil {
 		c.log.Printf("error deleting subscribers: %v", err)
 		return echo.NewHTTPError(http.StatusInternalServerError,

internal/core/subscriptions.go

@@ -35,12 +35,12 @@ func (c *Core) AddSubscriptions(subIDs, listIDs []int, status string) error {
 
 // AddSubscriptionsByQuery adds list subscriptions to subscribers by a given arbitrary query expression.
 // sourceListIDs is the list of list IDs to filter the subscriber query with.
-func (c *Core) AddSubscriptionsByQuery(query string, sourceListIDs, targetListIDs []int, status string) error {
+func (c *Core) AddSubscriptionsByQuery(query string, sourceListIDs, targetListIDs []int, status string, subStatus string) error {
 	if sourceListIDs == nil {
 		sourceListIDs = []int{}
 	}
 
-	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.AddSubscribersToListsByQuery, sourceListIDs, c.db, pq.Array(targetListIDs), status)
+	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.AddSubscribersToListsByQuery, sourceListIDs, c.db, subStatus, pq.Array(targetListIDs), status)
 	if err != nil {
 		c.log.Printf("error adding subscriptions by query: %v", err)
 		return echo.NewHTTPError(http.StatusInternalServerError,
@@ -64,12 +64,12 @@ func (c *Core) DeleteSubscriptions(subIDs, listIDs []int) error {
 
 // DeleteSubscriptionsByQuery deletes list subscriptions from subscribers by a given arbitrary query expression.
 // sourceListIDs is the list of list IDs to filter the subscriber query with.
-func (c *Core) DeleteSubscriptionsByQuery(query string, sourceListIDs, targetListIDs []int) error {
+func (c *Core) DeleteSubscriptionsByQuery(query string, sourceListIDs, targetListIDs []int, subStatus string) error {
 	if sourceListIDs == nil {
 		sourceListIDs = []int{}
 	}
 
-	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.DeleteSubscriptionsByQuery, sourceListIDs, c.db, pq.Array(targetListIDs))
+	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.DeleteSubscriptionsByQuery, sourceListIDs, c.db, subStatus, pq.Array(targetListIDs))
 	if err != nil {
 		c.log.Printf("error deleting subscriptions by query: %v", err)
 		return echo.NewHTTPError(http.StatusInternalServerError,
@@ -92,12 +92,12 @@ func (c *Core) UnsubscribeLists(subIDs, listIDs []int, listUUIDs []string) error
 
 // UnsubscribeListsByQuery sets list subscriptions to 'unsubscribed' by a given arbitrary query expression.
 // sourceListIDs is the list of list IDs to filter the subscriber query with.
-func (c *Core) UnsubscribeListsByQuery(query string, sourceListIDs, targetListIDs []int) error {
+func (c *Core) UnsubscribeListsByQuery(query string, sourceListIDs, targetListIDs []int, subStatus string) error {
 	if sourceListIDs == nil {
 		sourceListIDs = []int{}
 	}
 
-	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.UnsubscribeSubscribersFromListsByQuery, sourceListIDs, c.db, pq.Array(targetListIDs))
+	err := c.q.ExecSubQueryTpl(sanitizeSQLExp(query), c.q.UnsubscribeSubscribersFromListsByQuery, sourceListIDs, c.db, subStatus, pq.Array(targetListIDs))
 	if err != nil {
 		c.log.Printf("error unsubscribing from lists by query: %v", err)
 		return echo.NewHTTPError(http.StatusInternalServerError,

models/models.go

@@ -409,6 +409,55 @@ type TxMessage struct {
 	SubjectTpl *txttpl.Template   `json:"-"`
 }
 
+// ExternalTxMessage represents an e-mail campaign to an external recipient.
+type ExternalTxMessage struct {
+	RecipientEmails []string `json:"recipient_emails"`
+
+	// Deprecated.
+	RecipientEmail string `json:"recipient_email"`
+
+	TemplateID  int                    `json:"template_id"`
+	Data        map[string]interface{} `json:"data"`
+	FromEmail   string                 `json:"from_email"`
+	Headers     Headers                `json:"headers"`
+	ContentType string                 `json:"content_type"`
+	Messenger   string                 `json:"messenger"`
+
+	// File attachments added from multi-part form data.
+	Attachments []Attachment `json:"-"`
+
+	Subject    string             `json:"-"`
+	Body       []byte             `json:"-"`
+	Tpl        *template.Template `json:"-"`
+	SubjectTpl *txttpl.Template   `json:"-"`
+}
+
+func (m ExternalTxMessage) MapToTxMessage() TxMessage {
+	txMessage := TxMessage{}
+	txMessage.SubscriberEmails = m.RecipientEmails
+	txMessage.SubscriberIDs = []int{}
+
+	// Deprecated.
+	txMessage.SubscriberEmail = m.RecipientEmail
+	txMessage.SubscriberID = 0
+
+	txMessage.TemplateID = m.TemplateID
+	txMessage.Data = m.Data
+	txMessage.FromEmail = m.FromEmail
+	txMessage.Headers = m.Headers
+	txMessage.ContentType = m.ContentType
+	txMessage.Messenger = m.Messenger
+
+	// File attachments added from multi-part form data.
+	txMessage.Attachments = m.Attachments
+
+	txMessage.Subject = m.Subject
+	txMessage.Body = m.Body
+	txMessage.Tpl = m.Tpl
+	txMessage.SubjectTpl = m.SubjectTpl
+	return txMessage
+}
+
 // markdown is a global instance of Markdown parser and renderer.
 var markdown = goldmark.New(
 	goldmark.WithParserOptions(
@@ -651,27 +700,45 @@ func (m *TxMessage) Render(sub Subscriber, tpl *Template) error {
 		Tx         *TxMessage
 	}{sub, m}
 
+	message, err := renderMessage(data, tpl)
+	if err != nil {
+		return err
+	}
+	m.Body = message
+
+	subject, err := renderSubject(data, tpl)
+	if err != nil {
+		return err
+	}
+	m.Subject = subject
+
+	return nil
+}
+
+func renderMessage(data any, tpl *Template) ([]byte, error) {
 	// Render the body.
 	b := bytes.Buffer{}
 	if err := tpl.Tpl.ExecuteTemplate(&b, BaseTpl, data); err != nil {
-		return err
+		return nil, err
+	}
+	message := make([]byte, b.Len())
+	copy(message, b.Bytes())
+	return message, nil
 }
-	m.Body = make([]byte, b.Len())
-	copy(m.Body, b.Bytes())
-	b.Reset()
 
+func renderSubject(data any, tpl *Template) (string, error) {
 	// If the subject is also a template, render that.
+	b := bytes.Buffer{}
+	subject := ""
 	if tpl.SubjectTpl != nil {
 		if err := tpl.SubjectTpl.ExecuteTemplate(&b, BaseTpl, data); err != nil {
-			return err
+			return "", err
 		}
-		m.Subject = b.String()
-		b.Reset()
+		subject = b.String()
 	} else {
-		m.Subject = tpl.Subject
+		subject = tpl.Subject
 	}
-
-	return nil
+	return subject, nil
 }
 
 // FirstName splits the name by spaces and returns the first chunk
@@ -738,3 +805,33 @@ func (h Headers) Value() (driver.Value, error) {
 
 	return "[]", nil
 }
+
+func CreateMailMessage(sub Subscriber, m TxMessage) Message {
+	msg := Message{}
+	msg.Subscriber = sub
+	msg.To = []string{sub.Email}
+	msg.From = m.FromEmail
+	msg.Subject = m.Subject
+	msg.ContentType = m.ContentType
+	msg.Messenger = m.Messenger
+	msg.Body = m.Body
+	for _, a := range m.Attachments {
+		msg.Attachments = append(msg.Attachments, Attachment{
+			Name:    a.Name,
+			Header:  a.Header,
+			Content: a.Content,
+		})
+	}
+
+	// Optional headers.
+	if len(m.Headers) != 0 {
+		msg.Headers = make(textproto.MIMEHeader, len(m.Headers))
+		for _, set := range m.Headers {
+			for hdr, val := range set {
+				msg.Headers.Add(hdr, val)
+			}
+		}
+	}
+
+	return msg
+}

models/queries.go

@@ -114,7 +114,7 @@ type Queries struct {
 // out of it using the raw `query-subscribers-template` query template.
 // While doing this, a readonly transaction is created and the query is
 // dry run on it to ensure that it is indeed readonly.
-func (q *Queries) CompileSubscriberQueryTpl(exp string, db *sqlx.DB) (string, error) {
+func (q *Queries) CompileSubscriberQueryTpl(exp string, db *sqlx.DB, subStatus string) (string, error) {
 	tx, err := db.BeginTxx(context.Background(), &sql.TxOptions{ReadOnly: true})
 	if err != nil {
 		return "", err
@@ -126,19 +126,18 @@ func (q *Queries) CompileSubscriberQueryTpl(exp string, db *sqlx.DB) (string, er
 		exp = " AND " + exp
 	}
 	stmt := fmt.Sprintf(q.QuerySubscribersTpl, exp)
-	if _, err := tx.Exec(stmt, true, pq.Int64Array{}); err != nil {
+	if _, err := tx.Exec(stmt, true, pq.Int64Array{}, subStatus); err != nil {
 		return "", err
 	}
-
 	return stmt, nil
 }
 
 // compileSubscriberQueryTpl takes an arbitrary WHERE expressions and a subscriber
 // query template that depends on the filter (eg: delete by query, blocklist by query etc.)
 // combines and executes them.
-func (q *Queries) ExecSubQueryTpl(exp, tpl string, listIDs []int, db *sqlx.DB, args ...interface{}) error {
+func (q *Queries) ExecSubQueryTpl(exp, tpl string, listIDs []int, db *sqlx.DB, subStatus string, args ...interface{}) error {
 	// Perform a dry run.
-	filterExp, err := q.CompileSubscriberQueryTpl(exp, db)
+	filterExp, err := q.CompileSubscriberQueryTpl(exp, db, subStatus)
 	if err != nil {
 		return err
 	}
@@ -148,10 +147,9 @@ func (q *Queries) ExecSubQueryTpl(exp, tpl string, listIDs []int, db *sqlx.DB, a
 	}
 
 	// First argument is the boolean indicating if the query is a dry run.
-	a := append([]interface{}{false, pq.Array(listIDs)}, args...)
+	a := append([]interface{}{false, pq.Array(listIDs), subStatus}, args...)
 	if _, err := db.Exec(fmt.Sprintf(tpl, filterExp), a...); err != nil {
 		return err
 	}
-
 	return nil
 }

queries.sql

@@ -370,6 +370,7 @@ ON (
     -- Optional list filtering.
     (CASE WHEN CARDINALITY($2::INT[]) > 0 THEN true ELSE false END)
     AND subscriber_lists.subscriber_id = subscribers.id
+    AND ($3 = '' OR subscriber_lists.status = $3::subscription_status)
 )
 WHERE subscriber_lists.list_id = ALL($2::INT[]) %s
 LIMIT (CASE WHEN $1 THEN 1 END)
@@ -393,20 +394,20 @@ UPDATE subscriber_lists SET status='unsubscribed', updated_at=NOW()
 -- raw: true
 WITH subs AS (%s)
 INSERT INTO subscriber_lists (subscriber_id, list_id, status)
-    (SELECT a, b, (CASE WHEN $4 != '' THEN $4::subscription_status ELSE 'unconfirmed' END) FROM UNNEST(ARRAY(SELECT id FROM subs)) a, UNNEST($3::INT[]) b)
+    (SELECT a, b, (CASE WHEN $5 != '' THEN $5::subscription_status ELSE 'unconfirmed' END) FROM UNNEST(ARRAY(SELECT id FROM subs)) a, UNNEST($4::INT[]) b)
     ON CONFLICT (subscriber_id, list_id) DO NOTHING;
 
 -- name: delete-subscriptions-by-query
 -- raw: true
 WITH subs AS (%s)
 DELETE FROM subscriber_lists
-    WHERE (subscriber_id, list_id) = ANY(SELECT a, b FROM UNNEST(ARRAY(SELECT id FROM subs)) a, UNNEST($3::INT[]) b);
+    WHERE (subscriber_id, list_id) = ANY(SELECT a, b FROM UNNEST(ARRAY(SELECT id FROM subs)) a, UNNEST($4::INT[]) b);
 
 -- name: unsubscribe-subscribers-from-lists-by-query
 -- raw: true
 WITH subs AS (%s)
 UPDATE subscriber_lists SET status='unsubscribed', updated_at=NOW()
-    WHERE (subscriber_id, list_id) = ANY(SELECT a, b FROM UNNEST(ARRAY(SELECT id FROM subs)) a, UNNEST($3::INT[]) b);
+    WHERE (subscriber_id, list_id) = ANY(SELECT a, b FROM UNNEST(ARRAY(SELECT id FROM subs)) a, UNNEST($4::INT[]) b);
 
 
 -- lists