Merge pull request #37 from backpine/dep-injection

added server provider to jobs

Author: matthew-sessions

examples/effect-worker-v2/src/jobs/basic-debounce.ts

@@ -6,6 +6,12 @@ class Random extends Context.Tag("MyRandomService")<
   { readonly next: Effect.Effect<number> }
 >() {}
 
+import { Layer } from "effect";
+
+const RandomLive = Layer.succeed(Random, {
+  next: Effect.sync(() => Math.random()),
+});
+
 // =============================================================================
 // Debounce Job - Batches events and flushes after delay
 // =============================================================================
@@ -69,5 +75,5 @@ export const debounceExample = Debounce.make({
       yield* Effect.log(
         `Debounce flushed! Events: ${eventCount}, Last action: ${state?.actionId}, Reason: ${ctx.flushReason}`,
       );
-    }),
+    }).pipe(Effect.provide(RandomLive)),
 });

packages/jobs/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@durable-effect/jobs",
-  "version": "0.0.1-next.5",
+  "version": "0.0.1-next.6",
   "type": "module",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",

packages/jobs/src/definitions/continuous.ts

@@ -16,7 +16,7 @@ import type { JobRetryConfig } from "../retry/types";
 /**
  * Input config for creating a continuous job definition.
  */
-export interface ContinuousMakeConfig<S, E, R> {
+export interface ContinuousMakeConfig<S, E> {
   /**
    * Schema for validating and serializing state.
    * Accepts any Effect Schema (Struct, Class, etc.)
@@ -73,8 +73,20 @@ export interface ContinuousMakeConfig<S, E, R> {
 
   /**
    * The function to execute on schedule.
+   *
+   * Must return Effect<void, E, never> - all service requirements must be satisfied.
+   * If your effect requires services, provide them via .pipe(Effect.provide(layer)).
+   *
+   * @example
+   * ```ts
+   * execute: (ctx) =>
+   *   Effect.gen(function* () {
+   *     const random = yield* Random;
+   *     // ...
+   *   }).pipe(Effect.provide(RandomLive))
+   * ```
    */
-  execute(ctx: ContinuousContext<S>): Effect.Effect<void, E, R>;
+  execute(ctx: ContinuousContext<S>): Effect.Effect<void, E, never>;
 }
 
 /**
@@ -116,9 +128,9 @@ export const Continuous = {
    * @param config - Configuration for the job
    * @returns An UnregisteredContinuousDefinition that can be registered
    */
-  make: <S, E = never, R = never>(
-    config: ContinuousMakeConfig<S, E, R>
-  ): UnregisteredContinuousDefinition<S, E, R> => ({
+  make: <S, E = never>(
+    config: ContinuousMakeConfig<S, E>
+  ): UnregisteredContinuousDefinition<S, E> => ({
     _tag: "ContinuousDefinition",
     stateSchema: config.stateSchema,
     schedule: config.schedule,

packages/jobs/src/definitions/debounce.ts

@@ -15,8 +15,11 @@ import type { JobRetryConfig } from "../retry/types";
 
 /**
  * Configuration for creating a debounce job definition.
+ *
+ * Note: All handler functions must return Effect with R = never.
+ * If your effect requires services, provide them via .pipe(Effect.provide(layer)).
  */
-export interface DebounceMakeConfig<I, S, E, R> {
+export interface DebounceMakeConfig<I, S, E> {
   /**
    * Schema for validating incoming events.
    */
@@ -76,14 +79,24 @@ export interface DebounceMakeConfig<I, S, E, R> {
 
   /**
    * Reducer for each incoming event. Defaults to returning the latest event.
+   * Must return Effect<S, never, never> - all service requirements must be satisfied.
    */
-  onEvent?(ctx: DebounceEventContext<I, S>): Effect.Effect<S, never, R>;
+  onEvent?(ctx: DebounceEventContext<I, S>): Effect.Effect<S, never, never>;
 
   /**
    * Effect executed when the debounce flushes.
+   * Must return Effect<void, E, never> - all service requirements must be satisfied.
+   *
+   * @example
+   * ```ts
+   * execute: (ctx) =>
+   *   Effect.gen(function* () {
+   *     const random = yield* Random;
+   *     // ...
+   *   }).pipe(Effect.provide(RandomLive))
+   * ```
    */
-  execute(ctx: DebounceExecuteContext<S>): Effect.Effect<void, E, R>;
-
+  execute(ctx: DebounceExecuteContext<S>): Effect.Effect<void, E, never>;
 }
 
 /**
@@ -99,14 +112,14 @@ export interface DebounceMakeConfig<I, S, E, R> {
  * ```
  */
 export const Debounce = {
-  make: <I, S = I, E = never, R = never>(
-    config: DebounceMakeConfig<I, S, E, R>
-  ): UnregisteredDebounceDefinition<I, S, E, R> => ({
+  make: <I, S = I, E = never>(
+    config: DebounceMakeConfig<I, S, E>
+  ): UnregisteredDebounceDefinition<I, S, E> => ({
     _tag: "DebounceDefinition",
     eventSchema: config.eventSchema,
     stateSchema: (config.stateSchema ?? config.eventSchema) as Schema.Schema<
       S,
-      any,
+      unknown,
       never
     >,
     flushAfter: config.flushAfter,

packages/jobs/src/definitions/task.ts

@@ -22,12 +22,14 @@ import type {
  * - Execute runs when alarm fires
  * - User controls the full lifecycle via schedule/clear
  *
+ * Note: All handler functions must return Effect with R = never.
+ * If your effect requires services, provide them via .pipe(Effect.provide(layer)).
+ *
  * @typeParam S - State type (inferred from stateSchema)
  * @typeParam E - Event type (inferred from eventSchema)
  * @typeParam Err - Error type (inferred from handlers)
- * @typeParam R - Effect requirements (inferred from handlers)
  */
-export interface TaskMakeConfig<S, E, Err, R> {
+export interface TaskMakeConfig<S, E, Err> {
   /**
    * Schema for validating and serializing state.
    * State is persisted durably and survives restarts.
@@ -42,6 +44,7 @@ export interface TaskMakeConfig<S, E, Err, R> {
 
   /**
    * Handler called for each incoming event.
+   * Must return Effect<void, Err, never> - all service requirements must be satisfied.
    *
    * The event is passed as the first parameter (not on ctx) to make it
    * clear that it's a direct value, not an Effect that needs yielding.
@@ -76,10 +79,11 @@ export interface TaskMakeConfig<S, E, Err, R> {
    * })
    * ```
    */
-  onEvent(event: E, ctx: TaskEventContext<S>): Effect.Effect<void, Err, R>;
+  onEvent(event: E, ctx: TaskEventContext<S>): Effect.Effect<void, Err, never>;
 
   /**
    * Handler called when the scheduled alarm fires.
+   * Must return Effect<void, Err, never> - all service requirements must be satisfied.
    *
    * Responsibilities:
    * - Process the current state
@@ -100,11 +104,12 @@ export interface TaskMakeConfig<S, E, Err, R> {
    * })
    * ```
    */
-  execute(ctx: TaskExecuteContext<S>): Effect.Effect<void, Err, R>;
+  execute(ctx: TaskExecuteContext<S>): Effect.Effect<void, Err, never>;
 
   /**
    * Optional handler called when either `onEvent` or `execute` completes
    * and no alarm is scheduled.
+   * Must return Effect<void, never, never> - all service requirements must be satisfied.
    *
    * Use this to:
    * - Schedule delayed cleanup
@@ -119,10 +124,11 @@ export interface TaskMakeConfig<S, E, Err, R> {
    * })
    * ```
    */
-  readonly onIdle?: (ctx: TaskIdleContext<S>) => Effect.Effect<void, never, R>;
+  readonly onIdle?: (ctx: TaskIdleContext<S>) => Effect.Effect<void, never, never>;
 
   /**
    * Optional error handler for onEvent/execute failures.
+   * Must return Effect<void, never, never> - all service requirements must be satisfied.
    *
    * Use this to:
    * - Log errors
@@ -151,7 +157,7 @@ export interface TaskMakeConfig<S, E, Err, R> {
   readonly onError?: (
     error: Err,
     ctx: TaskErrorContext<S>
-  ) => Effect.Effect<void, never, R>;
+  ) => Effect.Effect<void, never, never>;
 
   /**
    * Control logging for this job.
@@ -269,9 +275,9 @@ export const Task = {
    * @param config - Configuration for the task
    * @returns An UnregisteredTaskDefinition that can be registered
    */
-  make: <S, E, Err = never, R = never>(
-    config: TaskMakeConfig<S, E, Err, R>
-  ): UnregisteredTaskDefinition<S, E, Err, R> => ({
+  make: <S, E, Err = never>(
+    config: TaskMakeConfig<S, E, Err>
+  ): UnregisteredTaskDefinition<S, E, Err> => ({
     _tag: "TaskDefinition",
     stateSchema: config.stateSchema,
     eventSchema: config.eventSchema,

packages/jobs/src/handlers/continuous/handler.ts

@@ -50,7 +50,7 @@ export const ContinuousHandlerLayer = Layer.effect(
 
     const getDefinition = (
       name: string,
-    ): Effect.Effect<StoredContinuousDefinition<any, any>, JobNotFoundError> => {
+    ): Effect.Effect<StoredContinuousDefinition, JobNotFoundError> => {
       const def = registryService.registry.continuous[name];
       if (!def) {
         return Effect.fail(new JobNotFoundError({ type: "continuous", name }));
@@ -78,7 +78,7 @@ export const ContinuousHandlerLayer = Layer.effect(
       });
 
     const scheduleNext = (
-      def: StoredContinuousDefinition<any, any>,
+      def: StoredContinuousDefinition,
     ): Effect.Effect<void, SchedulerError | ExecutionError> => {
       const schedule = def.schedule;
       switch (schedule._tag) {
@@ -94,7 +94,7 @@ export const ContinuousHandlerLayer = Layer.effect(
     };
 
     const runExecution = (
-      def: StoredContinuousDefinition<any, any>,
+      def: StoredContinuousDefinition,
       runCount: number,
       id?: string,
     ) =>
@@ -107,7 +107,7 @@ export const ContinuousHandlerLayer = Layer.effect(
         id,
         run: (ctx: ContinuousContext<unknown>) => def.execute(ctx),
         createContext: (base) => {
-          const proxyHolder: StateHolder<any> = {
+          const proxyHolder: StateHolder<unknown> = {
             get current() {
               return base.getState();
             },
@@ -132,9 +132,9 @@ export const ContinuousHandlerLayer = Layer.effect(
       });
 
     const handleStart = (
-      def: StoredContinuousDefinition<any, any>,
+      def: StoredContinuousDefinition,
       request: ContinuousRequest,
-    ): Effect.Effect<ContinuousResponse, HandlerError, any> =>
+    ): Effect.Effect<ContinuousResponse, HandlerError> =>
       Effect.gen(function* () {
         const existing = yield* metadata.get();
         if (existing) {
@@ -251,8 +251,8 @@ export const ContinuousHandlerLayer = Layer.effect(
       });
 
     const handleTrigger = (
-      def: StoredContinuousDefinition<any, any>,
-    ): Effect.Effect<ContinuousResponse, HandlerError, any> =>
+      def: StoredContinuousDefinition,
+    ): Effect.Effect<ContinuousResponse, HandlerError> =>
       Effect.gen(function* () {
         const existing = yield* metadata.get();
         if (
@@ -321,7 +321,7 @@ export const ContinuousHandlerLayer = Layer.effect(
       });
 
     const handleGetState = (
-      def: StoredContinuousDefinition<any, any>,
+      def: StoredContinuousDefinition,
     ): Effect.Effect<ContinuousResponse, HandlerError> =>
       Effect.gen(function* () {
         const stateService = yield* createEntityStateService(

packages/jobs/src/handlers/debounce/handler.ts

@@ -59,7 +59,7 @@ export const DebounceHandlerLayer = Layer.effect(
 
     const getDefinition = (
       name: string,
-    ): Effect.Effect<StoredDebounceDefinition<any, any, any>, JobNotFoundError> => {
+    ): Effect.Effect<StoredDebounceDefinition, JobNotFoundError> => {
       const def = registryService.registry.debounce[name];
       if (!def) {
         return Effect.fail(new JobNotFoundError({ type: "debounce", name }));
@@ -91,7 +91,7 @@ export const DebounceHandlerLayer = Layer.effect(
       });
 
     const runFlush = (
-      def: StoredDebounceDefinition<any, any, any>,
+      def: StoredDebounceDefinition,
       flushReason: "maxEvents" | "flushAfter" | "manual",
       id?: string,
     ) =>
@@ -102,7 +102,7 @@ export const DebounceHandlerLayer = Layer.effect(
         retryConfig: def.retry,
         runCount: 0, // Debounce doesn't track runCount persistently in same way
         id,
-        run: (ctx: DebounceExecuteContext<any>) => def.execute(ctx),
+        run: (ctx: DebounceExecuteContext<unknown>) => def.execute(ctx),
         createContext: (base) => {
           return {
             state: Effect.succeed(base.getState()), // Snapshotted state
@@ -116,14 +116,14 @@ export const DebounceHandlerLayer = Layer.effect(
             flushReason,
             attempt: base.attempt,
             isRetry: base.isRetry,
-          } as DebounceExecuteContext<any>;
+          } as DebounceExecuteContext<unknown>;
         },
       });
 
     const handleAdd = (
-      def: StoredDebounceDefinition<any, any, any>,
+      def: StoredDebounceDefinition,
       request: DebounceRequest,
-    ): Effect.Effect<DebounceResponse, HandlerError, any> =>
+    ): Effect.Effect<DebounceResponse, HandlerError> =>
       Effect.gen(function* () {
         const meta = yield* metadata.get();
         const created = !meta;
@@ -158,13 +158,12 @@ export const DebounceHandlerLayer = Layer.effect(
         const stateForContext = currentState ?? (validatedEvent as unknown);
 
         const onEvent = def.onEvent!;
-        // Cast is still needed unless we fix Definition generic constraints
         const reducedState = yield* onEvent({
           event: validatedEvent as unknown,
           state: stateForContext,
           eventCount: nextCount,
           instanceId: runtime.instanceId,
-        } as any);
+        });
 
         yield* stateService.set(reducedState);
         yield* setEventCount(nextCount);
@@ -244,9 +243,9 @@ export const DebounceHandlerLayer = Layer.effect(
       });
 
     const handleFlush = (
-      def: StoredDebounceDefinition<any, any, any>,
+      def: StoredDebounceDefinition,
       reason: "manual" | "flushAfter" | "maxEvents",
-    ): Effect.Effect<DebounceResponse, HandlerError, any> =>
+    ): Effect.Effect<DebounceResponse, HandlerError> =>
       Effect.gen(function* () {
         const meta = yield* metadata.get();
         if (!meta) {
@@ -353,7 +352,7 @@ export const DebounceHandlerLayer = Layer.effect(
       });
 
     const handleGetState = (
-      def: StoredDebounceDefinition<any, any, any>,
+      def: StoredDebounceDefinition,
     ): Effect.Effect<DebounceResponse, HandlerError> =>
       Effect.gen(function* () {
         const stateService = yield* withStorage(