Merge pull request #51 from parcelvoy/feat/campaign-fixes

Fixes campaign stats as a send goes out

db/migrations/20230210024314_add_project_timezone.js

@@ -0,0 +1,13 @@
+exports.up = function(knex) {
+    return knex.schema
+        .table('projects', function(table) {
+            table.string('timezone', 50).after('locale')
+        })
+}
+
+exports.down = function(knex) {
+    return knex.schema
+        .table('projects', function(table) {
+            table.dropColumn('timezone')
+        })
+}

src/campaigns/Campaign.ts

@@ -5,8 +5,8 @@ import List from '../lists/List'
 import Template from '../render/Template'
 import Subscription from '../subscriptions/Subscription'
 
-type CampaignState = 'draft' | 'scheduled' | 'running' | 'finished' | 'aborted'
-interface CampaignDelivery {
+export type CampaignState = 'draft' | 'scheduled' | 'running' | 'finished' | 'aborted'
+export interface CampaignDelivery {
     sent: number
     total: number
 }

src/campaigns/CampaignSendJob.ts

@@ -1,4 +1,6 @@
+import App from '../app'
 import { Job } from '../queue'
+import CampaignStateJob from './CampaignStateJob'
 import { campaignSendReadyQuery, sendCampaign } from './CampaignService'
 
 export default class CampaignSendJob extends Job {
@@ -11,5 +13,8 @@ export default class CampaignSendJob extends Job {
                     await sendCampaign(campaign, user_id)
                 }
             })
+            .then(() => {
+                App.main.queue.enqueue(CampaignStateJob.from())
+            })
     }
 }

src/campaigns/CampaignService.ts

@@ -3,7 +3,7 @@ import TextJob from '../channels/text/TextJob'
 import WebhookJob from '../channels/webhook/WebhookJob'
 import { UserEvent } from '../users/UserEvent'
 import { User } from '../users/User'
-import Campaign, { CampaignParams, CampaignSend, CampaignSendParams, CampaignSendState, SentCampaign } from './Campaign'
+import Campaign, { CampaignDelivery, CampaignParams, CampaignSend, CampaignSendParams, CampaignSendState, CampaignState, SentCampaign } from './Campaign'
 import { UserList } from '../lists/List'
 import Subscription from '../subscriptions/Subscription'
 import { RequestError } from '../core/errors'
@@ -15,9 +15,9 @@ import { allTemplates, duplicateTemplate } from '../render/TemplateService'
 import { utcToZonedTime } from 'date-fns-tz'
 import { getSubscription } from '../subscriptions/SubscriptionService'
 import { getProvider } from '../channels/ProviderRepository'
-import { isFuture } from 'date-fns'
 import { pick } from '../utilities'
 import { createTagSubquery } from '../tags/TagService'
+import { getProject } from '../projects/ProjectService'
 
 export const pagedCampaigns = async (params: SearchParams, projectId: number) => {
     return await Campaign.searchParams(
@@ -76,7 +76,7 @@ export const updateCampaign = async (id: number, projectId: number, params: Part
 
     // Calculate current state based on past properties
     if (params.send_at && data.state !== 'finished') {
-        data.state = isFuture(new Date(params.send_at)) ? 'scheduled' : 'running'
+        data.state = 'scheduled'
     } else {
         data.state = data.state === 'running' ? 'aborted' : 'draft'
     }
@@ -156,7 +156,8 @@ export const updateSendState = async (campaign: Campaign | number, user: User |
 
 export const sendList = async (campaign: SentCampaign) => {
 
-    if (!campaign.list_id) {
+    const project = await getProject(campaign.project_id)
+    if (!campaign.list_id || !project) {
         throw new RequestError('Unable to send to a campaign that does not have an associated list', 404)
     }
 
@@ -192,7 +193,7 @@ export const sendList = async (campaign: SentCampaign) => {
                     campaign_id: campaign.id,
                     state: 'pending',
                     send_at: campaign.send_in_user_timezone
-                        ? utcToZonedTime(campaign.send_at, timezone)
+                        ? utcToZonedTime(campaign.send_at, timezone ?? project.timezone)
                         : campaign.send_at,
                 })
                 i++
@@ -244,3 +245,23 @@ export const duplicateCampaign = async (campaign: Campaign) => {
     }
     return await getCampaign(cloneId, campaign.project_id)
 }
+
+export const campaignProgress = async (campaign: Campaign): Promise<CampaignDelivery> => {
+    const progress = await CampaignSend.first(
+        qb => qb.where('campaign_id', campaign.id)
+            .select(CampaignSend.raw("SUM(IF(state = 'sent', 1, 0)) AS sent, COUNT(*) AS total")),
+    ) as any
+    return {
+        sent: parseInt(progress.sent),
+        total: parseInt(progress.total),
+    }
+}
+
+export const updateCampaignProgress = async (
+    id: number,
+    projectId: number,
+    state: CampaignState,
+    delivery: CampaignDelivery,
+): Promise<void> => {
+    await Campaign.update(qb => qb.where('id', id).where('project_id', projectId), { state, delivery })
+}

src/campaigns/CampaignStateJob.ts

@@ -0,0 +1,16 @@
+import { Job } from '../queue'
+import Campaign from './Campaign'
+import { campaignProgress, updateCampaignProgress } from './CampaignService'
+
+export default class CampaignStateJob extends Job {
+    static $name = 'campaign_state_job'
+
+    static async handler() {
+        const campaigns = await Campaign.query()
+            .whereIn('state', ['running'])
+        for (const campaign of campaigns) {
+            const progress = await campaignProgress(campaign)
+            await updateCampaignProgress(campaign.id, campaign.project_id, 'finished', progress)
+        }
+    }
+}

src/channels/email/index.ts

@@ -13,8 +13,8 @@ const typeMap = {
     logger: LoggerEmailProvider,
 }
 
-export const providerMap = (record: { name: EmailProviderName }): EmailProvider => {
-    return typeMap[record.name].fromJson(record)
+export const providerMap = (record: { type: EmailProviderName }): EmailProvider => {
+    return typeMap[record.type].fromJson(record)
 }
 
 export const loadEmailChannel = async (providerId: number, projectId: number): Promise<EmailChannel | undefined> => {

src/channels/push/index.ts

@@ -11,8 +11,8 @@ const typeMap = {
     logger: LoggerPushProvider,
 }
 
-export const providerMap = (record: { name: PushProviderName }): PushProvider => {
-    return typeMap[record.name].fromJson(record)
+export const providerMap = (record: { type: PushProviderName }): PushProvider => {
+    return typeMap[record.type].fromJson(record)
 }
 
 export const loadPushChannel = async (providerId: number, projectId: number): Promise<PushChannel | undefined> => {

src/channels/text/index.ts

@@ -15,8 +15,8 @@ const typeMap = {
     logger: LoggerTextProvider,
 }
 
-export const providerMap = (record: { name: TextProviderName }): TextProvider => {
-    return typeMap[record.name].fromJson(record)
+export const providerMap = (record: { type: TextProviderName }): TextProvider => {
+    return typeMap[record.type].fromJson(record)
 }
 
 export const loadTextChannel = async (providerId: number, projectId: number): Promise<TextChannel | undefined> => {

src/channels/webhook/index.ts

@@ -11,8 +11,8 @@ const typeMap = {
     logger: LoggerWebhookProvider,
 }
 
-export const providerMap = (record: { name: WebhookProviderName }): WebhookProvider => {
-    return typeMap[record.name].fromJson(record)
+export const providerMap = (record: { type: WebhookProviderName }): WebhookProvider => {
+    return typeMap[record.type].fromJson(record)
 }
 
 export const loadWebhookChannel = async (providerId: number, projectId: number): Promise<WebhookChannel | undefined> => {

src/config/queue.ts

@@ -13,12 +13,14 @@ import ListPopulateJob from '../lists/ListPopulateJob'
 import ListStatsJob from '../lists/ListStatsJob'
 import ProcessListsJob from '../lists/ProcessListsJob'
 import CampaignSendJob from '../campaigns/CampaignSendJob'
+import CampaignStateJob from '../campaigns/CampaignStateJob'
 
 export type Queues = Record<number, Queue>
 
 export const loadJobs = (queue: Queue) => {
     queue.register(CampaignTriggerJob)
     queue.register(CampaignSendJob)
+    queue.register(CampaignStateJob)
     queue.register(EmailJob)
     queue.register(EventPostJob)
     queue.register(JourneyProcessJob)

src/projects/Project.ts

@@ -6,6 +6,7 @@ export default class Project extends Model {
     description?: string
     deleted_at?: Date
     locale?: string
+    timezone?: string
 }
 
 export type ProjectParams = Omit<Project, ModelParams | 'deleted_at'>