KKKK
R
F
B
createInstancenpm run cf-typegenMY_WORKFLOWFetcherService
Ac.env.COVE_WORKFLOW_VIDEO.create({
A
H
II
J
JYA
.create().status().terminate()to call into Workflows from other Workers scripts.USER_WORKFLOWwrangler devwrangler dev -c apps/rest/wrangler.jsonc -c apps/workflows/wrangler.jsonc -c apps/orm/wrangler.jsonc
But now I have the issue that I have multiple workers with an entry point, and I need to write a handler locally like described here: wrangler
A
Awrangler dev
DAAAA
TTTbut we don't want to implement our logic twiceWhy don't you share the workflow logic by abstracting it into a function which can be called by either the Workflow or synchronous endpoint?
TAATATS
S
F
wrangler dev --env dev --port 53769 --persist-to=../web-app/.wrangler/state
C
AA
IC
L
M
Instanceid
id
.get(
"/workflow",
validator(
"query",
z.object({
videoId: z.string().min(1),
instanceId: z.string().optional(),
}),
),
describeRoute({
tags: ["Video"],
}),
async (c) => {
const organization = c.get("organization");
const { instanceId, videoId } = c.req.valid("query");
const video = await db(c.env.COVE).video.findUnique({
where: { id: videoId, organizationId: organization?.id },
});
if (!video) {
return c.json({ error: "Video not found" }, 404);
}
if (instanceId) {
const instance =
await c.env.COVE_WORKFLOW_VIDEO.get(instanceId);
return c.json({
status: await instance.status(),
});
}
// Spawn a new instance and return the ID and status
const instance = await c.env.COVE_WORKFLOW_VIDEO.create({
params: {
key: video.key,
},
});
return c.json({
id: instance.id,
details: await instance.status(),
});
},
)// Inside the workflow (Maybe it could have some timeout option to fail the workflow if the event never comes within a timeframe)
const payload = await step.waitFor("event_name", "10 minutes")
// And outside the workflow, having the ability to check what event is being waited on would be great to avoid race conditions leading to stuck workflows
if("event_name" === await workflowInstance.waitedOnEvent()){
await workflowInstance.send("event_name", /* Optional payload */)
}
// And for events that we just want to queue for whenever they come up and don't care that the workflow will start waiting for them we can simply send them
// This would be useful for cases where multiple webhook events that we are interested in might come out of order so that we wouldn't need to care about when we send them to the workflow
await workflowInstance.send("some_event", /* Optional payload */)const a = await this.env.EXTENSION_HANDLER.get('id');
let state = await a.status();
while (state.status !== 'running') {
state = await a.status();
}export class HelloWorldWorkflow extends WorkflowEntrypoint {
async run(_: Readonly<WorkflowEvent<unknown>>, step: WorkflowStep) {
while (true) {
try {
await step.do('fetch data', async () => {
await fetch('data url')
})
await step.sleep('wait untill next data', '5 minutes')
} catch {
await step.sleep('handle too many subrequests', '15 minutes')
}
}
}
}// Start workflows
c.env.COVE_WORKFLOW_VIDEO_UPLOAD_STREAM.create({
params: {
videoId: id,
},
});
c.env.COVE_WORKFLOW_VIDEO_TRANSCRIPT.create({
params: {
videoId: id,
},
});