Carroway Research Note 05: aggregation pipelines
Read-only MongoDB aggregation pipelines, run on 23 September 2026 with allowDiskUse: true.
Collection names are replaced with neutral names; every stage is otherwise exactly as run:
  sent_log         one document per email the sending platform sent (fields: to_email, sender_email, sent_at, workspace_id)
  replies          one document per inbound reply (fields: timestamp_created, from_address_email_lower, workspace_id,
                   sentiment_label written by the platform's reply classifier, automated_reply flag)
  system_messages  the operator's own record of emails its system generated (fields: email_lower, workspace_id, status, sent_date)
Months are calendar months in UTC. Scope: 1 October 2025 00:00 UTC to 23 September 2026 00:00 UTC (exclusive).
The CSV files are built from these outputs with plain sums and ratios: rates are counts divided by sends or contacts, times 100.
Month status: small under 10,000 prospect sends; log incomplete when log capture is under 90%; September 2026 partial. Rates only for
complete and partial months. Same-workspace comparison: workspaces with at least 5,000 prospect sends in both May to June and July to
August 2026, linked replies grouped by sent_month. Position split: prospect emails sent June to August 2026, linked replies grouped by
sent_month and position.

// ----------------------------------------------------------------------------------------------------
// Q1. Sends per month and workspace, split into prospect emails (excluded: false) and emails to test or internal addresses (excluded: true). Feeds sends, workspaces, largest_workspace_share_pct and test_and_internal_sends_excluded in monthly-counts.csv, sends in monthly-rates.csv, pooled-rates.csv and same-workspace-trend.csv, and the send rows of coverage.csv.
db.sent_log.aggregate([
  {
    $match: { sent_at: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') } }
  },
  {
    $group: {
      _id: {
        contact: { $toLower: '$to_email' },
        month: { $dateToString: { format: '%Y-%m', date: '$sent_at' } },
        workspace: { $ifNull: ['$workspace_id', null] }
      },
      sends: { $sum: 1 }
    }
  },
  {
    $group: {
      _id: '$_id.contact',
      rows: { $push: { month: '$_id.month', workspace: '$_id.workspace', sends: '$sends' } },
      contact_sends: { $sum: '$sends' }
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: '_id',
      foreignField: 'sender_email',
      pipeline: [{ $limit: 1 }, { $project: { _id: 1 } }],
      as: 'as_sender'
    }
  },
  {
    $set: { excluded: { $or: [{ $gte: ['$contact_sends', 100] }, { $gt: [{ $size: '$as_sender' }, 0] }] } }
  },
  { $unwind: '$rows' },
  {
    $group: {
      _id: { month: '$rows.month', workspace: '$rows.workspace', excluded: '$excluded' },
      sends: { $sum: '$rows.sends' }
    }
  },
  { $sort: { '_id.month': 1 } }
], { allowDiskUse: true })

// ----------------------------------------------------------------------------------------------------
// Q2. Unique prospect contacts emailed per month. Feeds contacts_emailed in monthly-counts.csv and monthly-rates.csv.
db.sent_log.aggregate([
  {
    $match: { sent_at: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') } }
  },
  {
    $group: {
      _id: {
        contact: { $toLower: '$to_email' },
        month: { $dateToString: { format: '%Y-%m', date: '$sent_at' } },
        workspace: { $ifNull: ['$workspace_id', null] }
      },
      sends: { $sum: 1 }
    }
  },
  {
    $group: {
      _id: '$_id.contact',
      rows: { $push: { month: '$_id.month', workspace: '$_id.workspace', sends: '$sends' } },
      contact_sends: { $sum: '$sends' }
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: '_id',
      foreignField: 'sender_email',
      pipeline: [{ $limit: 1 }, { $project: { _id: 1 } }],
      as: 'as_sender'
    }
  },
  {
    $set: { excluded: { $or: [{ $gte: ['$contact_sends', 100] }, { $gt: [{ $size: '$as_sender' }, 0] }] } }
  },
  { $unwind: '$rows' },
  { $match: { excluded: false } },
  { $group: { _id: { month: '$rows.month', contact: '$_id' } } },
  { $group: { _id: '$_id.month', contacts: { $sum: 1 } } },
  { $sort: { _id: 1 } }
], { allowDiskUse: true })

// ----------------------------------------------------------------------------------------------------
// Q3. Prospect sends per month by the email position among all logged emails from the same workspace to the same address (4 = fourth or later). Feeds the sends in email-position.csv.
db.sent_log.aggregate([
  {
    $match: { sent_at: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') } }
  },
  {
    $group: {
      _id: { workspace: { $ifNull: ['$workspace_id', null] }, contact: { $toLower: '$to_email' } },
      times: { $push: '$sent_at' }
    }
  },
  {
    $setWindowFields: { partitionBy: '$_id.contact', output: { contact_sends: { $sum: { $size: '$times' } } } }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: '_id.contact',
      foreignField: 'sender_email',
      pipeline: [{ $limit: 1 }, { $project: { _id: 1 } }],
      as: 'as_sender'
    }
  },
  { $match: { contact_sends: { $lt: 100 }, as_sender: { $size: 0 } } },
  { $set: { times: { $sortArray: { input: '$times', sortBy: 1 } } } },
  { $unwind: { path: '$times', includeArrayIndex: 'idx' } },
  {
    $group: {
      _id: {
        month: { $dateToString: { format: '%Y-%m', date: '$times' } },
        position: { $min: [{ $add: ['$idx', 1] }, 4] }
      },
      sends: { $sum: 1 }
    }
  },
  { $sort: { '_id.month': 1, '_id.position': 1 } }
], { allowDiskUse: true })

// ----------------------------------------------------------------------------------------------------
// Q4. Replies by month received, reply type, workspace, test/internal flag, and the logged email each reply links to (sent_month, position). Feeds the reply counts in monthly-counts.csv, monthly-rates.csv, reply-types.csv, email-position.csv, same-workspace-trend.csv, pooled-rates.csv and coverage.csv.
db.replies.aggregate([
  {
    $match: {
      timestamp_created: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') }
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: 'from_address_email_lower',
      foreignField: 'to_email',
      pipeline: [
        {
          $match: { sent_at: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') } }
        },
        { $limit: 100 },
        { $project: { _id: 1 } }
      ],
      as: 'emails_to_address'
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: 'from_address_email_lower',
      foreignField: 'sender_email',
      pipeline: [{ $limit: 1 }, { $project: { _id: 1 } }],
      as: 'as_sender'
    }
  },
  {
    $set: {
      excluded: { $or: [{ $gte: [{ $size: '$emails_to_address' }, 100] }, { $gt: [{ $size: '$as_sender' }, 0] }] }
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: 'from_address_email_lower',
      foreignField: 'to_email',
      let: { ts: '$timestamp_created', ws: '$workspace_id' },
      pipeline: [
        {
          $match: {
            $expr: {
              $and: [
                { $eq: [{ $type: '$sent_at' }, 'date'] },
                { $lt: ['$sent_at', '$$ts'] },
                { $eq: ['$workspace_id', '$$ws'] }
              ]
            }
          }
        },
        { $group: { _id: null, n: { $sum: 1 }, last: { $max: '$sent_at' } } }
      ],
      as: 'prior'
    }
  },
  {
    $project: {
      month: { $dateToString: { format: '%Y-%m', date: '$timestamp_created' } },
      workspace: '$workspace_id',
      excluded: 1,
      type: {
        $switch: {
          branches: [
            { case: { $eq: ['$sentiment_label', 'out of office'] }, then: 'out_of_office' },
            { case: { $eq: ['$sentiment_label', 'error notification'] }, then: 'bounce_notice' },
            { case: { $eq: ['$sentiment_label', 'automated referral'] }, then: 'automated_referral' },
            {
              case: {
                $or: [
                  { $in: ['$sentiment_label', ['automated_reply', 'automated_replyerror notification']] },
                  { $eq: ['$automated_reply', true] }
                ]
              },
              then: 'auto_reply'
            },
            { case: { $eq: ['$sentiment_label', 'ongoing_conversation'] }, then: 'conversation_message' },
            {
              case: {
                $in: ['$sentiment_label', ['possibly interested', 'question', 'contact later', 'manual referral']]
              },
              then: 'human_positive'
            },
            {
              case: { $in: ['$sentiment_label', ['zero interest', 'unsubscribe me', 'remove me', 'not intested']] },
              then: 'human_negative'
            },
            { case: { $eq: ['$sentiment_label', 'wrong person/company'] }, then: 'human_wrong_person' },
            {
              case: {
                $in: [
                  '$sentiment_label',
                  [
                    'pitching me sth',
                    'unclear response',
                    'neutral',
                    'interesting statement',
                    'client already',
                    'unclear',
                    'out of business'
                  ]
                ]
              },
              then: 'human_other'
            }
          ],
          default: 'non_prospect_message'
        }
      },
      position: { $ifNull: [{ $arrayElemAt: ['$prior.n', 0] }, 0] },
      sent_month: {
        $cond: [
          { $gt: [{ $size: '$prior' }, 0] },
          { $dateToString: { format: '%Y-%m', date: { $arrayElemAt: ['$prior.last', 0] } } },
          null
        ]
      }
    }
  },
  {
    $group: {
      _id: {
        month: '$month',
        sent_month: '$sent_month',
        position: { $min: ['$position', 4] },
        type: '$type',
        workspace: '$workspace',
        excluded: '$excluded'
      },
      replies: { $sum: 1 }
    }
  },
  { $sort: { '_id.month': 1, '_id.sent_month': 1, '_id.position': 1 } }
], { allowDiskUse: true })

// ----------------------------------------------------------------------------------------------------
// Q5. Distinct replying contacts per month, test and internal addresses removed. Feeds the per-contact columns in monthly-rates.csv.
db.replies.aggregate([
  {
    $match: {
      timestamp_created: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') }
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: 'from_address_email_lower',
      foreignField: 'to_email',
      pipeline: [
        {
          $match: { sent_at: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') } }
        },
        { $limit: 100 },
        { $project: { _id: 1 } }
      ],
      as: 'emails_to_address'
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: 'from_address_email_lower',
      foreignField: 'sender_email',
      pipeline: [{ $limit: 1 }, { $project: { _id: 1 } }],
      as: 'as_sender'
    }
  },
  {
    $set: {
      excluded: { $or: [{ $gte: [{ $size: '$emails_to_address' }, 100] }, { $gt: [{ $size: '$as_sender' }, 0] }] }
    }
  },
  { $match: { excluded: false } },
  {
    $project: {
      month: { $dateToString: { format: '%Y-%m', date: '$timestamp_created' } },
      contact: '$from_address_email_lower',
      type: {
        $switch: {
          branches: [
            { case: { $eq: ['$sentiment_label', 'out of office'] }, then: 'out_of_office' },
            { case: { $eq: ['$sentiment_label', 'error notification'] }, then: 'bounce_notice' },
            { case: { $eq: ['$sentiment_label', 'automated referral'] }, then: 'automated_referral' },
            {
              case: {
                $or: [
                  { $in: ['$sentiment_label', ['automated_reply', 'automated_replyerror notification']] },
                  { $eq: ['$automated_reply', true] }
                ]
              },
              then: 'auto_reply'
            },
            { case: { $eq: ['$sentiment_label', 'ongoing_conversation'] }, then: 'conversation_message' },
            {
              case: {
                $in: ['$sentiment_label', ['possibly interested', 'question', 'contact later', 'manual referral']]
              },
              then: 'human_positive'
            },
            {
              case: { $in: ['$sentiment_label', ['zero interest', 'unsubscribe me', 'remove me', 'not intested']] },
              then: 'human_negative'
            },
            { case: { $eq: ['$sentiment_label', 'wrong person/company'] }, then: 'human_wrong_person' },
            {
              case: {
                $in: [
                  '$sentiment_label',
                  [
                    'pitching me sth',
                    'unclear response',
                    'neutral',
                    'interesting statement',
                    'client already',
                    'unclear',
                    'out of business'
                  ]
                ]
              },
              then: 'human_other'
            }
          ],
          default: 'non_prospect_message'
        }
      }
    }
  },
  {
    $group: {
      _id: { month: '$month', contact: '$contact' },
      automated: { $max: { $in: ['$type', ['out_of_office', 'bounce_notice', 'automated_referral', 'auto_reply']] } },
      human: {
        $max: { $in: ['$type', ['human_positive', 'human_negative', 'human_wrong_person', 'human_other']] }
      },
      positive: { $max: { $eq: ['$type', 'human_positive'] } }
    }
  },
  {
    $group: {
      _id: '$_id.month',
      contacts_any: { $sum: 1 },
      contacts_automated: { $sum: { $cond: ['$automated', 1, 0] } },
      contacts_human: { $sum: { $cond: ['$human', 1, 0] } },
      contacts_positive: { $sum: { $cond: ['$positive', 1, 0] } }
    }
  },
  { $sort: { _id: 1 } }
], { allowDiskUse: true })

// ----------------------------------------------------------------------------------------------------
// Q6. Completeness of the sent-email log against the operator's own record of emails its system marked sent. Feeds log-completeness.csv and log_capture_pct in monthly-counts.csv.
db.system_messages.aggregate([
  {
    $match: {
      status: 'sent',
      sent_date: { $gte: ISODate('2025-10-01T00:00:00.000Z'), $lt: ISODate('2026-09-23T00:00:00.000Z') }
    }
  },
  {
    $lookup: {
      from: 'sent_log',
      localField: 'email_lower',
      foreignField: 'to_email',
      let: { t: '$sent_date', ws: '$workspace_id' },
      pipeline: [
        {
          $match: {
            $expr: {
              $and: [
                { $eq: ['$workspace_id', '$$ws'] },
                { $gte: ['$sent_at', { $subtract: ['$$t', 259200000] }] },
                { $lte: ['$sent_at', { $add: ['$$t', 259200000] }] }
              ]
            }
          }
        },
        { $limit: 1 }
      ],
      as: 'logged'
    }
  },
  {
    $group: {
      _id: { $dateToString: { format: '%Y-%m', date: '$sent_date' } },
      system_sent: { $sum: 1 },
      found_in_log: { $sum: { $cond: [{ $gt: [{ $size: '$logged' }, 0] }, 1, 0] } }
    }
  },
  { $sort: { _id: 1 } }
], { allowDiskUse: true })

// ----------------------------------------------------------------------------------------------------
// Q7. Logged emails by date range. Feeds coverage.csv.
db.sent_log.aggregate([
  {
    $group: {
      _id: {
        $switch: {
          branches: [
            { case: { $ne: [{ $type: '$sent_at' }, 'date'] }, then: 'no_send_date' },
            { case: { $lt: ['$sent_at', ISODate('2025-10-01T00:00:00.000Z')] }, then: 'a_before_1_oct_2025' },
            {
              case: { $lt: ['$sent_at', ISODate('2026-01-06T00:00:00.000Z')] },
              then: 'b_1_oct_2025_to_5_jan_2026'
            },
            { case: { $lt: ['$sent_at', ISODate('2026-09-23T00:00:00.000Z')] }, then: 'c_6_jan_to_22_sep_2026' }
          ],
          default: 'd_after_22_sep_2026'
        }
      },
      sends: { $sum: 1 },
      first: { $min: '$sent_at' },
      last: { $max: '$sent_at' }
    }
  },
  { $sort: { _id: 1 } }
], { allowDiskUse: true })