Function dispatchDueDrains

apps/sim/lib/data-drains/dispatcher.ts:64–186

(now: Date = new Date())

 * belonging to orgs that have lapsed off the enterprise plan are skipped.
 */
export async function dispatchDueDrains(now: Date = new Date()): Promise<{
  candidates: number
  dispatched: number
  skipped: number
  reaped: number
}> {
  const { reaped } = await reapOrphanedRuns(now)

  const hourlyCutoff = new Date(now.getTime() - HOUR_MS + CADENCE_BUFFER_MS)
  const dailyCutoff = new Date(now.getTime() - DAY_MS + CADENCE_BUFFER_MS)

  const duePredicate = and(
    eq(dataDrains.enabled, true),
    or(
      isNull(dataDrains.lastRunAt),
      and(eq(dataDrains.scheduleCadence, 'hourly'), lt(dataDrains.lastRunAt, hourlyCutoff)),
      and(eq(dataDrains.scheduleCadence, 'daily'), lt(dataDrains.lastRunAt, dailyCutoff))
    )
  )

  const candidates = await db
    .select({
      id: dataDrains.id,
      organizationId: dataDrains.organizationId,
      lastRunAt: dataDrains.lastRunAt,
    })
    .from(dataDrains)
    .where(duePredicate)

  if (candidates.length === 0) {
    return { candidates: 0, dispatched: 0, skipped: 0, reaped }
  }

  // Self-hosted deployments have no subscription infra; `DATA_DRAINS_ENABLED`
  // is the global on/off there. Cache per-org so a multi-drain org pays one
  // billing lookup.
  const enterpriseCache = new Map<string, boolean>()
  const isEnterprise = async (orgId: string): Promise<boolean> => {
    if (!isBillingEnabled) return true
    const cached = enterpriseCache.get(orgId)
    if (cached !== undefined) return cached
    const result = await isOrganizationOnEnterprisePlan(orgId)
    enterpriseCache.set(orgId, result)
    return result
  }

  const queue = await getJobQueue()
  let dispatched = 0
  let skipped = 0

  for (const candidate of candidates) {
    let enterprise: boolean
    try {
      enterprise = await isEnterprise(candidate.organizationId)
    } catch (error) {
      // A billing-API failure for one org must not abort the whole batch —
      // skip this drain and let the next cron tick retry it.
      logger.warn('Enterprise check failed; skipping drain', {