Review Request Pulse Processor

Overview

The Review Request Pulse processor identifies agencies with at least 1 month of active subscription and creates pulses to request reviews. It prevents duplicate requests within a 30-day window to avoid over-solicitation.

Source File: queue-manager/services/projects/reviewRequestPulse.js
Queue File: queue-manager/queues/projects/reviewRequestPulse.js
Execution: Part of projects cron (every 5 minutes)
Business Impact: MEDIUM - Customer feedback & testimonials

Processing Flow

sequenceDiagram
    participant CRON as Projects Cron
    participant SVC as Review Service
    participant SUBS as Subscriptions
    participant PULSE as Projects Pulse
    participant QUEUE as Review Queue

    CRON->>SVC: reviewRequestPulse()
    SVC->>SUBS: Aggregate active subscriptions<br/>(1+ month old)
    SUBS-->>SVC: Agency subscription data

    loop For Each Agency
        SVC->>PULSE: Check existing pulse<br/>(last 31 days)

        alt No Recent Pulse
            SVC->>QUEUE: Add to queue
            QUEUE->>PULSE: Create review request
        end
    end

Key Logic

Eligibility Criteria

Agencies must meet ALL criteria:

• Subscription Status: Active
• Product Type: Managed subscriptions (excluding sites and listings)
• Duration: At least 1 month (30 days)
• Recent Pulse: None in last 31 days

Time Window

oneMonthAgo = Math.floor(Date.now() / 1000) - 30 * 24 * 60 * 60;

Product Filter

const UPDATED_MANAGED_SUBSCRIPTIONS = MANAGED_SUBSCRIPTIONS.filter(
  item => item !== 'listings' && item !== 'site',
);

MongoDB Aggregation Pipeline

Stage 1: Match Active Managed Subscriptions

{
  $match: {
    status: 'active',
    'plan.metadata.product_type': {
      $in: UPDATED_MANAGED_SUBSCRIPTIONS
    }
  }
}

Stage 2: Group by Agency

{
  $group: {
    _id: '$account',
    oldestSubscription: { $min: '$created_at' },
    subscriptionIds: { $push: '$_id' },
    metadataAccountId: { $first: '$metadata.account_id' }
  }
}

Stage 3: Filter by Minimum Duration

{
  $match: {
    oldestSubscription: {
      $lte: oneMonthAgo;
    }
  }
}

Stage 4: Project Account Data

{
  $project: {
    _id: 0,
    account_id: '$metadataAccountId',
    subscription_ids: '$subscriptionIds',
    parent_account: '$_id',
    oldestSubscription: 1
  }
}

Stage 5: Check Existing Pulses

{
  $lookup: {
    from: 'projects.pulse',
    let: { acc_id: '$account_id' },
    pipeline: [
      {
        $match: {
          $expr: {
            $and: [
              { $eq: ['$account_id', '$$acc_id'] },
              { $eq: ['$type', 'review_request'] },
              {
                $or: [
                  { $eq: ['$status', 'pending'] },
                  {
                    $and: [
                      { $eq: ['$status', 'completed'] },
                      { $gt: ['$created_at', new Date(oneMonthAgo * 1000 + 24 * 60 * 60 * 1000)] }
                    ]
                  }
                ]
              }
            ]
          }
        }
      }
    ],
    as: 'pulse'
  }
}

Stage 6: Filter No Recent Pulse

{
  $match: {
    pulse: {
      $size: 0;
    }
  }
}

Queue Processing

Adding to Queue

const queue = await reviewRequestPulseService.start();

for (const reviewRequest of reviewRequestPulsesInfo) {
  await queue.add(reviewRequest, {
    removeOnComplete: true,
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 2000,
    },
  });
}

Queue Options

• Remove on Complete: true
• Attempts: 3
• Backoff: Exponential with 2s delay

Pulse Structure

{
  account_id: ObjectId,
  type: 'review_request',
  status: 'pending',
  metadata: {
    subscription_ids: [...],
    parent_account: ObjectId,
    oldest_subscription: Date,
    request_reason: 'one_month_active'
  }
}

Collections Used

Input: store-subscription

• Queries active subscriptions
• Groups by account
• Filters by minimum duration

Check: projects-pulse

• Prevents duplicate requests within 31 days
• Checks both pending and recently completed

Output: projects-pulse (via queue)

• Creates review request pulses

Duplicate Prevention

Time-Based Logic

Prevents pulses if:

• Pending pulse exists
• OR completed pulse within last 31 days (30 days + 1 day buffer)

Example Timeline

Day 0: Review request sent
Day 30: Eligible again
Day 31: New pulse can be created (1-day grace period)

Error Handling

try {
  // Aggregation and queue processing
} catch (error) {
  logger.error({
    initiator: 'QM/projects/reviewRequestPulse',
    error: error,
  });
  throw error;
}

Business Rules

Why 1 Month?

• Agencies need time to experience value
• Reduces premature review requests
• Aligns with typical onboarding completion

Why Exclude Sites/Listings?

• Not managed services
• Different customer engagement model
• Lower touch point requirements

Why 31-Day Window?

• Prevents review fatigue
• Monthly cadence feels natural
• 1-day buffer prevents edge cases

Performance

Execution Time: 2-5 seconds
Frequency: Every 5 minutes
Typical Results: 5-20 agencies per run
Peak: 30-50 agencies (beginning of month)

---

Complexity: MEDIUM
Lines of Code: 136