Parallel Steps

import { defineWorkflow } from "openworkflow";

export const fetchAllData = defineWorkflow(
  { name: "fetch-all-data" },
  async ({ input, step }) => {
    const [user, subscription, settings] = await Promise.all([
      step.run({ name: "fetch-user" }, async () => {
        return await db.users.findOne({ id: input.userId });
      }),
      step.run({ name: "fetch-subscription" }, async () => {
        return await stripe.subscriptions.retrieve(input.subId);
      }),
      step.run({ name: "fetch-settings" }, async () => {
        return await db.settings.findOne({ userId: input.userId });
      }),
    ]);

    return { user, subscription, settings };
  },
);

const [a, b, c] = await Promise.all([
  step.run({ name: "step-a" }, ...),  // Completed before crash - cached
  step.run({ name: "step-b" }, ...),  // Completed before crash - cached
  step.run({ name: "step-c" }, ...),  // In progress during crash - re-runs
]);

try {
  const [user, settings] = await Promise.all([
    step.run({ name: "fetch-user" }, async () => {
      return await db.users.findOne({ id: input.userId });
    }),
    step.run({ name: "fetch-settings" }, async () => {
      throw new Error("Database unavailable");
    }),
  ]);
} catch (error) {
  // Handle the error
  console.error("Failed to fetch data:", error);
}

const results = await Promise.allSettled([
  step.run({ name: "fetch-user" }, ...),
  step.run({ name: "fetch-settings" }, ...),
]);

for (const result of results) {
  if (result.status === "fulfilled") {
    console.log("Success:", result.value);
  } else {
    console.log("Failed:", result.reason);
  }
}

const [user, orders, notifications] = await Promise.all([
  step.run({ name: "fetch-user" }, () => db.users.findOne(userId)),
  step.run({ name: "fetch-orders" }, () => db.orders.find({ userId })),
  step.run({ name: "fetch-notifications" }, () =>
    db.notifications.find({ userId }),
  ),
]);

await Promise.all([
  step.run({ name: "send-email" }, () =>
    emailService.send({ to: user.email, ... })
  ),
  step.run({ name: "send-sms" }, () =>
    smsService.send({ to: user.phone, ... })
  ),
  step.run({ name: "send-push" }, () =>
    pushService.send({ to: user.deviceToken, ... })
  ),
]);

const items = await step.run({ name: "fetch-items" }, () => db.items.find());

// Process items in parallel (be mindful of API rate limits)
await Promise.all(
  items.map((item) =>
    step.run({ name: `process-${item.id}` }, async () => {
      await processItem(item);
    }),
  ),
);

// Good - clear what each step does
await Promise.all([
  step.run({ name: "fetch-user-profile" }, ...),
  step.run({ name: "fetch-user-preferences" }, ...),
]);

// Bad - confusing names
await Promise.all([
  step.run({ name: "step-1" }, ...),
  step.run({ name: "step-2" }, ...),
]);

// Be careful with large arrays - you might hit rate limits
const items = await step.run({ name: "fetch-items" }, () => db.items.find());

// Process in smaller batches if needed
const batchSize = 10;
for (let i = 0; i < items.length; i += batchSize) {
  const batch = items.slice(i, i + batchSize);
  await Promise.all(
    batch.map((item) =>
      step.run({ name: `process-${item.id}` }, () => api.process(item)),
    ),
  );
}

// Good - independent fetches
const [user, products] = await Promise.all([
  step.run({ name: "fetch-user" }, ...),
  step.run({ name: "fetch-products" }, ...),
]);

// Bad - second step depends on first
const [user, orders] = await Promise.all([
  step.run({ name: "fetch-user" }, ...),
  step.run({ name: "fetch-orders" }, async () => {
    // This needs user.id, but we don't have it yet!
    return await db.orders.find({ userId: user.id });
  }),
]);

const user = await step.run({ name: "fetch-user" }, ...);
const orders = await step.run({ name: "fetch-orders" }, async () => {
  return await db.orders.find({ userId: user.id });
});