Event-Driven Patterns

let lastSeen: number | undefined = head
let stopped = false
 
// Listen for newly-ready tasks and notify assignees. Reconnect with the last sequence: the
// server ends each stream after a few minutes, and resuming loses nothing.
async function watchForReadyTasks() {
  while (!stopped) {
    for await (const event of client.events.stream({ types: ['task.ready'], sinceSequence: lastSeen })) {
      lastSeen = event.sequence
      const taskId = event.task_id!
      const { data: task } = await client.tasks.get(taskId)
      if (task.assigneePlatformUserId) {
        await sendPushNotification(task.assigneePlatformUserId, {
          title: 'New task ready',
          body: task.name,
          url: `/tasks/${task.id}`,
        })
      }
      if (taskId === job.refs.print) stopped = true // (so this demo ends)
      if (stopped) break
    }
  }
}

// Keep a screen in sync with a plan: one job, `schedule.changed` only.
const taskCards: Record<string, { start: string; end: string }> = {}
async function syncScreen() {
  for await (const event of client.events.stream({ workflowInstanceId: job.instance.id, types: ['schedule.changed'], sinceSequence: head })) {
    for (const change of event.data.changes as Array<{ taskId: string; current: { start: string; end: string } }>) taskCards[change.taskId] = change.current
    if (Object.keys(taskCards).length > 0) break
  }
}

async function processCascade(payload: StreamEvent) {
  if (payload.type === 'task.completed') {
    const { data: task } = await client.tasks.get(payload.task_id!)
 
    // When the "Proof sign-off" task completes, send the contract
    if (task.name === 'Proof sign-off') {
      const { data: instance } = await client.workflowInstances.get(task.workflowInstanceId!)
      await sendContractToCounterparty({ dealId: instance.subjectId })
    }
  }
}

async function processConflict(payload: StreamEvent) {
  if (payload.type === 'conflict.detected') {
    const conflicts = await client.conflicts.list({ resolved: false })
 
    for (const conflict of conflicts.data) {
      if (conflict.taskId !== payload.task_id) continue // this event's task (list is already open conflicts only)
      if (conflict.type === 'deadline_violated' && conflict.slackViolationSec > 3600) {
        // More than 1 hour over deadline — escalate
        await pageOnCall({
          message: `Deadline violated: ${conflict.description}`,
          severity: 'high',
          taskId: conflict.taskId,
        })
      }
    }
  }
}

const applied: number[] = []
const applyEventToLocalState = async (e: { sequence: number }) => { applied.push(e.sequence) }
let savedSequence = head
const getLastProcessedSequence = async () => savedSequence
const saveLastProcessedSequence = async (s: number) => { savedSequence = s }
 
async function rebuildState() {
  let sequence = await getLastProcessedSequence()
 
  while (true) {
    // `fromSequence` is exclusive: it returns events AFTER this sequence, so pass the last one you applied.
    const { data: events } = await client.events.replay({
      fromSequence: sequence,
      limit: 500,
    })
 
    if (events.length === 0) break
 
    for (const event of events) {
      await applyEventToLocalState(event)
      sequence = event.sequence
    }
 
    await saveLastProcessedSequence(sequence)
  }
}

async function onOrderReceived(orderId: string) {
  const { data: instance } = await client.workflowInstances.create({
    subjectType: 'order',
    subjectId: orderId,
  })
 
  console.log(`Started workflow ${instance.id} for order ${orderId}`)
}
 
// Monitor all in-progress orders
async function getDashboard() {
  const { data: instances } = await client.workflowInstances.list({
    status: 'running',
  })
 
  return instances.map(i => ({
    orderId: i.subjectId,
    status: i.status,
    startedAt: i.startedAt,
  }))
}