Email workflows

emailSend

File

lib/workflows/functions/email-send.ts

Event

email-campaign/send.requested

Config

retries: 2, concurrency: { limit: 20 }

A commented-out rateLimit keyed on event.data.email (1 per 24 h) remains in the source; the duplicate-send guard is currently step 1 plus the broadcaster’s queuedAt lock.

retries: 2 is safe because step 1 refuses any target that has already left pending, so a retry after a successful send is a no-op.

Event payload

subscriptionId, campaignId, targetId, email, subject, htmlBody, variationId, resendTemplateId, templateVariables, contactData, fromName, fromEmail, resendTopicId, unsubscribeUrl, firstName, lastName, sendingDomainId.

The broadcaster resolves all of this — variation, sending account, contact fields, unsubscribe token — so the workflow does not have to re-derive it.

Steps

1. check-already-sent

Refuses when the target is missing, or its status is any of requested, sent, opened, clicked, unsubscribed, bounced, blocked, suppressed, failed. Returns { skipped: true, reason: "already_sent" }.

The primary idempotency guard: a duplicate event, or a retry after a successful send, stops here.

2. check-eligibility

One read of target_emails through the target (getTargetEmailEligibility).

Condition Action

newsletterOptOut or marketingOptOut or isBlocked

Set the target unsubscribed; return reason: "unsubscribed"

newsletterLastSentAt newer than newsletterGapDays ago (default 7)

Clear the queue lock so the row can be re-leased later; return reason: "newsletter_gap"

The cadence gap is cross-campaign: without it, an address enrolled in three campaigns would get three emails the same morning.

Clearing the lock rather than failing the target is what lets the row come back once the gap has elapsed.

3. check-send-limits

Loads the campaign, the assigned variation, today’s send count and the per-variation hourly counts in one Promise.all.

Condition Result

Campaign or variation missing

skip: "not_found"

sentToday >= maxEmailsPerDay

skip: "daily_limit_reached"

Assigned variation under emailsPerVariationPerHour (default 100)

Proceed

Assigned variation over the rate, another active variation under it

Proceed with that fallback variation

All active variations over the rate

Release the queue lock; skip: "variation_rate_limited"

Limits are re-checked even though the broadcaster already checked them, because minutes pass between leasing and sending and other campaigns may have consumed the budget.

When a fallback is used, all of subject, HTML body, Resend template id and template variables switch to the fallback’s — not just the id.

4. send-email

Resolves template variables: for each { key, type, fallbackValue }, take contactData[key] when it is present and non-empty, else fallbackValue; coerce to a number when type is number (Number(value) || 0); drop the variable entirely when the result is null.

UNSUBSCRIBE_LINK is set last, so no user-defined variable can overwrite it.

from is "<fromName> <fromEmail>"; replyTo comes from buildReplyToForSendingDomain, which is what makes inbound reply correlation possible.

Sends through ResendEmailClient.sendCampaign with the resolved subject, HTML, optional resendTemplateId, topicId, variable values and unsubscribe URL, plus a target_id tag (RESEND_TARGET_ID_TAG) holding the target id. Returns the Resend id.

The tag lets ResendWebhookDispatcher match a webhook that arrives before step 5 saves resendEmailId. See Matching webhooks to targets.

5. update-target

  • markTargetRequested stores resendEmailId, variationId (the effective one), sendingDomainId and sentAt, and sets status → requested only while it is still pending, so a status a webhook already set (sent, opened, suppressed) is kept.

  • touchTargetEmailNewsletterLastSent(emailId) — stamps the cross-campaign cadence clock.

  • incrementSendingDomainDailyCount(sendingDomainId) — reflects the send against the daily cap immediately, rather than waiting for the email.sent webhook. The lifetime sentCount is bumped by that webhook, so the two do not double-count.

  • incrementVariationUsage(effectiveVariationId).

6. log-activity-and-crm

One read for contactId and hubspotContactId, reused twice:

  • Write an Activity row (type email, body truncated to 500 characters).

  • publishCrmNewActivity with direction: "EMAIL" — only when a HubSpot contact id exists.

Both are best-effort: failures are warned and the step still succeeds. The email has already been sent; failing the run over a log entry would be worse than losing the log entry.

emailCampaignDuplicate

File

lib/workflows/functions/email-campaign-duplicate.ts

Event

email-campaign/duplicate.requested

Config

retries: 2, concurrency: { limit: 5 }

POST /api/email-campaigns/{id}/duplicate creates the destination campaign shell synchronously — a draft copy named "<name> (copy)" with the source’s email configuration and a fresh Resend topic — then emits this event.

Payload

{ subscriptionId, sourceCampaignId, newCampaignId }

Behaviour

A cursor-paged loop, bounded by MAX_ITERATIONS:

let cursor = ZERO_UUID
for (let i = 0; i < MAX_ITERATIONS; i++) {
  const result = await step.run(`copy-chunk-${i}`, () =>
    emailRepo.copyValidTargetsChunk(sourceCampaignId, newCampaignId, cursor, CHUNK),
  )
  totalInserted += result.inserted
  totalScanned  += result.scanned
  if (result.scanned < CHUNK || !result.nextCursor) break
  cursor = result.nextCursor
}

Then a finalize step logs and returns { sourceCampaignId, newCampaignId, inserted, scanned }.

Points worth noting:

  • Each chunk is its own step.run, so a failure mid-copy resumes from the last completed chunk rather than restarting.

  • The cursor starts at ZERO_UUID and advances by target id, giving a stable ordering that tolerates concurrent inserts.

  • inserted and scanned differ because only valid targets are copied — suppressed and blocked addresses are scanned and skipped. Comparing the two tells an operator how much of the source list is no longer contactable.

  • The new campaign starts in draft, so it cannot begin sending before its recipients have arrived.

emailCampaignImportSegment

File

lib/workflows/functions/email-campaign-import-segment.ts

Event

email-campaign/import-segment.requested

Config

retries: 2, concurrency: { limit: 5 }

Emitted by POST /api/email-campaigns/{id}/targets/from-segment (202) and, optionally, by POST /api/email-campaigns when segment_id is set. This is a one-shot snapshot of the segment’s current active members — not a live audience, and Campaign does not store segmentId.

Payload

{ subscriptionId, campaignId, segmentId }

Behaviour

A verify step no-ops when the campaign is missing, not type = email, or not in the event’s subscription, or when the segment is missing or inactive.

Then a cursor-paged loop calling copySegmentMembersChunk, ordered by priorityScore DESC, id DESC with a composite (priorityScore, id) keyset:

let cursor: SegmentMemberCopyCursor | null = null
for (let i = 0; i < MAX_ITERATIONS; i++) {
  const result = await step.run(`copy-chunk-${i}`, () =>
    emailRepo.copySegmentMembersChunk(subscriptionId, segmentId, campaignId, cursor, CHUNK),
  )
  if (result.scanned < CHUNK || !result.nextCursor) break
  cursor = result.nextCursor
}

Eligibility is SQL-only (no HubSpot hydration): open membership (removedAt IS NULL), a local Contact (by contactId, or HubSpot id when contactId is null), Contact.email, and a sendable target_emails row (not deleted / opted out / blocked). One address per contact, matching @@unique([campaignId, contactId]). ON CONFLICT DO NOTHING makes re-imports and step replays idempotent.

finalize logs campaign/segment ids and { inserted, scanned } only — no PII.