export class AgentEndpointResolver {
  static readonly make: Effect.Effect<AgentEndpointResolver> = Effect.map(
    Ref.make<ResolverState>(emptyState),
    (state) => new AgentEndpointResolver(state),
  );

  private readonly state: Ref.Ref<ResolverState>;

  private constructor(state: Ref.Ref<ResolverState>) {
    this.state = state;
  }

  /**
   * Atomically associate `(agentId, connectionId)` and the reverse
   * `(connectionId → agentId)` entry.
   *
   * Idempotent on the forward set: re-adding the same connection to the
   * same agent leaves the set unchanged ({@link HashSet.add} is set-union
   * semantics).
   *
   * Cross-agent ownership conflict: if `connectionId` is already in the
   * reverse index for a *different* agent, the new add takes ownership —
   * the connection is removed from
   * the prior agent's forward set inside the same `Ref.update` so the
   * forward and reverse views stay invariant. Practically unreachable
   * but the detection is cheap and the alternative is a silent
   * forward-map leak.
   * @param agentId Identifier of the agent targeted by the operation.
   * @param connId Value supplied to the operation.
   * @returns The prior result.
   */
  add(agentId: AgentId, connId: ConnectionId): Effect.Effect<void> {
    return Ref.update(this.state, (s) => {
      const prior = HashMap.get(s.byConnection, connId);
      let byAgent = s.byAgent;
      if (Option.isSome(prior) && prior.value !== agentId) {
        byAgent = HashMap.modifyAt(byAgent, prior.value, (existing) =>
          Option.flatMap(existing, (set) => {
            const next = HashSet.remove(set, connId);
            return HashSet.size(next) === 0 ? Option.none() : Option.some(next);
          }),
        );
      }
      return {
        byAgent: HashMap.modifyAt(byAgent, agentId, (existing) =>
          Option.some(
            Option.match(existing, {
              onNone: () => HashSet.make(connId),
              onSome: (set) => HashSet.add(set, connId),
            }),
          ),
        ),
        byConnection: HashMap.set(s.byConnection, connId, agentId),
      };
    });
  }

  /**
   * Atomically drop `(agentId, connectionId)` from the forward multimap
   * and, if the pair was actually present in the agent's set, drop
   * `connectionId` from the reverse index too.
   *
   * Idempotent. Calling `remove` for a `(agentId, connectionId)` pair
   * that was never added is a no-op — the disconnect path can fire it
   * unconditionally when the connection authed. For never-authed
   * connections, the disconnect path simply skips the call (no agentId
   * to address it with) and the resolver state is unchanged.
   *
   * Tearing the invariant matters when `connectionId` is genuinely owned
   * by a *different* agent than the caller asserts. The reverse index
   * is only cleared when `byAgent[agentId]` actually held `connectionId`;
   * a stray `remove(WRONG_AGENT, conn)` therefore cannot evict
   * `byConnection[conn]` from under the rightful owner. This guarantees
   * the two maps stay consistent under any sequence of mis-targeted
   * removes (programmer error or a re-issued lifecycle hook).
   *
   * If removing `connectionId` empties the agent's set, the agent key
   * itself is dropped from the forward map so {@link resolveAll} returns
   * the empty set rather than hitting an empty bucket.
   * @param agentId Identifier of the agent targeted by the operation.
   * @param connId Value supplied to the operation.
   * @returns The existed result.
   */
  remove(agentId: AgentId, connId: ConnectionId): Effect.Effect<void> {
    return Ref.update(this.state, (s) => {
      const existed = Option.match(HashMap.get(s.byAgent, agentId), {
        onNone: () => false,
        onSome: (set) => HashSet.has(set, connId),
      });
      if (!existed) {
        return s;
      }
      return {
        byAgent: HashMap.modifyAt(s.byAgent, agentId, (existing) =>
          Option.flatMap(existing, (set) => {
            const next = HashSet.remove(set, connId);
            return HashSet.size(next) === 0 ? Option.none() : Option.some(next);
          }),
        ),
        byConnection: HashMap.remove(s.byConnection, connId),
      };
    });
  }

  /**
   * Hot-path fan-out lookup. Returns every connection id currently
   * associated with `agentId`. Read-only snapshot — the `HashSet` is
   * immutable and the caller cannot mutate the resolver through it.
   * @param agentId Identifier of the agent targeted by the operation.
   * @returns The resolve all result.
   */
  resolveAll(agentId: AgentId): Effect.Effect<HashSet.HashSet<ConnectionId>> {
    return Effect.map(Ref.get(this.state), (s) =>
      Option.getOrElse(HashMap.get(s.byAgent, agentId), () =>
        HashSet.empty<ConnectionId>(),
      ),
    );
  }
}

export const agentEndpointResolverLive = Layer.effect(
  AgentEndpointResolverTag,
  AgentEndpointResolver.make,
)

export class AgentEndpointResolverTag extends Context.Tag(
  "moltzap/AgentEndpointResolver",
)<AgentEndpointResolverTag, AgentEndpointResolver>() {}

export const broadcastNotificationToAgents = <
  D extends AnyNotificationDefinition,
>(
  agentIds: readonly AgentId[],
  definition: D,
  params: NotificationParamsOf<D>,
  options?: BroadcastOptions,
): Effect.Effect<void, never, NetworkSendServiceTag>

export const connectAgent: ServerHandler<typeof agentConnect> = (params)

export const connectionId: (value: string)

export class DeliveryAck extends Data.TaggedClass("DeliveryAck")<{
  readonly to: AgentId;
}> {}

export type DeliveryError = RecipientNotResolved | WriteFailed;

export class NetworkSendService {
  private readonly resolver: AgentEndpointResolver;
  private readonly connections: ConnectionManager;

  constructor(resolver: AgentEndpointResolver, connections: ConnectionManager) {
    this.resolver = resolver;
    this.connections = connections;
  }

  /**
   * Route `payload` to one live connection of `agentId`. Iterates the
   * resolver set so a stale entry does not poison the send when a
   * sibling connection is still live. {@link RecipientNotResolved}
   * Folds "no resolver entry" and "every resolved connection has gone
   * away" — callers can't act on the distinction without poking
   * internal state.
   * @param to Value supplied to the operation.
   * @param payload Value supplied to the operation.
   * @returns The conns result.
   */
  send(
    to: AgentId,
    payload: OpaquePayload,
  ): Effect.Effect<DeliveryAck, DeliveryError> {
    return Effect.gen(
      function* (this: NetworkSendService) {
        const conns = yield* this.resolver.resolveAll(to);
        for (const candidate of HashSet.values(conns)) {
          const conn = yield* this.connections.peek(candidate);
          if (Option.isNone(conn)) {
            continue;
          }
          yield* conn.value.socket.write(payload).pipe(
            Effect.either,
            Effect.flatMap(
              Either.match({
                onLeft: (cause) => Effect.fail(new WriteFailed({ to, cause })),
                onRight: () => Effect.void,
              }),
            ),
          );
          return new DeliveryAck({ to });
        }
        return yield* new RecipientNotResolved({ to });
      }.bind(this),
    );
  }

  /**
   * Fan out `payload` across every live connection of every agent in
   * `agentIds`. Per-CONNECTION (multi-tab agents receive one frame per
   * live connection); writes are forked so a hung recipient does not
   * extend the caller's RPC latency.
   *
   * Filter options:
   * - `forConversation` — apply the server-side conversation subscription
   *   index gate; absent, every connection of every listed agent receives.
   * - `excludeConnectionId` — skip the named connection. The
   *   `agent/message/send` author uses this to avoid echoing the RPC reply
   *   back as a notification.
   *
   * `delivered` lists agents whose at-least-one connection was scheduled to
   * receive a write; the message service records that set on its delivery
   * trace.
   * @param agentIds Value supplied to the operation.
   * @param payload Value supplied to the operation.
   * @param opts Value supplied to the operation.
   * @returns The delivered result.
   */
  broadcast(
    agentIds: readonly AgentId[],
    payload: OpaquePayload,
    opts: BroadcastOptions = {},
  ): Effect.Effect<{ readonly delivered: readonly AgentId[] }> {
    return this.fanOut(agentIds, opts, (conn, cid, target) =>
      this.forkBroadcastWrite({ cid, conn, target, payload, options: opts }),
    );
  }

  /**
   * Shared per-agent / per-connection fan-out driver. For every agent in
   * `agentIds`, resolves its live connections, runs the `connectionCanReceive`
   * gate, and invokes `fire` on each gate-passing connection. An agent lands in
   * `delivered` when at least one of its connections passed the gate.
   * @param agentIds Value supplied to the operation.
   * @param options Options that control the operation.
   * @param fire Value supplied to the operation.
   * @returns The delivered result.
   */
  private fanOut(
    agentIds: readonly AgentId[],
    options: BroadcastOptions,
    fire: (
      conn: AgentConnection,
      cid: ConnectionId,
      target: AgentId,
    ) => Effect.Effect<void>,
  ): Effect.Effect<{ readonly delivered: readonly AgentId[] }> {
    return Effect.gen(
      function* (this: NetworkSendService) {
        const delivered: AgentId[] = [];
        for (const target of agentIds) {
          const connIds = yield* this.resolver.resolveAll(target);
          let reached = false;
          for (const cid of HashSet.values(connIds)) {
            const connOpt = yield* this.connectionCanReceive(cid, options);
            if (Option.isNone(connOpt)) {
              continue;
            }
            yield* fire(connOpt.value, cid, target);
            reached = true;
          }
          if (reached) {
            delivered.push(target);
          }
        }
        return { delivered };
      }.bind(this),
    );
  }

export const networkSendServiceLive = Layer.effect(
  NetworkSendServiceTag,
  Effect.gen(function* () {
    const resolver = yield* AgentEndpointResolverTag;
    const connections = yield* ConnectionManagerTag;
    return new NetworkSendService(resolver, connections);
  }).pipe(Effect.withSpan("NetworkSendServiceLive")),
)

export class NetworkSendServiceTag extends Context.Tag(
  "moltzap/NetworkSendService",
)<NetworkSendServiceTag, NetworkSendService>() {}

export type OpaquePayload = string & Brand.Brand<"OpaquePayload">;