Hyperlinkv0.9.0-beta.0

← Hyperlink

Hyperlink.peersLayerconstsrc/Hyperlink.ts:6045
<Self, S extends Spec, EIn = never, RIn = never>(
  tag: HyperlinkTag<Self, S>,
  self: AnyNode,
  options?: {
    readonly nodes?: ReadonlyArray<AnyNode>
    readonly url?: (
      node: AnyNode
    ) => Effect.Effect<string | undefined, EIn, RIn>
  }
): Layer.Layer<PeersId<Self> | SelfNodeId<Self>, EIn, RIn>

Provide the peers capability on this node: connect every OTHER node in the tag's distributed / nodes set and expose them as the peer clients. Also provides the selfNode capability (this node's key) for byNode-style folds. The opt-in mesh — add it to a node's serve only where the HyperService's own logic reaches across nodes. self is the node you are, so you're excluded from your own peer set.

Membership (D3):

  • Fixed — non-empty options.nodes or stamped nodes([…]) / distributed([…]).
  • Directory — stamped empty set (bare .pipe(Hyperlink.distributed) / nodes([])): read Lookup Directory.nodesServing(tag.key) at layer build. Soft empty map when Directory is absent.
  • Undeclared — no nodesSym and no options.nodes → empty static peers (not directory).

Peer addresses: each Node's own url / path is the default. Pass options.url to override the url per node — an env-specific port, a tunnel, or a value from Effect Config — falling back to Node.url when the resolver returns undefined. A node with no dialable address is skipped (never a throw), so a partial mesh degrades cleanly. IpcSocket peers dial via protocolIpc when only path is set. The resolver's error and requirements flow to the layer (typed).

nodes & fleetpeersdistributednodesselfNodeNodeprotocolIpc
Source src/Hyperlink.ts:6045108 lines
export const peersLayer = <Self, S extends Spec, EIn = never, RIn = never>(
  tag: HyperlinkTag<Self, S>,
  self: AnyNode,
  options?: {
    /** The fleet (including `self`) — supply it **at the use site** so a shared resource can be defined
     *  node-free and exported; falls back to the tag's baked-in {@link distributed} set when omitted.
     *  An explicit empty array is directory-backed (same as bare {@link distributed}). */
    readonly nodes?: ReadonlyArray<AnyNode>;
    readonly url?: (node: AnyNode) => Effect.Effect<string | undefined, EIn, RIn>;
  },
): Layer.Layer<PeersId<Self> | SelfNodeId<Self>, EIn, RIn> =>
  Layer.merge(
    Layer.effect(
      tag[peersSym],
      Effect.gen(function* () {
        const stamped =
          options?.nodes !== undefined ? options.nodes : tag[nodesSym];

        // D3: stamped empty set → Lookup directory membership (soft if Directory absent).
        if (stamped !== undefined && stamped.length === 0) {
          const Lookup = yield* Effect.promise(() => import("./Lookup"));
          const dirOpt = yield* Effect.serviceOption(Lookup.Directory);
          if (Option.isNone(dirOpt)) {
            return {} as Record<string, PeerServiceOf<S>>;
          }
          const rows = yield* Lookup.nodesServing(tag).pipe(
            Effect.provideService(Lookup.Directory, dirOpt.value),
          );
          type DialTarget = {
            readonly key: string;
            readonly kind: ProtocolKind;
            readonly url?: string;
            readonly path?: string;
          };
          const dialable: Array<DialTarget> = [];
          for (const row of rows) {
            if (row.nodeKey === self.key) continue;
            if (row.kind === "IpcSocket" && row.path !== undefined) {
              dialable.push({
                key: row.nodeKey,
                kind: row.kind,
                path: row.path,
              });
              continue;
            }
            if (row.url !== undefined) {
              dialable.push({
                key: row.nodeKey,
                kind: row.kind,
                url: row.url,
              });
            }
          }
          const discovered = yield* Effect.forEach(dialable, (target) =>
            Effect.map(
              buildPeerClientAt(tag, target),
              (client) => [target.key, client] as const,
            ),
          );
          return Object.fromEntries(discovered) as unknown as Record<
            string,
            PeerServiceOf<S>
          >;
        }

        // Fixed fleet (or undeclared → []); drop self to get the peers.
        const fleet = stamped ?? [];
        const others = fleet.filter((node) => node.key !== self.key);
        const resolveUrl = (
          node: AnyNode,
        ): Effect.Effect<string | undefined, EIn, RIn> =>
          options?.url === undefined
            ? Effect.succeed(node.url)
            : Effect.map(options.url(node), (override) => override ?? node.url);
        const resolved = yield* Effect.forEach(others, (node) =>
          Effect.map(resolveUrl(node), (url) => ({
            key: node.key,
            kind: node.kind,
            url,
            path: node.path,
          })),
        );
        const entries = yield* Effect.forEach(
          // no dialable address → skip (partial mesh); ipc path counts when url absent
          resolved.filter(
            (entry) =>
              entry.url !== undefined ||
              (entry.kind === "IpcSocket" && entry.path !== undefined),
          ),
          (target) =>
            Effect.map(
              buildPeerClientAt(tag, {
                key: target.key,
                kind: target.kind,
                ...(target.url !== undefined ? { url: target.url } : {}),
                ...(target.path !== undefined ? { path: target.path } : {}),
              }),
              (client) => [target.key, client] as const,
            ),
        );
        // Boundary: each peer client is a full `ServiceOf<S>` — a width-supertype of the leaf
        // `PeerServiceOf<S>` the capability exposes — but the mapped types don't reduce under a generic
        // `S`, so TS can't see the overlap; the erasure through `unknown` is the honest boundary.
        return Object.fromEntries(entries) as unknown as Record<string, PeerServiceOf<S>>;
      }),
    ),
    selfNodeLayer(tag, self),
  );