import type { env, WorkflowEvent, WorkflowStep } from "cloudflare:workers";
import { WorkflowEntrypoint } from "cloudflare:workers";
import { NonRetryableError } from "cloudflare:workflows";

type OrderParams = { orderId: string; userId: string };

export class OrderWorkflow extends WorkflowEntrypoint<typeof env, OrderParams> {
  async run(event: WorkflowEvent<OrderParams>, step: WorkflowStep) {
    // Dynamic import for database (required for proper connection handling)
    const { db } = await import("@/db");

    const user = await step.do("fetch-user", async () => {
      return await db
        .selectFrom("user")
        .where("id", "=", event.payload.userId)
        .executeTakeFirst();
    });

    if (!user) {
      throw new NonRetryableError("User not found");
    }

    await step.do(
      "process-order",
      {
        retries: { limit: 3, delay: "5 seconds", backoff: "exponential" },
        timeout: "2 minutes",
      },
      async () => {
        return await processOrder(event.payload.orderId);
      },
    );

    await step.sleep("wait-for-review-period", "24 hours");

    await step.do("send-confirmation", async () => {
      await sendEmail(user.email, "Order complete!");
    });

    return { status: "complete" };
  }
}

# wrangler.toml
[[workflows]]
name = "order-workflow"
binding = "MY_WORKFLOW"
class_name = "OrderWorkflow"

await step.do(
  "call-external-api",
  {
    retries: { limit: 5, delay: "10 seconds", backoff: "exponential" },
    timeout: "5 minutes",
  },
  async () => {
    /* ... */
  },
);

import { NonRetryableError } from "cloudflare:workflows";

await step.do("validate-input", async () => {
  if (!isValid) throw new NonRetryableError("Invalid input");
});

// From fetch handler
export default {
  async fetch(
    req: Request,
    env: typeof import("cloudflare:workers").env,
  ): Promise<Response> {
    // Create instance
    const instance = await env.MY_WORKFLOW.create({
      id: `order-${orderId}`, // Must be unique
      params: { orderId, userId },
    });

    // Get existing instance
    const existing = await env.MY_WORKFLOW.get(instanceId);
    return Response.json(await existing.status());
  },
};

// Instance methods
await instance.status(); // { status: 'running' | 'complete' | ... }
await instance.pause();
await instance.resume();
await instance.terminate();

// Send event to waiting instance
await instance.sendEvent({
  type: "payment-confirmed",
  payload: { status: "success" },
});

// Batch creation (use instead of loops)
await env.MY_WORKFLOW.createBatch([
  { id: "user-1", params: { name: "Alice" } },
  { id: "user-2", params: { name: "Bob" } },
]);

// Trigger another workflow from within a workflow
const subInstance = await step.do("create-sub-workflow", async () => {
  return await this.env.OTHER_WORKFLOW.create({
    params: {
      /* ... */
    },
  });
});

// ✅ Check before mutating
await step.do("charge-customer", async () => {
  const sub = await getSubscription(id);
  if (sub.charged) return; // Already done
  await chargeCustomer(id);
});

// ❌ State lost on hibernation
let results = [];
await step.do("fetch-data", async () => {
  results.push(data);
});
await step.sleep("wait-period", "1 hour");
// results is empty here!

// ✅ Build state from step returns
const results = await Promise.all([
  step.do("fetch-item-1", () => getData(1)),
  step.do("fetch-item-2", () => getData(2)),
]);

// ❌ Creates multiple instances on replay
const sub = await this.env.OTHER_WORKFLOW.create();
const random = Math.random();

// ✅ Wrap in steps
const sub = await step.do("create-sub-workflow", () =>
  this.env.OTHER_WORKFLOW.create(),
);
const random = await step.do("generate-random", () => Math.random());

// ❌ Different name on each replay
await step.do(`step-${Date.now()}`, async () => {
  /* ... */
});

// ✅ Stable, descriptive names (kebab-case)
await step.do("process-payment", async () => {
  /* ... */
});

// ✅ Dynamic but deterministic (based on prior step output)
const items = await step.do("get-items", () => fetchItems());
for (const item of items) {
  await step.do(`process-item-${item.id}`, () => processItem(item));
}

// ❌ Dangling promise causes race conditions
step.do("fetch-data", async () => {
  /* ... */
});

// ✅ Always await
await step.do("fetch-data", async () => {
  /* ... */
});

// ❌ Reusing ID prevents multiple runs for same user
await env.MY_WORKFLOW.create({ id: userId });

// ✅ Composite ID or naturally unique (transactionId, orderId)
await env.MY_WORKFLOW.create({
  id: `${userId}-${crypto.randomUUID().slice(0, 6)}`,
});

// ❌ Individual creates hit rate limits
for (const user of users) {
  await env.MY_WORKFLOW.create({ id: user.id, params: user });
}

// ✅ Batch creation
await env.MY_WORKFLOW.createBatch(users.map((u) => ({ id: u.id, params: u })));

// ✅ Consistent caching
const result = await step.do("race-fetches", async () => {
  return await Promise.race([fetchFast(), fetchSlow()]);
});

export class MyWorkflow extends WorkflowEntrypoint<typeof env, Params> {
  async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
    // ✅ Dynamic import at start of run()
    const { db } = await import("@/db");

    await step.do("query-database", async () => {
      return await db.selectFrom("users").execute();
    });
  }
}

// Start workflow, wait for external webhook
const confirmation = await step.waitForEvent<{ status: string }>(
  "await-payment-webhook",
  {
    type: "payment-confirmed",
    timeout: "1 hour",
  },
);

// Webhook handler sends event
app.post("/webhook", async (req, env) => {
  const instance = await env.MY_WORKFLOW.get(req.body.orderId);
  await instance.sendEvent({ type: "payment-confirmed", payload: req.body });
});

await step.do("send-welcome-email", () => sendEmail(user, "Welcome!"));
await step.sleep("wait-for-engagement", "7 days");
await step.do("send-followup-email", () =>
  sendEmail(user, "How are things going?"),
);

try {
  await step.do("risky-operation", async () => {
    throw new NonRetryableError("failed");
  });
} catch (e) {
  await step.do("cleanup-after-failure", () => rollback());
}
// Workflow continues
await step.do("alternative-approach", () => alternativeApproach());

await step.do("update-status-processing", async () => {
  await updateJobStatus(db, jobId, {
    status: "processing",
    currentStep: "Validating input",
    progress: 20,
  });
  await triggerRealtimeUpdate(this.env.REALTIME_DO, jobId);
});

const result = await step.do(
  "process-data",
  {
    retries: { limit: 3, delay: "10 seconds", backoff: "exponential" },
    timeout: "5 minutes",
  },
  async () => {
    return await processData();
  },
);

await step.do("update-status-completed", async () => {
  await updateJobStatus(db, jobId, {
    status: "completed",
    progress: 100,
    completedAt: new Date().toISOString(),
  });
});