Add typed flow spec accessors

ts/packages/base/src/spec/producer-spec.ts

@@ -8,6 +8,11 @@ import { Effect } from "effect";
 import type { Spec } from "./types.js";
 import type { Flow, FlowDefinition } from "../processor/flow.js";
 import {
+  flowResourceNotFoundError,
+  type FlowResourceNotFoundError,
+} from "../errors.js";
+import {
+  type EffectProducer,
   ProducerFactory,
 } from "../messaging/runtime.js";
 
@@ -15,9 +20,41 @@ declare const ProducerSpecType: unique symbol;
 
 export interface ProducerSpec<T> extends Spec {
   readonly [ProducerSpecType]?: (_: T) => T;
+  readonly producerEffect: <Requirements = never>(
+    flow: Flow<Requirements>,
+  ) => Effect.Effect<EffectProducer<T>, FlowResourceNotFoundError>;
 }
 
 export function makeProducerSpec<T>(name: string): ProducerSpec<T> {
+  const producers = new WeakMap<object, EffectProducer<T>>();
+
+  const registerProducer = <Requirements>(
+    flow: Flow<Requirements>,
+    producer: EffectProducer<T>,
+  ) =>
+    Effect.sync(() => {
+      producers.set(flow, producer);
+    });
+
+  const unregisterProducer = <Requirements>(
+    flow: Flow<Requirements>,
+    producer: EffectProducer<T>,
+  ) =>
+    Effect.sync(() => {
+      if (producers.get(flow) === producer) {
+        producers.delete(flow);
+      }
+    });
+
+  const producerEffect = <Requirements>(
+    flow: Flow<Requirements>,
+  ): Effect.Effect<EffectProducer<T>, FlowResourceNotFoundError> => {
+    const producer = producers.get(flow);
+    return producer === undefined
+      ? Effect.fail(flowResourceNotFoundError(flow.name, "producer", name))
+      : Effect.succeed(producer);
+  };
+
   const addEffect = Effect.fn("ProducerSpec.addEffect")(function* (
     flow: Flow,
     definition: FlowDefinition,
@@ -26,10 +63,13 @@ export function makeProducerSpec<T>(name: string): ProducerSpec<T> {
       const factory = yield* ProducerFactory;
       const producer = yield* factory.make<T>({ topic });
       flow.registerProducer(name, producer);
+      yield* registerProducer(flow, producer);
+      yield* Effect.addFinalizer(() => unregisterProducer(flow, producer));
   });
 
   return {
     name,
+    producerEffect,
     addEffect,
     add: (flow, pubsub, definition, context) =>
       flow.runInCompatibilityScope(addEffect(flow, definition), pubsub, context),

ts/packages/base/src/spec/request-response-spec.ts

@@ -11,6 +11,11 @@ import { Effect } from "effect";
 import type { Spec } from "./types.js";
 import type { Flow, FlowDefinition } from "../processor/flow.js";
 import {
+  flowResourceNotFoundError,
+  type FlowResourceNotFoundError,
+} from "../errors.js";
+import {
+  type EffectRequestResponse,
   RequestResponseFactory,
 } from "../messaging/runtime.js";
 
@@ -21,6 +26,9 @@ export interface RequestResponseSpec<TReq, TRes> extends Spec {
     readonly request: TReq;
     readonly response: TRes;
   };
+  readonly requestorEffect: <Requirements = never>(
+    flow: Flow<Requirements>,
+  ) => Effect.Effect<EffectRequestResponse<TReq, TRes>, FlowResourceNotFoundError>;
 }
 
 export function makeRequestResponseSpec<TReq, TRes>(
@@ -28,6 +36,35 @@ export function makeRequestResponseSpec<TReq, TRes>(
   requestTopicName: string,
   responseTopicName: string,
 ): RequestResponseSpec<TReq, TRes> {
+  const requestors = new WeakMap<object, EffectRequestResponse<TReq, TRes>>();
+
+  const registerRequestor = <Requirements>(
+    flow: Flow<Requirements>,
+    requestor: EffectRequestResponse<TReq, TRes>,
+  ) =>
+    Effect.sync(() => {
+      requestors.set(flow, requestor);
+    });
+
+  const unregisterRequestor = <Requirements>(
+    flow: Flow<Requirements>,
+    requestor: EffectRequestResponse<TReq, TRes>,
+  ) =>
+    Effect.sync(() => {
+      if (requestors.get(flow) === requestor) {
+        requestors.delete(flow);
+      }
+    });
+
+  const requestorEffect = <Requirements>(
+    flow: Flow<Requirements>,
+  ): Effect.Effect<EffectRequestResponse<TReq, TRes>, FlowResourceNotFoundError> => {
+    const requestor = requestors.get(flow);
+    return requestor === undefined
+      ? Effect.fail(flowResourceNotFoundError(flow.name, "requestor", name))
+      : Effect.succeed(requestor);
+  };
+
   const addEffect = Effect.fn("RequestResponseSpec.addEffect")(function* (
     flow: Flow,
     definition: FlowDefinition,
@@ -41,10 +78,13 @@ export function makeRequestResponseSpec<TReq, TRes>(
         subscription: `${flow.processorId}-${flow.name}-${name}`,
       });
       flow.registerRequestor(name, requestor);
+      yield* registerRequestor(flow, requestor);
+      yield* Effect.addFinalizer(() => unregisterRequestor(flow, requestor));
   });
 
   return {
     name,
+    requestorEffect,
     addEffect,
     add: (flow, pubsub, definition, context) =>
       flow.runInCompatibilityScope(addEffect(flow, definition), pubsub, context),