A durable workflow orchestrator for Firebase Cloud Functions.
Firequeue lets you write multi-step workflows that persist state to Firestore. Each step runs exactly once, and workflows can pause and resume across function invocations.
# pnpm
pnpm add @fireq/firequeue
# npm
npm install @fireq/firequeue
# yarn
yarn add @fireq/firequeue- Invoke: Call
firequeue.invokeTask()to create a task document in Firestore. - Trigger: A Cloud Function created by
firequeue.createTask()is triggered by the document write. - Execute: The function runs through your workflow. Each
step.run()call:- Creates a step document in a subcollection (if it doesn't exist)
- Executes the step function and saves the result
- In default mode: stops execution and re-triggers the function for the next step
- In speculative mode: continues to the next step until timeout approaches
- Resume: On subsequent runs, completed steps return their saved result without re-executing.
| Mode | Behavior |
|---|---|
serializable (default) | Executes one step per function invocation, then re-triggers |
speculative | Executes as many steps as possible until timeoutSeconds - 5s |
| Status | Description |
|---|---|
scheduled | Task is queued for execution |
running | Task is currently executing |
waiting | Task is waiting for an event via step.waitForEvent() |
completed | All steps finished successfully |
cancelled | Task was cancelled |
error | A step failed |
| Status | Description |
|---|---|
scheduled | Step is ready to execute |
running | Step is currently executing |
completed | Step finished and result is saved |
cancelled | Step was cancelled |
error | Step execution failed |
// src/init.tsimport{createFirequeue}from"@fireq/firequeue";import*asadminfrom"firebase-admin";admin.initializeApp();// Define your task input typesinterfaceTaskRegistry{"order-processing": {orderId: string;customerId: string};"send-email": {to: string;subject: string};}exportconstfirequeue=createFirequeue<TaskRegistry>({firestore: admin.firestore(),logLevel: "debug",// optional});// src/functions.tsimport{firequeue}from"./init";exportconstprocessOrder=firequeue.createTask("order-processing",{collectionPath: "queue",timeoutSeconds: 120,executionMode: "speculative",// optional, defaults to "serializable"},async({ step, input, taskInstanceId })=>{// Step 1constpayment=awaitstep.run("process-payment",async()=>{returnprocessPayment(input.orderId);});// Step 2awaitstep.run("update-inventory",async()=>{returnupdateInventory(input.orderId);});// Wait for external event (e.g., webhook confirmation)awaitstep.waitForEvent({event: "payment-confirmed"});// Step 3awaitstep.run("send-confirmation",async()=>{returnsendEmail(input.customerId,payment);});});import{firequeue}from"./init";awaitfirequeue.invokeTask({taskId: "order-processing",collectionPath: "queue",input: {orderId: "123",customerId: "456"},});When a task is waiting for an event via step.waitForEvent(), send the event to resume execution:
awaitfirequeue.sendEvent({taskInstanceId: "abc123",collectionPath: "queue",event: "payment-confirmed",});// Cancel an entire taskawaitfirequeue.cancelTask({taskInstanceId: "abc123",collectionPath: "queue",});// Cancel specific stepsawaitfirequeue.cancelSteps({taskInstanceId: "abc123",collectionPath: "queue",stepIds: ["step-1","step-2"],});Use invalidateTask to reschedule failed steps:
awaitfirequeue.invalidateTask({taskInstanceId: "abc123",collectionPath: "queue",stepIds: ["failed-step"],events: ["event-to-recreate"],// optional});Creates a Firequeue instance.
| Option | Type | Description |
|---|---|---|
firestore | FirebaseFirestore.Firestore | Firestore instance from firebase-admin |
serializer | Serializer | Optional. Custom serializer for step results. Default handles undefined, null, NaN |
logLevel | LogSeverity | Optional. Log level (debug, info, warn, error) |
Creates a Firestore-triggered Cloud Function.
Options:
| Option | Type | Description |
|---|---|---|
collectionPath | string | Firestore collection path for task documents |
executionMode | "serializable" | "speculative" | Optional. Default: "serializable" |
timeoutSeconds | number | Optional. Function timeout |
concurrency | number | Optional. Max concurrent instances |
secrets | string[] | Optional. Secret names to expose |
Run function parameters:
| Parameter | Type | Description |
|---|---|---|
step | StepFactory | Step execution utilities |
input | T | null | Input data passed to invokeTask |
taskInstanceId | string | Unique ID for this task instance |
event | FirestoreEvent | The Firestore trigger event |
Executes a durable step.
| Parameter | Type | Description |
|---|---|---|
stepId | string | Unique step identifier within the task |
fn | () => Promise<T> | Async function to execute. Return value is persisted |
Returns the step result (from execution or cache).
Pauses task execution until an event is received.
| Option | Type | Description |
|---|---|---|
event | string | Event name to wait for |
timeout | TimeString | Optional. Timeout (e.g., "5m", "1h") |
Starts a new task execution.
| Option | Type | Description |
|---|---|---|
taskId | string | Task identifier (must match createTask) |
collectionPath | string | Collection path (must match createTask) |
input | T | Optional. Input data for the task |
Sends an event to a waiting task.
| Option | Type | Description |
|---|---|---|
taskInstanceId | string | Task instance ID |
collectionPath | string | Collection path |
event | string | Event name |
Cancels a task.
| Option | Type | Description |
|---|---|---|
taskInstanceId | string | Task instance ID |
collectionPath | string | Collection path |
Cancels specific steps and sets task status to cancelled.
| Option | Type | Description |
|---|---|---|
taskInstanceId | string | Task instance ID |
collectionPath | string | Collection path |
stepIds | string[] | Step IDs to cancel |
Reschedules steps for retry and sets task status to scheduled.
| Option | Type | Description |
|---|---|---|
taskInstanceId | string | Task instance ID |
collectionPath | string | Collection path |
stepIds | string[] | Step IDs to reschedule |
events | string[] | Optional. Events to recreate |
In speculative mode, multiple steps run in a single function invocation. If the workflow takes longer than timeoutSeconds, the function will timeout and the workflow will not automatically resume. The task will be left in running status.
Use speculative execution only for short workflows where you want the step structure (idempotency, result caching) but don't need the reliability of per-step re-invocation.
// Good: small workflow, steps are fastfirequeue.createTask("send-notification",{collectionPath: "queue",executionMode: "speculative",},async({ step })=>{constuser=awaitstep.run("get-user",()=>getUser());awaitstep.run("send-email",()=>sendEmail(user));});// Bad: long workflow with slow stepsfirequeue.createTask("process-video",{collectionPath: "queue",executionMode: "speculative",// Don't do this},async({ step })=>{awaitstep.run("download",()=>downloadVideo());// 30sawaitstep.run("transcode",()=>transcodeVideo());// 60sawaitstep.run("upload",()=>uploadVideo());// 30s});Each step.run() call must have a unique stepId. If you use the same ID twice, the second call will return the cached result from the first execution.
// Wrong: both steps have the same IDawaitstep.run("process",()=>processA());awaitstep.run("process",()=>processB());// Returns result of processA()// Correctawaitstep.run("process-a",()=>processA());awaitstep.run("process-b",()=>processB());Step return values are stored in Firestore as JSON. Functions, symbols, circular references, and other non-serializable values will cause errors or be lost.
// Wrong: returning a functionawaitstep.run("bad",()=>{return{callback: ()=>{}};// Will fail or be lost});// Correct: return plain dataawaitstep.run("good",()=>{return{id: "123",name: "test"};});If you need to serialize types that JSON doesn't support (e.g., Date, BigInt, custom classes), provide a custom serializer:
importsuperjsonfrom"superjson";constfirequeue=createFirequeue({firestore: admin.firestore(),serializer: {stringify: (data)=>superjson.stringify(data),parse: (str)=>superjson.parse(str),},});If you have conditional logic, make sure step IDs are unique across all branches:
// Wrong: step ID collision between branchesif(condition){awaitstep.run("send",()=>sendEmail());}else{awaitstep.run("send",()=>sendSms());// ID collision!}// Correctif(condition){awaitstep.run("send-email",()=>sendEmail());}else{awaitstep.run("send-sms",()=>sendSms());}If you send an event before the task reaches step.waitForEvent(), the event will be consumed immediately. If the task hasn't started yet or is still on earlier steps, make sure your event sender waits for the appropriate state.
If a step is in running status for longer than timeoutSeconds + 10s, it's marked as error (zombie detection). This handles cases where a function crashed mid-step without updating the status.
{collectionPath}/
{taskInstanceId}/ # Task document
steps/
{stepId}/ # Step documents
events/
{eventId}/ # Event documents