Add Unified Resource watcher (gravitational#28296)

* Adds a UnifiedResource watcher to the auth server. This watcher
will watch all the types that are displayed in the web UI and store
them in-memory to allow us to search/filter/query and get multiple
kinds returned at the same time.
[RFD](gravitational#28162)

* Add ListUnifiedResources gRPC and web endpoints (gravitational#29661)

Adds the `ListUnifiedResources` grpc endpoint that returns paginated unified resources and exposes it in the web apiserver

* Fix test

api/client/client.go

@@ -3076,12 +3076,32 @@ func (c *Client) GetResources(ctx context.Context, req *proto.ListResourcesReque
 	return resp, trail.FromGRPC(err)
 }
 
+// ListUnifiedResources returns a paginated list of unified resources that the user has access to.
+// `nextKey` is used as `startKey` in another call to ListUnifiedResources to retrieve
+// the next page.
+// It will return a `trace.LimitExceeded` error if the page exceeds gRPC max
+// message size.
+func (c *Client) ListUnifiedResources(ctx context.Context, req *proto.ListUnifiedResourcesRequest) (*proto.ListUnifiedResourcesResponse, error) {
+	if err := req.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	resp, err := c.grpc.ListUnifiedResources(ctx, req)
+	return resp, trail.FromGRPC(err)
+}
+
 // GetResourcesClient is an interface used by GetResources to abstract over implementations of
 // the ListResources method.
 type GetResourcesClient interface {
 	GetResources(ctx context.Context, req *proto.ListResourcesRequest) (*proto.ListResourcesResponse, error)
 }
 
+// ListUnifiedResourcesClient is an interface used by ListUnifiedResources to abstract over implementations of
+// the ListUnifiedResources method.
+type ListUnifiedResourcesClient interface {
+	ListUnifiedResources(ctx context.Context, req *proto.ListUnifiedResourcesRequest) (*proto.ListUnifiedResourcesResponse, error)
+}
+
 // ResourcePage holds a page of results from [GetResourcePage].
 type ResourcePage[T types.ResourceWithLabels] struct {
 	// Resources retrieved for a single [proto.ListResourcesRequest]. The length of
@@ -3094,6 +3114,86 @@ type ResourcePage[T types.ResourceWithLabels] struct {
 	NextKey string
 }
 
+// getResourceFromProtoPage extracts the resource from the PaginatedResource returned
+// from the rpc ListUnifiedResources
+func getResourceFromProtoPage(resource *proto.PaginatedResource) (types.ResourceWithLabels, error) {
+	var out types.ResourceWithLabels
+	if r := resource.GetNode(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetDatabaseServer(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetDatabaseService(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetAppServerOrSAMLIdPServiceProvider(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetWindowsDesktop(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetWindowsDesktopService(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetKubeCluster(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetKubernetesServer(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetUserGroup(); r != nil {
+		out = r
+		return out, nil
+	} else if r := resource.GetAppServer(); r != nil {
+		out = r
+		return out, nil
+	} else {
+		return nil, trace.BadParameter("received unsupported resource %T", resource.Resource)
+	}
+}
+
+// ListUnifiedResourcePage is a helper for getting a single page of unified resources that match the provided request.
+func ListUnifiedResourcePage(ctx context.Context, clt ListUnifiedResourcesClient, req *proto.ListUnifiedResourcesRequest) (ResourcePage[types.ResourceWithLabels], error) {
+	var out ResourcePage[types.ResourceWithLabels]
+
+	// Set the limit to the default size if one was not provided within
+	// an acceptable range.
+	if req.Limit == 0 || req.Limit > int32(defaults.DefaultChunkSize) {
+		req.Limit = int32(defaults.DefaultChunkSize)
+	}
+
+	for {
+		resp, err := clt.ListUnifiedResources(ctx, req)
+		if err != nil {
+			if trace.IsLimitExceeded(err) {
+				// Cut chunkSize in half if gRPC max message size is exceeded.
+				req.Limit /= 2
+				// This is an extremely unlikely scenario, but better to cover it anyways.
+				if req.Limit == 0 {
+					return out, trace.Wrap(trail.FromGRPC(err), "resource is too large to retrieve")
+				}
+
+				continue
+			}
+
+			return out, trail.FromGRPC(err)
+		}
+
+		for _, respResource := range resp.Resources {
+			resource, err := getResourceFromProtoPage(respResource)
+			if err != nil {
+				return out, trace.Wrap(err)
+			}
+			out.Resources = append(out.Resources, resource)
+		}
+
+		out.NextKey = resp.NextKey
+
+		return out, nil
+	}
+}
+
 // GetResourcePage is a helper for getting a single page of resources that match the provide request.
 func GetResourcePage[T types.ResourceWithLabels](ctx context.Context, clt GetResourcesClient, req *proto.ListResourcesRequest) (ResourcePage[T], error) {
 	var out ResourcePage[T]