feat: Add support for cloud tasks

Author: fwal

README.md

@@ -129,7 +129,10 @@ const program = Effect.gen(function* () {
   const posts = yield* repo.query(Query.where('status', '==', 'published'));
 
   return { postId, posts };
-}).pipe(Effect.provide(PostRepository), Effect.provide(Client.layerFromApp(app)));
+}).pipe(
+  Effect.provide(PostRepository),
+  Effect.provide(Client.layerFromApp(app))
+);
 
 Effect.runPromise(program).then(console.log);
 ```
@@ -225,6 +228,8 @@ Platform-agnostic schemas for Firestore types:
 - `onDocumentUpdated` - Firestore trigger
 - `onDocumentDeleted` - Firestore trigger
 - `onDocumentWritten` - Firestore trigger
+- `onMessagePublished` - Pub/Sub trigger
+- `onTaskDispatched` - Cloud Tasks trigger
 
 ### Planned Features
 

packages/admin/README.md

@@ -12,6 +12,7 @@ Firebase Admin SDK integration for Effect Firebase. This package provides the Fi
 - 📝 **Cloud Logging** - Automatic logging integration
 - 🎯 **Type-Safe Function Handlers** - Schema validation for function inputs/outputs
 - 🔄 **Firestore Triggers** - Document lifecycle event handlers
+- 📮 **Cloud Tasks Triggers** - Typed task queue handlers with optional payload decoding
 - 🚀 **Effect Native** - Built on Effect's powerful composition and error handling
 
 ## Installation
@@ -28,7 +29,11 @@ npm install firebase-admin firebase-functions
 ```typescript
 import { Effect } from 'effect';
 import { initializeApp } from 'firebase-admin/app';
-import { Admin, FunctionsRuntime, onRequestEffect } from '@effect-firebase/admin';
+import {
+  Admin,
+  FunctionsRuntime,
+  onRequestEffect,
+} from '@effect-firebase/admin';
 
 // Create the runtime with your layers
 const runtime = FunctionsRuntime.make(Admin.layerFromApp(initializeApp()));
@@ -89,6 +94,41 @@ export const createPost = onCallEffect(
 );
 ```
 
+### Cloud Tasks Functions (onTaskDispatched)
+
+```typescript
+import { Effect, Schema } from 'effect';
+import { initializeApp } from 'firebase-admin/app';
+import {
+  Admin,
+  FunctionsRuntime,
+  onTaskDispatchedEffect,
+} from '@effect-firebase/admin';
+
+const runtime = FunctionsRuntime.make(Admin.layerFromApp(initializeApp()));
+
+const ProcessEmailTask = Schema.Struct({
+  email: Schema.String,
+  template: Schema.String,
+});
+
+export const processEmail = onTaskDispatchedEffect(
+  {
+    runtime,
+    retryConfig: { maxAttempts: 5 },
+    rateLimits: { maxConcurrentDispatches: 10 },
+    dataSchema: ProcessEmailTask,
+  },
+  (task, request) =>
+    Effect.gen(function* () {
+      yield* Effect.log('Processing task', {
+        email: task.email,
+        queue: request.queueName,
+      });
+    })
+);
+```
+
 ## Cloud Logging
 
 The Admin layer automatically provides Cloud Logging integration:
@@ -153,6 +193,8 @@ export const myFunction = onCallEffect(
 - `onDocumentUpdatedEffect` - Document updated trigger
 - `onDocumentDeletedEffect` - Document deleted trigger
 - `onDocumentWrittenEffect` - Document written (any change) trigger
+- `onMessagePublishedEffect` - Pub/Sub message published trigger
+- `onTaskDispatchedEffect` - Cloud Tasks dispatch trigger
 - `FunctionsRuntime.make(layer)` - Create an Effect runtime from a layer
 - `FunctionsRuntime.Default(app)` - Create a runtime from the provided Firebase Admin app
 

packages/admin/src/lib/functions/functions.ts

@@ -7,3 +7,4 @@ export * from './on-document-updated.js';
 export * from './on-document-deleted.js';
 export * from './on-document-written.js';
 export * from './on-message-published.js';
+export * from './on-task-dispatched.js';

packages/admin/src/lib/functions/on-task-dispatched.ts

@@ -0,0 +1,87 @@
+import { Effect, Schema } from 'effect';
+import {
+  onTaskDispatched,
+  Request,
+  TaskQueueFunction,
+  TaskQueueOptions,
+} from 'firebase-functions/v2/tasks';
+import { logger } from 'firebase-functions';
+import { run, Runtime } from './run.js';
+
+interface TaskDispatchedEffectOptions<R> extends TaskQueueOptions {
+  runtime: Runtime<R>;
+}
+
+interface TaskDispatchedEffectOptionsWithSchema<R, S extends Schema.Schema.Any>
+  extends TaskDispatchedEffectOptions<R | Schema.Schema.Context<S>> {
+  dataSchema: S;
+}
+
+/**
+ * Decode task payload JSON data using the provided schema.
+ */
+function decodeTaskData<S extends Schema.Schema.Any>(
+  schema: S,
+  request: Request<unknown>
+): Effect.Effect<Schema.Schema.Type<S>, Error, Schema.Schema.Context<S>> {
+  return Schema.decodeUnknown(schema)(request.data).pipe(
+    Effect.mapError(
+      (error) => new Error(`Failed to decode task payload: ${error.message}`)
+    )
+  ) as Effect.Effect<Schema.Schema.Type<S>, Error, Schema.Schema.Context<S>>;
+}
+
+/**
+ * Create a Firebase Functions Cloud Tasks trigger that runs an effect when a task is dispatched.
+ *
+ * @param options - The options for the Cloud Tasks trigger including optional payload schema.
+ * @param handler - The handler function that runs the effect.
+ * @returns The Firebase Functions Cloud Tasks trigger.
+ */
+// Overload: with payload schema
+export function onTaskDispatchedEffect<R, S extends Schema.Schema.Any, E>(
+  options: TaskDispatchedEffectOptionsWithSchema<R, S>,
+  handler: (
+    data: Schema.Schema.Type<S>,
+    request: Request<Schema.Schema.Encoded<S>>
+  ) => Effect.Effect<void, E, R>
+): TaskQueueFunction<Schema.Schema.Encoded<S>>;
+
+// Overload: without payload schema (full control)
+export function onTaskDispatchedEffect<R, T, E>(
+  options: TaskDispatchedEffectOptions<R>,
+  handler: (request: Request<T>) => Effect.Effect<void, E, R>
+): TaskQueueFunction<T>;
+
+// Implementation
+export function onTaskDispatchedEffect<R>(
+  options: TaskDispatchedEffectOptions<R> & {
+    dataSchema?: Schema.Schema.Any;
+  },
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  handler: (...args: any[]) => Effect.Effect<void, unknown, R>
+): TaskQueueFunction<unknown> {
+  const { dataSchema } = options;
+
+  return onTaskDispatched(options, async (request) => {
+    const effect = Effect.gen(function* () {
+      if (dataSchema) {
+        // Decode task payload and pass both parsed payload and request to handler
+        const taskData = yield* decodeTaskData(dataSchema, request);
+        return yield* handler(taskData, request);
+      } else {
+        // Pass raw request to handler
+        return yield* handler(request);
+      }
+    });
+
+    await run(options.runtime, effect as Effect.Effect<void, never, R>).catch(
+      (error) => {
+        logger.error('Defect in onTaskDispatched', {
+          inner: error,
+          stack: error instanceof Error ? error.stack : undefined,
+        });
+      }
+    );
+  });
+}