feat: Pipeline 3000: Configuring new plugin and batch exports (#20561)

frontend/src/lib/api.ts

@@ -1490,6 +1490,9 @@ const api = {
         async update(id: PluginConfigTypeNew['id'], data: FormData): Promise<PluginConfigWithPluginInfoNew> {
             return await new ApiRequest().pluginConfig(id).update({ data })
         },
+        async create(data: FormData): Promise<PluginConfigWithPluginInfoNew> {
+            return await new ApiRequest().pluginConfigs().create({ data })
+        },
         async list(): Promise<PaginatedResponse<PluginConfigTypeNew>> {
             return await new ApiRequest().pluginConfigs().get()
         },

frontend/src/scenes/pipeline/PipelineNode.tsx

@@ -1,3 +1,4 @@
+import { Spinner } from '@posthog/lemon-ui'
 import { useValues } from 'kea'
 import { ActivityLog } from 'lib/components/ActivityLog/ActivityLog'
 import { NotFound } from 'lib/components/NotFound'
@@ -59,7 +60,16 @@ export function PipelineNode(params: { stage?: string; id?: string } = {}): JSX.
         return <NotFound object="pipeline app stage" />
     }
 
-    if (!nodeLoading && !node) {
+    if (nodeLoading) {
+        return <Spinner />
+    }
+
+    if (id === 'new') {
+        // If it's new we don't want to show any tabs
+        return <PipelineNodeConfiguration />
+    }
+
+    if (!node) {
         return <NotFound object={stage} />
     }
 

frontend/src/scenes/pipeline/PipelineNodeConfiguration.tsx

@@ -1,7 +1,8 @@
 import { IconLock } from '@posthog/icons'
-import { LemonButton, LemonSkeleton, Tooltip } from '@posthog/lemon-ui'
+import { LemonButton, LemonInput, LemonSelect, LemonSkeleton, Tooltip } from '@posthog/lemon-ui'
 import { useActions, useValues } from 'kea'
 import { Form } from 'kea-forms'
+import { NotFound } from 'lib/components/NotFound'
 import { LemonField } from 'lib/lemon-ui/LemonField'
 import { LemonMarkdown } from 'lib/lemon-ui/LemonMarkdown'
 import React from 'react'
@@ -10,17 +11,57 @@ import { BatchExportConfigurationForm } from 'scenes/batch_exports/batchExportEd
 import { getConfigSchemaArray, isValidField } from 'scenes/pipeline/configUtils'
 import { PluginField } from 'scenes/plugins/edit/PluginField'
 
+import { PluginType } from '~/types'
+
 import { pipelineNodeLogic } from './pipelineNodeLogic'
-import { PipelineBackend, PipelineNode } from './types'
 
 export function PipelineNodeConfiguration(): JSX.Element {
-    const { node, savedConfiguration, configuration, isConfigurationSubmitting, isConfigurable } =
-        useValues(pipelineNodeLogic)
-    const { resetConfiguration, submitConfiguration } = useActions(pipelineNodeLogic)
+    const {
+        stage,
+        node,
+        savedConfiguration,
+        configuration,
+        isConfigurationSubmitting,
+        isConfigurable,
+        newConfigurationPlugins,
+        newConfigurationBatchExports,
+        newConfigurationServiceOrPluginID,
+        isNew,
+        maybeNodePlugin,
+    } = useValues(pipelineNodeLogic)
+    const { resetConfiguration, submitConfiguration, setNewConfigurationServiceOrPluginID } =
+        useActions(pipelineNodeLogic)
+
+    let selector = <></>
+
+    if (isNew) {
+        if (!stage) {
+            return <NotFound object="pipeline app stage" />
+        }
+        const pluginsOptions = Object.values(newConfigurationPlugins).map((plugin) => ({
+            value: plugin.id,
+            label: plugin.name, // TODO: Ideally this would show RenderApp or MinimalAppView
+        }))
+        const batchExportsOptions = Object.entries(newConfigurationBatchExports).map(([key, name]) => ({
+            value: key,
+            label: name, // TODO: same render with a picture ?
+        }))
+
+        selector = (
+            <LemonSelect
+                value={newConfigurationServiceOrPluginID}
+                onChange={(newValue) => {
+                    setNewConfigurationServiceOrPluginID(newValue) // TODO: this should change the URL so we can link new specific plugin/batch export
+                }}
+                options={[...pluginsOptions, ...batchExportsOptions]}
+            />
+        )
+    }
 
     return (
         <div className="space-y-3">
-            {!node ? (
+            {selector}
+            {!node && !newConfigurationServiceOrPluginID ? (
                 Array(2)
                     .fill(null)
                     .map((_, index) => (
@@ -29,13 +70,29 @@ export function PipelineNodeConfiguration(): JSX.Element {
                             <LemonSkeleton className="h-9" />
                         </div>
                     ))
-            ) : isConfigurable ? (
+            ) : (
                 <>
                     <Form logic={pipelineNodeLogic} formKey="configuration" className="space-y-3">
-                        {node.backend === PipelineBackend.Plugin ? (
-                            <PluginConfigurationFields node={node} formValues={configuration} />
+                        <LemonField
+                            name="name"
+                            label="Name"
+                            info="Customising the name can be useful if multiple instances of the same type are used."
+                        >
+                            <LemonInput type="text" />
+                        </LemonField>
+                        <LemonField
+                            name="description"
+                            label="Description"
+                            info="Add a description to share context with other team members"
+                        >
+                            <LemonInput type="text" />
+                        </LemonField>
+                        {!isConfigurable ? (
+                            <span>This {stage} isn't configurable.</span>
+                        ) : maybeNodePlugin ? (
+                            <PluginConfigurationFields plugin={maybeNodePlugin} formValues={configuration} />
                         ) : (
-                            <BatchExportConfigurationFields node={node} formValues={configuration} />
+                            <BatchExportConfigurationFields isNew={isNew} formValues={configuration} />
                         )}
                         <div className="flex gap-2">
                             <LemonButton
@@ -44,7 +101,7 @@ export function PipelineNodeConfiguration(): JSX.Element {
                                 onClick={() => resetConfiguration(savedConfiguration || {})}
                                 disabledReason={isConfigurationSubmitting ? 'Saving in progress…' : undefined}
                             >
-                                Cancel
+                                {isNew ? 'Reset' : 'Cancel'}
                             </LemonButton>
                             <LemonButton
                                 type="primary"
@@ -52,27 +109,20 @@ export function PipelineNodeConfiguration(): JSX.Element {
                                 onClick={submitConfiguration}
                                 loading={isConfigurationSubmitting}
                             >
-                                Save
+                                {isNew ? 'Create' : 'Save'}
                             </LemonButton>
                         </div>
                     </Form>
                 </>
-            ) : (
-                <span>This {node.stage} isn't configurable.</span>
             )}
         </div>
     )
 }
 
-function PluginConfigurationFields({
-    node,
-}: {
-    node: PipelineNode & { backend: PipelineBackend.Plugin }
-    formValues: Record<string, any>
-}): JSX.Element {
+function PluginConfigurationFields({ plugin }: { plugin: PluginType; formValues: Record<string, any> }): JSX.Element {
     const { hiddenFields, requiredFields } = useValues(pipelineNodeLogic)
 
-    const configSchemaArray = getConfigSchemaArray(node.plugin.config_schema)
+    const configSchemaArray = getConfigSchemaArray(plugin.config_schema)
     const fields = configSchemaArray.map((fieldConfig, index) => (
         <React.Fragment key={fieldConfig.key || `__key__${index}`}>
             {fieldConfig.key &&
@@ -116,14 +166,15 @@ function PluginConfigurationFields({
 }
 
 function BatchExportConfigurationFields({
+    isNew,
     formValues,
 }: {
-    node: PipelineNode & { backend: PipelineBackend.BatchExport }
+    isNew: boolean
     formValues: Record<string, any>
 }): JSX.Element {
     return (
         <BatchExportsEditFields
-            isNew={false /* TODO */}
+            isNew={isNew}
             isPipeline
             batchExportConfigForm={formValues as BatchExportConfigurationForm}
         />

frontend/src/scenes/pipeline/pipelineNodeLogic.tsx

@@ -1,4 +1,4 @@
-import { actions, afterMount, kea, key, listeners, path, props, reducers, selectors } from 'kea'
+import { actions, afterMount, connect, kea, key, listeners, path, props, reducers, selectors } from 'kea'
 import { forms } from 'kea-forms'
 import { loaders } from 'kea-loaders'
 import { actionToUrl, urlToAction } from 'kea-router'
@@ -6,9 +6,10 @@ import api from 'lib/api'
 import { capitalizeFirstLetter } from 'lib/utils'
 import { batchExportFormFields } from 'scenes/batch_exports/batchExportEditLogic'
 import { Scene } from 'scenes/sceneTypes'
+import { teamLogic } from 'scenes/teamLogic'
 import { urls } from 'scenes/urls'
 
-import { Breadcrumb, PipelineNodeTab, PipelineStage } from '~/types'
+import { Breadcrumb, PipelineNodeTab, PipelineStage, PluginType } from '~/types'
 
 import {
     defaultConfigForPlugin,
@@ -17,7 +18,11 @@ import {
     getConfigSchemaArray,
     getPluginConfigFormData,
 } from './configUtils'
+import { pipelineDestinationsLogic } from './destinationsLogic'
+import { frontendAppsLogic } from './frontendAppsLogic'
+import { importAppsLogic } from './importAppsLogic'
 import type { pipelineNodeLogicType } from './pipelineNodeLogicType'
+import { pipelineTransformationsLogic } from './transformationsLogic'
 import {
     BatchExportBasedStep,
     convertToPipelineNode,
@@ -42,35 +47,57 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
     props({} as PipelineNodeLogicProps),
     key(({ id }) => id),
     path((id) => ['scenes', 'pipeline', 'pipelineNodeLogic', id]),
+    connect(() => ({
+        values: [
+            teamLogic,
+            ['currentTeamId'],
+            pipelineDestinationsLogic,
+            ['plugins as destinationPlugins'],
+            pipelineTransformationsLogic,
+            ['plugins as transformationPlugins'],
+            frontendAppsLogic,
+            ['plugins as frontendAppsPlugins'],
+            importAppsLogic,
+            ['plugins as importAppsPlugins'],
+        ],
+    })),
     actions({
         setCurrentTab: (tab: PipelineNodeTab = PipelineNodeTab.Configuration) => ({ tab }),
         loadNode: true,
         updateNode: (payload: PluginUpdatePayload | BatchExportUpdatePayload) => ({
             payload,
         }),
+        createNode: (payload: PluginUpdatePayload | BatchExportUpdatePayload) => ({
+            payload,
+        }),
+        setNewConfigurationServiceOrPluginID: (id: number | string | null) => ({ id }),
     }),
-    reducers({
+    reducers(() => ({
         currentTab: [
             PipelineNodeTab.Configuration as PipelineNodeTab,
             {
                 setCurrentTab: (_, { tab }) => tab,
             },
         ],
-    }),
+        newConfigurationServiceOrPluginID: [
+            // TODO: this doesn't clear properly if I exit out of the page and more importantly switching to a different stage
+            null as null | number | string,
+            {
+                setNewConfigurationServiceOrPluginID: (_, { id }) => id,
+            },
+        ],
+    })),
     loaders(({ props, values }) => ({
         node: [
             null as PipelineNode | null,
             {
                 loadNode: async (_, breakpoint) => {
-                    if (!props.stage) {
+                    if (!props.stage || props.id === 'new') {
                         return null
                     }
                     let node: PipelineNode | null = null
                     try {
                         if (typeof props.id === 'string') {
-                            if (props.stage !== PipelineStage.Destination) {
-                                return null
-                            }
                             const batchExport = await api.batchExports.get(props.id)
                             node = convertToPipelineNode(batchExport, props.stage)
                         } else {
@@ -85,6 +112,33 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
                     breakpoint()
                     return node
                 },
+                createNode: async ({ payload }) => {
+                    if (!values.newConfigurationServiceOrPluginID || !values.stage) {
+                        return null
+                    }
+                    if (values.nodeBackend === PipelineBackend.BatchExport) {
+                        payload = payload as BatchExportUpdatePayload
+                        const batchExport = await api.batchExports.create({
+                            paused: !payload.enabled,
+                            name: payload.name,
+                            interval: payload.interval,
+                            destination: payload.service,
+                        })
+                        return convertToPipelineNode(batchExport, values.stage)
+                    } else if (values.maybeNodePlugin) {
+                        payload = payload as PluginUpdatePayload
+                        const formdata = getPluginConfigFormData(
+                            values.maybeNodePlugin.config_schema,
+                            defaultConfigForPlugin(values.maybeNodePlugin),
+                            { ...payload, enabled: true } // Default enable on creation
+                        )
+                        formdata.append('plugin', values.maybeNodePlugin.id.toString())
+                        formdata.append('order', '99') // TODO: fix this should be at the end of latest here for transformations
+                        const pluginConfig = await api.pluginConfigs.create(formdata)
+                        return convertToPipelineNode(pluginConfig, values.stage)
+                    }
+                    return null
+                },
                 updateNode: async ({ payload }) => {
                     if (!values.node) {
                         return null
@@ -112,7 +166,7 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
     })),
     forms(({ props, values, asyncActions }) => ({
         configuration: {
-            defaults: {} as Record<string, any>,
+            defaults: { name: '', description: '' } as Record<string, any>,
             errors: (form) => {
                 if (values.nodeBackend === PipelineBackend.BatchExport) {
                     return batchExportFormFields(props.id === 'new', form as any, { isPipeline: true })
@@ -126,12 +180,18 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
                 }
             },
             submit: async (formValues) => {
-                // @ts-expect-error - Sadly Kea logics can't be generic based on props, so TS complains here
-                await asyncActions.updateNode(formValues)
+                if (values.isNew) {
+                    // @ts-expect-error - Sadly Kea logics can't be generic based on props, so TS complains here
+                    return await asyncActions.createNode(formValues)
+                } else {
+                    // @ts-expect-error - Sadly Kea logics can't be generic based on props, so TS complains here
+                    await asyncActions.updateNode(formValues)
+                }
             },
         },
     })),
     selectors(() => ({
+        isNew: [(_, p) => [p.id], (id): boolean => id === 'new'],
         breadcrumbs: [
             (s, p) => [p.id, p.stage, s.node, s.nodeLoading],
             (id, stage, node, nodeLoading): Breadcrumb[] => [
@@ -152,12 +212,87 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
             ],
         ],
         nodeBackend: [
-            (_, p) => [p.id],
-            (id): PipelineBackend => (typeof id === 'string' ? PipelineBackend.BatchExport : PipelineBackend.Plugin),
+            (s, p) => [s.node, p.id, s.newConfigurationServiceOrPluginID],
+            (node, id, newConfigurationServiceOrPluginID): PipelineBackend | null => {
+                if (node) {
+                    return node.backend
+                }
+                if (id === 'new') {
+                    if (newConfigurationServiceOrPluginID === null) {
+                        return null
+                    } else if (typeof newConfigurationServiceOrPluginID === 'string') {
+                        return PipelineBackend.BatchExport
+                    }
+                    return PipelineBackend.Plugin
+                }
+                if (typeof id === 'string') {
+                    return PipelineBackend.BatchExport
+                }
+                return PipelineBackend.Plugin
+            },
+        ],
+        maybeNodePlugin: [
+            (s) => [s.node, s.newConfigurationServiceOrPluginID, s.newConfigurationPlugins],
+            (node, maybePluginId, plugins): PluginType | null => {
+                if (node) {
+                    return node.backend === PipelineBackend.Plugin ? node.plugin : null
+                }
+                if (typeof maybePluginId === 'number') {
+                    // in case of new config creations
+                    return plugins[maybePluginId] || null
+                }
+                return null
+            },
+        ],
+        newConfigurationBatchExports: [
+            (_, p) => [p.stage],
+            (stage): Record<string, string> => {
+                if (stage === PipelineStage.Destination) {
+                    return {
+                        BigQuery: 'BigQuery',
+                        Postgres: 'PostgreSQL',
+                        Redshift: 'Redshift',
+                        S3: 'S3',
+                        Snowflake: 'Snowflake',
+                    }
+                }
+                return {}
+            },
+        ],
+        newConfigurationPlugins: [
+            (s, p) => [
+                p.stage,
+                s.destinationPlugins,
+                s.transformationPlugins,
+                s.frontendAppsPlugins,
+                s.importAppsPlugins,
+            ],
+            (
+                stage,
+                destinationPlugins,
+                transformationPlugins,
+                frontendAppsPlugins,
+                importAppsPlugins
+            ): Record<string, PluginType> => {
+                if (stage === PipelineStage.Transformation) {
+                    return transformationPlugins
+                } else if (stage === PipelineStage.Destination) {
+                    return destinationPlugins
+                } else if (stage === PipelineStage.SiteApp) {
+                    return frontendAppsPlugins
+                } else if (stage === PipelineStage.ImportApp) {
+                    return importAppsPlugins
+                }
+                return {}
+            },
         ],
         tabs: [
             (_, p) => [p.id],
             (id) => {
+                if (id === 'new') {
+                    // not used, but just in case
+                    return [PipelineNodeTab.Configuration]
+                }
                 const tabs = Object.values(PipelineNodeTab)
                 if (typeof id === 'string') {
                     // Batch export
@@ -167,36 +302,41 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
             },
         ],
         savedConfiguration: [
-            (s) => [s.node],
-            (node): Record<string, any> | null =>
-                node
-                    ? node.backend === PipelineBackend.Plugin
+            (s) => [s.node, s.maybeNodePlugin],
+            (node, maybeNodePlugin): Record<string, any> | null => {
+                if (node) {
+                    return node.backend === PipelineBackend.Plugin
                         ? node.config || defaultConfigForPlugin(node.plugin)
                         : { interval: node.interval, destination: node.service.type, ...node.service.config }
-                    : null,
+                }
+                if (maybeNodePlugin) {
+                    return defaultConfigForPlugin(maybeNodePlugin)
+                }
+                return null
+            },
         ],
         hiddenFields: [
-            (s) => [s.node, s.configuration],
-            (node, configuration): string[] => {
-                if (node?.backend === PipelineBackend.Plugin) {
-                    return determineInvisibleFields((fieldName) => configuration[fieldName], node.plugin)
+            (s) => [s.maybeNodePlugin, s.configuration],
+            (maybeNodePlugin, configuration): string[] => {
+                if (maybeNodePlugin) {
+                    return determineInvisibleFields((fieldName) => configuration[fieldName], maybeNodePlugin)
                 }
                 return []
             },
         ],
         requiredFields: [
-            (s) => [s.node, s.configuration],
-            (node, configuration): string[] => {
-                if (node?.backend === PipelineBackend.Plugin) {
-                    return determineRequiredFields((fieldName) => configuration[fieldName], node.plugin)
+            (s) => [s.maybeNodePlugin, s.configuration],
+            (maybeNodePlugin, configuration): string[] => {
+                if (maybeNodePlugin) {
+                    return determineRequiredFields((fieldName) => configuration[fieldName], maybeNodePlugin)
                 }
                 return []
             },
         ],
         isConfigurable: [
-            (s) => [s.node],
-            (node): boolean =>
-                node?.backend === PipelineBackend.Plugin && getConfigSchemaArray(node.plugin.config_schema).length > 0,
+            (s) => [s.maybeNodePlugin],
+            (maybeNodePlugin): boolean =>
+                !maybeNodePlugin || getConfigSchemaArray(maybeNodePlugin.config_schema).length > 0,
         ],
         id: [(_, p) => [p.id], (id) => id],
         stage: [(_, p) => [p.stage], (stage) => stage],
@@ -206,6 +346,9 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
             actions.resetConfiguration(values.savedConfiguration || {})
             // TODO: Update entry in the relevant list logic
         },
+        setNewSelected: () => {
+            actions.resetConfiguration({}) // If the user switches to a different plugin/batch export, then clear the form
+        },
         setConfigurationValue: async ({ name, value }) => {
             if (name[0] === 'json_config_file' && value) {
                 try {
@@ -244,7 +387,9 @@ export const pipelineNodeLogic = kea<pipelineNodeLogicType>([
             }
         },
     })),
-    afterMount(({ actions }) => {
-        actions.loadNode()
+    afterMount(({ values, actions }) => {
+        if (!values.isNew) {
+            actions.loadNode()
+        }
     }),
 ])

posthog/api/plugin.py

@@ -646,11 +646,15 @@ class PluginConfigSerializer(serializers.ModelSerializer):
             raise ValidationError("Plugin configuration is not available for the current organization!")
         validated_data["team_id"] = self.context["team_id"]
         _fix_formdata_config_json(self.context["request"], validated_data)
-        existing_config = PluginConfig.objects.filter(
-            team_id=validated_data["team_id"], plugin_id=validated_data["plugin"]
-        )
-        if existing_config.exists():
-            return self.update(existing_config.first(), validated_data)  # type: ignore
+        # Legacy pipeline UI doesn't show multiple plugin configs per plugin, so we don't allow it
+        # pipeline 3000 UI does, but to keep things simple we for now pass this flag to not break old users
+        # name field is something that only the new UI sends
+        if "config" not in validated_data or "name" not in validated_data["config"]:
+            existing_config = PluginConfig.objects.filter(
+                team_id=validated_data["team_id"], plugin_id=validated_data["plugin"]
+            )
+            if existing_config.exists():
+                return self.update(existing_config.first(), validated_data)  # type: ignore
 
         validated_data["web_token"] = generate_random_token()
         plugin_config = super().create(validated_data)