Add optional per-campaign subscriber query for send-time segmentation

cmd/campaigns.go

@@ -288,6 +288,11 @@ func (a *App) CreateCampaign(c echo.Context) error {
 		o.ArchiveTemplateID = o.TemplateID
 	}
 
+	// Normalize, permission-gate, and validate the optional segment query.
+	if err := a.processCampaignSegment(c, &o, ""); err != nil {
+		return err
+	}
+
 	out, err := a.core.CreateCampaign(o.Campaign, o.ListIDs, o.MediaIDs)
 	if err != nil {
 		return err
@@ -340,6 +345,11 @@ func (a *App) UpdateCampaign(c echo.Context) error {
 		o = c
 	}
 
+	// Normalize, permission-gate, and validate the optional segment query.
+	if err := a.processCampaignSegment(c, &o, cm.SubscriberQuery.String); err != nil {
+		return err
+	}
+
 	out, err := a.core.UpdateCampaign(id, o.Campaign, o.ListIDs, o.MediaIDs)
 	if err != nil {
 		return err
@@ -348,6 +358,68 @@ func (a *App) UpdateCampaign(c echo.Context) error {
 	return c.JSON(http.StatusOK, okResp{out})
 }
 
+// processCampaignSegment normalizes, permission-gates, and validates a campaign's optional
+// subscriber_query segment in place. Segments apply only to regular campaigns; the permission
+// and validation run only when the query is newly set or changed from oldQuery.
+func (a *App) processCampaignSegment(c echo.Context, o *campReq, oldQuery string) error {
+	q := formatSQLExp(o.SubscriberQuery.String)
+
+	// Segments only apply to regular campaigns.
+	if o.Type != models.CampaignTypeRegular {
+		o.SubscriberQuery = null.String{}
+		return nil
+	}
+
+	if q != "" && q != formatSQLExp(oldQuery) {
+		// Writing an arbitrary SQL segment needs the same permission as subscriber SQL queries.
+		user := auth.GetUser(c)
+		if !user.HasPerm(auth.PermSubscribersSqlQuery) {
+			return echo.NewHTTPError(http.StatusForbidden,
+				a.i18n.Ts("globals.messages.permissionDenied", "name", auth.PermSubscribersSqlQuery))
+		}
+		if err := a.core.ValidateCampaignQuery(o.ListIDs, models.CampaignTypeRegular, q); err != nil {
+			return err
+		}
+	}
+
+	if q == "" {
+		o.SubscriberQuery = null.String{}
+	} else {
+		o.SubscriberQuery = null.NewString(q, true)
+	}
+
+	return nil
+}
+
+// PreviewCampaignRecipients returns how many subscribers would receive a campaign sent to the
+// given lists with an optional ad-hoc segment query, applying send-time consent rules.
+func (a *App) PreviewCampaignRecipients(c echo.Context) error {
+	user := auth.GetUser(c)
+
+	var req struct {
+		Lists []int  `json:"lists"`
+		Query string `json:"query"`
+	}
+	if err := c.Bind(&req); err != nil {
+		return err
+	}
+
+	listIDs := user.GetPermittedListIDs(req.Lists)
+	if len(listIDs) == 0 {
+		return echo.NewHTTPError(http.StatusBadRequest,
+			a.i18n.Ts("globals.messages.missingFields", "name", "{globals.terms.lists}"))
+	}
+
+	count, err := a.core.CountCampaignRecipients(listIDs, models.CampaignTypeRegular, formatSQLExp(req.Query))
+	if err != nil {
+		return err
+	}
+
+	return c.JSON(http.StatusOK, okResp{struct {
+		Count int `json:"count"`
+	}{count}})
+}
+
 // UpdateCampaignStatus handles campaign status modification.
 func (a *App) UpdateCampaignStatus(c echo.Context) error {
 	// Get the campaign ID.

cmd/handlers.go

@@ -168,6 +168,7 @@ func initHTTPHandlers(e *echo.Echo, a *App) {
 		g.POST("/api/campaigns/:id/content", pm(hasID(a.CampaignContent), "campaigns:manage_all", "campaigns:manage"))
 		g.POST("/api/campaigns/:id/text", pm(hasID(a.PreviewCampaign), "campaigns:get"))
 		g.POST("/api/campaigns/:id/test", pm(hasID(a.TestCampaign), "campaigns:manage_all", "campaigns:manage"))
+		g.POST("/api/campaigns/preview-recipients", pm(a.PreviewCampaignRecipients, "subscribers:sql_query"))
 		g.POST("/api/campaigns", pm(a.CreateCampaign, "campaigns:manage_all", "campaigns:manage"))
 		g.PUT("/api/campaigns/:id", pm(hasID(a.UpdateCampaign), "campaigns:manage_all", "campaigns:manage"))
 		g.PUT("/api/campaigns/:id/status", pm(hasID(a.UpdateCampaignStatus), "campaigns:send"))

cmd/install.go

@@ -270,6 +270,7 @@ func installCampaign(campTplID, archiveTplID int, q *models.Queries) {
 		`{"name": "Subscriber"}`,
 		nil,
 		nil,
+		nil,
 	); err != nil {
 		lo.Fatalf("error creating sample campaign: %v", err)
 	}

cmd/manager_store.go

@@ -1,12 +1,15 @@
 package main
 
 import (
+	"strings"
+
 	"github.com/gofrs/uuid/v5"
 	"github.com/knadh/listmonk/internal/core"
 	"github.com/knadh/listmonk/internal/manager"
 	"github.com/knadh/listmonk/internal/media"
 	"github.com/knadh/listmonk/models"
 	"github.com/lib/pq"
+	null "gopkg.in/volatiletech/null.v6"
 )
 
 // store implements DataSource over the primary
@@ -18,11 +21,12 @@ type store struct {
 }
 
 type runningCamp struct {
-	CampaignID       int    `db:"campaign_id"`
-	CampaignType     string `db:"campaign_type"`
-	LastSubscriberID int    `db:"last_subscriber_id"`
-	MaxSubscriberID  int    `db:"max_subscriber_id"`
-	ListID           int    `db:"list_id"`
+	CampaignID       int         `db:"campaign_id"`
+	CampaignType     string      `db:"campaign_type"`
+	LastSubscriberID int         `db:"last_subscriber_id"`
+	MaxSubscriberID  int         `db:"max_subscriber_id"`
+	ListID           int         `db:"list_id"`
+	SubscriberQuery  null.String `db:"subscriber_query"`
 }
 
 func newManagerStore(q *models.Queries, c *core.Core, m media.Store) *store {
@@ -61,9 +65,23 @@ func (s *store) NextSubscribers(campID, limit int) ([]models.Subscriber, error)
 		return nil, nil
 	}
 
-	var out []models.Subscriber
-	err := s.queries.NextCampaignSubscribers.Select(&out, camps[0].CampaignID, camps[0].CampaignType, camps[0].LastSubscriberID, camps[0].MaxSubscriberID, pq.Array(listIDs), limit)
-	return out, err
+	rc := camps[0]
+
+	// No segment: unchanged prepared-statement fast path.
+	if !rc.SubscriberQuery.Valid || strings.TrimSpace(rc.SubscriberQuery.String) == "" {
+		var out []models.Subscriber
+		err := s.queries.NextCampaignSubscribers.Select(&out, rc.CampaignID, rc.CampaignType, rc.LastSubscriberID, rc.MaxSubscriberID, pq.Array(listIDs), limit)
+		return out, err
+	}
+
+	// Segment present: splice the (validated-at-save) query into the filtered template.
+	return s.core.NextCampaignFilteredSubscribers(rc.CampaignID, rc.CampaignType, rc.LastSubscriberID, rc.MaxSubscriberID, listIDs, rc.SubscriberQuery.String, limit)
+}
+
+// SetCampaignToSend recomputes and persists to_send for a filtered campaign and returns the
+// new value. Called once when a filtered campaign starts.
+func (s *store) SetCampaignToSend(c *models.Campaign) (int, error) {
+	return s.core.SetCampaignFilteredToSend(c.ID, c.SubscriberQuery.String)
 }
 
 // GetCampaign fetches a campaign from the database.

cmd/upgrade.go

@@ -45,6 +45,7 @@ var migList = []migFunc{
 	{"v6.0.0", migrations.V6_0_0},
 	{"v6.1.0", migrations.V6_1_0},
 	{"v6.2.0", migrations.V6_2_0},
+	{"v6.3.0", migrations.V6_3_0},
 }
 
 // upgrade upgrades the database to the current version by running SQL migration files

deploy/README.md

@@ -0,0 +1,51 @@
+# Deploying this fork
+
+This is a fork of listmonk with extra backend features (e.g. per-campaign subscriber
+segments). It is deployed to the prealpha `selfhosted-tools` stack (Docker Compose + Traefik
+on a VM). There is **no registry**: a self-contained image is built locally and shipped to the
+server over SSH.
+
+## One-time: point the stack at the fork image
+
+In `selfhosted-tools/services/listmonk.yml`:
+
+```yaml
+  listmonk:
+    image: listmonk-fork:prod      # was: listmonk/listmonk:latest
+    pull_policy: never             # use the locally-loaded image, never a registry
+    # ... unchanged ...
+    labels:
+      # remove: com.centurylinklabs.watchtower.scope=auto-update
+```
+
+Commit and push to `main` so it goes through the normal GitLab pipeline. `pull_policy: never`
+means a deploy fails loudly if the image was not shipped first (instead of pulling upstream).
+
+## Each deploy
+
+1. **Back up the DB** (`selfhosted-tools/scripts/backup-restore.sh`).
+2. **Build + ship** the image from this checkout:
+   ```sh
+   DEPLOY_HOST=<server-host> ./deploy/ship-fork-image.sh
+   ```
+   Needs Docker (buildx), Go, and Node + Yarn on the build machine. It cross-compiles a
+   linux/amd64 self-contained binary, builds `listmonk-fork:prod`, and `docker save | ssh
+   docker load`s it onto the server. Add `RUN_DEPLOY=1` to also run the remote `deploy.sh`.
+3. **Roll out** (if not using `RUN_DEPLOY=1`): trigger the normal selfhosted-tools deploy
+   (push to `main`), or on the server `docker compose ... up -d listmonk`. The container's
+   command runs `--upgrade`, applying pending migrations automatically.
+
+## Verify
+
+- Admin UI footer shows `v6.3.0-fork+<sha>`.
+- New campaign form shows the **Segment** field and **Preview recipients** button.
+
+## Notes
+
+- The `subscriber_query` migration is `v6.3.0` and is an `ADD COLUMN IF NOT EXISTS` (nullable):
+  instant and non-destructive even on large subscriber tables.
+- Fork maintenance: if a future upstream release also ships a `v6.3.0` migration, reconcile it
+  (ours is idempotent and safe to re-run; rename if you need upstream's `v6.3.0` to still apply).
+- `ship-fork-image.sh` overwrites `listmonk-fork:prod` each run; the old image becomes dangling
+  and is cleaned by the server's `docker image prune -f`. The `:<sha>` tags accumulate; prune
+  occasionally.

deploy/ship-fork-image.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+# Build the listmonk fork into a self-contained linux/amd64 image and ship it to the
+# server over SSH (docker save | docker load), no registry. KISS deploy for the fork.
+#
+# Usage:
+#   DEPLOY_HOST=1.2.3.4 ./deploy/ship-fork-image.sh            # build + ship
+#   BUILD_ONLY=1 ./deploy/ship-fork-image.sh                   # build locally only
+#   DEPLOY_HOST=1.2.3.4 RUN_DEPLOY=1 ./deploy/ship-fork-image.sh  # ship + run remote deploy.sh
+#
+# Requires on the build host: docker (with buildx), Go, Node + Yarn (frontend toolchain).
+set -euo pipefail
+
+IMAGE="${IMAGE:-listmonk-fork}"
+PLATFORM="${PLATFORM:-linux/amd64}"
+DEPLOY_USER="${DEPLOY_USER:-deploy}"
+DEPLOY_HOST="${DEPLOY_HOST:-}"
+REMOTE_INFRA_DIR="${REMOTE_INFRA_DIR:-/home/deploy/infra}"
+BUILD_ONLY="${BUILD_ONLY:-0}"
+RUN_DEPLOY="${RUN_DEPLOY:-0}"
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT"
+
+SHA="$(git rev-parse --short HEAD)"
+VERSION="${LISTMONK_VERSION:-v6.3.0-fork+$SHA}"
+TAG="$IMAGE:prod"
+TAG_SHA="$IMAGE:$SHA"
+
+# corepack ships Yarn 1.x with Node; fall back to it if yarn isn't on PATH.
+YARN="${YARN:-yarn}"
+command -v "$YARN" >/dev/null 2>&1 || YARN="corepack yarn"
+
+echo "==> building self-contained binary ($PLATFORM, version $VERSION)"
+# stuffbin packs assets onto the binary and runs on the host, so install it for the host
+# arch first; otherwise the cross-compile env below would build a linux stuffbin make can't run.
+GOOS="$(go env GOHOSTOS)" GOARCH="$(go env GOHOSTARCH)" go install github.com/knadh/stuffbin/...@latest
+# Drop any stale (host-arch) binary so make actually cross-compiles instead of reusing it.
+rm -f listmonk
+GOOS="${PLATFORM%%/*}" GOARCH="${PLATFORM##*/}" LISTMONK_VERSION="$VERSION" \
+  make dist YARN="$YARN"
+
+echo "==> building image $TAG"
+docker buildx build --platform "$PLATFORM" -t "$TAG" -t "$TAG_SHA" --load .
+
+if [[ "$BUILD_ONLY" == "1" ]]; then
+  echo "==> build-only: $TAG ready locally ($VERSION)"
+  exit 0
+fi
+
+[[ -n "$DEPLOY_HOST" ]] || { echo "ERROR: set DEPLOY_HOST (or BUILD_ONLY=1)" >&2; exit 1; }
+
+echo "==> shipping $TAG to $DEPLOY_USER@$DEPLOY_HOST"
+docker save "$TAG" "$TAG_SHA" | ssh "$DEPLOY_USER@$DEPLOY_HOST" 'docker load'
+
+if [[ "$RUN_DEPLOY" == "1" ]]; then
+  echo "==> running remote deploy.sh"
+  ssh "$DEPLOY_USER@$DEPLOY_HOST" "bash $REMOTE_INFRA_DIR/scripts/deploy.sh"
+else
+  echo "==> image loaded. Deploy via your pipeline (push selfhosted-tools), or re-run with RUN_DEPLOY=1."
+fi

frontend/src/api/index.js

@@ -302,6 +302,12 @@ export const createCampaign = async (data) => http.post(
   { loading: models.campaigns },
 );
 
+export const previewCampaignRecipients = async (data) => http.post(
+  '/api/campaigns/preview-recipients',
+  data,
+  { loading: models.campaigns },
+);
+
 export const getCampaignViewCounts = async (params) => http.get(
   '/api/campaigns/analytics/views',
   { params, loading: models.campaigns },

frontend/src/views/Campaign.vue

@@ -81,6 +81,22 @@
                 <list-selector v-model="form.lists" :selected="form.lists" :all="lists.results" :disabled="!canEdit"
                   :label="$t('globals.terms.lists')" :placeholder="$t('campaigns.sendToLists')" />
 
+                <b-field v-if="form.messenger.startsWith('email') && $can('subscribers:sql_query')"
+                  :label="$t('campaigns.segment')" label-position="on-border" :message="$t('campaigns.segmentHelp')">
+                  <b-input v-model="form.subscriberQuery" type="textarea" :disabled="!canEdit" name="subscriber_query"
+                    placeholder="subscribers.attribs->>'city' = 'Berlin'" />
+                </b-field>
+                <div v-if="form.messenger.startsWith('email') && $can('subscribers:sql_query') && form.subscriberQuery"
+                  class="mb-4">
+                  <b-button icon-left="account-search-outline" :disabled="form.lists.length === 0"
+                    @click="onPreviewRecipients">
+                    {{ $t('campaigns.previewRecipients') }}
+                  </b-button>
+                  <span v-if="recipientCount !== null" class="ml-2 has-text-grey">
+                    {{ $t('campaigns.recipientCount', { num: recipientCount }) }}
+                  </span>
+                </div>
+
                 <div class="columns">
                   <div class="column is-6">
                     <b-field :label="$tc('globals.terms.messenger')" label-position="on-border">
@@ -359,6 +375,9 @@ export default Vue.extend({
       isPreviewingArchive: false,
       activeTab: 'campaign',
 
+      // Transient count from the "Preview recipients" button for the segment query.
+      recipientCount: null,
+
       data: {},
 
       // IDs from ?list_id query param.
@@ -375,6 +394,7 @@ export default Vue.extend({
         attribsStr: '{}',
         messenger: 'email',
         lists: [],
+        subscriberQuery: '',
         tags: [],
         sendAt: null,
         content: {
@@ -571,6 +591,7 @@ export default Vue.extend({
         name: this.form.name,
         subject: this.form.subject,
         lists: this.form.lists.map((l) => l.id),
+        subscriber_query: this.form.subscriberQuery || null,
         from_email: this.form.fromEmail,
         content_type: this.form.content.contentType,
         messenger: this.form.messenger,
@@ -588,12 +609,22 @@ export default Vue.extend({
       return false;
     },
 
+    onPreviewRecipients() {
+      this.$api.previewCampaignRecipients({
+        lists: this.form.lists.map((l) => l.id),
+        query: this.form.subscriberQuery,
+      }).then((d) => {
+        this.recipientCount = d.count;
+      });
+    },
+
     async updateCampaign(typ) {
       const data = {
         archive_slug: this.form.archiveSlug,
         name: this.form.name,
         subject: this.form.subject,
         lists: this.form.lists.map((l) => l.id),
+        subscriber_query: this.form.subscriberQuery || null,
         from_email: this.form.fromEmail,
         messenger: this.form.messenger,
         type: 'regular',

i18n/en.json

@@ -86,6 +86,10 @@
     "campaigns.sendTest": "Send test message",
     "campaigns.sendTestHelp": "Hit Enter after typing an address to add multiple recipients. The addresses must belong to existing subscribers.",
     "campaigns.sendToLists": "Lists to send to",
+    "campaigns.segment": "Segment",
+    "campaigns.segmentHelp": "Optional SQL expression to send only to a subset of the selected lists' subscribers. Combined (AND) with list and consent rules at send time.",
+    "campaigns.previewRecipients": "Preview recipients",
+    "campaigns.recipientCount": "{num} recipients match",
     "campaigns.sent": "Sent",
     "campaigns.start": "Start campaign",
     "campaigns.started": "\"{name}\" started",