Support dekaf materializations

src/api/connectors.ts

@@ -1,17 +1,17 @@
 import { supabaseClient } from 'context/GlobalProviders';
 import {
-    CONNECTOR_WITH_TAG_QUERY,
     ConnectorWithTagDetailQuery,
+    CONNECTOR_WITH_TAG_QUERY,
 } from 'hooks/connectors/shared';
 import {
+    CONNECTOR_DETAILS,
     CONNECTOR_NAME,
     CONNECTOR_RECOMMENDED,
-    TABLES,
     defaultTableFilter,
     handleFailure,
     handleSuccess,
     supabaseRetry,
-    CONNECTOR_DETAILS,
+    TABLES,
 } from 'services/supabase';
 import { SortDirection } from 'types';
 import { requiredConnectorColumnsExist } from 'utils/connector-utils';
@@ -57,14 +57,14 @@ export interface ConnectorTag_Base {
 export interface ConnectorsQuery_DetailsForm {
     id: string;
     image_name: string;
-    logo_url: string;
+    image: string;
     connector_tags: ConnectorTag_Base[];
 }
 
 const DETAILS_FORM_QUERY = `
     id,
     image_name,
-    logo_url:logo_url->>en-US::text,
+    image:logo_url->>en-US::text,
     connector_tags !inner(
         id,
         connector_id,

src/components/capture/useCaptureDraftUpdate.ts

@@ -23,6 +23,7 @@ import {
     useEndpointConfigStore_setEncryptedEndpointConfig,
     useEndpointConfigStore_setPreviousEndpointConfig,
 } from 'stores/EndpointConfig/hooks';
+import { isDekafConnector } from 'utils/connector-utils';
 
 import { modifyExistingCaptureDraftSpec } from 'utils/workflow-utils';
 
@@ -38,8 +39,10 @@ function useDiscoverDraftUpdate(options?: {
     const persistedDraftId = useEditorStore_persistedDraftId();
     const setDraftId = useEditorStore_setId();
 
-    const imagePath = useDetailsFormStore(
-        (state) => state.details.data.connectorImage.imagePath
+    const imagePath = useDetailsFormStore((state) =>
+        isDekafConnector(state.details.data.connectorImage)
+            ? undefined
+            : state.details.data.connectorImage.imagePath
     );
 
     const setEncryptedEndpointConfig =
@@ -67,10 +70,13 @@ function useDiscoverDraftUpdate(options?: {
             //      Doing a rediscovery with no new changes
 
             if (
-                (!draftId && persistedDraftId && !options?.initiateDiscovery) ||
-                (persistedDraftId &&
-                    !options?.initiateDiscovery &&
-                    !options?.initiateRediscovery)
+                imagePath &&
+                ((!draftId &&
+                    persistedDraftId &&
+                    !options?.initiateDiscovery) ||
+                    (persistedDraftId &&
+                        !options?.initiateDiscovery &&
+                        !options?.initiateRediscovery))
             ) {
                 const existingDraftSpecResponse =
                     await getDraftSpecsByCatalogName(

src/components/materialization/useGenerateCatalog.ts

@@ -30,22 +30,25 @@ import {
 } from 'stores/Binding/hooks';
 import { useDetailsFormStore } from 'stores/DetailsForm/Store';
 import {
-    useEndpointConfig_serverUpdateRequired,
     useEndpointConfigStore_encryptedEndpointConfig_data,
     useEndpointConfigStore_endpointConfig_data,
     useEndpointConfigStore_endpointSchema,
     useEndpointConfigStore_errorsExist,
     useEndpointConfigStore_setEncryptedEndpointConfig,
     useEndpointConfigStore_setPreviousEndpointConfig,
+    useEndpointConfig_serverUpdateRequired,
 } from 'stores/EndpointConfig/hooks';
 import {
     useFormStateStore_setFormState,
     useFormStateStore_updateStatus,
 } from 'stores/FormState/hooks';
 import { FormStatus } from 'stores/FormState/types';
 import { useSourceCaptureStore_sourceCaptureDefinition } from 'stores/SourceCapture/hooks';
+import { DekafConfig } from 'types';
+import { isDekafConnector } from 'utils/connector-utils';
 import { encryptEndpointConfig } from 'utils/sops-utils';
 import { generateTaskSpec } from 'utils/workflow-utils';
+import { ConnectorConfig } from '../../../deps/flow/flow';
 
 const ENTITY_TYPE = 'materialization';
 
@@ -63,8 +66,17 @@ function useGenerateCatalog() {
     const imageConnectorId = useDetailsFormStore(
         (state) => state.details.data.connectorImage.connectorId
     );
-    const imagePath = useDetailsFormStore(
-        (state) => state.details.data.connectorImage.imagePath
+    const endpointConfig: ConnectorConfig | DekafConfig = useDetailsFormStore(
+        (state) =>
+            isDekafConnector(state.details.data.connectorImage)
+                ? {
+                      config: {},
+                      variant: state.details.data.connectorImage.variant,
+                  }
+                : {
+                      config: {},
+                      image: state.details.data.connectorImage.imagePath,
+                  }
     );
     const setDraftedEntityName = useDetailsFormStore(
         (state) => state.setDraftedEntityName
@@ -209,12 +221,11 @@ function useGenerateCatalog() {
                     { overrideJsonFormDefaults: true }
                 );
 
+                endpointConfig.config = encryptedEndpointConfig.data;
+
                 const draftSpec = generateTaskSpec(
                     ENTITY_TYPE,
-                    {
-                        image: imagePath,
-                        config: encryptedEndpointConfig.data,
-                    },
+                    endpointConfig,
                     resourceConfigs,
                     resourceConfigServerUpdateRequired,
                     bindings,
@@ -279,8 +290,13 @@ function useGenerateCatalog() {
                 // Update all the store state
                 setCatalogName(processedEntityName);
                 setEncryptedEndpointConfig({
-                    data: draftSpecsResponse.data[0].spec.endpoint.connector
-                        .config,
+                    data: Object.hasOwn(
+                        draftSpecsResponse.data[0].spec.endpoint,
+                        'dekaf'
+                    )
+                        ? draftSpecsResponse.data[0].spec.endpoint.dekaf.config
+                        : draftSpecsResponse.data[0].spec.endpoint.connector
+                              .config,
                 });
                 setPreviousEndpointConfig({ data: endpointConfigData });
 
@@ -314,14 +330,14 @@ function useGenerateCatalog() {
             bindings,
             callFailed,
             detailsFormsErrorsExist,
+            endpointConfig,
             endpointConfigData,
             endpointConfigErrorsExist,
             endpointSchema,
             fullSourceConfigs,
             fullSourceErrorsExist,
             imageConnectorId,
             imageConnectorTagId,
-            imagePath,
             persistedDraftId,
             prefillBindingDependentState,
             processedEntityName,

src/components/shared/Entity/DetailsForm/Form.tsx

@@ -5,37 +5,19 @@ import { useEditorStore_isSaving } from 'components/editor/Store/hooks';
 import AlertBox from 'components/shared/AlertBox';
 import { Props } from 'components/shared/Entity/DetailsForm/types';
 import { CONNECTOR_IMAGE_SCOPE } from 'forms/renderers/Connectors';
-import { ConnectorWithTagDetailQuery } from 'hooks/connectors/shared';
 import { FormattedMessage, useIntl } from 'react-intl';
 import defaultRenderers from 'services/jsonforms/defaultRenderers';
 import { defaultOptions, showValidation } from 'services/jsonforms/shared';
 import { useDetailsFormStore } from 'stores/DetailsForm/Store';
-import { Details } from 'stores/DetailsForm/types';
 import {
     useFormStateStore_displayValidation,
     useFormStateStore_isActive,
     useFormStateStore_messagePrefix,
 } from 'stores/FormState/hooks';
-import { evaluateConnectorVersions } from 'utils/workflow-utils';
 import useFormFields from './useFormFields';
 
 export const CONFIG_EDITOR_ID = 'endpointConfigEditor';
 
-export const getConnectorImageDetails = (
-    connector: ConnectorWithTagDetailQuery,
-    options?: { connectorId: string; existingImageTag: string }
-): Details['data']['connectorImage'] => {
-    const connectorTag = evaluateConnectorVersions(connector, options);
-
-    return {
-        connectorId: connector.id,
-        id: connectorTag.id,
-        imageName: connector.image_name,
-        imagePath: `${connector.image_name}${connectorTag.image_tag}`,
-        iconPath: connector.image,
-    };
-};
-
 function DetailsFormForm({ connectorTags, entityType, readOnly }: Props) {
     const intl = useIntl();
 

src/components/shared/Entity/DetailsForm/useConnectorField.ts

@@ -6,18 +6,18 @@ import useGlobalSearchParams, {
 } from 'hooks/searchParams/useGlobalSearchParams';
 import { useCallback, useEffect, useMemo } from 'react';
 import { useIntl } from 'react-intl';
-import { useDetailsFormStore } from 'stores/DetailsForm/Store';
 import { useDetailsForm_changed_connectorId } from 'stores/DetailsForm/hooks';
+import { useDetailsFormStore } from 'stores/DetailsForm/Store';
 import { Details } from 'stores/DetailsForm/types';
 import { EntityWithCreateWorkflow } from 'types';
+import { getConnectorMetadata } from 'utils/connector-utils';
 import { hasLength } from 'utils/misc-utils';
 import {
     ConnectorVersionEvaluationOptions,
     evaluateConnectorVersions,
 } from 'utils/workflow-utils';
 import { MAC_ADDR_RE } from 'validation';
 import useEntityCreateNavigate from '../hooks/useEntityCreateNavigate';
-import { getConnectorImageDetails } from './Form';
 
 export default function useConnectorField(
     connectorTags: ConnectorWithTagDetailQuery[],
@@ -33,8 +33,8 @@ export default function useConnectorField(
     const originalConnectorImage = useDetailsFormStore(
         (state) => state.details.data.connectorImage
     );
-    const connectorImagePath = useDetailsFormStore(
-        (state) => state.details.data.connectorImage.imagePath
+    const connectorImageTag = useDetailsFormStore(
+        (state) => state.details.data.connectorImage.imageTag
     );
     const connectorIdChanged = useDetailsForm_changed_connectorId();
     const setDetails_connector = useDetailsFormStore(
@@ -53,7 +53,7 @@ export default function useConnectorField(
                     connectorTag.connector_id === connectorId;
 
                 if (connectorLocated) {
-                    setDetails_connector(getConnectorImageDetails(connector));
+                    setDetails_connector(getConnectorMetadata(connector));
                 }
 
                 return connectorLocated;
@@ -66,29 +66,25 @@ export default function useConnectorField(
         | undefined = useMemo(() => {
         // This is rare but can happen so being safe.
         // If you remove the connector id from the create URL
-        if (!connectorImagePath) {
+        if (!connectorImageTag) {
             return undefined;
         }
 
-        const imageTagStartIndex = connectorImagePath.indexOf(':');
-        return isEdit && hasLength(connectorId) && imageTagStartIndex > 0
+        return isEdit && hasLength(connectorId)
             ? {
                   connectorId,
-                  existingImageTag: connectorImagePath.substring(
-                      imageTagStartIndex,
-                      connectorImagePath.length
-                  ),
+                  existingImageTag: connectorImageTag,
               }
             : undefined;
-    }, [connectorId, connectorImagePath, isEdit]);
+    }, [connectorId, connectorImageTag, isEdit]);
 
     const connectorsOneOf = useMemo(() => {
         const response = [] as { title: string; const: Object }[];
 
         if (connectorTags.length > 0) {
             connectorTags.forEach((connector) => {
                 response.push({
-                    const: getConnectorImageDetails(
+                    const: getConnectorMetadata(
                         connector,
                         versionEvaluationOptions
                     ),

src/components/shared/Entity/EndpointConfig/index.tsx

@@ -253,6 +253,16 @@ function EndpointConfig({
                         </Box>
                     ) : null}
 
+                    {canBeEmpty ? (
+                        <Box sx={{ mb: 3 }}>
+                            <AlertBox severity="info" short>
+                                {intl.formatMessage({
+                                    id: 'workflows.alert.endpointConfigEmpty',
+                                })}
+                            </AlertBox>
+                        </Box>
+                    ) : null}
+
                     <EndpointConfigForm readOnly={readOnly} />
                 </ErrorBoundryWrapper>
             </WrapperWithHeader>

src/lang/en-US/Workflows.ts

@@ -202,6 +202,8 @@ export const Workflows: Record<string, string> = {
     'workflows.dataPlane.description': `Choose the data plane you would like to use.`,
     'workflows.dataPlane.label': `Data Plane`,
 
+    'workflows.alert.endpointConfigEmpty': `This endpoint requires configuration of the external system only.`,
+
     // Field Selection
     'fieldSelection.header': `Field Selection`,
     'fieldSelection.table.label': `Field Selection Editor`,

src/stores/DetailsForm/Store.ts

@@ -22,6 +22,7 @@ import {
     getInitialHydrationData,
     getStoreWithHydrationSettings,
 } from 'stores/extensions/Hydration';
+import { getConnectorMetadata } from 'utils/connector-utils';
 import {
     getDataPlaneScope,
     parseDataPlaneName,
@@ -30,10 +31,7 @@ import {
 import { defaultDataPlaneSuffix, isProduction } from 'utils/env-utils';
 import { hasLength } from 'utils/misc-utils';
 import { devtoolsOptions } from 'utils/store-utils';
-import {
-    ConnectorVersionEvaluationOptions,
-    evaluateConnectorVersions,
-} from 'utils/workflow-utils';
+import { ConnectorVersionEvaluationOptions } from 'utils/workflow-utils';
 import { NAME_RE } from 'validation';
 import { create, StoreApi } from 'zustand';
 import { devtools, NamedSet } from 'zustand/middleware';
@@ -48,20 +46,11 @@ const getConnectorImage = async (
 
     if (!error && data && data.length > 0) {
         const connector = data[0];
-        const { image_name, logo_url } = connector;
 
         const options: ConnectorVersionEvaluationOptions | undefined =
             existingImageTag ? { connectorId, existingImageTag } : undefined;
 
-        const connectorTag = evaluateConnectorVersions(connector, options);
-
-        return {
-            connectorId,
-            id: connectorTag.id,
-            imageName: image_name,
-            imagePath: `${image_name}${connectorTag.image_tag}`,
-            iconPath: logo_url,
-        };
+        return getConnectorMetadata(connector, options);
     }
 
     return null;
@@ -103,6 +92,7 @@ const initialDetails: Details = {
             iconPath: '',
             imageName: '',
             imagePath: '',
+            imageTag: '',
         },
         entityName: '',
     },
@@ -425,6 +415,7 @@ export const getInitialState = (
                     iconPath: '',
                     imageName: '',
                     imagePath: '',
+                    imageTag: '',
                     connectorId,
                 });
                 get().setHydrationErrorsExist(true);