fix(alerts): address new PR comments

superset-frontend/src/features/alerts/AlertReportModal.tsx

@@ -2600,6 +2600,7 @@ const AlertReportModal: FunctionComponent<AlertReportModalProps> = ({
                         email_subject={currentAlert?.email_subject || ''}
                         defaultSubject={emailSubject || ''}
                         setErrorSubject={handleErrorUpdate}
+                        addDangerToast={addDangerToast}
                       />
                     </StyledNotificationMethodWrapper>
                   ))}

superset-frontend/src/features/alerts/components/NotificationMethod.tsx

@@ -1,3 +1,4 @@
+/* eslint-disable camelcase */
 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -20,23 +21,25 @@ import {
   FunctionComponent,
   useState,
   ChangeEvent,
-  useEffect,
   useMemo,
+  useRef,
+  useEffect,
+  useCallback,
 } from 'react';
+import { debounce } from 'lodash';
 import rison from 'rison';
 
 import {
   FeatureFlag,
-  JsonResponse,
   SupersetClient,
   isFeatureEnabled,
+  logging,
   styled,
   t,
   useTheme,
 } from '@superset-ui/core';
 import { Icons } from '@superset-ui/core/components/Icons';
-import { Input, Select } from '@superset-ui/core/components';
+import { Input, Select, AsyncSelect } from '@superset-ui/core/components';
-import RefreshLabel from '@superset-ui/core/components/RefreshLabel';
 import {
   NotificationMethodOption,
   NotificationSetting,
@@ -77,6 +80,17 @@ const StyledNotificationMethod = styled.div`
         margin-left: ${theme.sizeUnit * 2}px;
         padding-top: ${theme.sizeUnit}px;
       }
+
+      .refresh-button {
+        margin-left: ${theme.sizeUnit * 2}px;
+        cursor: pointer;
+        color: ${theme.colorTextSecondary};
+
+        &:hover {
+          color: ${theme.colorPrimary};
+        }
+      }
+
       .anticon {
         margin-left: ${theme.sizeUnit}px;
       }
@@ -127,6 +141,7 @@ interface NotificationMethodProps {
   email_subject: string;
   defaultSubject: string;
   setErrorSubject: (hasError: boolean) => void;
+  addDangerToast?: (msg: string) => void;
 }
 
 export const mapSlackValues = ({
@@ -150,47 +165,6 @@ export const mapSlackValues = ({
     .filter(val => !!val) as { label: string; value: string }[];
 };
 
-export const mapChannelsToOptions = (result: SlackChannel[]) => {
-  const publicChannels: SlackChannel[] = [];
-  const privateChannels: SlackChannel[] = [];
-
-  result.forEach(channel => {
-    if (channel.is_private) {
-      privateChannels.push(channel);
-    } else {
-      publicChannels.push(channel);
-    }
-  });
-
-  return [
-    {
-      label: 'Public Channels',
-      options: publicChannels.map((channel: SlackChannel) => ({
-        label: `${channel.name} ${
-          channel.is_member ? '' : t('(Bot not in channel)')
-        }`,
-        value: channel.id,
-        key: channel.id,
-      })),
-      key: 'public',
-    },
-    {
-      label: t('Private Channels (Bot in channel)'),
-      options: privateChannels.map((channel: SlackChannel) => ({
-        label: channel.name,
-        value: channel.id,
-        key: channel.id,
-      })),
-      key: 'private',
-    },
-  ];
-};
-
-type SlackOptionsType = {
-  label: string;
-  options: { label: string; value: string }[];
-}[];
-
 export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
   setting = null,
   index,
@@ -200,6 +174,7 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
   email_subject,
   defaultSubject,
   setErrorSubject,
+  addDangerToast,
 }) => {
   const { method, recipients, cc, bcc, options } = setting || {};
   const [recipientValue, setRecipientValue] = useState<string>(
@@ -214,24 +189,24 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
   const [ccValue, setCcValue] = useState<string>(cc || '');
   const [bccValue, setBccValue] = useState<string>(bcc || '');
   const theme = useTheme();
-  const [methodOptionsLoading, setMethodOptionsLoading] =
-    useState<boolean>(true);
-  const [slackOptions, setSlackOptions] = useState<SlackOptionsType>([
-    {
-      label: '',
-      options: [],
-    },
-  ]);
-
   const [useSlackV1, setUseSlackV1] = useState<boolean>(false);
-  const [isSlackChannelsLoading, setIsSlackChannelsLoading] =
+  const cursorRef = useRef<Record<string, string | null>>({});
-    useState<boolean>(true);
+  const dataCache = useRef<
+    Record<
+      string,
+      { data: { label: string; value: string }[]; totalCount: number }
+    >
+  >({});
+  const forceRefreshRef = useRef(false);
+  const [isRefreshing, setIsRefreshing] = useState(false);
+  const hasShownErrorToast = useRef(false);
+  const [searchGeneration, setSearchGeneration] = useState(0);
+  const lastSearchValueRef = useRef('');
 
   const onMethodChange = (selected: {
     label: string;
     value: NotificationMethodOption;
   }) => {
-    // Since we're swapping the method, reset the recipients
     setRecipientValue('');
     setCcValue('');
     setBccValue('');
@@ -249,84 +224,209 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
     }
   };
 
-  const fetchSlackChannels = async ({
+  const fetchSlackChannels = useCallback(
-    searchString = '',
+    async (
-    types = [],
+      search: string,
-    exactMatch = false,
+      page: number,
-    force = false,
+      pageSize: number,
-  }: {
+    ): Promise<{
-    searchString?: string | undefined;
+      data: { label: string; value: string }[];
-    types?: string[];
+      totalCount: number;
-    exactMatch?: boolean | undefined;
+    }> => {
-    force?: boolean | undefined;
+      try {
-  } = {}): Promise<JsonResponse> => {
+        const cacheKey = `${search}:${page}`;
-    const queryString = rison.encode({
-      searchString,
-      types,
-      exactMatch,
-      force,
-    });
-    const endpoint = `/api/v1/report/slack_channels/?q=${queryString}`;
-    return SupersetClient.get({ endpoint });
-  };
 
-  const updateSlackOptions = async ({
+        if (dataCache.current[cacheKey] && !forceRefreshRef.current) {
-    force,
+          return dataCache.current[cacheKey];
-  }: {
-    force?: boolean | undefined;
-  } = {}) => {
-    setIsSlackChannelsLoading(true);
-    fetchSlackChannels({ types: ['public_channel', 'private_channel'], force })
-      .then(({ json }) => {
-        const { result } = json;
-        const options: SlackOptionsType = mapChannelsToOptions(result);
-
-        setSlackOptions(options);
-
-        if (isFeatureEnabled(FeatureFlag.AlertReportSlackV2)) {
-          // for edit mode, map existing ids to names for display if slack v2
-          // or names to ids if slack v1
-          const [publicOptions, privateOptions] = options;
-          if (
-            method &&
-            [
-              NotificationMethodOption.SlackV2,
-              NotificationMethodOption.Slack,
-            ].includes(method)
-          ) {
-            setSlackRecipients(
-              mapSlackValues({
-                method,
-                recipientValue,
-                slackOptions: [
-                  ...publicOptions.options,
-                  ...privateOptions.options,
-                ],
-              }),
-            );
-          }
         }
-      })
+
-      .catch(e => {
+        // Clear cache for previous searches to prevent stale data
-        // Fallback to slack v1 if slack v2 is not compatible
+        if (page === 0) {
+          Object.keys(dataCache.current).forEach(key => {
+            if (!key.startsWith(`${search}:`)) {
+              delete dataCache.current[key];
+              delete cursorRef.current[key];
+            }
+          });
+        }
+
+        const cursor = page > 0 ? cursorRef.current[cacheKey] : null;
+
+        const params: Record<string, any> = {
+          types: ['public_channel', 'private_channel'],
+          limit: pageSize,
+        };
+
+        if (page === 0 && forceRefreshRef.current) {
+          params.force = true;
+          forceRefreshRef.current = false;
+        }
+
+        if (search) {
+          params.search_string = search;
+        }
+
+        if (cursor) {
+          params.cursor = cursor;
+        }
+
+        const queryString = rison.encode(params);
+        const endpoint = `/api/v1/report/slack_channels/?q=${queryString}`;
+        const response = await SupersetClient.get({ endpoint });
+
+        const { result, next_cursor, has_more } = response.json;
+
+        if (next_cursor) {
+          cursorRef.current[`${search}:${page + 1}`] = next_cursor;
+        }
+
+        const options = result.map((channel: SlackChannel) => ({
+          label: channel.name,
+          value: channel.id,
+        }));
+
+        const totalCount = has_more
+          ? (page + 1) * pageSize + 1
+          : page * pageSize + options.length;
+
+        const responseData = {
+          data: options,
+          totalCount,
+        };
+
+        dataCache.current[cacheKey] = responseData;
+
+        return responseData;
+      } catch (error) {
+        logging.error('Failed to fetch Slack channels:', error);
+
+        // Show user-friendly error message
+        if (addDangerToast && !hasShownErrorToast.current) {
+          addDangerToast(
+            t(
+              'Unable to load Slack channels. Please check your Slack API token configuration. ' +
+                'Switching to manual channel input.',
+            ),
+          );
+          hasShownErrorToast.current = true;
+        }
+
+        // Fallback to Slack v1 without clearing recipients to prevent data loss
         setUseSlackV1(true);
-      })
+
-      .finally(() => {
+        // Auto-switch to Slack V1 in the notification method dropdown
-        setMethodOptionsLoading(false);
+        if (
-        setIsSlackChannelsLoading(false);
+          onUpdate &&
-      });
+          setting &&
+          setting.method === NotificationMethodOption.SlackV2
+        ) {
+          onUpdate(index, {
+            ...setting,
+            method: NotificationMethodOption.Slack, // Switch from SlackV2 to Slack V1
+          });
+        }
+
+        return {
+          data: [],
+          totalCount: 0,
+        };
+      }
+    },
+    // Note: searchGeneration is intentionally included even though not used in function body
+    // Purpose: Trigger function reference change when search changes (after debounce)
+    // Effect: Forces AsyncSelect to clear internal state and show fresh results
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+    [addDangerToast, onUpdate, setting, index, searchGeneration],
+  );
+
+  const handleRefreshSlackChannels = async () => {
+    setIsRefreshing(true);
+
+    try {
+      // Clear cache and cursor to force fresh data fetch
+      forceRefreshRef.current = true;
+      cursorRef.current = {};
+      dataCache.current = {};
+
+      // Fetch fresh channel list without clearing user selections
+      await fetchSlackChannels('', 0, 100);
+    } catch (error) {
+      logging.error('Error refreshing channels:', error);
+
+      // Show error toast if available
+      if (addDangerToast) {
+        addDangerToast(
+          t('Failed to refresh Slack channels. Please try again.'),
+        );
+      }
+    } finally {
+      setIsRefreshing(false);
+    }
   };
 
+  // Reset error toast flag when method changes
   useEffect(() => {
-    const slackEnabled = options?.some(
+    hasShownErrorToast.current = false;
-      option =>
+  }, [method]);
-        option === NotificationMethodOption.Slack ||
+
-        option === NotificationMethodOption.SlackV2,
+  // Initialize slackRecipients when editing an existing alert with SlackV2
-    );
+  useEffect(() => {
-    if (slackEnabled && !slackOptions[0]?.options.length) {
+    const initializeSlackRecipients = async () => {
-      updateSlackOptions();
+      if (
-    }
+        method === NotificationMethodOption.SlackV2 &&
-  }, []);
+        recipients &&
+        slackRecipients.length === 0
+      ) {
+        try {
+          // Fetch first page of channels to map IDs to names
+          const channelData = await fetchSlackChannels('', 0, 100);
+          const recipientIds = recipients.split(',').map(id => id.trim());
+
+          // Map recipient IDs to {label, value} format
+          const mappedRecipients = recipientIds
+            .map(id => {
+              const channel = channelData.data.find(
+                ch => ch.value.toLowerCase() === id.toLowerCase(),
+              );
+              return channel || { label: id, value: id };
+            })
+            .filter(r => r.value); // Filter out empty values
+
+          setSlackRecipients(mappedRecipients);
+        } catch (error) {
+          const recipientIds = recipients.split(',').map(id => id.trim());
+          const fallbackRecipients = recipientIds.map(id => ({
+            label: id,
+            value: id,
+          }));
+          setSlackRecipients(fallbackRecipients);
+        }
+      }
+    };
+
+    initializeSlackRecipients();
+    // Note: fetchSlackChannels and slackRecipients are intentionally omitted
+    // - fetchSlackChannels: Would cause re-initialization on every search change
+    // - slackRecipients: Would cause infinite loop (we modify it inside)
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [method, recipients]);
+
+  const debouncedSearchUpdate = useMemo(
+    () =>
+      debounce((search: string) => {
+        const trimmedSearch = search.trim();
+        if (lastSearchValueRef.current !== trimmedSearch) {
+          lastSearchValueRef.current = trimmedSearch;
+          setSearchGeneration(prev => prev + 1);
+        }
+      }, 500),
+    [],
+  );
+
+  useEffect(
+    () => () => debouncedSearchUpdate.cancel(),
+    [debouncedSearchUpdate],
+  );
 
   const methodOptions = useMemo(
     () =>
@@ -385,6 +485,10 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
     }
   };
 
+  const handleSlackSearch = (search: string) => {
+    debouncedSearchUpdate(search);
+  };
+
   const onSubjectChange = (
     event: ChangeEvent<HTMLTextAreaElement | HTMLInputElement>,
   ) => {
@@ -459,10 +563,8 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
               options={methodOptions}
               showSearch
               value={methodOptions.find(option => option.value === method)}
-              loading={methodOptionsLoading}
             />
             {index !== 0 && !!onRemove ? (
-              // eslint-disable-next-line jsx-a11y/control-has-associated-label
               <span
                 role="button"
                 tabIndex={0}
@@ -541,24 +643,33 @@ export const NotificationMethod: FunctionComponent<NotificationMethodProps> = ({
                 ) : (
                   // for SlackV2
                   <div className="input-container">
-                    <Select
+                    <AsyncSelect
                       ariaLabel={t('Select channels')}
                       mode="multiple"
                       name="recipients"
                       value={slackRecipients}
-                      options={slackOptions}
+                      options={fetchSlackChannels}
                       onChange={onSlackRecipientsChange}
+                      onSearch={handleSlackSearch}
                       allowClear
                       data-test="recipients"
-                      loading={isSlackChannelsLoading}
+                      fetchOnlyOnSearch={false}
-                      allowSelectAll={false}
+                      pageSize={100}
-                      labelInValue
+                      placeholder={t('Select Slack channels')}
-                    />
+                      tokenSeparators={[]}
-                    <RefreshLabel
+                      filterOption={() => true}
-                      onClick={() => updateSlackOptions({ force: true })}
-                      tooltipContent={t('Force refresh Slack channels list')}
-                      disabled={isSlackChannelsLoading}
                     />
+                    <span
+                      role="button"
+                      tabIndex={0}
+                      className="refresh-button"
+                      onClick={handleRefreshSlackChannels}
+                      data-test="refresh-slack-channels"
+                      title={t('Refresh channels')}
+                      style={{ opacity: isRefreshing ? 0.5 : 1 }}
+                    >
+                      <Icons.SyncOutlined iconSize="l" spin={isRefreshing} />
+                    </span>
                   </div>
                 )}
               </div>

superset/commands/report/execute.py

@@ -142,7 +142,7 @@ class BaseReportState:
                     channel_names = (slack_recipients["target"] or "").replace("#", "")
                     # we need to ensure that existing reports can also fetch
                     # ids from private channels
-                    channels = get_channels_with_search(
+                    channels_data = get_channels_with_search(
                         search_string=channel_names,
                         types=[
                             SlackChannelTypes.PRIVATE,
@@ -150,6 +150,7 @@ class BaseReportState:
                         ],
                         exact_match=True,
                     )
+                    channels = channels_data["result"]
                     channels_list = recipients_string_to_list(channel_names)
                     if len(channels_list) != len(channels):
                         missing_channels = set(channels_list) - {

superset/reports/api.py

@@ -577,14 +577,17 @@ class ReportScheduleRestApi(BaseSupersetModelRestApi):
             search_string = params.get("search_string")
             types = params.get("types", [])
             exact_match = params.get("exact_match", False)
-            force = params.get("force", False)
+            cursor = params.get("cursor")
-            channels = get_channels_with_search(
+            limit = params.get("limit", 100)
+
+            channels_data = get_channels_with_search(
                 search_string=search_string,
                 types=types,
                 exact_match=exact_match,
-                force=force,
+                cursor=cursor,
+                limit=limit,
             )
-            return self.response(200, result=channels)
+            return self.response(200, **channels_data)
         except SupersetException as ex:
             logger.error("Error fetching slack channels %s", str(ex))
             return self.response_422(message=str(ex))

superset/reports/schemas.py

@@ -58,6 +58,9 @@ get_slack_channels_schema = {
             "items": {"type": "string", "enum": ["public_channel", "private_channel"]},
         },
         "exact_match": {"type": "boolean"},
+        "cursor": {"type": ["string", "null"]},
+        "limit": {"type": "integer", "default": 100, "minimum": 1, "maximum": 1000},
+        "force": {"type": "boolean"},
     },
 }
 

superset/tasks/slack.py

@@ -15,17 +15,24 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
+from typing import Optional
 
 from flask import current_app
 
-from superset.extensions import celery_app
+from superset.extensions import cache_manager, celery_app
-from superset.utils.slack import get_channels
+from superset.utils.slack import get_channels_with_search, SlackChannelTypes
 
 logger = logging.getLogger(__name__)
 
 
 @celery_app.task(name="slack.cache_channels")
 def cache_channels() -> None:
+    """
+    Celery task to warm up the Slack channels cache.
+
+    This task fetches all Slack channels using pagination and stores them in cache.
+    Useful for large workspaces where the initial channel fetch can be slow.
+    """
     cache_timeout = current_app.config["SLACK_CACHE_TIMEOUT"]
     retry_count = current_app.config.get("SLACK_API_RATE_LIMIT_RETRY_COUNT", 2)
 
@@ -37,7 +44,45 @@ def cache_channels() -> None:
     )
 
     try:
-        get_channels(force=True, cache_timeout=cache_timeout)
+        all_channels = []
+        cursor: Optional[str] = None
+        page_count = 0
+
+        while True:
+            page_count += 1
+
+            result = get_channels_with_search(
+                search_string="",
+                types=list(SlackChannelTypes),
+                cursor=cursor,
+                limit=1000,
+            )
+
+            page_channels = result["result"]
+            all_channels.extend(page_channels)
+
+            logger.debug(
+                "Fetched page %d: %d channels (total: %d)",
+                page_count,
+                len(page_channels),
+                len(all_channels),
+            )
+
+            cursor = result.get("next_cursor")
+            if not cursor or not result.get("has_more"):
+                break
+
+        logger.info(
+            "Successfully fetched %d Slack channels in %d pages. Caching results.",
+            len(all_channels),
+            page_count,
+        )
+
+        cache_key = "slack_conversations_list"
+        cache_manager.cache.set(cache_key, all_channels, timeout=cache_timeout)
+
+        logger.info("Slack channels cache warm-up completed successfully")
+
     except Exception as ex:
         logger.exception(
             "Failed to cache Slack channels: %s. "

superset/utils/slack.py

@@ -17,7 +17,7 @@
 
 
 import logging
-from typing import Callable, Optional
+from typing import Any, Optional
 
 from flask import current_app as app
 from slack_sdk import WebClient
@@ -26,11 +26,8 @@ from slack_sdk.http_retry.builtin_handlers import RateLimitErrorRetryHandler
 
 from superset import feature_flag_manager
 from superset.exceptions import SupersetException
-from superset.extensions import cache_manager
 from superset.reports.schemas import SlackChannelSchema
-from superset.utils import cache as cache_util
 from superset.utils.backports import StrEnum
-from superset.utils.core import recipients_string_to_list
 
 logger = logging.getLogger(__name__)
 
@@ -59,80 +56,162 @@ def get_slack_client() -> WebClient:
     return client
 
 
-@cache_util.memoized_func(
+def _fetch_channels_without_search(
-    key="slack_conversations_list",
+    client: WebClient,
-    cache=cache_manager.cache,
+    channel_schema: SlackChannelSchema,
-)
+    types_param: str,
-def get_channels() -> list[SlackChannelSchema]:
+    cursor: Optional[str],
-    """
+    limit: int,
-    Retrieves a list of all conversations accessible by the bot
+) -> dict[str, Any]:
-    from the Slack API, and caches results (to avoid rate limits).
+    """Fetch channels without search filtering, paginating for large limits."""
-
-    The Slack API does not provide search so to apply a search use
-    get_channels_with_search instead.
-    """
-    client = get_slack_client()
-    channel_schema = SlackChannelSchema()
     channels: list[SlackChannelSchema] = []
-    extra_params = {"types": ",".join(SlackChannelTypes)}
+    slack_cursor = cursor
-    cursor = None
+    page_size = min(limit, 1000)
-    page_count = 0
 
-    logger.info("Starting Slack channels fetch")
+    while True:
+        response = client.conversations_list(
+            limit=page_size,
+            cursor=slack_cursor,
+            exclude_archived=True,
+            types=types_param,
+        )
 
-    try:
+        page_channels = [
-        while True:
+            channel_schema.load(channel) for channel in response.data["channels"]
-            page_count += 1
+        ]
+        channels.extend(page_channels)
 
-            response = client.conversations_list(
+        slack_cursor = response.data.get("response_metadata", {}).get("next_cursor")
-                limit=999, cursor=cursor, exclude_archived=True, **extra_params
-            )
-            page_channels = response.data["channels"]
-            channels.extend(channel_schema.load(channel) for channel in page_channels)
 
-            logger.debug(
+        if not slack_cursor or len(page_channels) < page_size or len(channels) >= limit:
-                "Fetched page %d: %d channels (total: %d)",
+            break
-                page_count,
-                len(page_channels),
-                len(channels),
-            )
 
-            cursor = response.data.get("response_metadata", {}).get("next_cursor")
+    return {
-            if not cursor:
+        "result": channels[:limit],
+        "next_cursor": slack_cursor,
+        "has_more": bool(slack_cursor),
+    }
+
+
+def _fetch_channels_with_search(
+    client: WebClient,
+    channel_schema: SlackChannelSchema,
+    types_param: str,
+    search_string: str,
+    exact_match: bool,
+    cursor: Optional[str],
+    limit: int,
+) -> dict[str, Any]:
+    """Fetch channels with search filtering, streaming through pages."""
+    matches: list[SlackChannelSchema] = []
+    slack_cursor = cursor
+    search_terms = [
+        term.strip().lower() for term in search_string.split(",") if term.strip()
+    ]
+
+    while len(matches) < limit:
+        response = client.conversations_list(
+            limit=1000,
+            cursor=slack_cursor,
+            exclude_archived=True,
+            types=types_param,
+        )
+
+        for channel_data in response.data["channels"]:
+            channel_name_lower = channel_data["name"].lower()
+            channel_id_lower = channel_data["id"].lower()
+
+            is_match = False
+            for search_term in search_terms:
+                if exact_match:
+                    if (
+                        search_term == channel_name_lower
+                        or search_term == channel_id_lower
+                    ):
+                        is_match = True
+                        break
+                else:
+                    if (
+                        search_term in channel_name_lower
+                        or search_term in channel_id_lower
+                    ):
+                        is_match = True
+                        break
+
+            if is_match:
+                channel = channel_schema.load(channel_data)
+                matches.append(channel)
+
+            if len(matches) >= limit:
                 break
 
-        logger.info(
+        slack_cursor = response.data.get("response_metadata", {}).get("next_cursor")
-            "Successfully fetched %d Slack channels in %d pages",
+        if not slack_cursor:
-            len(channels),
+            break
-            page_count,
+
-        )
+    has_more = bool(slack_cursor) and len(matches) >= limit
-        return channels
+
-    except SlackApiError as ex:
+    return {
-        logger.error(
+        "result": matches[:limit],
-            "Failed to fetch Slack channels after %d pages: %s",
+        "next_cursor": slack_cursor if has_more else None,
-            page_count,
+        "has_more": has_more,
-            str(ex),
+    }
-            exc_info=True,
-        )
-        raise
 
 
 def get_channels_with_search(
     search_string: str = "",
     types: Optional[list[SlackChannelTypes]] = None,
     exact_match: bool = False,
-    force: bool = False,
+    cursor: Optional[str] = None,
-) -> list[SlackChannelSchema]:
+    limit: int = 100,
+) -> dict[str, Any]:
     """
-    The slack api is paginated but does not include search, so we need to fetch
+    Fetches Slack channels with pagination and search support.
-    all channels and filter them ourselves
+
-    This will search by slack name or id
+    Args:
+        search_string: Search term(s). Can be comma-separated for OR logic.
+                      e.g., "engineering,marketing" matches channels containing
+                      "engineering" OR "marketing"
+        types: Channel types to filter (public_channel, private_channel)
+        exact_match: If True, search term must exactly match channel name/ID
+        cursor: Pagination cursor for fetching next page
+        limit: Maximum number of channels to return
+
+    Returns a dict with:
+    - result: list of channels
+    - next_cursor: cursor for next page (None if no more pages)
+    - has_more: boolean indicating if more pages exist
+
+    The Slack API is paginated but does not include search.
+    We handle two cases:
+    1. WITHOUT search: Fetch single page from Slack API
+    2. WITH search: Stream through Slack API pages until we find enough matches
     """
     try:
-        channels = get_channels(
+        client = get_slack_client()
-            force=force,
+        channel_schema = SlackChannelSchema()
-            cache_timeout=app.config["SLACK_CACHE_TIMEOUT"],
+
+        types_param = (
+            ",".join(types)
+            if types and len(types) < len(SlackChannelTypes)
+            else ",".join(SlackChannelTypes)
         )
+
+        if not search_string:
+            return _fetch_channels_without_search(
+                client, channel_schema, types_param, cursor, limit
+            )
+
+        return _fetch_channels_with_search(
+            client,
+            channel_schema,
+            types_param,
+            search_string,
+            exact_match,
+            cursor,
+            limit,
+        )
+
     except SlackApiError as ex:
         # Check if it's a rate limit error
         status_code = getattr(ex.response, "status_code", None)
@@ -146,38 +225,6 @@ def get_channels_with_search(
     except SlackClientError as ex:
         raise SupersetException(f"Failed to list channels: {ex}") from ex
 
-    if types and not len(types) == len(SlackChannelTypes):
-        conditions: list[Callable[[SlackChannelSchema], bool]] = []
-        if SlackChannelTypes.PUBLIC in types:
-            conditions.append(lambda channel: not channel["is_private"])
-        if SlackChannelTypes.PRIVATE in types:
-            conditions.append(lambda channel: channel["is_private"])
-
-        channels = [
-            channel for channel in channels if any(cond(channel) for cond in conditions)
-        ]
-
-    # The search string can be multiple channels separated by commas
-    if search_string:
-        search_array = recipients_string_to_list(search_string)
-        channels = [
-            channel
-            for channel in channels
-            if any(
-                (
-                    search.lower() == channel["name"].lower()
-                    or search.lower() == channel["id"].lower()
-                    if exact_match
-                    else (
-                        search.lower() in channel["name"].lower()
-                        or search.lower() in channel["id"].lower()
-                    )
-                )
-                for search in search_array
-            )
-        ]
-    return channels
-
 
 def should_use_v2_api() -> bool:
     if not feature_flag_manager.is_feature_enabled("ALERT_REPORT_SLACK_V2"):

tests/integration_tests/reports/api_tests.py

@@ -2049,3 +2049,215 @@ class TestReportSchedulesApi(SupersetTestCase):
         )
 
         assert json.loads(report_schedule.extra_json) == extra_json
+
+    @patch("superset.reports.api.get_channels_with_search")
+    def test_slack_channels_api_without_pagination(self, mock_get_channels):
+        """
+        Test /api/v1/report/slack_channels/ endpoint without pagination
+        """
+        self.login(ADMIN_USERNAME)
+
+        # Mock response without pagination
+        mock_get_channels.return_value = {
+            "result": [
+                {
+                    "id": "C001",
+                    "name": "general",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "id": "C002",
+                    "name": "random",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+        uri = "api/v1/report/slack_channels/"
+        rv = self.client.get(uri)
+        assert rv.status_code == 200
+
+        data = json.loads(rv.data.decode("utf-8"))
+        assert len(data["result"]) == 2
+        assert data["result"][0]["name"] == "general"
+        assert data["next_cursor"] is None
+        assert data["has_more"] is False
+
+        # Verify get_channels_with_search was called with defaults
+        mock_get_channels.assert_called_once_with(
+            search_string=None,
+            types=[],
+            exact_match=False,
+            cursor=None,
+            limit=100,
+        )
+
+    @patch("superset.reports.api.get_channels_with_search")
+    def test_slack_channels_api_with_pagination(self, mock_get_channels):
+        """
+        Test /api/v1/report/slack_channels/ endpoint with pagination
+        """
+        self.login(ADMIN_USERNAME)
+
+        # Mock response with pagination
+        mock_get_channels.return_value = {
+            "result": [
+                {
+                    "id": f"C{i:03d}",
+                    "name": f"channel-{i}",
+                    "is_private": False,
+                    "is_member": True,
+                }
+                for i in range(100)
+            ],
+            "next_cursor": "page_1",
+            "has_more": True,
+        }
+
+        uri = "api/v1/report/slack_channels/?q=(limit:100)"
+        rv = self.client.get(uri)
+        assert rv.status_code == 200
+
+        data = json.loads(rv.data.decode("utf-8"))
+        assert len(data["result"]) == 100
+        assert data["next_cursor"] == "page_1"
+        assert data["has_more"] is True
+
+    @patch("superset.reports.api.get_channels_with_search")
+    def test_slack_channels_api_with_cursor(self, mock_get_channels):
+        """
+        Test /api/v1/report/slack_channels/ endpoint with cursor for next page
+        """
+        self.login(ADMIN_USERNAME)
+
+        # Mock response for second page
+        mock_get_channels.return_value = {
+            "result": [
+                {
+                    "id": "C100",
+                    "name": "channel-100",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+        uri = "api/v1/report/slack_channels/?q=(cursor:'page_1',limit:100)"
+        rv = self.client.get(uri)
+        assert rv.status_code == 200
+
+        data = json.loads(rv.data.decode("utf-8"))
+        assert len(data["result"]) == 1
+        assert data["next_cursor"] is None
+
+        # Verify cursor was passed
+        mock_get_channels.assert_called_once_with(
+            search_string=None,
+            types=[],
+            exact_match=False,
+            cursor="page_1",
+            limit=100,
+        )
+
+    @patch("superset.reports.api.get_channels_with_search")
+    def test_slack_channels_api_with_search(self, mock_get_channels):
+        """
+        Test /api/v1/report/slack_channels/ endpoint with search parameter
+        """
+        self.login(ADMIN_USERNAME)
+
+        # Mock response with search results
+        mock_get_channels.return_value = {
+            "result": [
+                {
+                    "id": "C001",
+                    "name": "engineering",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "id": "C002",
+                    "name": "engineering-ops",
+                    "is_private": True,
+                    "is_member": True,
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+        uri = "api/v1/report/slack_channels/?q=(search_string:'engineering')"
+        rv = self.client.get(uri)
+        assert rv.status_code == 200
+
+        data = json.loads(rv.data.decode("utf-8"))
+        assert len(data["result"]) == 2
+        assert all("engineering" in ch["name"] for ch in data["result"])
+
+        # Verify search_string was passed
+        mock_get_channels.assert_called_once_with(
+            search_string="engineering",
+            types=[],
+            exact_match=False,
+            cursor=None,
+            limit=100,
+        )
+
+    @patch("superset.reports.api.get_channels_with_search")
+    def test_slack_channels_api_with_types(self, mock_get_channels):
+        """
+        Test /api/v1/report/slack_channels/ endpoint with channel types filter
+        """
+        self.login(ADMIN_USERNAME)
+
+        # Mock response with filtered types
+        mock_get_channels.return_value = {
+            "result": [
+                {
+                    "id": "C001",
+                    "name": "general",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+        uri = "api/v1/report/slack_channels/?q=(types:!('public_channel'))"
+        rv = self.client.get(uri)
+        assert rv.status_code == 200
+
+        # Verify types were passed
+        mock_get_channels.assert_called_once_with(
+            search_string=None,
+            types=["public_channel"],
+            exact_match=False,
+            cursor=None,
+            limit=100,
+        )
+
+    @patch("superset.reports.api.get_channels_with_search")
+    def test_slack_channels_api_error_handling(self, mock_get_channels):
+        """
+        Test /api/v1/report/slack_channels/ endpoint error handling
+        """
+        self.login(ADMIN_USERNAME)
+
+        # Mock Slack API error
+        from superset.exceptions import SupersetException
+
+        mock_get_channels.side_effect = SupersetException("Slack API error")
+
+        uri = "api/v1/report/slack_channels/"
+        rv = self.client.get(uri)
+        assert rv.status_code == 422
+
+        data = json.loads(rv.data.decode("utf-8"))
+        assert "Slack API error" in data["message"]

tests/integration_tests/reports/commands_tests.py

@@ -186,7 +186,7 @@ def create_test_table_context(database: Database):
 def create_report_email_chart():
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
-        email_target="target@email.com", chart=chart
+        email_target="target@email.com", chart=chart, name="report_email_chart"
     )
     yield report_schedule
 
@@ -201,6 +201,7 @@ def create_report_email_chart_with_cc_and_bcc():
         ccTarget="cc@email.com",
         bccTarget="bcc@email.com",
         chart=chart,
+        name="report_email_chart_with_cc_and_bcc",
     )
     yield report_schedule
 
@@ -212,7 +213,10 @@ def create_report_email_chart_alpha_owner(get_user):
     owners = [get_user("alpha")]
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
-        email_target="target@email.com", chart=chart, owners=owners
+        email_target="target@email.com",
+        chart=chart,
+        owners=owners,
+        name="report_email_chart_alpha_owner",
     )
     yield report_schedule
 
@@ -223,7 +227,10 @@ def create_report_email_chart_alpha_owner(get_user):
 def create_report_email_chart_force_screenshot():
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
-        email_target="target@email.com", chart=chart, force_screenshot=True
+        email_target="target@email.com",
+        chart=chart,
+        force_screenshot=True,
+        name="report_email_chart_force_screenshot",
     )
     yield report_schedule
 
@@ -238,6 +245,7 @@ def create_report_email_chart_with_csv():
         email_target="target@email.com",
         chart=chart,
         report_format=ReportDataFormat.CSV,
+        name="report_email_chart_with_csv",
     )
     yield report_schedule
     cleanup_report_schedule(report_schedule)
@@ -251,6 +259,7 @@ def create_report_email_chart_with_text():
         email_target="target@email.com",
         chart=chart,
         report_format=ReportDataFormat.TEXT,
+        name="report_email_chart_with_text",
     )
     yield report_schedule
     cleanup_report_schedule(report_schedule)
@@ -274,7 +283,9 @@ def create_report_email_chart_with_csv_no_query_context():
 def create_report_email_dashboard():
     dashboard = db.session.query(Dashboard).first()
     report_schedule = create_report_notification(
-        email_target="target@email.com", dashboard=dashboard
+        email_target="target@email.com",
+        dashboard=dashboard,
+        name="report_email_dashboard",
     )
     yield report_schedule
 
@@ -285,7 +296,10 @@ def create_report_email_dashboard():
 def create_report_email_dashboard_force_screenshot():
     dashboard = db.session.query(Dashboard).first()
     report_schedule = create_report_notification(
-        email_target="target@email.com", dashboard=dashboard, force_screenshot=True
+        email_target="target@email.com",
+        dashboard=dashboard,
+        force_screenshot=True,
+        name="report_email_dashboard_force_screenshot",
     )
     yield report_schedule
 
@@ -296,7 +310,7 @@ def create_report_email_dashboard_force_screenshot():
 def create_report_slack_chart():
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
-        slack_channel="slack_channel", chart=chart
+        slack_channel="slack_channel", chart=chart, name="report_slack_chart"
     )
     yield report_schedule
 
@@ -325,6 +339,7 @@ def create_report_slack_chart_with_csv():
         slack_channel="slack_channel",
         chart=chart,
         report_format=ReportDataFormat.CSV,
+        name="report_slack_chart_with_csv",
     )
     yield report_schedule
 
@@ -339,6 +354,7 @@ def create_report_slack_chart_with_text():
         slack_channel="slack_channel",
         chart=chart,
         report_format=ReportDataFormat.TEXT,
+        name="report_slack_chart_with_text",
     )
     yield report_schedule
 
@@ -349,7 +365,7 @@ def create_report_slack_chart_with_text():
 def create_report_slack_chart_working():
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
-        slack_channel="slack_channel", chart=chart
+        slack_channel="slack_channel", chart=chart, name="report_slack_chart_working"
     )
     report_schedule.last_state = ReportState.WORKING
     report_schedule.last_eval_dttm = datetime(2020, 1, 1, 0, 0)
@@ -381,6 +397,7 @@ def create_alert_slack_chart_success():
         slack_channel="slack_channel",
         chart=chart,
         report_type=ReportScheduleType.ALERT,
+        name="alert_slack_chart_success",
     )
     report_schedule.last_state = ReportState.SUCCESS
     report_schedule.last_eval_dttm = datetime(2020, 1, 1, 0, 0)
@@ -423,6 +440,7 @@ def create_alert_slack_chart_grace(request):
             sql=param_config[request.param]["sql"],
             validator_type=param_config[request.param]["validator_type"],
             validator_config_json=param_config[request.param]["validator_config_json"],
+            name=f"alert_slack_chart_grace_{request.param}",
         )
         report_schedule.last_state = ReportState.GRACE
         report_schedule.last_eval_dttm = datetime(2020, 1, 1, 0, 0)
@@ -508,6 +526,7 @@ def create_alert_email_chart(request):
             validator_type=param_config[request.param]["validator_type"],
             validator_config_json=param_config[request.param]["validator_config_json"],
             force_screenshot=True,
+            name=f"alert_email_chart_{request.param}",
         )
         yield report_schedule
 
@@ -586,6 +605,7 @@ def create_no_alert_email_chart(request):
             sql=param_config[request.param]["sql"],
             validator_type=param_config[request.param]["validator_type"],
             validator_config_json=param_config[request.param]["validator_config_json"],
+            name=f"no_alert_email_chart_{request.param}",
         )
         yield report_schedule
 
@@ -617,6 +637,7 @@ def create_mul_alert_email_chart(request):
             sql=param_config[request.param]["sql"],
             validator_type=param_config[request.param]["validator_type"],
             validator_config_json=param_config[request.param]["validator_config_json"],
+            name=f"mul_alert_email_chart_{request.param}",
         )
         yield report_schedule
 
@@ -649,6 +670,7 @@ def create_invalid_sql_alert_email_chart(request, app_context: AppContext):
             validator_type=param_config[request.param]["validator_type"],
             validator_config_json=param_config[request.param]["validator_config_json"],
             grace_period=60 * 60,
+            name=f"invalid_sql_alert_email_chart_{request.param}",
         )
         yield report_schedule
 
@@ -1213,6 +1235,7 @@ def test_email_dashboard_report_schedule_with_tab_anchor(
                         "urlParams": [["native_filters", "()"]],
                     }
                 },
+                name="dashboard_report_with_tab_anchor",
             )
             AsyncExecuteReportScheduleCommand(
                 TEST_ID, report_schedule.id, datetime.utcnow()
@@ -1269,6 +1292,7 @@ def test_email_dashboard_report_schedule_disabled_tabs(
                         "urlParams": [["native_filters", "()"]],
                     }
                 },
+                name="dashboard_report_disabled_tabs",
             )
             AsyncExecuteReportScheduleCommand(
                 TEST_ID, report_schedule.id, datetime.utcnow()
@@ -1341,14 +1365,18 @@ def test_slack_chart_report_schedule_converts_to_v2(
     # setup screenshot mock
     screenshot_mock.return_value = SCREENSHOT_FILE
     channel_id = "slack_channel_id"
-    get_channels_with_search_mock.return_value = [
+    get_channels_with_search_mock.return_value = {
-        {
+        "result": [
-            "id": channel_id,
+            {
-            "name": "slack_channel",
+                "id": channel_id,
-            "is_member": True,
+                "name": "slack_channel",
-            "is_private": False,
+                "is_member": True,
-        },
+                "is_private": False,
-    ]
+            },
+        ],
+        "next_cursor": None,
+        "has_more": False,
+    }
 
     with freeze_time("2020-01-01T00:00:00Z"):
         with patch(
@@ -1404,16 +1432,22 @@ def test_slack_chart_report_schedule_converts_to_v2_channel_with_hash(
     channel_id = "slack_channel_id"
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
-        slack_channel="#slack_channel", chart=chart
+        slack_channel="#slack_channel",
+        chart=chart,
+        name="slack_converts_to_v2_with_hash",
     )
-    get_channels_with_search_mock.return_value = [
+    get_channels_with_search_mock.return_value = {
-        {
+        "result": [
-            "id": channel_id,
+            {
-            "name": "slack_channel",
+                "id": channel_id,
-            "is_member": True,
+                "name": "slack_channel",
-            "is_private": False,
+                "is_member": True,
-        },
+                "is_private": False,
-    ]
+            },
+        ],
+        "next_cursor": None,
+        "has_more": False,
+    }
 
     with freeze_time("2020-01-01T00:00:00Z"):
         with patch(
@@ -1467,16 +1501,22 @@ def test_slack_chart_report_schedule_fails_to_converts_to_v2(
     channel_id = "slack_channel_id"
     chart = db.session.query(Slice).first()
     report_schedule = create_report_notification(
-        slack_channel="#slack_channel,my_member_ID", chart=chart
+        slack_channel="#slack_channel,my_member_ID",
+        chart=chart,
+        name="slack_fails_to_convert_to_v2",
     )
-    get_channels_with_search_mock.return_value = [
+    get_channels_with_search_mock.return_value = {
-        {
+        "result": [
-            "id": channel_id,
+            {
-            "name": "slack_channel",
+                "id": channel_id,
-            "is_member": True,
+                "name": "slack_channel",
-            "is_private": False,
+                "is_member": True,
-        },
+                "is_private": False,
-    ]
+            },
+        ],
+        "next_cursor": None,
+        "has_more": False,
+    }
 
     with pytest.raises(ReportScheduleSystemErrorsException):
         AsyncExecuteReportScheduleCommand(

tests/unit_tests/commands/report/execute_test.py

@@ -515,20 +515,24 @@ def test_update_recipient_to_slack_v2(mocker: MockerFixture):
     """
     mocker.patch(
         "superset.commands.report.execute.get_channels_with_search",
-        return_value=[
+        return_value={
-            {
+            "result": [
-                "id": "abc124f",
+                {
-                "name": "channel-1",
+                    "id": "abc124f",
-                "is_member": True,
+                    "name": "channel-1",
-                "is_private": False,
+                    "is_member": True,
-            },
+                    "is_private": False,
-            {
+                },
-                "id": "blah_!channel_2",
+                {
-                "name": "Channel_2",
+                    "id": "blah_!channel_2",
-                "is_member": True,
+                    "name": "Channel_2",
-                "is_private": False,
+                    "is_member": True,
-            },
+                    "is_private": False,
-        ],
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        },
     )
     mock_report_schedule = ReportSchedule(
         recipients=[
@@ -558,14 +562,18 @@ def test_update_recipient_to_slack_v2_missing_channels(mocker: MockerFixture):
     """
     mocker.patch(
         "superset.commands.report.execute.get_channels_with_search",
-        return_value=[
+        return_value={
-            {
+            "result": [
-                "id": "blah_!channel_2",
+                {
-                "name": "Channel 2",
+                    "id": "blah_!channel_2",
-                "is_member": True,
+                    "name": "Channel 2",
-                "is_private": False,
+                    "is_member": True,
-            },
+                    "is_private": False,
-        ],
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        },
     )
     mock_report_schedule = ReportSchedule(
         name="Test Report",

tests/unit_tests/utils/slack_test.py

@@ -34,7 +34,14 @@ class TestGetChannelsWithSearch:
     def test_fetch_all_channels_no_search_string(self, mocker):
         # Mock data
         mock_data = {
-            "channels": [{"name": "general", "id": "C12345"}],
+            "channels": [
+                {
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
             "response_metadata": {"next_cursor": None},
         }
 
@@ -46,12 +53,30 @@ class TestGetChannelsWithSearch:
         mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
 
         result = get_channels_with_search()
-        assert result == [{"name": "general", "id": "C12345"}]
+        assert result == {
+            "result": [
+                {
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
 
     # Handle an empty search string gracefully
     def test_handle_empty_search_string(self, mocker):
         mock_data = {
-            "channels": [{"name": "general", "id": "C12345"}],
+            "channels": [
+                {
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
             "response_metadata": {"next_cursor": None},
         }
 
@@ -61,15 +86,41 @@ class TestGetChannelsWithSearch:
         mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
 
         result = get_channels_with_search(search_string="")
-        assert result == [{"name": "general", "id": "C12345"}]
+        assert result == {
+            "result": [
+                {
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
 
     def test_handle_exact_match_search_string_single_channel(self, mocker):
         # Mock data with multiple channels
         mock_data = {
             "channels": [
-                {"name": "general", "id": "C12345"},
+                {
-                {"name": "general2", "id": "C13454"},
+                    "name": "general",
-                {"name": "random", "id": "C67890"},
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "general2",
+                    "id": "C13454",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "random",
+                    "id": "C67890",
+                    "is_private": False,
+                    "is_member": True,
+                },
             ],
             "response_metadata": {"next_cursor": None},
         }
@@ -81,17 +132,45 @@ class TestGetChannelsWithSearch:
         mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
 
         # Call the function with a search string that matches a single channel
-        result = get_channels_with_search(search_string="general", exact_match=True)
+        result = get_channels_with_search(
+            search_string="general", exact_match=True, limit=100
+        )
 
-        # Assert that the result is a list with a single channel dictionary
+        # Assert that the result is a dict with proper structure
-        assert result == [{"name": "general", "id": "C12345"}]
+        assert result == {
+            "result": [
+                {
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
 
     def test_handle_exact_match_search_string_multiple_channels(self, mocker):
         mock_data = {
             "channels": [
-                {"name": "general", "id": "C12345"},
+                {
-                {"name": "general2", "id": "C13454"},
+                    "name": "general",
-                {"name": "random", "id": "C67890"},
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "general2",
+                    "id": "C13454",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "random",
+                    "id": "C67890",
+                    "is_private": False,
+                    "is_member": True,
+                },
             ],
             "response_metadata": {"next_cursor": None},
         }
@@ -102,19 +181,42 @@ class TestGetChannelsWithSearch:
         mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
 
         result = get_channels_with_search(
-            search_string="general,random", exact_match=True
+            search_string="general", exact_match=True, limit=100
         )
-        assert result == [
+        assert result == {
-            {"name": "general", "id": "C12345"},
+            "result": [
-            {"name": "random", "id": "C67890"},
+                {
-        ]
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
 
     def test_handle_loose_match_search_string_multiple_channels(self, mocker):
         mock_data = {
             "channels": [
-                {"name": "general", "id": "C12345"},
+                {
-                {"name": "general2", "id": "C13454"},
+                    "name": "general",
-                {"name": "random", "id": "C67890"},
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "general2",
+                    "id": "C13454",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "random",
+                    "id": "C67890",
+                    "is_private": False,
+                    "is_member": True,
+                },
             ],
             "response_metadata": {"next_cursor": None},
         }
@@ -124,12 +226,25 @@ class TestGetChannelsWithSearch:
         mock_client.conversations_list.return_value = mock_response_instance
         mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
 
-        result = get_channels_with_search(search_string="general,random")
+        result = get_channels_with_search(search_string="general", limit=100)
-        assert result == [
+        assert result == {
-            {"name": "general", "id": "C12345"},
+            "result": [
-            {"name": "general2", "id": "C13454"},
+                {
-            {"name": "random", "id": "C67890"},
+                    "name": "general",
-        ]
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "general2",
+                    "id": "C13454",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
 
     def test_handle_slack_client_error_listing_channels(self, mocker):
         from slack_sdk.errors import SlackApiError
@@ -165,21 +280,30 @@ The server responded with: missing scope: channels:read"""
     def test_filter_channels_by_specified_types(
         self, types: list[SlackChannelTypes], expected_channel_ids: set[str], mocker
     ):
+        # Determine which channels to return based on types parameter
+        public_channel = {
+            "id": "public_channel_id",
+            "name": "open",
+            "is_member": False,
+            "is_private": False,
+        }
+        private_channel = {
+            "id": "private_channel_id",
+            "name": "secret",
+            "is_member": False,
+            "is_private": True,
+        }
+
+        # Mock should return channels matching the requested types
+        # (simulating Slack API's type filtering)
+        channels = []
+        if not types or SlackChannelTypes.PUBLIC in types:
+            channels.append(public_channel)
+        if not types or SlackChannelTypes.PRIVATE in types:
+            channels.append(private_channel)
+
         mock_data = {
-            "channels": [
+            "channels": channels,
-                {
-                    "id": "public_channel_id",
-                    "name": "open",
-                    "is_member": False,
-                    "is_private": False,
-                },
-                {
-                    "id": "private_channel_id",
-                    "name": "secret",
-                    "is_member": False,
-                    "is_private": True,
-                },
-            ],
             "response_metadata": {"next_cursor": None},
         }
 
@@ -189,15 +313,93 @@ The server responded with: missing scope: channels:read"""
         mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
 
         result = get_channels_with_search(types=types)
-        assert {channel["id"] for channel in result} == expected_channel_ids
+        assert {channel["id"] for channel in result["result"]} == expected_channel_ids
 
-    def test_handle_pagination_multiple_pages(self, mocker):
+    def test_handle_pagination_without_search(self, mocker):
+        """Test pagination returns single page with cursor"""
         mock_data_page1 = {
-            "channels": [{"name": "general", "id": "C12345"}],
+            "channels": [
+                {
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
+            "response_metadata": {"next_cursor": "page2_cursor"},
+        }
+
+        mock_response_instance_page1 = MockResponse(mock_data_page1)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance_page1
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(limit=100)
+        assert result == {
+            "result": [
+                {
+                    "name": "general",
+                    "id": "C12345",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
+            "next_cursor": "page2_cursor",
+            "has_more": True,
+        }
+
+    def test_handle_pagination_with_cursor(self, mocker):
+        """Test pagination with cursor fetches next page"""
+        mock_data_page2 = {
+            "channels": [
+                {
+                    "name": "random",
+                    "id": "C67890",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response_instance_page2 = MockResponse(mock_data_page2)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response_instance_page2
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(cursor="page2_cursor", limit=100)
+        assert result == {
+            "result": [
+                {
+                    "name": "random",
+                    "id": "C67890",
+                    "is_private": False,
+                    "is_member": True,
+                }
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+    def test_streaming_search_pagination(self, mocker):
+        """Test search mode streams through pages until limit is reached"""
+        mock_data_page1 = {
+            "channels": [
+                {"name": "general", "id": "C1", "is_private": False, "is_member": True},
+                {"name": "random", "id": "C2", "is_private": False, "is_member": True},
+            ],
             "response_metadata": {"next_cursor": "page2"},
         }
         mock_data_page2 = {
-            "channels": [{"name": "random", "id": "C67890"}],
+            "channels": [
+                {
+                    "name": "general-2",
+                    "id": "C3",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {"name": "other", "id": "C4", "is_private": False, "is_member": True},
+            ],
             "response_metadata": {"next_cursor": None},
         }
 
@@ -211,8 +413,517 @@ The server responded with: missing scope: channels:read"""
         ]
         mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
 
-        result = get_channels_with_search()
+        # Search for "general" - should find 2 channels across 2 pages
-        assert result == [
+        result = get_channels_with_search(search_string="general", limit=100)
-            {"name": "general", "id": "C12345"},
+        assert result == {
-            {"name": "random", "id": "C67890"},
+            "result": [
+                {"name": "general", "id": "C1", "is_private": False, "is_member": True},
+                {
+                    "name": "general-2",
+                    "id": "C3",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+    def test_search_with_no_matches(self, mocker):
+        """Test search that finds no matching channels"""
+
+        mock_data = {
+            "channels": [
+                {"name": "general", "id": "C1", "is_private": False, "is_member": True},
+                {"name": "random", "id": "C2", "is_private": False, "is_member": True},
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+        mock_response = MockResponse(mock_data)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Search for non-existent channel
+        result = get_channels_with_search(search_string="nonexistent", limit=100)
+
+        assert result == {
+            "result": [],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+    def test_search_returns_exactly_limit(self, mocker):
+        """Test search that returns exactly the requested limit"""
+
+        # Create 100 matching channels
+        channels = [
+            {"name": f"test-{i}", "id": f"C{i}", "is_private": False, "is_member": True}
+            for i in range(100)
         ]
+
+        mock_data = {
+            "channels": channels,
+            "response_metadata": {"next_cursor": "next_page"},
+        }
+        mock_response = MockResponse(mock_data)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Search that matches all channels
+        result = get_channels_with_search(search_string="test", limit=100)
+
+        # Should return exactly 100 channels
+        assert len(result["result"]) == 100
+        # Should indicate more results available
+        assert result["has_more"] is True
+        assert result["next_cursor"] == "next_page"
+
+    def test_partial_page_results(self, mocker):
+        """Test pagination with partial page (less than limit)"""
+
+        # Only 50 channels returned (less than default 100 limit)
+        channels = [
+            {
+                "name": f"channel-{i}",
+                "id": f"C{i}",
+                "is_private": False,
+                "is_member": True,
+            }
+            for i in range(50)
+        ]
+
+        mock_data = {
+            "channels": channels,
+            "response_metadata": {"next_cursor": None},
+        }
+        mock_response = MockResponse(mock_data)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(limit=100)
+
+        # Should return all 50 channels
+        assert len(result["result"]) == 50
+        # Should indicate no more results
+        assert result["has_more"] is False
+        assert result["next_cursor"] is None
+
+    def test_streaming_search_stops_when_limit_reached(self, mocker):
+        """Test that streaming search stops immediately when limit is reached"""
+
+        # First page with 60 matching channels
+        page1_channels = [
+            {"name": f"test-{i}", "id": f"C{i}", "is_private": False, "is_member": True}
+            for i in range(60)
+        ]
+        # Second page with 60 more matching channels (should not be fully processed)
+        page2_channels = [
+            {
+                "name": f"test-{i}",
+                "id": f"C{i + 60}",
+                "is_private": False,
+                "is_member": True,
+            }
+            for i in range(60)
+        ]
+
+        mock_data_page1 = {
+            "channels": page1_channels,
+            "response_metadata": {"next_cursor": "page2"},
+        }
+        mock_data_page2 = {
+            "channels": page2_channels,
+            "response_metadata": {"next_cursor": "page3"},
+        }
+
+        mock_response_page1 = MockResponse(mock_data_page1)
+        mock_response_page2 = MockResponse(mock_data_page2)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = [
+            mock_response_page1,
+            mock_response_page2,
+        ]
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Request limit of 100
+        result = get_channels_with_search(search_string="test", limit=100)
+
+        # Should return exactly 100 channels (60 from page1 + 40 from page2)
+        assert len(result["result"]) == 100
+        # Should indicate more results available (next cursor points to page3)
+        assert result["has_more"] is True
+        assert result["next_cursor"] == "page3"
+
+    def test_cursor_format_with_special_characters(self, mocker):
+        """Test that cursor with special characters is handled correctly"""
+
+        # Slack cursors are base64 encoded strings that might contain special chars
+        special_cursor = "dGVhbTpDMDYxRkE1UEw="
+
+        mock_data = {
+            "channels": [
+                {"name": "test", "id": "C123", "is_private": False, "is_member": True},
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+        mock_response = MockResponse(mock_data)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Call with special cursor
+        get_channels_with_search(cursor=special_cursor, limit=100)
+
+        # Verify cursor was passed to Slack API
+        mock_client.conversations_list.assert_called_once()
+        call_kwargs = mock_client.conversations_list.call_args[1]
+        assert call_kwargs["cursor"] == special_cursor
+
+    def test_empty_channel_list_response(self, mocker):
+        """Test handling of empty channels list from Slack API"""
+
+        mock_data = {
+            "channels": [],
+            "response_metadata": {"next_cursor": None},
+        }
+        mock_response = MockResponse(mock_data)
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search()
+
+        assert result == {
+            "result": [],
+            "next_cursor": None,
+            "has_more": False,
+        }
+
+    def test_custom_limit_parameter(self, mocker):
+        """Test that custom limit parameter is respected"""
+
+        all_channels = [
+            {
+                "name": f"channel-{i}",
+                "id": f"C{i}",
+                "is_private": False,
+                "is_member": True,
+            }
+            for i in range(200)
+        ]
+
+        # Mock should respect the limit parameter (simulating Slack API behavior)
+        def mock_conversations_list(**kwargs):
+            limit = kwargs.get("limit", 100)
+            return MockResponse(
+                {
+                    "channels": all_channels[:limit],
+                    "response_metadata": {"next_cursor": "next_page"},
+                }
+            )
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = mock_conversations_list
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Request custom limit of 50
+        result = get_channels_with_search(limit=50)
+
+        # Should return exactly 50 channels
+        assert len(result["result"]) == 50
+        assert result["has_more"] is True
+
+    def test_non_search_pagination_over_1000_limit(self, mocker):
+        """Test non-search queries paginate correctly for limits > 1000"""
+        # Create 2500 channels
+        all_channels = [
+            {
+                "name": f"channel-{i}",
+                "id": f"C{i}",
+                "is_private": False,
+                "is_member": True,
+            }
+            for i in range(2500)
+        ]
+
+        call_count = 0
+
+        def mock_conversations_list(**kwargs):
+            nonlocal call_count
+            limit = kwargs.get("limit", 100)
+            cursor = kwargs.get("cursor")
+
+            # Simulate Slack API pagination (max 1000 per page)
+            if cursor is None:
+                start = 0
+            elif cursor == "cursor_1000":
+                start = 1000
+            elif cursor == "cursor_2000":
+                start = 2000
+            else:
+                start = 3000
+
+            end = min(start + limit, 2500)
+            next_cursor = f"cursor_{end}" if end < 2500 else None
+
+            call_count += 1
+            return MockResponse(
+                {
+                    "channels": all_channels[start:end],
+                    "response_metadata": {"next_cursor": next_cursor},
+                }
+            )
+
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.side_effect = mock_conversations_list
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Request 1500 channels (requires 2 pages of 1000 each)
+        result = get_channels_with_search(limit=1500)
+
+        # Should return exactly 1500 channels
+        assert len(result["result"]) == 1500
+        assert result["has_more"] is True
+        assert result["next_cursor"] == "cursor_2000"
+        # Should have made 2 API calls
+        assert call_count == 2
+
+    def test_search_with_exact_match_optimization(self, mocker):
+        """Test exact match search uses optimized string comparison"""
+        channels = [
+            {"name": "test", "id": "C1", "is_private": False, "is_member": True},
+            {"name": "test-dev", "id": "C2", "is_private": False, "is_member": True},
+            {"name": "testing", "id": "C3", "is_private": False, "is_member": True},
+        ]
+        mock_data = {"channels": channels, "response_metadata": {"next_cursor": None}}
+
+        mock_response = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(
+            search_string="TEST", exact_match=True, limit=100
+        )
+
+        # Only "test" should match (case-insensitive exact match)
+        assert len(result["result"]) == 1
+        assert result["result"][0]["name"] == "test"
+
+    def test_search_substring_match_optimization(self, mocker):
+        """Test substring search uses optimized string comparison"""
+        channels = [
+            {"name": "prod-api", "id": "C1", "is_private": False, "is_member": True},
+            {"name": "dev-api", "id": "C2", "is_private": False, "is_member": True},
+            {"name": "staging", "id": "C3", "is_private": False, "is_member": True},
+        ]
+        mock_data = {"channels": channels, "response_metadata": {"next_cursor": None}}
+
+        mock_response = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(search_string="API", limit=100)
+
+        # Both "prod-api" and "dev-api" should match (case-insensitive)
+        assert len(result["result"]) == 2
+        assert {ch["name"] for ch in result["result"]} == {"prod-api", "dev-api"}
+
+    def test_search_by_channel_id(self, mocker):
+        """Test search can match by channel ID"""
+        channels = [
+            {"name": "general", "id": "C12345", "is_private": False, "is_member": True},
+            {"name": "random", "id": "C67890", "is_private": False, "is_member": True},
+        ]
+        mock_data = {"channels": channels, "response_metadata": {"next_cursor": None}}
+
+        mock_response = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(
+            search_string="c12345", exact_match=True, limit=100
+        )
+
+        # Should match by ID (case-insensitive)
+        assert len(result["result"]) == 1
+        assert result["result"][0]["id"] == "C12345"
+
+    def test_non_search_empty_result_handling(self, mocker):
+        """Test non-search query handles empty channel list"""
+        mock_data = {
+            "channels": [],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        result = get_channels_with_search(limit=100)
+
+        assert len(result["result"]) == 0
+        assert result["has_more"] is False
+        assert result["next_cursor"] is None
+
+    def test_comma_separated_search_strings(self, mocker):
+        """Test search with comma-separated search strings (OR logic)"""
+        mock_data = {
+            "channels": [
+                {
+                    "name": "engineering",
+                    "id": "C1",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "marketing",
+                    "id": "C2",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "sales",
+                    "id": "C3",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "design",
+                    "id": "C4",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "general",
+                    "id": "C5",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Search for "engineering,marketing,sales"
+        result = get_channels_with_search(
+            search_string="engineering,marketing,sales", limit=100
+        )
+
+        # Should match 3 channels: engineering, marketing, sales
+        assert len(result["result"]) == 3
+        channel_names = [channel["name"] for channel in result["result"]]
+        assert "engineering" in channel_names
+        assert "marketing" in channel_names
+        assert "sales" in channel_names
+        assert "design" not in channel_names
+        assert "general" not in channel_names
+
+    def test_comma_separated_search_with_whitespace(self, mocker):
+        """Test comma-separated search handles extra whitespace correctly"""
+        mock_data = {
+            "channels": [
+                {
+                    "name": "engineering-team",
+                    "id": "C1",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "marketing-ops",
+                    "id": "C2",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "general",
+                    "id": "C3",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Search with extra whitespace: " engineering , marketing "
+        result = get_channels_with_search(
+            search_string=" engineering , marketing ", limit=100
+        )
+
+        # Should match 2 channels, whitespace should be stripped
+        assert len(result["result"]) == 2
+        channel_names = [channel["name"] for channel in result["result"]]
+        assert "engineering-team" in channel_names
+        assert "marketing-ops" in channel_names
+        assert "general" not in channel_names
+
+    def test_comma_separated_exact_match(self, mocker):
+        """Test comma-separated search with exact_match=True"""
+        mock_data = {
+            "channels": [
+                {
+                    "name": "engineering",
+                    "id": "C1",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "engineering-team",
+                    "id": "C2",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "sales",
+                    "id": "C3",
+                    "is_private": False,
+                    "is_member": True,
+                },
+                {
+                    "name": "general",
+                    "id": "C4",
+                    "is_private": False,
+                    "is_member": True,
+                },
+            ],
+            "response_metadata": {"next_cursor": None},
+        }
+
+        mock_response = MockResponse(mock_data)
+        mock_client = mocker.Mock()
+        mock_client.conversations_list.return_value = mock_response
+        mocker.patch("superset.utils.slack.get_slack_client", return_value=mock_client)
+
+        # Exact match search for "engineering,sales"
+        result = get_channels_with_search(
+            search_string="engineering,sales", exact_match=True, limit=100
+        )
+
+        # Should match only exact names: engineering and sales (not engineering-team)
+        assert len(result["result"]) == 2
+        channel_names = [channel["name"] for channel in result["result"]]
+        assert "engineering" in channel_names
+        assert "sales" in channel_names
+        assert "engineering-team" not in channel_names
+        assert "general" not in channel_names