refactor(standalone): dispatch internal work in-process instead of over HTTP

ai-context/core-features-reference.md

@@ -76,7 +76,7 @@ This document provides the correct import paths and type definitions for commonl
 
 - **Import:** `import { TaskService } from "@webiny/background-tasks/api/domain/TaskService.js"`
 - **Interface Type:** See `packages/background-tasks/src/api/domain/TaskService.ts`
-- **Usage:** Low-level transport abstraction (`send` + `fetch`). Implemented by `StepFunctionService` (AWS) and `WorkerService` (server). Registered by `BackgroundTasksAwsFeature` or `BackgroundTasksStandaloneFeature`. Namespace types: `TaskService.Interface`, `.SendTaskParams`, `.Task`.
+- **Usage:** Low-level transport abstraction (`send` + `fetch`). Implemented by `StepFunctionService` (AWS) and `InProcessTaskService` (standalone, which hands the task to a root `TaskLoop` that dispatches one `BackgroundTaskEvent` per iteration). Registered by `BackgroundTasksAwsFeature` or `BackgroundTasksStandaloneFeature`. Namespace types: `TaskService.Interface`, `.SendTaskParams`, `.Task`.
 
 ### Timer
 

packages/api-core/__tests__/tenancy/tenantContext.test.ts

@@ -81,3 +81,47 @@ describe("TenantContext.withTenant", () => {
         expect(tenantContext.getTenant().id).toBe("root");
     });
 });
+
+/*
+ * Work outside an HTTP request (a background task, a scheduled action, a scheduled EventBridge
+ * event) starts with no tenant. Restoring that afterwards used to throw, after the callback had
+ * already done its work.
+ */
+describe("TenantContext with no tenant set", () => {
+    let tenantContext: TenantContext.Interface;
+
+    beforeEach(() => {
+        const container = new Container();
+        TenancyFeature.register(container, new MockTenancyStorageOperations());
+        tenantContext = container.resolve(TenantContext);
+    });
+
+    it("withTenant runs the callback and leaves no tenant set", async () => {
+        const tenant = createTenant("sub", "root");
+        const result = await tenantContext.withTenant(tenant, async () => {
+            return tenantContext.getTenant().id;
+        });
+
+        expect(result).toBe("sub");
+        expect(tenantContext.getTenant()).toBeNull();
+    });
+
+    it("withRootTenant runs the callback and leaves no tenant set", async () => {
+        const result = await tenantContext.withRootTenant(async () => {
+            return tenantContext.getTenant().id;
+        });
+
+        expect(result).toBe("root");
+        expect(tenantContext.getTenant()).toBeNull();
+    });
+
+    it("withEachTenant runs the callback per tenant and leaves no tenant set", async () => {
+        const tenants = [createTenant("a", "root"), createTenant("b", "root")];
+        const result = await tenantContext.withEachTenant(tenants, async tenant => {
+            return tenant.id;
+        });
+
+        expect(result).toEqual(["a", "b"]);
+        expect(tenantContext.getTenant()).toBeNull();
+    });
+});

packages/api-core/src/features/tenancy/TenantContext/TenantContext.ts

@@ -20,8 +20,17 @@ class TenantContextImpl implements Abstraction.Interface {
         this.currentTenant = tenant;
     }
 
+    /*
+     * Puts back the tenant that was current before a `with*` call. That can be none at all: work
+     * outside an HTTP request (a background task, a scheduled action) starts without a tenant.
+     * Restoring is not switching, so it skips the checks `setTenant` makes.
+     */
+    private restoreTenant(tenant: Tenant | null): void {
+        this.currentTenant = tenant;
+    }
+
     async withRootTenant<T>(cb: () => T): Promise<T> {
-        const initialTenant = this.getTenant();
+        const initialTenant = this.currentTenant;
         const rootTenant = await this.getRootTenant.execute();
         if (!rootTenant.isOk()) {
             return rootTenant.error as T;
@@ -34,22 +43,22 @@ class TenantContextImpl implements Abstraction.Interface {
             return await cb();
         } finally {
             // Make sure that, whatever happens in the callback, the tenant is set back to the initial one.
-            this.setTenant(initialTenant);
+            this.restoreTenant(initialTenant);
         }
     }
 
     async withEachTenant<TReturn>(
         tenants: Tenant[],
         cb: (tenant: Tenant) => Promise<TReturn>
     ): Promise<TReturn[]> {
-        const initialTenant = this.getTenant();
+        const initialTenant = this.currentTenant;
         const results = [];
         for (const tenant of tenants) {
             this.setTenant(tenant);
             try {
                 results.push(await cb(tenant));
             } finally {
-                this.setTenant(initialTenant);
+                this.restoreTenant(initialTenant);
             }
         }
         return results;
@@ -59,13 +68,13 @@ class TenantContextImpl implements Abstraction.Interface {
         tenant: Tenant,
         cb: (tenant: Tenant) => Promise<TReturn>
     ): Promise<TReturn> {
-        const initialTenant = this.getTenant();
+        const initialTenant = this.currentTenant;
         this.setTenant(tenant);
         try {
             return await cb(tenant);
         } finally {
             // Make sure that, whatever happens in the callback, the tenant is set back to the initial one.
-            this.setTenant(initialTenant);
+            this.restoreTenant(initialTenant);
         }
     }
 }

packages/api-event-handler-aws/__tests__/registerInboundEventTypes.test.ts

@@ -19,7 +19,7 @@
 import { describe, expect, it } from "vitest";
 import { Container } from "@webiny/di";
 import { EventType } from "@webiny/event-handler-core";
-import { SCHEDULED_ACTION_EVENT_IDENTIFIER } from "@webiny/event-handler-aws/eventTypes/ScheduledActionEventType.js";
+import { SCHEDULED_ACTION_EVENT_IDENTIFIER } from "@webiny/event-handler-core";
 import { registerInboundEventTypes } from "~/composition/registerInboundEventTypes.js";
 
 /** One representative payload per inbound transport, shaped as the real invoker sends it. */

packages/api-event-handler-aws/src/composition/registerInboundEventTypes.ts

@@ -1,10 +1,8 @@
 import type { Container } from "@webiny/di";
-import {
-    BackgroundTaskEventType,
-    EventBridgeEventType,
-    ScheduledActionEventType,
-    WebSocketEventType
-} from "@webiny/event-handler-aws";
+import { BackgroundTaskEventType } from "@webiny/event-handler-core";
+import { ScheduledActionEventType } from "@webiny/event-handler-core";
+import { EventBridgeEventType } from "@webiny/event-handler-aws";
+import { WebSocketEventType } from "@webiny/event-handler-aws";
 import { BackgroundTasksAwsFeature } from "@webiny/background-tasks-aws";
 import { BulkActionsEventBridgeLambdaHandlerFeature } from "@webiny/api-headless-cms-bulk-actions-aws";
 import { ScheduledActionLambdaHandler } from "@webiny/api-scheduler";

packages/api-event-handler-standalone/package.json

@@ -30,8 +30,7 @@
     "@webiny/event-handler-core": "0.0.0",
     "@webiny/event-handler-standalone": "0.0.0",
     "@webiny/feature": "0.0.0",
-    "@webiny/handler": "0.0.0",
-    "@webiny/stdlib": "^0.0.17"
+    "@webiny/handler": "0.0.0"
   },
   "peerDependencies": {
     "@webiny/di": "^1.1.0"

packages/api-event-handler-standalone/src/bulkActions/bulkActionsServer.ts

@@ -1,35 +1,30 @@
 import type { Container } from "@webiny/di";
-import { BulkActionsInternalToken } from "@webiny/api-headless-cms-bulk-actions-standalone";
+import { EventDispatcher } from "@webiny/event-handler-core";
+import { EMPTY_TRASH_BINS_EVENT_IDENTIFIER } from "@webiny/api-headless-cms-bulk-actions-standalone";
+import type { IEmptyTrashBinsEvent } from "@webiny/api-headless-cms-bulk-actions-standalone";
 
-const BULK_ACTIONS_HEADER = "x-webiny-bulk-actions-token";
-const serverBase = () => `http://localhost:${process.env.PORT || "3002"}`;
-
-// Default: trigger every 6 hours (in milliseconds).
+// Every 6 hours.
 const EMPTY_TRASH_INTERVAL_MS = 6 * 60 * 60 * 1000;
 
+// Shortly after boot, so the first run doesn't compete with startup.
+const FIRST_RUN_DELAY_MS = 5000;
+
+/**
+ * Empties the trash bins on a timer by dispatching an `EmptyTrashBinsEvent`. The standalone
+ * counterpart of the EventBridge rule that does it on AWS.
+ */
 export function startBulkActionsServer(rootContainer: Container): void {
-    const token = rootContainer.resolve(BulkActionsInternalToken).value;
+    const dispatcher = rootContainer.resolve(EventDispatcher);
+    const event: IEmptyTrashBinsEvent = { [EMPTY_TRASH_BINS_EVENT_IDENTIFIER]: true };
 
     const trigger = async () => {
         try {
-            const res = await fetch(`${serverBase()}/empty-trash-bins`, {
-                method: "POST",
-                headers: { "content-type": "application/json", [BULK_ACTIONS_HEADER]: token },
-                body: JSON.stringify({})
-            });
-            if (!res.ok) {
-                const body = await res.text().catch(() => "");
-                console.error(
-                    `[bulk-actions] empty-trash-bins returned HTTP ${res.status}: ${body}`
-                );
-            }
+            await dispatcher.dispatch(event);
         } catch (err) {
-            const message = err instanceof Error ? err.message : String(err);
-            console.error(`[bulk-actions] failed to reach empty-trash-bins route: ${message}`);
+            console.error("[bulk-actions] failed to start emptying the trash bins:", err);
         }
     };
 
-    // Deferred initial trigger + periodic interval.
-    setTimeout(trigger, 5000);
+    setTimeout(trigger, FIRST_RUN_DELAY_MS);
     setInterval(trigger, EMPTY_TRASH_INTERVAL_MS);
 }

packages/api-event-handler-standalone/src/createWebinyApiHandler.ts

@@ -37,7 +37,7 @@ import { NodeHttpIdentityLoaderDecorator } from "~/handlers/NodeHttpIdentityLoad
 import { NodeHttpTenantLoaderDecorator } from "~/handlers/NodeHttpTenantLoaderDecorator.js";
 import { NodeHttpAssumePermissionsDecorator } from "~/handlers/NodeHttpAssumePermissionsDecorator.js";
 import { createWebsocketsAuthenticator } from "~/websockets/createWebsocketsAuthenticator.js";
-import { EmptyTrashBinRouteFeature } from "@webiny/api-headless-cms-bulk-actions-standalone";
+import { EmptyTrashBinsFeature } from "@webiny/api-headless-cms-bulk-actions-standalone";
 
 export interface CreateWebinyApiHandlerConfig {
     /**
@@ -99,30 +99,25 @@ export function createWebinyApiHandler(config: CreateWebinyApiHandlerConfig) {
             JwksStoreFeature.register(rootContainer);
 
             // ── Background tasks (root) ────────────────────────────────
-            // Mirrors the AWS handler registering its background-task transport at root. There is no
-            // Step Functions / Lambda re-invocation in a single process, so the transport is in-process:
-            // WorkerService (the BackgroundTasks/TaskService dispatch abstraction) runs each triggered
-            // task in a Node worker_thread that POSTs back to this server's `/background-task` HTTP route
-            // (BackgroundTaskRoute), which runs the task loop (continue/timeout/abort) in-process. Root,
-            // not per-request, is load-bearing: the shared InternalToken (registered here as a singleton)
-            // gates the route against the worker's callback, so dispatcher and route MUST see the SAME
-            // token — a per-request registration would mint a fresh token per request and always 403.
-            // The route is an HttpRoute; the per-request HttpRouter collects it via the parent chain.
+            // Mirrors the AWS handler registering its background-task transport at root. Where AWS
+            // has Step Functions invoke the Lambda once per iteration, a root task loop dispatches a
+            // BackgroundTaskEvent per iteration through the EventDispatcher (registered by
+            // createServerHandler), and each one runs in a fresh request container.
             BackgroundTasksStandaloneFeature.register(rootContainer);
 
             // ── Scheduler (root) ───────────────────────────────────────
             // The Bree scheduler is a single long-lived instance for ALL tenants, started once at boot
             // (onServer, below) — the counterpart of the WebSockets connection manager, NOT a per-request
             // transport. Registered here as SchedulerService (per-request create/update/delete during
-            // mutations manipulate this one live timer set) plus the run/recover HTTP routes + internal
-            // token. When a timer fires (outside any request) it POSTs `/scheduled-action-run`, which
-            // rebuilds the tenant's request context and executes the action.
+            // mutations manipulate this one live timer set) plus the run/recover event handlers. When a
+            // timer fires (outside any request) it dispatches a ScheduledActionEvent, handled in a fresh
+            // request container by the same handler AWS uses.
             registerSchedulerServer(rootContainer);
 
             // ── Bulk actions (root) ───────────────────────────────────
-            // Registers the `/empty-trash-bins` HTTP route + internal token so the periodic trigger
-            // (startBulkActionsServer, onServer below) can POST to it after the server is listening.
-            EmptyTrashBinRouteFeature.register(rootContainer);
+            // Registers the EmptyTrashBinsEvent handler that the periodic trigger
+            // (startBulkActionsServer, onServer below) dispatches.
+            EmptyTrashBinsFeature.register(rootContainer);
         },
 
         child: async container => {
@@ -175,11 +170,11 @@ export function createWebinyApiHandler(config: CreateWebinyApiHandlerConfig) {
             });
             await websockets.start();
 
-            // Start the in-process scheduler timers, then re-arm persisted schedules (deferred until
-            // the server is listening — see startSchedulerServer).
+            // Start the in-process scheduler timers, then re-arm persisted schedules in the
+            // background (see startSchedulerServer).
             await startSchedulerServer(rootContainer);
 
-            // Start the periodic empty-trash-bin trigger (deferred until the server is listening).
+            // Start the periodic empty-trash-bin trigger.
             startBulkActionsServer(rootContainer);
         }
     });

packages/api-event-handler-standalone/src/scheduler/RecoverScheduledActionsHandler.ts

@@ -0,0 +1,96 @@
+import { RawTenantId } from "@webiny/api-core/features/requestContext/index.js";
+import { RequestTenantLoader } from "@webiny/api-core/features/requestContext/index.js";
+import { GetRootTenantUseCase } from "@webiny/api-core/features/tenancy/GetRootTenant/index.js";
+import { IdentityContext } from "@webiny/api-core/exports/api/security.js";
+import { ListScheduledActionsUseCase } from "@webiny/api-scheduler/features/ListScheduledActions/index.js";
+import { ScheduledActionRecoverEventHandler } from "./ScheduledActionRecoverEventHandler.js";
+import type { IPendingAction } from "@webiny/api-scheduler-standalone";
+import { SchedulerSingleton } from "./abstractions/SchedulerSingleton.js";
+
+// Pending actions read per page.
+const PAGE_SIZE = 1000;
+
+/**
+ * Re-arms the root tenant's pending scheduled actions in the Bree singleton. Overdue ones run at
+ * once (see `BreeSchedulerService.recover`).
+ *
+ * Root tenant only. Recovering other tenants means enumerating them and recovering each; until
+ * then a restart drops their pending schedules.
+ */
+class RecoverScheduledActionsHandlerImpl implements ScheduledActionRecoverEventHandler.Interface {
+    public constructor(
+        private readonly getRootTenant: GetRootTenantUseCase.Interface,
+        private readonly rawTenantId: RawTenantId.Interface,
+        private readonly tenantLoader: RequestTenantLoader.Interface,
+        private readonly identityContext: IdentityContext.Interface,
+        private readonly listScheduledActions: ListScheduledActionsUseCase.Interface,
+        private readonly scheduler: SchedulerSingleton.Interface
+    ) {}
+
+    public async execute(): Promise<ScheduledActionRecoverEventHandler.Result> {
+        const rootResult = await this.getRootTenant.execute();
+        if (rootResult.isFail()) {
+            // No root tenant yet: a fresh install has nothing to recover.
+            return { recovered: 0 };
+        }
+        const tenant = rootResult.value.id;
+
+        this.rawTenantId.set(tenant);
+        await this.tenantLoader.establish();
+
+        /*
+         * Nobody is signed in at boot. Listing without authorization is what the run side does too
+         * (ExecuteScheduledActionUseCase); otherwise the anonymous identity fails the permission check.
+         */
+        const pending = await this.identityContext.withoutAuthorization(() => {
+            return this.listAllPending(tenant);
+        });
+
+        await this.scheduler.recover(pending);
+
+        return { recovered: pending.length };
+    }
+
+    private async listAllPending(tenant: string): Promise<IPendingAction[]> {
+        const pending: IPendingAction[] = [];
+        let after: string | undefined;
+
+        do {
+            const listResult = await this.listScheduledActions.execute({
+                where: {},
+                limit: PAGE_SIZE,
+                after
+            });
+            if (listResult.isFail()) {
+                throw listResult.error;
+            }
+
+            for (const action of listResult.value.items) {
+                pending.push({
+                    id: action.id,
+                    namespace: action.namespace,
+                    tenant,
+                    scheduledFor: action.scheduledFor
+                });
+            }
+
+            const { meta } = listResult.value;
+            after = meta.hasMoreItems && meta.cursor ? meta.cursor : undefined;
+        } while (after);
+
+        return pending;
+    }
+}
+
+export const RecoverScheduledActionsHandler =
+    ScheduledActionRecoverEventHandler.createImplementation({
+        implementation: RecoverScheduledActionsHandlerImpl,
+        dependencies: [
+            GetRootTenantUseCase,
+            RawTenantId,
+            RequestTenantLoader,
+            IdentityContext,
+            ListScheduledActionsUseCase,
+            SchedulerSingleton
+        ]
+    });

packages/api-event-handler-standalone/src/scheduler/ScheduledActionRecoverEventHandler.ts

@@ -0,0 +1,20 @@
+import { Abstraction } from "@webiny/di";
+import type { IEventHandler } from "@webiny/event-handler-core";
+import type { IScheduledActionRecoverEvent } from "./ScheduledActionRecoverEventType.js";
+
+export interface IScheduledActionRecoverResult {
+    recovered: number;
+}
+
+export interface IScheduledActionRecoverEventHandler extends IEventHandler<
+    IScheduledActionRecoverEvent,
+    IScheduledActionRecoverResult
+> {}
+
+export const ScheduledActionRecoverEventHandler =
+    new Abstraction<IScheduledActionRecoverEventHandler>("ScheduledActionRecoverEventHandler");
+
+export namespace ScheduledActionRecoverEventHandler {
+    export type Interface = IScheduledActionRecoverEventHandler;
+    export type Result = IScheduledActionRecoverResult;
+}