Invoke a service from a "serverless" fsm

[bokzor]

I'm using xsate on a Node.JS backend.
Here is the current flow :

• State is rehydrated (Initialized or fetched from DB)
• Event is sent to the FSM
• State is serialized to DB

Here is some pseudo code

const state  = db.fetch(flowId) ?? machine.initialState;
 // Use State.create() to restore state from a plain object
const previousState = State.create<DefaultContext, MyEvent>(stateDefinition);
// Use machine.resolveState() to resolve the state definition to a new State instance relative to the machine
const resolvedState = machine.resolveState(previousState);
const interpreter = interpret(machine).start(resolvedState);
onst newState: State<DefaultContext, MyEvent, any, Typestate<DefaultContext>> = interpreter.send(event);
db.saveState(flowId, newState);

My question is :
Is it possible to Invoke a Promise ?

I would like to keep my FSM "alive" while I have pending promises.
The goal is to modify the context based on the promise result.
Is there some hook I could use ?

Thanks for any advise.

[Andarist]

In other words - you would like to await some kind of a "settled" state when the inner invoked promise is already resolved?

[bokzor]

yes

[davidkpiano]

One way you can do this now is by adding a tag to your "pending" states:

loading: {
  tags: ['pending'],
  invoke: { ... }
},
// ...

Then you can wait for the state machine to not have that tag anymore:

// ...
if (!state.hasTag('pending')) {
  resolve(...);
}

[tamlyn]

Is this still the recommended approach in v5?

[christophe-g]

my 2 cents, I use waitFor to wait until a machine as reached a state where a user triggered event is required.

Smth along (running as a firebase cloud function):

/**
 * Processes an event for the actor, ensuring that only one process runs at a time for a given actor.
 * 
 * @template TEvent - The type of the event to be processed.
 * @template ProcessT - The type of the process data.
 * 
 * @param {TEvent} event - The event to be processed.
 * @param {ActorOptions<any>} [options={}] - Optional settings for the actor.
 * 
 * @returns {Promise<ProcessT>} - A promise that resolves to the process data.
 * 
 * @throws {Error} - Throws an error if the event does not have a type.
 * 
 * @remarks
 * This method ensures that only one process is running at a time for a given actor by using a queue.
 * It waits for the actor to reach a non-transient state before stopping the actor and persisting its snapshot.
 * 
 */
async process(event: TEvent, options: ActorOptions<any> = {}) {
	this.#processID = (await this.enqueue(event)).id
	const query = this.queueCollection.where('status', '==', 'queuing').orderBy('timestamp', 'asc').limit(1)
	let unsubscribe: () => void
	// we need a queue here 
	// to make sure that only one process is running at a time for a given actor
	const promise = new Promise<ProcessT>((resolve, reject) => {
		unsubscribe = query.onSnapshot(async (snapshot) => {
			if (snapshot.empty) {
				reject(`no more process in queue for ${this.machineId} [${this.docId}, ${this.#processID}]`)
			}
			const process = snapshot.docs[0];
			if (process.id === this.#processID) {
				resolve(process.data() as ProcessT)
			}
		}, reject)
	})
	await promise
	unsubscribe!()
	if (!this.actor) {
		debug('CREATE ACTOR', this.machineId, this.docId, options)
		await this.createActor(options)
	}
	await this.start()

	if (event) {
		if (!event.type) {
			throw new Error(`Event must have a type for ${this.machineId} [${this.docId}, ${this.#processID}]`)
		}
		await this.send(event);
	} else {
		// if no event is provided, it means this is new actor
		await this.addHistory('init', this.actor?.getSnapshot())
	}
	
	debug('WAITFOR', this.machineId, this.docId) 
	// we need to know if a state has its own events to stop waiting.
	// internal xstate events start with 'xstate.', we disregard them
	await waitFor(this.actor!, (snap) => {
		const stopWait = () => {
				const allOwnEvents =  [
					...new Set([...snap._nodes.flatMap((sn: StateNode) => sn.ownEvents)])
				].filter((e) => !e.startsWith('xstate.'));
			debug('OWN EVENTS', this.docId,  allOwnEvents, snap.value)
			return allOwnEvents.length !== 0
		}
		return snap.status === 'done'
			|| snap.status === 'error'
			|| stopWait()
	})
	debug('WAITFOR DONE', this.machineId, this.docId) 
	
	// we need to get persisted snapshot before stopping the actor 
	// see https://github.com/statelyai/xstate/issues/4583
	const snapshot = await this.persistSnapshot()
	await this.stop()
	return snapshot

}