class SignupInput(BaseModel):
    email: str
    user_id: str


class WelcomeEmailResult(BaseModel):
    user_id: str
    welcome_sent: bool
    follow_up_sent: bool

export type SignupInput = {
  email: string;
  user_id: string;
};

export type WelcomeEmailResult = {
  userId: string;
  welcomeSent: boolean;
  followUpSent: boolean;
};

@hatchet.durable_task(
    name="welcome-email",
    on_events=["user:signup"],
    input_validator=SignupInput,
    execution_timeout=timedelta(minutes=5),
)
async def welcome_email(input: SignupInput, ctx: DurableContext) -> WelcomeEmailResult:
    # Step 1: Send the welcome email
    print(f"Sending welcome email to {input.email}: finish your first onboarding step")

    # Step 2: Wait for the user to complete onboarding, or time out
    # (use a longer duration for a more realistic workflow)
    now = await ctx.aio_now()
    consider_events_since = now - timedelta(minutes=LOOKBACK_MINUTES)

    wait_result = await ctx.aio_wait_for(
        "onboarding-or-timeout",
        or_(
            SleepCondition(timedelta(seconds=TIMEOUT_SECONDS)),
            # Scope the event condition to this user so that another user's
            # onboarding-completed event does not resolve this wait.
            UserEventCondition(
                event_key=ONBOARDING_EVENT_KEY,
                scope=input.user_id,
                consider_events_since=consider_events_since,
            ),
        ),
    )

    # The or-group result is {"CREATE": {"<condition_key>": ...}}.
    # Check whether the onboarding event was the one that resolved.
    resolved_key = list(wait_result["CREATE"].keys())[0]
    onboarding_completed = resolved_key == ONBOARDING_EVENT_KEY

    if onboarding_completed:
        # Step 3a: User completed onboarding -> skip follow-up
        print(f"User {input.user_id} completed onboarding, skipping follow-up")
        return WelcomeEmailResult(
            user_id=input.user_id,
            welcome_sent=True,
            follow_up_sent=False,
        )

    # Step 3b: Timeout -> send follow-up email
    print(f"Sending follow-up email to {input.email}: need help finishing onboarding?")
    return WelcomeEmailResult(
        user_id=input.user_id,
        welcome_sent=True,
        follow_up_sent=True,
    )

export const welcomeEmail = hatchet.durableTask<SignupInput, WelcomeEmailResult>({
  name: 'welcome-email',
  onEvents: ['user:signup'],
  executionTimeout: '5m',
  fn: async (input, ctx) => {
    // Step 1: Send the welcome email
    console.log(`Sending welcome email to ${input.email}: finish your first onboarding step`);

    // Step 2: Wait for the user to complete onboarding, or time out
    // (use a longer duration for a more realistic workflow)
    const now = await ctx.now();
    const considerEventsSince = new Date(
      now.getTime() - durationToMs(LOOKBACK_WINDOW)
    ).toISOString();

    const waitResult = await ctx.waitFor(
      Or(
        { sleepFor: `${TIMEOUT_SECONDS}s` },
        // Scope the event condition to this user so that another user's
        // onboarding-completed event does not resolve this wait.
        { eventKey: ONBOARDING_EVENT_KEY, scope: input.user_id, considerEventsSince }
      )
    );

    // The or-group result is { CREATE: { <condition_key>: ... } }.
    // Check whether the onboarding event was the one that resolved.
    const create = (waitResult as Record<string, Record<string, unknown>>)['CREATE'] ?? waitResult;
    const resolvedKey = Object.keys(create as Record<string, unknown>)[0] ?? '';
    const onboardingCompleted = resolvedKey === ONBOARDING_EVENT_KEY;

    if (onboardingCompleted) {
      // Step 3a: User completed onboarding -> skip follow-up
      console.log(`User ${input.user_id} completed onboarding, skipping follow-up`);
      return { userId: input.user_id, welcomeSent: true, followUpSent: false };
    }

    // Step 3b: Timeout -> send follow-up email
    console.log(`Sending follow-up email to ${input.email}: need help finishing onboarding?`);
    return { userId: input.user_id, welcomeSent: true, followUpSent: true };
  },
});

def main() -> None:
    worker = hatchet.worker("welcome-email-worker", workflows=[welcome_email])
    worker.start()


if __name__ == "__main__":
    main()

from examples.welcome_email.worker import (
    ONBOARDING_EVENT_KEY,
    SignupInput,
    hatchet,
    welcome_email,
)

signup = SignupInput(
    email="alice@example.com",
    user_id="user-123",
)

# Start the welcome-email workflow
ref = welcome_email.run(signup, wait_for_result=False)
print(f"Started workflow run: {ref.workflow_run_id}")

# Push onboarding-completed event (scoped to this user)
print("Pushing onboarding-completed event...")
hatchet.event.push(
    ONBOARDING_EVENT_KEY,
    {"status": "done"},
    scope=signup.user_id,
)

# Wait for the workflow to complete
result = ref.result()
print(f"Workflow completed: {result}")

import { hatchet } from '../hatchet-client';
import { welcomeEmail, ONBOARDING_EVENT_KEY, SignupInput } from './workflow';

async function main() {
  const input: SignupInput = {
    email: 'alice@example.com',
    user_id: 'user-123',
  };

  // Start the welcome-email workflow
  const ref = await welcomeEmail.runNoWait(input);
  const runId = await ref.getWorkflowRunId();
  console.log(`Started workflow run: ${runId}`);

  // Push onboarding-completed event (scoped to this user)
  console.log('Pushing onboarding-completed event...');
  await hatchet.events.push(ONBOARDING_EVENT_KEY, { status: 'done' }, { scope: input.user_id });

  // Wait for the workflow to complete
  const result = await ref.output;
  console.log('Workflow completed:', result);
}