feat: Initial commit

.gitignore

@@ -0,0 +1,3 @@
+config.toml
+payload.json
+*.bin

Makefile

@@ -0,0 +1,16 @@
+BIN := listmonk-messenger.bin
+
+LAST_COMMIT := $(shell git rev-parse --short HEAD)
+LAST_COMMIT_DATE := $(shell git show -s --format=%ci ${LAST_COMMIT})
+VERSION := $(shell git describe --abbrev=1)
+BUILDSTR := ${VERSION} (build "\\\#"${LAST_COMMIT} $(shell date '+%Y-%m-%d %H:%M:%S'))
+
+build:
+	go build -o ${BIN} -ldflags="-X 'main.buildString=${BUILDSTR}'" *.go
+.PHONY: build
+
+run: build
+	@./${BIN}
+.PHONY: run
+
+.DEFAULT_GOAL := build

README.md

@@ -0,0 +1,33 @@
+## listmonk-messenger
+
+Lightweight HTTP server to handle webhooks from [listmonk](https://listmonk.app) and forward it to different messengers.
+
+### Supported messengers
+
+* Pinpoint
+
+### Development
+
+* Build binary
+```
+make build
+```
+
+* Change config.toml and tweak messenger config
+
+Run the binary which starts a server on :8082
+```
+./listmonk-messenger.bin --config config.toml --messenger pinpoint
+```
+
+* Setting up webhooks
+![](/screenshots/listmonk-setting-up-webhook.png)
+
+* Add messenger specific subscriber atrributes in listmonk
+![](/screenshots/listmonk-add-subsriber-attrib.png)
+
+* Add plain text template
+![](/screenshots/listmonk-plain-text-template.png)
+
+* Change campaign messenger
+![](/screenshots/listmonk-change-campaign-mgr.png)

config.sample.toml

@@ -0,0 +1,16 @@
+[server]
+address = ":8082"
+read_timeout = "5s"
+write_timeout = "5s"
+
+[messenger.pinpoint]
+config = '''
+{
+    "app_id": "",
+    "access_key": "",
+    "secret_key": "",
+    "region": "",
+    "message_type": "",
+    "sender_id": ""
+}
+'''

go.mod

@@ -0,0 +1,25 @@
+module github.com/joeirimpan/listmonk-messenger
+
+go 1.17
+
+require (
+	github.com/aws/aws-sdk-go v1.42.26
+	github.com/go-chi/chi v1.5.4
+	github.com/knadh/koanf v1.4.0
+	github.com/knadh/listmonk v1.1.0
+	github.com/spf13/pflag v1.0.5
+)
+
+require (
+	github.com/fsnotify/fsnotify v1.5.1 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/jmoiron/sqlx v1.2.0 // indirect
+	github.com/lib/pq v1.3.0 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/mapstructure v1.4.3 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/pelletier/go-toml v1.9.4 // indirect
+	github.com/yuin/goldmark v1.3.4 // indirect
+	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e // indirect
+	gopkg.in/volatiletech/null.v6 v6.0.0-20170828023728-0bef4e07ae1b // indirect
+)

handlers.go

@@ -0,0 +1,136 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/go-chi/chi"
+	"github.com/joeirimpan/listmonk-messenger/messenger"
+	"github.com/knadh/listmonk/models"
+)
+
+type postback struct {
+	Subject     string      `json:"subject"`
+	ContentType string      `json:"content_type"`
+	Body        string      `json:"body"`
+	Recipients  []recipient `json:"recipients"`
+	Campaign    *campaign   `json:"campaign"`
+}
+
+type campaign struct {
+	UUID string   `db:"uuid" json:"uuid"`
+	Name string   `db:"name" json:"name"`
+	Tags []string `db:"tags" json:"tags"`
+}
+
+type recipient struct {
+	UUID    string                   `db:"uuid" json:"uuid"`
+	Email   string                   `db:"email" json:"email"`
+	Name    string                   `db:"name" json:"name"`
+	Attribs models.SubscriberAttribs `db:"attribs" json:"attribs"`
+	Status  string                   `db:"status" json:"status"`
+}
+
+type httpResp struct {
+	Status  string      `json:"status"`
+	Message string      `json:"message,omitempty"`
+	Data    interface{} `json:"data,omitempty"`
+}
+
+// handlePostback picks the messager based on url params and pushes message using it.
+func handlePostback(w http.ResponseWriter, r *http.Request) {
+	var (
+		app      = r.Context().Value("app").(*App)
+		provider = chi.URLParam(r, "provider")
+	)
+
+	// Decode body
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		sendErrorResponse(w, "invalid body", http.StatusBadRequest, nil)
+		return
+	}
+	defer r.Body.Close()
+
+	data := &postback{}
+	if err := json.Unmarshal(body, &data); err != nil {
+		sendErrorResponse(w, "invalid body", http.StatusBadRequest, nil)
+		return
+	}
+
+	// Get the provider.
+	p, ok := app.messengers[provider]
+	if !ok {
+		sendErrorResponse(w, "unknown provider", http.StatusBadRequest, nil)
+		return
+	}
+
+	if len(data.Recipients) > 1 {
+		sendErrorResponse(w, "invalid recipients", http.StatusBadRequest, nil)
+		return
+	}
+
+	rec := data.Recipients[0]
+	message := messenger.Message{
+		Subject:     data.Subject,
+		ContentType: data.ContentType,
+		Body:        []byte(data.Body),
+		Subscriber: models.Subscriber{
+			UUID:    rec.UUID,
+			Email:   rec.Email,
+			Name:    rec.Name,
+			Status:  rec.Status,
+			Attribs: rec.Attribs,
+		},
+	}
+
+	if data.Campaign != nil {
+		message.Campaign = &models.Campaign{
+			UUID: data.Campaign.UUID,
+			Name: data.Campaign.Name,
+			Tags: data.Campaign.Tags,
+		}
+	}
+
+	// Send message.
+	if err := p.Push(message); err != nil {
+		sendErrorResponse(w, "error sending message", http.StatusInternalServerError, nil)
+		return
+	}
+
+	sendResponse(w, "OK")
+}
+
+// wrap is a middleware that wraps HTTP handlers and injects the "app" context.
+func wrap(app *App, next http.HandlerFunc) http.HandlerFunc {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx := context.WithValue(r.Context(), "app", app)
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}
+
+// sendResponse sends a JSON envelope to the HTTP response.
+func sendResponse(w http.ResponseWriter, data interface{}) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	out, err := json.Marshal(httpResp{Status: "success", Data: data})
+	if err != nil {
+		sendErrorResponse(w, "Internal Server Error", http.StatusInternalServerError, nil)
+		return
+	}
+
+	w.Write(out)
+}
+
+// sendErrorResponse sends a JSON error envelope to the HTTP response.
+func sendErrorResponse(w http.ResponseWriter, message string, code int, data interface{}) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(code)
+
+	resp := httpResp{Status: "error",
+		Message: message,
+		Data:    data}
+	out, _ := json.Marshal(resp)
+	w.Write(out)
+}

main.go

@@ -0,0 +1,120 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+
+	"github.com/go-chi/chi"
+	"github.com/joeirimpan/listmonk-messenger/messenger"
+	"github.com/knadh/koanf"
+	"github.com/knadh/koanf/parsers/toml"
+	"github.com/knadh/koanf/providers/file"
+	"github.com/knadh/koanf/providers/posflag"
+	flag "github.com/spf13/pflag"
+)
+
+var (
+	logger = log.New(os.Stdout, "", log.Ldate|log.Ltime|log.Lshortfile)
+	ko     = koanf.New(".")
+
+	// Version of the build injected at build time.
+	buildString = "unknown"
+)
+
+type MessengerCfg struct {
+	Config string `koanf:"config"`
+}
+
+type App struct {
+	messengers map[string]messenger.Messenger
+}
+
+func init() {
+	f := flag.NewFlagSet("config", flag.ContinueOnError)
+	f.Usage = func() {
+		fmt.Println(f.FlagUsages())
+		os.Exit(0)
+	}
+	f.StringSlice("config", []string{"config.toml"},
+		"Path to one or more TOML config files to load in order")
+	f.StringSlice("msgr", []string{"pinpoint"},
+		"Name of messenger. Can specify multiple values.")
+	f.Bool("version", false, "Show build version")
+	if err := f.Parse(os.Args[1:]); err != nil {
+		log.Fatalf("error parsing flags: %v", err)
+	}
+
+	// Display version.
+	if ok, _ := f.GetBool("version"); ok {
+		fmt.Println(buildString)
+		os.Exit(0)
+	}
+
+	// Read the config files.
+	cFiles, _ := f.GetStringSlice("config")
+	for _, f := range cFiles {
+		log.Printf("reading config: %s", f)
+		if err := ko.Load(file.Provider(f), toml.Parser()); err != nil {
+			log.Printf("error reading config: %v", err)
+		}
+	}
+
+	if err := ko.Load(posflag.Provider(f, ".", ko), nil); err != nil {
+		log.Fatalf("error loading flags: %v", err)
+	}
+}
+
+// loadMessengers loads all messages mentioned in posflag into application.
+func loadMessengers(msgrs []string, app *App) {
+	app.messengers = make(map[string]messenger.Messenger)
+
+	for _, m := range msgrs {
+		var cfg MessengerCfg
+		if err := ko.Unmarshal("messenger."+m, &cfg); err != nil {
+			log.Fatalf("error reading %s messenger config: %v", m, err)
+		}
+
+		var (
+			msgr messenger.Messenger
+			err  error
+		)
+		switch m {
+		case "pinpoint":
+			msgr, err = messenger.NewPinpoint([]byte(cfg.Config))
+		default:
+			log.Fatalf("invalid provider: %s", m)
+		}
+
+		if err != nil {
+			log.Fatalf("error creating %s messenger: %v", m, err)
+		}
+
+		app.messengers[m] = msgr
+		log.Printf("loaded %s\n", m)
+	}
+}
+
+func main() {
+	// load messengers
+	app := &App{}
+
+	loadMessengers(ko.Strings("msgr"), app)
+
+	r := chi.NewRouter()
+	r.Post("/webhook/{provider}", wrap(app, handlePostback))
+
+	// HTTP Server.
+	srv := &http.Server{
+		Addr:         ko.String("server.address"),
+		ReadTimeout:  ko.Duration("server.read_timeout"),
+		WriteTimeout: ko.Duration("server.write_timeout"),
+		Handler:      r,
+	}
+
+	logger.Printf("starting on %s", srv.Addr)
+	if err := srv.ListenAndServe(); err != nil {
+		logger.Fatalf("couldn't start server: %v", err)
+	}
+}

messenger/messenger.go

@@ -0,0 +1,39 @@
+package messenger
+
+import (
+	"net/textproto"
+
+	"github.com/knadh/listmonk/models"
+)
+
+type Messenger interface {
+	Name() string
+	Push(Message) error
+	Flush() error
+	Close() error
+}
+
+// Message is the message pushed to a Messenger.
+type Message struct {
+	From        string
+	To          []string
+	Subject     string
+	ContentType string
+	Body        []byte
+	AltBody     []byte
+	Headers     textproto.MIMEHeader
+	Attachments []Attachment
+
+	Subscriber models.Subscriber
+
+	// Campaign is generally the same instance for a large number of subscribers.
+	Campaign *models.Campaign
+}
+
+// Attachment represents a file or blob attachment that can be
+// sent along with a message by a Messenger.
+type Attachment struct {
+	Name    string
+	Header  textproto.MIMEHeader
+	Content []byte
+}