feat: Global events

[ramedina86]

This PR is preparation for async jobs. This feature will allow an API endpoint such as POST /api/job/my_workflow and later get the result via GET /api/job/3.

To facilitate triggering events at this level, this PR introduces "global events" which are WriterEvent that don't target a specific component. Therefore, turning instancePath into an optional field. Instead, a new handler field can be specified to run arbitrary event handlers.

There's of course potential for abuse, so that's why only global events marked as "safe" can be executed. Safe events are produced in "edit" mode. In an upcoming PR, the creation of async jobs will also trigger safe events to run workflows.

[ramedina86]

For the moment, type in WriterEvent isn't being used when specifying a global event (event with no instancePath).

[ramedina86]

Job Vault will be used to store results of async jobs.

A basic dict-based structure for single-node deployments. There'll be an option to use Redis for bigger deployments (we can then use this same Redis to do other interesting stuff).

[FabienArcellier]

I have one concern. I don't know how to save the result of the job in the job vault.

If I understand correctly the goal is to create an API endpoint for triggering a job decoupled from a user session. WF retrieves it and executes it on an Event handler of the AppProcess

In this case, the job vault will serve as memory. When the job is finished, it will allow the user to retrieve the result. It is only used for "synchronous" jobs for which we are interested in the result.

My concern is the job vault is in the primary process, the thread pool that run a job in async in WF is in the secondary process. You can write something into the job vault from the secondary process. I have tried to explain what I have understand in the schema below.

[ramedina86]

The idea is that a job is always a single event. We dispatch to app_runner, then we get the result and, instead of sending via WebSockets, we save it in the JobVault. The vault stays completely independent from the internals (AppProcess, AppRunner, etc) - it all happens at serve level.

I wrote some working code that I didn't include as part of the PR so that this PR can be focused on foundations.

    @app.post("/api/job/workflow/{workflow_key}")
    async def create_workflow_job(workflow_key: str, request: Request, response: Response):
        def save_job_result(task, job_id: str):
            try:
                apsr: Optional[AppProcessServerResponse] = None
                apsr = task.result()
                app.state.job_vault.set(job_id, apsr)
            except Exception as e:
                app.state.job_vault.set(job_id, e)

        app_response = await app_runner.init_session(InitSessionRequestPayload(
            cookies=dict(request.cookies),
            headers=dict(request.headers),
            proposedSessionId=None
        ))

        session_id = app_response.payload.sessionId
        is_session_ok = await app_runner.check_session(session_id)
        if not is_session_ok:
            return

        loop = asyncio.get_running_loop()
        task = loop.create_task(app_runner.handle_event(
            session_id, WriterEvent(
                type="wf-builtin-run",
                isSafe=True,
                handler=f"$runWorkflow_{workflow_key}"
            )))

        job_id = app.state.job_vault.generate_job_id()
        task.add_done_callback(lambda t: save_job_result(t, job_id))

    @app.get("/api/job/{job_id}")
    async def get_workflow_job(job_id: str, request: Request, response: Response):
        return app.state.job_vault.get(job_id)
        ```

[FabienArcellier]

Ok, you rely on loop.create_task. That was the part I wasn't sure. Looks ok for me then.

loop = asyncio.get_running_loop()
task = loop.create_task(app_runner.handle_event(
    session_id, WriterEvent(
        type="wf-builtin-run",
        isSafe=True,
        handler=f"$runWorkflow_{workflow_key}"
    )))

[ramedina86]

@FabienArcellier mostly interested in getting a review given that it's the core, around middleware. Also, please verify the idea of the JobVault.