Selectively refresh stats for opened campaigns

apps/platform/src/campaigns/CampaignInteractJob.ts

@@ -1,7 +1,8 @@
+import App from '../app'
 import { Job } from '../queue'
 import { unsubscribe } from '../subscriptions/SubscriptionService'
 import { CampaignSend } from './Campaign'
-import { getCampaignSend, updateCampaignSend } from './CampaignService'
+import { CacheKeys, getCampaignSend, updateCampaignSend } from './CampaignService'
 
 interface CampaignIteraction {
     user_id: number
@@ -25,6 +26,7 @@ export default class CampaignInteractJob extends Job {
 
         if (type === 'opened' && !send.opened_at) {
             await updateCampaignSend(send.id, { opened_at: new Date() })
+            await App.main.redis.sadd(CacheKeys.pendingStats, campaign_id)
         }
 
         if (type === 'clicked') {

apps/platform/src/campaigns/CampaignService.ts

@@ -23,6 +23,10 @@ import Template from '../render/Template'
 import { differenceInDays, subDays } from 'date-fns'
 import { raw } from '../core/Model'
 
+export const CacheKeys = {
+    pendingStats: 'campaigns:pending_stats',
+}
+
 export const pagedCampaigns = async (params: PageParams, projectId: number) => {
     const result = await Campaign.search(
         { ...params, fields: ['name'] },

apps/platform/src/campaigns/CampaignStateJob.ts

@@ -1,24 +1,29 @@
 import { subDays } from 'date-fns'
 import { Job } from '../queue'
 import Campaign from './Campaign'
-import { updateCampaignProgress } from './CampaignService'
+import { CacheKeys, updateCampaignProgress } from './CampaignService'
+import App from '../app'
 
 export default class CampaignStateJob extends Job {
     static $name = 'campaign_state_job'
 
     static async handler() {
 
         // Fetch anything that is currently running or has finished
-        // within the last two days
+        // within the last two days or has activity
+        const openedCampaignIds = await App.main.redis.smembers(CacheKeys.pendingStats)
         const campaigns = await Campaign.query()
             .whereIn('state', ['scheduled', 'running'])
             .orWhere(function(qb) {
                 qb.where('state', 'finished')
                     .where('send_at', '>', subDays(Date.now(), 2))
             })
+            .orWhereIn('id', openedCampaignIds.map(id => parseInt(id, 10))) as Campaign[]
 
         for (const campaign of campaigns) {
             await updateCampaignProgress(campaign)
         }
+
+        await App.main.redis.srem(CacheKeys.pendingStats, ...campaigns.map(c => c.id))
     }
 }