listmonk/subscribers.go

342 lines
9.2 KiB
Go
Raw Normal View History

2018-10-25 15:51:47 +02:00
package main
// !!!!!!!!!!! TODO
// For non-flat JSON attribs, show the advanced editor instead of the key-value editor
import (
"context"
"database/sql"
"fmt"
"net/http"
"strconv"
"strings"
"github.com/asaskevich/govalidator"
"github.com/knadh/listmonk/models"
"github.com/knadh/listmonk/subimporter"
"github.com/labstack/echo"
"github.com/lib/pq"
uuid "github.com/satori/go.uuid"
)
type subsWrap struct {
Results models.Subscribers `json:"results"`
Query string `json:"query"`
Total int `json:"total"`
PerPage int `json:"per_page"`
Page int `json:"page"`
}
type queryAddResp struct {
Count int64 `json:"count"`
}
type queryAddReq struct {
Query string `json:"query"`
SourceList int `json:"source_list"`
TargetLists pq.Int64Array `json:"target_lists"`
}
var jsonMap = []byte("{}")
// handleGetSubscriber handles the retrieval of a single subscriber by ID.
func handleGetSubscriber(c echo.Context) error {
var (
app = c.Get("app").(*App)
out models.Subscribers
id, _ = strconv.Atoi(c.Param("id"))
)
// Fetch one list.
if id < 1 {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid subscriber ID.")
}
err := app.Queries.GetSubscriber.Select(&out, id, nil)
if err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error fetching subscriber: %s", pqErrMsg(err)))
} else if err == sql.ErrNoRows {
return echo.NewHTTPError(http.StatusBadRequest, "Subscriber not found.")
}
out.LoadLists(app.Queries.GetSubscriberLists)
return c.JSON(http.StatusOK, okResp{out[0]})
}
// handleQuerySubscribers handles querying subscribers based on arbitrary conditions in SQL.
func handleQuerySubscribers(c echo.Context) error {
var (
app = c.Get("app").(*App)
pg = getPagination(c.QueryParams())
// Limit the subscribers to a particular list?
listID, _ = strconv.Atoi(c.FormValue("list_id"))
hasList bool
// The "WHERE ?" bit.
query = c.FormValue("query")
out subsWrap
)
if listID < 0 {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid `list_id`.")
} else if listID > 0 {
hasList = true
}
// There's an arbitrary query condition from the frontend.
cond := ""
if query != "" {
cond = " AND " + query
}
// The SQL queries to be executed are different for global subscribers
// and subscribers belonging to a specific list.
var (
stmt = ""
stmtCount = ""
)
if hasList {
stmt = fmt.Sprintf(app.Queries.QuerySubscribersByList,
listID, cond, pg.Offset, pg.Limit)
stmtCount = fmt.Sprintf(app.Queries.QuerySubscribersByListCount,
listID, cond)
} else {
stmt = fmt.Sprintf(app.Queries.QuerySubscribers,
cond, pg.Offset, pg.Limit)
stmtCount = fmt.Sprintf(app.Queries.QuerySubscribersCount, cond)
}
// Create a readonly transaction to prevent mutations.
tx, err := app.DB.BeginTxx(context.Background(), &sql.TxOptions{ReadOnly: true})
if err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error preparing query: %v", pqErrMsg(err)))
}
// Run the actual query.
if err := tx.Select(&out.Results, stmt); err != nil {
tx.Rollback()
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error querying subscribers: %v", pqErrMsg(err)))
}
// Run the query count.
if err := tx.Get(&out.Total, stmtCount); err != nil {
tx.Rollback()
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error running count query: %v", pqErrMsg(err)))
}
if err := tx.Commit(); err != nil {
tx.Rollback()
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error in subscriber query transaction: %v", pqErrMsg(err)))
}
// Lazy load lists for each subscriber.
if err := out.Results.LoadLists(app.Queries.GetSubscriberLists); err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error fetching subscriber lists: %v", pqErrMsg(err)))
}
if len(out.Results) == 0 {
out.Results = make(models.Subscribers, 0)
return c.JSON(http.StatusOK, okResp{out})
}
// Meta.
out.Query = query
out.Page = pg.Page
out.PerPage = pg.PerPage
return c.JSON(http.StatusOK, okResp{out})
}
// handleCreateSubscriber handles subscriber creation.
func handleCreateSubscriber(c echo.Context) error {
var (
app = c.Get("app").(*App)
req subimporter.SubReq
)
// Get and validate fields.
if err := c.Bind(&req); err != nil {
return err
} else if err := subimporter.ValidateFields(req); err != nil {
return echo.NewHTTPError(http.StatusBadRequest, err.Error())
}
req.Email = strings.ToLower(strings.TrimSpace(req.Email))
// Insert and read ID.
var newID int
err := app.Queries.UpsertSubscriber.Get(&newID,
uuid.NewV4(),
req.Email,
req.Name,
req.Status,
req.Attribs,
true,
req.Lists)
if err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error creating subscriber: %v", err))
}
// Hand over to the GET handler to return the last insertion.
c.SetParamNames("id")
c.SetParamValues(fmt.Sprintf("%d", newID))
return c.JSON(http.StatusOK, handleGetSubscriber(c))
}
// handleUpdateSubscriber handles subscriber modification.
func handleUpdateSubscriber(c echo.Context) error {
var (
app = c.Get("app").(*App)
id, _ = strconv.ParseInt(c.Param("id"), 10, 64)
req subimporter.SubReq
)
// Get and validate fields.
if err := c.Bind(&req); err != nil {
return err
}
if id < 1 {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid ID.")
}
if req.Email != "" && !govalidator.IsEmail(req.Email) {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid `email`.")
}
if req.Name != "" && !govalidator.IsByteLength(req.Name, 1, stdInputMaxLen) {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid length for `name`.")
}
req.Email = strings.ToLower(strings.TrimSpace(req.Email))
_, err := app.Queries.UpdateSubscriber.Exec(req.ID,
req.Email,
req.Name,
req.Status,
req.Attribs,
req.Lists)
if err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Update failed: %v", pqErrMsg(err)))
}
return handleGetSubscriber(c)
}
// handleDeleteSubscribers handles subscriber deletion,
// either a single one (ID in the URI), or a list.
func handleDeleteSubscribers(c echo.Context) error {
var (
app = c.Get("app").(*App)
id, _ = strconv.ParseInt(c.Param("id"), 10, 64)
ids pq.Int64Array
)
// Read the list IDs if they were sent in the body.
c.Bind(&ids)
if id < 1 && len(ids) == 0 {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid ID.")
}
if id > 0 {
ids = append(ids, id)
}
if _, err := app.Queries.DeleteSubscribers.Exec(ids); err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Delete failed: %v", err))
}
return c.JSON(http.StatusOK, okResp{true})
}
// handleQuerySubscribersIntoLists handles querying subscribers based on arbitrary conditions in SQL
// and adding them to given lists.
func handleQuerySubscribersIntoLists(c echo.Context) error {
var (
app = c.Get("app").(*App)
req queryAddReq
)
// Get and validate fields.
if err := c.Bind(&req); err != nil {
return echo.NewHTTPError(http.StatusBadRequest,
fmt.Sprintf("Error parsing request: %v", err))
}
if len(req.TargetLists) < 1 {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid `target_lists`.")
}
if req.SourceList < 0 {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid `source_list`.")
}
if req.Query == "" {
return echo.NewHTTPError(http.StatusBadRequest, "Invalid subscriber `query`.")
}
cond := " AND " + req.Query
// The SQL queries to be executed are different for global subscribers
// and subscribers belonging to a specific list.
var (
stmt = ""
stmtDry = ""
)
if req.SourceList > 0 {
stmt = fmt.Sprintf(app.Queries.QuerySubscribersByList, req.SourceList, cond)
stmtDry = fmt.Sprintf(app.Queries.QuerySubscribersByList, req.SourceList, cond, 0, 1)
} else {
stmt = fmt.Sprintf(app.Queries.QuerySubscribersIntoLists, cond)
stmtDry = fmt.Sprintf(app.Queries.QuerySubscribers, cond, 0, 1)
}
// Create a readonly transaction to prevent mutations.
// This is used to dry-run the arbitrary query before it's used to
// insert subscriptions.
tx, err := app.DB.BeginTxx(context.Background(), &sql.TxOptions{ReadOnly: true})
if err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error preparing query (dry-run): %v", pqErrMsg(err)))
}
// Perform the dry run.
if _, err := tx.Exec(stmtDry); err != nil {
tx.Rollback()
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error querying (dry-run) subscribers: %v", pqErrMsg(err)))
}
if err := tx.Commit(); err != nil {
tx.Rollback()
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error in subscriber dry-run query transaction: %v", pqErrMsg(err)))
}
// Prepare the query.
q, err := app.DB.Preparex(stmt)
if err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error preparing query: %v", pqErrMsg(err)))
}
// Run the query.
res, err := q.Exec(req.TargetLists)
if err != nil {
return echo.NewHTTPError(http.StatusInternalServerError,
fmt.Sprintf("Error adding subscribers to lists: %v", pqErrMsg(err)))
}
num, _ := res.RowsAffected()
return c.JSON(http.StatusOK, okResp{queryAddResp{num}})
}