Skip to main content
This example demonstrates a customer onboarding process using Upstash Workflow. The following example workflow registers a new user, sends welcome emails, and periodically checks and responds to the user’s activity state.

​
Use Case

Our workflow will:
  1. Register a new user to our service
  2. Send them a welcome email
  3. Wait for a certain time
  4. Periodically check the user’s state
  5. Send appropriate emails based on the user’s activity

​
Code Example

import { serve } from "@upstash/workflow/nextjs"

type InitialData = {
  email: string
}

export const { POST } = serve<InitialData>(async (context) => {
  const { email } = context.requestPayload

  await context.run("new-signup", async () => {
    await sendEmail("Welcome to the platform", email)
  })

  await context.sleep("wait-for-3-days", 60 * 60 * 24 * 3)

  while (true) {
    const state = await context.run("check-user-state", async () => {
      return await getUserState()
    })

    if (state === "non-active") {
      await context.run("send-email-non-active", async () => {
        await sendEmail("Email to non-active users", email)
      })
    } else if (state === "active") {
      await context.run("send-email-active", async () => {
        await sendEmail("Send newsletter to active users", email)
      })
    }

    await context.sleep("wait-for-1-month", 60 * 60 * 24 * 30)
  }
})

async function sendEmail(message: string, email: string) {
  // Implement email sending logic here
  console.log(`Sending ${message} email to ${email}`)
}

type UserState = "non-active" | "active"

const getUserState = async (): Promise<UserState> => {
  // Implement user state logic here
  return "non-active"
}

from fastapi import FastAPI
from typing import Literal, TypedDict
from upstash_workflow.fastapi import Serve
from upstash_workflow import AsyncWorkflowContext

app = FastAPI()
serve = Serve(app)

UserState = Literal["non-active", "active"]


class InitialData(TypedDict):
    email: str


async def send_email(message: str, email: str) -> None:
    # Implement email sending logic here
    print(f"Sending {message} email to {email}")


async def get_user_state() -> UserState:
    # Implement user state logic here
    return "non-active"


@serve.post("/customer-onboarding")
async def customer_onboarding(context: AsyncWorkflowContext[InitialData]) -> None:
    email = context.request_payload["email"]

    async def _new_signup() -> None:
        await send_email("Welcome to the platform", email)

    await context.run("new-signup", _new_signup)

    await context.sleep("wait-for-3-days", 60 * 60 * 24 * 3)

    while True:

        async def _check_user_state() -> UserState:
            return await get_user_state()

        state: UserState = await context.run("check-user-state", _check_user_state)

        if state == "non-active":

            async def _send_email_non_active() -> None:
                await send_email("Email to non-active users", email)

            await context.run("send-email-non-active", _send_email_non_active)
        else:

            async def _send_email_active() -> None:
                await send_email("Send newsletter to active users", email)

            await context.run("send-email-active", _send_email_active)

        await context.sleep("wait-for-1-month", 60 * 60 * 24 * 30)

​
Code Breakdown

​
1. New User Signup

We start by sending a newly signed-up user a welcome email: 
await context.run("new-signup", async () => {
  await sendEmail("Welcome to the platform", email)
})

async def _new_signup() -> None:
    await send_email("Welcome to the platform", email)

await context.run("new-signup", _new_signup)

​
2. Initial Waiting Period

To leave time for the user to interact with our platform, we use context.sleep to pause our workflow for 3 days: 
await context.sleep("wait-for-3-days", 60 * 60 * 24 * 3)

await context.sleep("wait-for-3-days", 60 * 60 * 24 * 3)

​
3. Periodic State Check

We enter an infinite loop to periodically (every month) check the user’s engagement level with our platform and send appropriate emails: 
while (true) {
  const state = await context.run("check-user-state", async () => {
    return await getUserState()
  })

  if (state === "non-active") {
    await context.run("send-email-non-active", async () => {
      await sendEmail("Email to non-active users", email)
    })
  } else if (state === "active") {
    await context.run("send-email-active", async () => {
      await sendEmail("Send newsletter to active users", email)
    })
  }

  await context.sleep("wait-for-1-month", 60 * 60 * 24 * 30)
}

while True:

    async def _check_user_state() -> UserState:
        return await get_user_state()

    state: UserState = await context.run("check-user-state", _check_user_state)

    if state == "non-active":

        async def _send_email_non_active() -> None:
            await send_email("Email to non-active users", email)

        await context.run("send-email-non-active", _send_email_non_active)
    else:

        async def _send_email_active() -> None:
            await send_email("Send newsletter to active users", email)

        await context.run("send-email-active", _send_email_active)

    await context.sleep("wait-for-1-month", 60 * 60 * 24 * 30)

​
Key Features

  1. Non-blocking sleep: We use context.sleep for pausing the workflow without consuming execution time (great for optimizing serverless cost).
  2. Long-running task: This workflow runs indefinitely, checking and responding to a users engagement state every month.