← Cloudflare Workflows / workflows / build

События и параметры

При запуске Workflow может получить необязательное событие. Это событие может содержать данные, с которыми будет работать ваш Workflow, включая сведения о запросе, данные пользователя, полученные из базы данных (например, D1 или KV) или через вебхук, либо сообщения от потребителя Queue.

События являются важной частью Workflow, поскольку Workflow часто должен обрабатывать данные. Поскольку конкретный экземпляр Workflow выполняется устойчиво, события удобно использовать для передачи Workflow данных, которые должны оставаться неизменными и/или представляют собой данные, необходимые Workflow для работы в этот конкретный момент времени.

Передача данных в Workflow

Передать параметры в Workflow можно тремя способами:

В качестве необязательного аргумента create метод у привязка Workflow при запуске Workflow из Worker.
Через --params флаг при использовании wrangler CLI для запуска Workflow.
Через step.waitForEvent API, который позволяет экземпляру Workflow дождаться получения события (и необязательных данных) пока он выполняется. Экземплярам Workflow можно отправлять события из внешних сервисов по HTTP или через Workers API для Workflows.

В качестве параметра можно передать любой объект, сериализуемый в JSON.

export default {
	async fetch(req, env) {
		let someEvent = { url: req.url, createdTimestamp: Date.now() };
		// Trigger our Workflow
		// Pass our event as the second parameter to the `create` method
		// on our Workflow binding.
		let instance = await env.MY_WORKFLOW.create({
			id: crypto.randomUUID(),
			params: someEvent,
		});

		return Response.json({
			id: instance.id,
			details: await instance.status(),
		});
	},
};

export default {
	async fetch(req: Request, env: Env) {
		let someEvent = { url: req.url, createdTimestamp: Date.now() };
		// Trigger our Workflow
		// Pass our event as the second parameter to the `create` method
		// on our Workflow binding.
		let instance = await env.MY_WORKFLOW.create({
			id: crypto.randomUUID(),
			params: someEvent,
		});

		return Response.json({
			id: instance.id,
			details: await instance.status(),
		});
	},
};

Чтобы передать параметры через wrangler интерфейс командной строки, передайте строку JSON в качестве второго параметра для workflows trigger подкоманда:

npx wrangler@latest workflows trigger workflows-starter '{"some":"data"}'

🚀 Workflow instance "57c7913b-8e1d-4a78-a0dd-dce5a0b7aa30" has been queued successfully

Ожидание событий

Выполняющийся Workflow может ожидать событие (или события), вызвав step.waitForEvent внутри Workflow, что позволяет отправлять события в Workflow одним из двух способов:

Через Привязка Workers API: вызовите instance.sendEvent для отправки событий в конкретные экземпляры Workflow.
С помощью эндпоинта REST API (HTTP API) Эндпоинт событий.

Поскольку waitForEvent является частью WorkflowStep API, вы можете вызывать его несколько раз в рамках Workflow и использовать управляющие конструкции для условного ожидания события.

Вызов waitForEvent требует указать type (до 100 символов ¹), которая используется для сопоставления с соответствующим type при отправке события в экземпляр Workflow.

Например, чтобы дождаться вебхука выставления счета:

export class MyWorkflow extends WorkflowEntrypoint {
	async run(event, step) {
		// Other steps in your Workflow
		let stripeEvent = await step.waitForEvent(
			"receive invoice paid webhook from Stripe",
			{ type: "stripe-webhook", timeout: "1 hour" },
		);
		// Rest of your Workflow
	}
}

export class MyWorkflow extends WorkflowEntrypoint<Env, Params> {
	async run(event: WorkflowEvent<Params>, step: WorkflowStep) {
		// Other steps in your Workflow
		let stripeEvent = await step.waitForEvent<IncomingStripeWebhook>(
			"receive invoice paid webhook from Stripe",
			{ type: "stripe-webhook", timeout: "1 hour" },
		);
		// Rest of your Workflow
	}
}

В приведённом выше примере:

Вызовы waitForEvent с type stripe-webhook : соответствующий sendEvent вызов, таким образом, будет await instance.sendEvent({type: "stripe-webhook", payload: webhookPayload}).
Использует TypeScript параметр типа ↗ для указания типа возвращаемого значения step.waitForEvent в качестве нашего IncomingStripeWebhook.
Продолжает выполнение остальной части Workflow.

Тайм-аут по умолчанию для waitForEvent вызов составляет 24 часа, это значение можно изменить, передав { timeout: WorkflowTimeoutDuration } в качестве второго аргумента вашего waitForEvent вызов.

let event = await step.waitForEvent("wait for human approval", {
	type: "approval-flow",
	timeout: "15 minutes",
});

let event = await step.waitForEvent(
		"wait for human approval",
		{ type: "approval-flow", timeout: "15 minutes" },
	);

Можно задать тайм-аут от 1 секунды до 365 дней.

Поведение тайм-аута

Когда waitForEvent истекает, Workflow выбросит ошибку, и экземпляр завершится со сбоем. Если нужно, чтобы Workflow продолжал работу даже при отсутствии события, оберните waitForEvent вызов в try...catch блок:

try {
	const event = await step.waitForEvent("wait for approval", {
		type: "approval",
		timeout: "1 hour",
	});
	// Handle the received event
} catch (e) {
	// Timeout occurred - handle the case where no event was received
	console.log("No approval received, proceeding with default action");
}

try {
	const event = await step.waitForEvent("wait for approval", {
		type: "approval",
		timeout: "1 hour",
	});
	// Handle the received event
} catch (e) {
	// Timeout occurred - handle the case where no event was received
	console.log("No approval received, proceeding with default action");
}

Отправляйте события в выполняющиеся workflows

Экземпляры Workflow, которые ожидают события с помощью waitForEvent API можно отправлять события с помощью instance.sendEvent API:

export default {
	async fetch(req, env) {
		const instanceId = new URL(req.url).searchParams.get("instanceId");
		const webhookPayload = await req.json();

		let instance = await env.MY_WORKFLOW.get(instanceId);
		// Send our event, with `type` matching the event type defined in
		// our step.waitForEvent call
		await instance.sendEvent({
			type: "stripe-webhook",
			payload: webhookPayload,
		});

		return Response.json({
			status: await instance.status(),
		});
	},
};

export default {
	async fetch(req: Request, env: Env) {
		const instanceId = new URL(req.url).searchParams.get("instanceId");
		const webhookPayload = await req.json<Payload>();

		let instance = await env.MY_WORKFLOW.get(instanceId);
		// Send our event, with `type` matching the event type defined in
		// our step.waitForEvent call
		await instance.sendEvent({
			type: "stripe-webhook",
			payload: webhookPayload,
		});

		return Response.json({
			status: await instance.status(),
		});
	},
};

Аналогично waitForEvent примера в этом руководстве, type свойство в нашем waitForEvent и sendEvent поля должны совпадать.
Чтобы отправить несколько событий в Workflow с несколькими waitForEvent вызовов, вызовите sendEvent с соответствующим type свойство (до 100 символов ¹).
События также можно отправлять с помощью REST API (HTTP API) через Эндпоинт событий.

TypeScript и параметры типов

По умолчанию WorkflowEvent передаётся в run метод в определении вашего Workflow имеет тип, соответствующий следующему:

export type WorkflowCronSchedule = {
	/** Cron expression that triggered this event. */
	cron: string;
	/** Timestamp of the scheduled trigger, in milliseconds since the Unix epoch. */
	scheduledTime: number;
};

export type WorkflowEvent<T> = {
	/** The data passed as the parameter when the Workflow instance was triggered. */
	payload: Readonly<T>;
	/** The timestamp that the Workflow was triggered. */
	timestamp: Date;
	/** ID of the current Workflow instance. */
	instanceId: string;
	/** Name of the current Workflow. */
	workflowName: string;
	/** Metadata for Workflow instances created by a cron schedule. */
	schedule?: WorkflowCronSchedule;
};

Когда экземпляр Workflow создаётся по cron-расписанию, настроенному в привязке Workflow, event.schedule включает cron-выражение, создавшее инстанс, и запланированное время запуска:

export class MyWorkflow extends WorkflowEntrypoint<Env> {
	async run(event: WorkflowEvent<unknown>, step: WorkflowStep) {
		if (event.schedule) {
			console.log(event.schedule.cron);
			console.log(new Date(event.schedule.scheduledTime));
		}
	}
}

При желании можно типизировать эти события, определив собственный тип и передав его как параметр типа ↗ к WorkflowEvent:

// Define a type that conforms to the events your Workflow instance is
// instantiated with
interface YourEventType {
	userEmail: string;
	createdTimestamp: number;
	metadata?: Record<string, string>;
}

Когда вы передаёте свой YourEventType к WorkflowEvent в качестве параметра типа, event.payload свойство теперь имеет тип YourEventType на протяжении всего определения вашего workflow:

src/index.ts

// Import the Workflow definition
import { WorkflowEntrypoint, WorkflowStep, WorkflowEvent} from 'cloudflare:workers';

export class MyWorkflow extends WorkflowEntrypoint {
	// Pass your type as a type parameter to WorkflowEvent
	// The 'payload' property will have the type of your parameter.
	async run(event: WorkflowEvent<YourEventType>, step: WorkflowStep) {
		let state = await step.do("my first step", async () => {
			// Access your properties via event.payload
          let userEmail = event.payload.userEmail
          let createdTimestamp = event.payload.createdTimestamp
        })

        await step.do("my second step", async () => { /* your code here */ })
	}
}

Вы также можете указать параметр type для Workflows тип при создании (запуске) экземпляра Workflow с помощью create метод объекта API Workers. Обратите внимание, что это не передать информацию о типах в сам Workflow, поскольку типы TypeScript существуют только на этапе сборки.

Сноски

Шаблон соответствия: ^[a-zA-Z0-9_][a-zA-Z0-9-_]*$ ↩ ↩²