Merge branch 'fix-campaign-sent'

cmd/campaigns.go

@@ -299,6 +299,10 @@ func handleUpdateCampaignStatus(c echo.Context) error {
 		return err
 	}
 
+	if o.Status == models.CampaignStatusPaused || o.Status == models.CampaignStatusCancelled {
+		app.manager.StopCampaign(id)
+	}
+
 	return c.JSON(http.StatusOK, okResp{out})
 }
 

cmd/events.go

@@ -38,8 +38,6 @@ func handleEventStream(c echo.Context) error {
 				continue
 			}
 
-			fmt.Printf("data: %s\n\n", b)
-
 			c.Response().Write([]byte(fmt.Sprintf("retry: 3000\ndata: %s\n\n", b)))
 			c.Response().Flush()
 

cmd/manager_store.go

@@ -28,10 +28,12 @@ func newManagerStore(q *models.Queries, c *core.Core, m media.Store) *store {
 	}
 }
 
-// NextCampaigns retrieves active campaigns ready to be processed.
-func (s *store) NextCampaigns(excludeIDs []int64) ([]*models.Campaign, error) {
+// NextCampaigns retrieves active campaigns ready to be processed excluding
+// campaigns that are also being processed. Additionally, it takes a map of campaignID:sentCount
+// of campaigns that are being processed and updates them in the DB.
+func (s *store) NextCampaigns(currentIDs []int64, sentCounts []int64) ([]*models.Campaign, error) {
 	var out []*models.Campaign
-	err := s.queries.NextCampaigns.Select(&out, pq.Int64Array(excludeIDs))
+	err := s.queries.NextCampaigns.Select(&out, pq.Int64Array(currentIDs), pq.Int64Array(sentCounts))
 	return out, err
 }
 
@@ -58,6 +60,12 @@ func (s *store) UpdateCampaignStatus(campID int, status string) error {
 	return err
 }
 
+// UpdateCampaignStatus updates a campaign's status.
+func (s *store) UpdateCampaignCounts(campID int, toSend int, sent int, lastSubID int) error {
+	_, err := s.queries.UpdateCampaignCounts.Exec(campID, toSend, sent, lastSubID)
+	return err
+}
+
 // GetAttachment fetches a media attachment blob.
 func (s *store) GetAttachment(mediaID int) (models.Attachment, error) {
 	m, err := s.core.GetMedia(mediaID, "", s.media)