will/eec-campaigns
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix campaign/listmonk API shapes against confirmed source, not guesses

Browse Source 
Read knadh/listmonk's actual Go source (cmd/campaigns.go, cmd/subscribers.go)
instead of relying on incomplete public docs. Two corrections: the media
attach field is "media" (plain IDs), not "media_ids" as originally guessed;
and there's a single query-based bulk list action
(PUT /api/subscribers/query/lists) that applies a segment_query server-side,
so resolveSegment no longer fetches subscriber IDs client-side before
bulk-adding them. Create-status-defaults-to-draft and the test-send
"subscribers" field are now confirmed rather than flagged TODO(verify).
This commit is contained in:
Will Estes
2026-07-10 08:37:07 -04:00
parent b16d2b6c5f
commit bcf8df5579
5 changed files with 109 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/campaign/sync.go
+7 -9
View File
@@ -157,21 +157,19 @@ func resolveLists(lm *listmonk.Client, names []string) ([]int, error) {
// resolveSegment materializes a segment_query into list membership, since a
// listmonk campaign can only target whole list(s), not an arbitrary query
// directly (still an open, unmerged upstream feature request — see the
// plan's Open items). The list is a point-in-time snapshot: re-running sync
// re-evaluates the query and re-syncs membership, so a later push before
// send picks up newly-matching subscribers.
// plan's Open items). listmonk applies the query server-side in one call
// (QueryAddToList) — no subscriber IDs are ever fetched client-side. The
// list is a point-in-time snapshot: re-running sync re-evaluates the query
// and re-syncs membership, so a later push before send picks up
// newly-matching subscribers.
func resolveSegment(lm *listmonk.Client, slug, query string) (int, error) {
ids, err := lm.QuerySubscriberIDs(query)
if err != nil {
return 0, fmt.Errorf("segment_query: %w", err)
}
listName := "segment:" + slug
listID, err := lm.FindOrCreateListByName(listName)
if err != nil {
return 0, fmt.Errorf("segment list %q: %w", listName, err)
}
if err := lm.BulkAddToList(ids, listID); err != nil {
return 0, fmt.Errorf("segment list %q: %w", listName, err)
if err := lm.QueryAddToList(query, listID); err != nil {
return 0, fmt.Errorf("segment_query: %w", err)
}
return listID, nil
}
internal/campaign/sync_test.go
+42 -45
View File
@@ -49,25 +49,25 @@ type testCall struct {
Emails []string
}
type bulkAddCall struct {
IDs []int
type segmentQueryCall struct {
Query string
ListID int
}
type fakeListmonk struct {
mu sync.Mutex
t *testing.T
nextID int
lists []fakeList
campaigns []fakeCampaign
media []fakeMedia
subscriberIDsForQuery map[string][]int
testCalls []testCall
bulkAddCalls []bulkAddCall
mu sync.Mutex
t *testing.T
nextID int
lists []fakeList
campaigns []fakeCampaign
media []fakeMedia
testCalls []testCall
segmentQueryCalls []segmentQueryCall
failSegmentQuery bool
}
func newFakeListmonk(t *testing.T) *fakeListmonk {
return &fakeListmonk{t: t, nextID: 1, subscriberIDsForQuery: map[string][]int{}}
return &fakeListmonk{t: t, nextID: 1}
}
func (f *fakeListmonk) id() int {
@@ -105,10 +105,8 @@ func (f *fakeListmonk) handle(w http.ResponseWriter, r *http.Request) {
f.writeMedia(w)
case r.Method == http.MethodPost && r.URL.Path == "/api/media":
f.uploadMedia(w, r)
case r.Method == http.MethodGet && r.URL.Path == "/api/subscribers":
f.querySubscribers(w, r)
case r.Method == http.MethodPut && r.URL.Path == "/api/subscribers/lists":
f.bulkAdd(w, r)
case r.Method == http.MethodPut && r.URL.Path == "/api/subscribers/query/lists":
f.queryAddToList(w, r)
default:
f.t.Errorf("fakeListmonk: unhandled request %s %s", r.Method, r.URL.Path)
w.WriteHeader(http.StatusNotFound)
@@ -120,10 +118,14 @@ func (c fakeCampaign) toJSON() map[string]any {
for _, id := range c.ListIDs {
lists = append(lists, map[string]any{"id": id})
}
media := make([]map[string]any, 0, len(c.MediaIDs))
for _, id := range c.MediaIDs {
media = append(media, map[string]any{"id": id})
}
return map[string]any{
"id": c.ID, "name": c.Name, "status": c.Status, "subject": c.Subject,
"body": c.Body, "from_email": c.FromEmail, "template_id": c.TemplateID,
"type": c.Type, "tags": c.Tags, "lists": lists, "media_ids": c.MediaIDs,
"type": c.Type, "tags": c.Tags, "lists": lists, "media": media,
}
}
@@ -142,7 +144,7 @@ func (f *fakeListmonk) createCampaign(w http.ResponseWriter, r *http.Request) {
Subject: str(body["subject"]), Body: str(body["body"]),
FromEmail: str(body["from_email"]), TemplateID: toInt(body["template_id"]),
Type: str(body["type"]), Tags: toStrings(body["tags"]),
ListIDs: toInts(body["lists"]), MediaIDs: toInts(body["media_ids"]),
ListIDs: toInts(body["lists"]), MediaIDs: toInts(body["media"]),
}
f.campaigns = append(f.campaigns, c)
json.NewEncoder(w).Encode(map[string]any{"data": c.toJSON()})
@@ -160,7 +162,7 @@ func (f *fakeListmonk) updateCampaign(w http.ResponseWriter, r *http.Request) {
f.campaigns[i].Type = str(body["type"])
f.campaigns[i].Tags = toStrings(body["tags"])
f.campaigns[i].ListIDs = toInts(body["lists"])
f.campaigns[i].MediaIDs = toInts(body["media_ids"])
f.campaigns[i].MediaIDs = toInts(body["media"])
json.NewEncoder(w).Encode(map[string]any{"data": f.campaigns[i].toJSON()})
return
}
@@ -226,24 +228,19 @@ func (f *fakeListmonk) uploadMedia(w http.ResponseWriter, r *http.Request) {
json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"id": m.ID}})
}
func (f *fakeListmonk) querySubscribers(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query().Get("query")
ids := f.subscriberIDsForQuery[q]
results := make([]map[string]any, 0, len(ids))
for _, id := range ids {
results = append(results, map[string]any{"id": id})
func (f *fakeListmonk) queryAddToList(w http.ResponseWriter, r *http.Request) {
if f.failSegmentQuery {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte(`{"message":"invalid SQL expression"}`))
return
}
json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"results": results}})
}
func (f *fakeListmonk) bulkAdd(w http.ResponseWriter, r *http.Request) {
body := decodeBody(r)
targetIDs := toInts(body["target_list_ids"])
var listID int
if len(targetIDs) > 0 {
listID = targetIDs[0]
}
f.bulkAddCalls = append(f.bulkAddCalls, bulkAddCall{IDs: toInts(body["ids"]), ListID: listID})
f.segmentQueryCalls = append(f.segmentQueryCalls, segmentQueryCall{Query: str(body["query"]), ListID: listID})
w.Write([]byte(`{}`))
}
@@ -434,7 +431,6 @@ func TestSyncDir_SegmentQueryMaterializesIntoManagedList(t *testing.T) {
f := newFakeListmonk(t)
f.lists = []fakeList{{ID: 3, Name: "Newsletter"}}
query := "subscribers.attribs->>'source' = 'workshop'"
f.subscriberIDsForQuery[query] = []int{10, 11, 12}
lm := f.client()
root := t.TempDir()
@@ -448,11 +444,11 @@ func TestSyncDir_SegmentQueryMaterializesIntoManagedList(t *testing.T) {
if len(result.Rejected) != 0 {
t.Fatalf("expected no rejections, got %v", result.Rejected)
}
if len(f.bulkAddCalls) != 1 {
t.Fatalf("expected 1 bulk-add call, got %d", len(f.bulkAddCalls))
if len(f.segmentQueryCalls) != 1 {
t.Fatalf("expected 1 segment query call, got %d", len(f.segmentQueryCalls))
}
if len(f.bulkAddCalls[0].IDs) != 3 {
t.Errorf("expected 3 subscriber IDs bulk-added, got %v", f.bulkAddCalls[0].IDs)
if f.segmentQueryCalls[0].Query != query {
t.Errorf("expected the query sent verbatim, got %q", f.segmentQueryCalls[0].Query)
}
var segList *fakeList
@@ -464,6 +460,9 @@ func TestSyncDir_SegmentQueryMaterializesIntoManagedList(t *testing.T) {
if segList == nil {
t.Fatal("expected a managed 'segment:launch' list to be created")
}
if f.segmentQueryCalls[0].ListID != segList.ID {
t.Errorf("expected the segment query call to target list %d, got %d", segList.ID, f.segmentQueryCalls[0].ListID)
}
if len(f.campaigns) != 1 {
t.Fatalf("expected 1 campaign, got %d", len(f.campaigns))
}
@@ -478,27 +477,25 @@ func TestSyncDir_SegmentQueryMaterializesIntoManagedList(t *testing.T) {
}
}
func TestSyncDir_InvalidSegmentQueryRejectsWithoutSideEffects(t *testing.T) {
func TestSyncDir_InvalidSegmentQueryRejectsWithoutCreatingCampaign(t *testing.T) {
f := newFakeListmonk(t)
f.lists = []fakeList{{ID: 3, Name: "Newsletter"}}
f.failSegmentQuery = true
lm := f.client()
root := t.TempDir()
// No entry seeded in subscriberIDsForQuery for this exact string simulates
// a query listmonk would reject — here we just confirm an empty/no-match
// result still flows through cleanly without creating a segment list
// mistakenly treated as an error path; a real invalid-SQL rejection from
// listmonk itself surfaces as a plain HTTP-error wrapped by
// QuerySubscriberIDs, exercised at the listmonk package's own test level.
fm := baseFrontmatter + "segment_query: \"subscribers.attribs->>'nope' = 'nothing'\"\n"
fm := baseFrontmatter + "segment_query: \"not valid sql\"\n"
writeCampaignDir(t, root, "launch", fm, "Hello world.")
result, err := SyncDir(lm, root, nil)
if err != nil {
t.Fatalf("SyncDir: %v", err)
}
if len(result.Rejected) != 0 {
t.Fatalf("expected no rejections for a zero-match segment query, got %v", result.Rejected)
if len(result.Rejected) != 1 || !strings.Contains(result.Rejected[0], "segment_query") {
t.Fatalf("expected a segment_query rejection, got %v", result.Rejected)
}
if len(f.campaigns) != 0 {
t.Errorf("expected no campaign created when segment_query fails, got %d", len(f.campaigns))
}
}