Async Jobs & Polling

import { setTimeout as sleep } from 'node:timers/promises'

async function fireAndPoll() {
  // 1. Fire, returns 200 with the run ids (trigger id is in the PATH)
  const fire = await fetch('https://brew.new/api/v1/automations/triggers/tri_signup/fire', {
    method: 'POST',
    headers: {
      Authorization: `Bearer ${process.env.BREW_API_KEY!}`,
      'Content-Type': 'application/json',
      'Idempotency-Key': `signup-${userId}-${eventTimestamp}`,
    },
    body: JSON.stringify({
      payload: { email: 'jane@example.com', firstName: 'Jane' },
    }),
  })
  const fireBody = await fire.json() as {
    details: { automationRunIds: string[]; triggerInstanceId?: string }
  }
  const [runId] = fireBody.details.automationRunIds

  // 2. Poll the run until it terminates (logs are always attached)
  for (let attempt = 0; attempt < 60; attempt += 1) {
    const res = await fetch(
      `https://brew.new/api/v1/automations/runs?automationRunId=${runId}&include=logs`,
      { headers: { Authorization: `Bearer ${process.env.BREW_API_KEY!}` } }
    )
    const body = await res.json() as {
      data: Array<{
        automationRunId: string
        status: 'pending' | 'running' | 'completed' | 'failed' | 'canceled'
        logs?: Array<{ nodeName: string; status: string; durationMs?: number }>
      }>
    }
    const run = body.data[0]!
    if (run.status === 'completed' || run.status === 'failed' || run.status === 'canceled') {
      console.log(`Run ${run.automationRunId} → ${run.status}`)
      for (const log of run.logs ?? []) {
        console.log(`  ${log.nodeName}: ${log.status} (${log.durationMs ?? '?'}ms)`)
      }
      return run
    }
    await sleep(2_000) // poll every 2 seconds
  }
  throw new Error('Run did not terminate within 2 minutes')
}

const send = await brew.emails.send({
  emailId: 'eml_promo',
  emailVersionId: 'emv_promo_v2',
  domainId: 'dom_brand_primary',
  audienceId: 'aud_subscribers',
  subject: 'Black Friday, 40% off',
})

console.log(send)
// { status: 'queued', sendId: 'snd_…', runId: 'wrun_…' }

const targets = await loadNewSignups()         // 50 users

const fires = await Promise.all(
  targets.map((user) =>
    brew.automations.triggers.fire(
      {
        triggerEventId: 'tri_signup',
        payload: { email: user.email, firstName: user.firstName },
      },
      { idempotencyKey: `signup-${user.id}-${user.signedUpAt}` }
    )
  )
)

// Each fire returns its own automationRunIds[]; collect for later inspection
const allRunIds = fires.flatMap((f) => f.automationRunIds)

async function* watchRun(runId: string) {
  while (true) {
    const { data } = await brew.automations.runs.list({
      automationRunId: runId,
      include: 'logs',
    })
    const run = data[0]!
    yield run
    if (['completed', 'failed', 'canceled'].includes(run.status)) return
    await new Promise((r) => setTimeout(r, 2_000))
  }
}

for await (const snapshot of watchRun(runId)) {
  await mySideEffect(snapshot)   // e.g. update a UI badge
}

const { data: runs } = await brew.automations.runs.list({
  automationId: 'auto_abc',
  limit: 100,
})
const failed = runs.filter((r) => r.status === 'failed')