Use scoped Effect producer runtime

ts/packages/base/src/__tests__/producer.test.ts

@@ -0,0 +1,86 @@
+import { describe, expect, it } from "vitest";
+import {
+  makeProducer,
+  type BackendConsumer,
+  type BackendProducer,
+  type CreateConsumerOptions,
+  type CreateProducerOptions,
+  type PubSubBackend,
+} from "../index.js";
+
+class ProducerBackend implements PubSubBackend {
+  readonly sent: Array<{ readonly message: unknown; readonly properties?: Record<string, string> }> = [];
+  readonly producerTopics: Array<string> = [];
+  closeCount = 0;
+  flushCount = 0;
+  failFlush = false;
+
+  async createProducer<T>(options: CreateProducerOptions<T>): Promise<BackendProducer<T>> {
+    this.producerTopics.push(options.topic);
+
+    return {
+      send: async (message, properties) => {
+        this.sent.push(properties === undefined ? { message } : { message, properties });
+      },
+      flush: async () => {
+        this.flushCount += 1;
+        if (this.failFlush) {
+          return Promise.reject("flush failed");
+        }
+      },
+      close: async () => {
+        this.closeCount += 1;
+      },
+    };
+  }
+
+  createConsumer<T>(_options: CreateConsumerOptions<T>): Promise<BackendConsumer<T>> {
+    return Promise.reject("consumer not supported");
+  }
+
+  async close(): Promise<void> {}
+}
+
+describe("Producer", () => {
+  it("routes the compatibility facade through the scoped Effect producer", async () => {
+    const backend = new ProducerBackend();
+    const producer = makeProducer<string>(backend, "tg.test.producer");
+
+    await producer.start();
+    await producer.send("message-1", "hello");
+    await producer.stop();
+
+    expect(backend.producerTopics).toEqual(["tg.test.producer"]);
+    expect(backend.sent).toEqual([
+      { message: "hello", properties: { id: "message-1" } },
+    ]);
+    expect(backend.flushCount).toBe(1);
+    expect(backend.closeCount).toBe(1);
+    await expect(producer.stop()).resolves.toBeUndefined();
+
+    const error = await producer.send("message-2", "late").catch((caught: unknown) => caught);
+    expect(error).toMatchObject({
+      _tag: "MessagingLifecycleError",
+      operation: "send",
+      resource: "tg.test.producer",
+    });
+  });
+
+  it("closes the scoped producer when flush fails during stop", async () => {
+    const backend = new ProducerBackend();
+    const producer = makeProducer<string>(backend, "tg.test.producer");
+
+    await producer.start();
+    backend.failFlush = true;
+
+    const error = await producer.stop().catch((caught: unknown) => caught);
+
+    expect(error).toMatchObject({
+      _tag: "MessagingDeliveryError",
+      operation: "flush",
+      topic: "tg.test.producer",
+    });
+    expect(backend.flushCount).toBe(1);
+    expect(backend.closeCount).toBe(1);
+  });
+});

ts/packages/base/src/messaging/producer.ts

@@ -6,8 +6,9 @@
 
 import type { PubSubBackend } from "../backend/types.js";
 import type { ProducerMetrics } from "../metrics/prometheus.js";
-import { Effect } from "effect";
-import { makeEffectProducerHandle, type EffectProducer } from "./runtime.js";
+import { Effect, Exit, Scope } from "effect";
+import { PubSub } from "../backend/pubsub.js";
+import { makeEffectProducerFromPubSub, type EffectProducer } from "./runtime.js";
 import { messagingLifecycleError } from "../errors.js";
 
 export interface Producer<T> {
@@ -16,46 +17,61 @@ export interface Producer<T> {
   readonly stop: () => Promise<void>;
 }
 
+interface ProducerRuntime<T> {
+  readonly scope: Scope.Closeable;
+  readonly producer: EffectProducer<T>;
+}
+
 export function makeProducer<T>(
   pubsub: PubSubBackend,
   topic: string,
   metrics?: ProducerMetrics,
 ): Producer<T> {
-  let effectProducer: EffectProducer<T> | null = null;
+  let runtime: ProducerRuntime<T> | null = null;
 
   return {
     start: () =>
-      Effect.runPromise(
-        Effect.gen(function* () {
-          const backend = yield* Effect.tryPromise({
-            try: () => pubsub.createProducer<T>({ topic }),
-            catch: (error) => messagingLifecycleError(topic, "create-producer", error),
-          });
-          effectProducer = makeEffectProducerHandle(backend, {
-            topic,
-            ...(metrics === undefined ? {} : { metrics }),
-          });
-        }),
-      ),
-    send: (id, message) =>
-      effectProducer === null
+      runtime !== null
+        ? Promise.resolve()
+        : Effect.runPromise(
+            Effect.gen(function* () {
+              const scope = yield* Scope.make();
+              const startProducer = Effect.gen(function* () {
+                const producer = yield* makeEffectProducerFromPubSub<T>(
+                  PubSub.fromBackend(pubsub),
+                  {
+                    topic,
+                    ...(metrics === undefined ? {} : { metrics }),
+                  },
+                ).pipe(
+                  Scope.provide(scope),
+                  Effect.mapError((error) => messagingLifecycleError(topic, "create-producer", error)),
+                );
+
+                runtime = { scope, producer };
+              });
+
+              yield* startProducer.pipe(
+                Effect.onError((cause) => Scope.close(scope, Exit.failCause(cause))),
+              );
+            }),
+          ),
+    send: (id, message) => {
+      const current = runtime;
+      return current === null
         ? Effect.runPromise(Effect.fail(messagingLifecycleError(topic, "send", "Producer not started")))
-        : Effect.runPromise(effectProducer.send(id, message)),
-    stop: () =>
-      Effect.runPromise(
-        Effect.gen(function* () {
-          if (effectProducer !== null) {
-            const producer = effectProducer;
-            yield* producer.flush.pipe(
-              Effect.flatMap(() => producer.close),
-              Effect.ensuring(
-                Effect.sync(() => {
-                  effectProducer = null;
-                }),
-              ),
-            );
-          }
-        }),
-      ),
+        : Effect.runPromise(current.producer.send(id, message));
+    },
+    stop: () => {
+      const current = runtime;
+      runtime = null;
+      return current === null
+        ? Promise.resolve()
+        : Effect.runPromise(
+            current.producer.flush.pipe(
+              Effect.ensuring(Scope.close(current.scope, Exit.void)),
+            ),
+          );
+    },
   };
 }