.opencode/node_modules/effect/dist/unstable/cluster/Entity.js

/**
 * Defines addressable entity types for Effect Cluster.
 *
 * An entity gives a stable name and RPC protocol to a group of values that are
 * addressed by id. The cluster uses that information to choose a shard and
 * route each request to the runner responsible for that id. This module
 * includes constructors for entity definitions, helpers for creating sharded
 * clients, layer builders for registering handlers, and services that expose
 * the current entity address while a request is being handled.
 *
 * @since 4.0.0
 */
import * as Arr from "../../Array.js";
import * as Context from "../../Context.js";
import * as Data from "../../Data.js";
import * as Effect from "../../Effect.js";
import * as Equal from "../../Equal.js";
import * as Exit from "../../Exit.js";
import { identity } from "../../Function.js";
import * as Hash from "../../Hash.js";
import * as Layer from "../../Layer.js";
import * as Option from "../../Option.js";
import * as Predicate from "../../Predicate.js";
import * as Queue from "../../Queue.js";
import { Scope } from "../../Scope.js";
import * as Stream from "../../Stream.js";
import * as Headers from "../http/Headers.js";
import * as Rpc from "../rpc/Rpc.js";
import * as RpcClient from "../rpc/RpcClient.js";
import * as RpcGroup from "../rpc/RpcGroup.js";
import * as RpcSchema from "../rpc/RpcSchema.js";
import * as RpcServer from "../rpc/RpcServer.js";
import { Persisted, ShardGroup, Uninterruptible } from "./ClusterSchema.js";
import { EntityAddress } from "./EntityAddress.js";
import { EntityType } from "./EntityType.js";
import * as Envelope from "./Envelope.js";
import { hashString } from "./internal/hash.js";
import { ResourceMap } from "./internal/resourceMap.js";
import * as Message from "./Message.js";
import { RunnerAddress } from "./RunnerAddress.js";
import * as ShardId from "./ShardId.js";
import { ShardingConfig } from "./ShardingConfig.js";
import * as Snowflake from "./Snowflake.js";
const TypeId = "~effect/cluster/Entity";
/**
 * Returns `true` when the supplied value is a cluster `Entity`.
 *
 * **Details**
 *
 * The check is based on the internal entity type identifier.
 *
 * @category refinements
 * @since 4.0.0
 */
export const isEntity = u => Predicate.hasProperty(u, TypeId);
const Proto = {
  [TypeId]: TypeId,
  [Hash.symbol]() {
    return Hash.structure({
      type: this.type
    });
  },
  [Equal.symbol](that) {
    return isEntity(that) && this.type === that.type;
  },
  annotate(key, value) {
    return fromRpcGroup(this.type, this.protocol.annotate(key, value));
  },
  annotateRpcs(key, value) {
    return fromRpcGroup(this.type, this.protocol.annotateRpcs(key, value));
  },
  annotateMerge(annotations) {
    return fromRpcGroup(this.type, this.protocol.annotateMerge(annotations));
  },
  annotateRpcsMerge(annotations) {
    return fromRpcGroup(this.type, this.protocol.annotateRpcsMerge(annotations));
  },
  getShardId(entityId) {
    return Effect.map(shardingTag, sharding => sharding.getShardId(entityId, this.getShardGroup(entityId)));
  },
  get client() {
    return shardingTag.pipe(Effect.flatMap(sharding => sharding.makeClient(this)));
  },
  toLayer(build, options) {
    return shardingTag.pipe(Effect.flatMap(sharding => sharding.registerEntity(this, Effect.isEffect(build) ? build : Effect.succeed(build), options)), Layer.effectDiscard);
  },
  of: identity,
  toLayerQueue(build, options) {
    const buildHandlers = Effect.gen({
      self: this
    }, function* () {
      const behaviour = Effect.isEffect(build) ? yield* build : build;
      const queue = yield* Queue.make();
      // create the rpc handlers for the entity
      const handler = envelope => Effect.callback(resume => {
        Queue.offerUnsafe(queue, envelope);
        resumes.set(envelope, resume);
      });
      const streamHandler = envelope => Effect.callback(resume => {
        Queue.offerUnsafe(queue, envelope);
        resumes.set(envelope, resume);
      }).pipe(Effect.map(streamOrQueue => Stream.isStream(streamOrQueue) ? streamOrQueue : Stream.fromQueue(streamOrQueue)), Stream.unwrap);
      const handlers = {};
      for (const rpc_ of this.protocol.requests.values()) {
        const rpc = rpc_;
        handlers[rpc._tag] = RpcSchema.isStreamSchema(rpc.successSchema) ? streamHandler : handler;
      }
      // make the Replier for the behaviour
      const resumes = new Map();
      const complete = (request, exit) => Effect.sync(() => {
        const resume = resumes.get(request);
        if (resume) {
          resumes.delete(request);
          resume(exit);
        }
      });
      const replier = {
        succeed: (request, value) => complete(request, Exit.succeed(value)),
        fail: (request, error) => complete(request, Exit.fail(error)),
        failCause: (request, cause) => complete(request, Exit.failCause(cause)),
        complete
      };
      // fork the behaviour into the layer scope
      yield* behaviour(queue, replier).pipe(Effect.catchCause(cause => {
        const exit = Exit.failCause(cause);
        for (const resume of resumes.values()) {
          resume(exit);
        }
        return Effect.void;
      }), Effect.interruptible, Effect.forkScoped);
      return handlers;
    });
    return this.toLayer(buildHandlers, {
      ...options,
      concurrency: "unbounded"
    });
  }
};
/**
 * Creates a new `Entity` of the specified `type` which will accept messages
 * that adhere to the provided `RpcGroup`.
 *
 * @category constructors
 * @since 4.0.0
 */
export const fromRpcGroup = (
/**
 * The entity type name.
 */
type,
/**
 * The schema definition for messages that the entity is capable of
 * processing.
 */
protocol) => {
  const self = Object.create(Proto);
  self.type = EntityType.make(type);
  self.protocol = protocol;
  self.getShardGroup = Context.get(protocol.annotations, ShardGroup);
  return self;
};
/**
 * Creates a new `Entity` of the specified `type` which will accept messages
 * that adhere to the provided schemas.
 *
 * **When to use**
 *
 * Use to define a cluster entity from individual `Rpc` definitions, giving the
 * cluster runtime a typed protocol for handlers and per-entity clients.
 *
 * **Details**
 *
 * The `type` argument is stored as the entity `EntityType`, and the RPC array
 * is grouped into the entity's `protocol`.
 *
 * **Gotchas**
 *
 * RPC tags should be unique within the array. If multiple definitions use the
 * same tag, the resulting protocol keeps the later definition for that tag.
 *
 * @see {@link fromRpcGroup} for creating an entity from an existing `RpcGroup`
 *
 * @category constructors
 * @since 4.0.0
 */
export const make = (
/**
 * The entity type name.
 */
type,
/**
 * The schema definition for messages that the entity is capable of
 * processing.
 */
protocol) => fromRpcGroup(type, RpcGroup.make(...protocol));
/**
 * Service tag for the entity address currently being processed.
 *
 * **When to use**
 *
 * Use to read the current entity identity and shard address from entity
 * handlers and keep-alive logic.
 *
 * @category context
 * @since 4.0.0
 */
export class CurrentAddress extends /*#__PURE__*/Context.Service()("effect/cluster/Entity/EntityAddress") {}
/**
 * Service tag for the runner address currently registering entity handlers.
 *
 * **When to use**
 *
 * Use to read the runner address associated with the current entity handler
 * registration.
 *
 * @category context
 * @since 4.0.0
 */
export class CurrentRunnerAddress extends /*#__PURE__*/Context.Service()("effect/cluster/Entity/RunnerAddress") {}
/**
 * Represents an entity request envelope delivered to entity handlers.
 *
 * **Details**
 *
 * It includes the underlying request envelope plus the last stream reply chunk
 * that was sent, allowing handlers to resume chunk sequencing after a restart.
 *
 * @category request
 * @since 4.0.0
 */
export class Request extends Data.Class {
  /**
   * Most recent success chunk value sent by the entity, when one exists.
   *
   * @since 4.0.0
   */
  get lastSentChunkValue() {
    return Option.map(this.lastSentChunk, chunk => Arr.lastNonEmpty(chunk.values));
  }
  /**
   * Sequence number to use for the entity's next outgoing success chunk.
   *
   * @since 4.0.0
   */
  get nextSequence() {
    if (Option.isNone(this.lastSentChunk)) {
      return 0;
    }
    return this.lastSentChunk.value.sequence + 1;
  }
}
const shardingTag = /*#__PURE__*/Context.Service("effect/cluster/Sharding");
/**
 * Builds an in-memory test client for an entity layer.
 *
 * **Details**
 *
 * The returned function creates a no-serialization RPC client for each entity ID,
 * using a test sharding service instead of the cluster transport.
 *
 * @category testing
 * @since 4.0.0
 */
export const makeTestClient = /*#__PURE__*/Effect.fnUntraced(function* (entity, layer) {
  const config = yield* ShardingConfig;
  const makeShardId = entityId => ShardId.make(entity.getShardGroup(entityId), Math.abs(hashString(entityId) % config.shardsPerGroup) + 1);
  const snowflakeGen = yield* Snowflake.makeGenerator;
  const runnerAddress = new RunnerAddress({
    host: "localhost",
    port: 3000
  });
  const entityMap = new Map();
  const sharding = shardingTag.of({
    ...{},
    registerEntity: (entity, handlers, options) => Effect.contextWith(context => {
      entityMap.set(entity.type, {
        context: context,
        concurrency: options?.concurrency ?? 1,
        build: entity.protocol.toHandlers(handlers).pipe(Effect.provideContext(Context.mutate(context, context => context.pipe(Context.add(CurrentRunnerAddress, runnerAddress), Context.omit(Scope)))))
      });
      return Effect.void;
    })
  });
  yield* Layer.build(Layer.provide(layer, Layer.succeed(shardingTag)(sharding)));
  const entityEntry = entityMap.get(entity.type);
  if (!entityEntry) {
    return yield* Effect.die(`Entity.makeTestClient: ${entity.type} was not registered by layer`);
  }
  const map = yield* ResourceMap.make(Effect.fnUntraced(function* (entityId) {
    const address = new EntityAddress({
      entityType: entity.type,
      entityId: entityId,
      shardId: makeShardId(entityId)
    });
    const handlers = yield* entityEntry.build.pipe(Effect.provideService(CurrentAddress, address));
    // oxlint-disable-next-line prefer-const
    let client;
    const server = yield* RpcServer.makeNoSerialization(entity.protocol, {
      concurrency: entityEntry.concurrency,
      onFromServer(response) {
        return client.write(response);
      }
    }).pipe(Effect.provide(handlers));
    client = yield* RpcClient.makeNoSerialization(entity.protocol, {
      supportsAck: true,
      generateRequestId: () => snowflakeGen.nextUnsafe(),
      onFromClient({
        message
      }) {
        if (message._tag === "Request") {
          return server.write(0, {
            ...message,
            payload: new Request({
              ...message,
              [Envelope.TypeId]: Envelope.TypeId,
              address,
              requestId: Snowflake.Snowflake(message.id),
              lastSentChunk: Option.none()
            })
          });
        }
        return server.write(0, message);
      }
    });
    return client.client;
  }));
  return entityId => map.get(entityId);
});
/**
 * Enables or disables keep-alive for the current entity.
 *
 * **Details**
 *
 * When enabled it sends the internal keep-alive RPC for the current address; when
 * disabled it releases the keep-alive latch if one is present.
 *
 * @category Keep alive
 * @since 4.0.0
 */
export const keepAlive = /*#__PURE__*/Effect.fnUntraced(function* (enabled) {
  const olatch = yield* Effect.serviceOption(KeepAliveLatch);
  if (olatch._tag === "None") return;
  if (!enabled) {
    yield* olatch.value.open;
    return;
  }
  const sharding = yield* shardingTag;
  const address = yield* CurrentAddress;
  const requestId = yield* sharding.getSnowflake;
  const span = yield* Effect.orDie(Effect.currentSpan);
  olatch.value.closeUnsafe();
  yield* Effect.orDie(sharding.sendOutgoing(new Message.OutgoingRequest({
    annotations: KeepAliveRpc.annotations,
    rpc: KeepAliveRpc,
    context: Context.empty(),
    envelope: Envelope.makeRequest({
      requestId,
      address,
      tag: KeepAliveRpc._tag,
      payload: void 0,
      headers: Headers.empty,
      traceId: span.traceId,
      spanId: span.spanId,
      sampled: span.sampled
    }),
    lastReceivedReply: Option.none(),
    respond: () => Effect.void
  }), true));
}, (effect, enabled) => Effect.withSpan(effect, "Entity/keepAlive", {
  attributes: {
    enabled
  },
  captureStackTrace: false
}));
/**
 * RPC used internally to keep an entity active while a resource is held.
 *
 * **Details**
 *
 * The RPC is marked as persisted and uninterruptible so the keep-alive signal
 * survives normal entity restarts.
 *
 * @category Keep alive
 * @since 4.0.0
 */
export const KeepAliveRpc = /*#__PURE__*/Rpc.make("Cluster/Entity/keepAlive").annotate(Persisted, true).annotate(Uninterruptible, true);
/**
 * Service tag for the latch that coordinates entity keep-alive state.
 *
 * **Details**
 *
 * `keepAlive` closes the latch when keep-alive is active and opens it again when
 * the resource no longer needs to keep the entity alive.
 *
 * @category Keep alive
 * @since 4.0.0
 */
export class KeepAliveLatch extends /*#__PURE__*/Context.Service()("effect/cluster/Entity/KeepAliveLatch") {}
//# sourceMappingURL=Entity.js.map