export class MoltZapService {
private client: MoltZapAgentClient | null = null;
private connectedValue = false;
private shutdownCompletion: Deferred.Deferred<undefined> | null = null;
/**
* Service-owned scope. Opened in `connect()`, owns the
* `subscribeAll → Stream.runForEach` fan-out fiber. Closed in `close()` so
* the fiber terminates with the service.
*
* Held off the public `connect()` signature so callers do not need to
* thread a `Scope` requirement.
*/
private serviceScope: Scope.CloseableScope | null = null;
private readonly presentationState = new PresentationState();
/**
* The branded outer and inner keys keep conversation and message ids from
* crossing accidentally while each conversation owns its eviction window.
*/
private readonly seenMessageIds = new Map<
ConversationId,
BoundedMap<MessageId, true>
>();
private readonly handlers: {
[K in ServiceHandlerName]: Array<
NotificationHandler<ServiceHandlerPayloads[K]>
>;
} = {
message: [],
rawNotification: [],
disconnect: [],
};
private readonly ownAgentIdValue: AgentId;
private readonly opts: ServiceOptions;
protected constructor(opts: ServiceOptions) {
this.opts = opts;
// The empty HelloOk carries no identity; `ownAgentId` is the client's
// registered/stored id, available before the handshake.
this.ownAgentIdValue = opts.agentId;
}
static fromConfig(config: MoltzapServiceConfig): MoltZapService {
return new MoltZapService(config);
}
static make(
profileName: string,
): Effect.Effect<MoltZapService, ServiceConfigError> {
return loadServiceConfig(profileName).pipe(
Effect.map((config) => MoltZapService.fromConfig(config)),
);
}
get connected(): boolean {
return this.connectedValue;
}
get ownAgentId(): AgentId | undefined {
return this.ownAgentIdValue;
}
/**
* Effect-native: compose via `yield*` or bridge at the edge via `Effect.runPromise`.
* @returns The client result.
*/
connect(): Effect.Effect<HelloOk, ServiceRpcError> {
return Effect.gen(
function* (this: MoltZapService) {
// A new connection never takes ownership while resources from the
// preceding lifecycle are still closing.
while (this.shutdownCompletion !== null) {
const priorShutdown = this.shutdownCompletion;
yield* Deferred.await(priorShutdown);
if (this.shutdownCompletion === priorShutdown) {
this.shutdownCompletion = null;
}
}
const client = new MoltZapAgentClient({
serverUrl: this.opts.serverUrl,
agentKey: this.opts.agentKey,
// The body doesn't branch on close metadata today; the signature is
// kept explicit so a future disconnect-handler chain can plumb
// code/reason through.
onDisconnect: () => {
this.connectedValue = false;
fanout(this.handlers.disconnect, undefined);
},
});
this.client = client;
// `subscribeAll().pipe(Stream.runForEach, …)` is forked into a
// service-owned scope. The Stream is materialized BEFORE `connect()` so
// subscriptions are registered with the registry pre-handshake (a
// pre-connect-legal operation).
//
// Stream errors of type `NotConnectedError` are surfaced on the
// fiber's failure channel only when the client transitions to
// terminal closed state (close() path); `Effect.catchAll` here
// would swallow them silently, so we route through `Effect.logError`
// before the fiber exits.
const serviceScope = yield* Scope.make();
this.serviceScope = serviceScope;
const fanoutEffect = client.subscribeAll().pipe(
Stream.runForEach((notification) =>
Effect.sync(() => {
this.handleNotification(notification);
}),
),
Effect.catchAll((cause) =>
Effect.logWarning(
"MoltZapService notification fan-out terminated",
cause,
),
),
Effect.asVoid,
