import { zodResolver } from '@hookform/resolvers/zod' import { useParams } from 'common' import { useCreateTenantSourceMutation } from 'data/replication/create-tenant-source-mutation' import { useReplicationPublicationsQuery } from 'data/replication/publications-query' import { useStartPipelineMutation } from 'data/replication/start-pipeline-mutation' import { useForm } from 'react-hook-form' import { toast } from 'sonner' import { Accordion_Shadcn_, AccordionContent_Shadcn_, AccordionItem_Shadcn_, AccordionTrigger_Shadcn_, Alert_Shadcn_, AlertDescription_Shadcn_, AlertTitle_Shadcn_, Button, Form_Shadcn_, FormControl_Shadcn_, FormField_Shadcn_, Input_Shadcn_, Select_Shadcn_, SelectContent_Shadcn_, SelectGroup_Shadcn_, SelectItem_Shadcn_, SelectTrigger_Shadcn_, Sheet, SheetContent, SheetDescription, SheetFooter, SheetHeader, SheetSection, SheetTitle, Switch, TextArea_Shadcn_, WarningIcon, Label_Shadcn_ as Label, } from 'ui' import * as z from 'zod' import PublicationsComboBox from './PublicationsComboBox' import NewPublicationPanel from './NewPublicationPanel' import { useState, useMemo, useEffect } from 'react' import { useReplicationSinkByIdQuery } from 'data/replication/sink-by-id-query' import { useReplicationPipelineByIdQuery } from 'data/replication/pipeline-by-id-query' import { useStopPipelineMutation } from 'data/replication/stop-pipeline-mutation' import { FormItemLayout } from 'ui-patterns/form/FormItemLayout/FormItemLayout' import { useCreateSinkPipelineMutation } from 'data/replication/create-sink-pipeline-mutation' import { useUpdateSinkPipelineMutation } from 'data/replication/update-sink-pipeline-mutation' interface DestinationPanelProps { visible: boolean sourceId: number | undefined onClose: () => void existingDestination?: { sourceId?: number sinkId: number pipelineId?: number enabled: boolean } } const DestinationPanel = ({ visible, sourceId, onClose, existingDestination, }: DestinationPanelProps) => { const { ref: projectRef } = useParams() const [publicationPanelVisible, setPublicationPanelVisible] = useState(false) const { mutateAsync: createTenantSource, isLoading: creatingTenantSource } = useCreateTenantSourceMutation() const { mutateAsync: createSinkPipeline, isLoading: creatingSinkPipeline } = useCreateSinkPipelineMutation() const { mutateAsync: startPipeline, isLoading: startingPipeline } = useStartPipelineMutation() const { mutateAsync: stopPipeline, isLoading: stoppingPipeline } = useStopPipelineMutation() const { mutateAsync: updateSinkPipeline, isLoading: updatingSinkPipeline } = useUpdateSinkPipelineMutation() const { data: publications, isLoading: loadingPublications } = useReplicationPublicationsQuery({ projectRef, sourceId, }) const { data: sinkData } = useReplicationSinkByIdQuery({ projectRef, sinkId: existingDestination?.sinkId, }) const { data: pipelineData } = useReplicationPipelineByIdQuery({ projectRef, pipelineId: existingDestination?.pipelineId, }) const isCreating = creatingTenantSource || creatingSinkPipeline || startingPipeline const isUpdating = updatingSinkPipeline || stoppingPipeline || startingPipeline const isSubmitting = isCreating || isUpdating const editMode = !!existingDestination const formId = 'destination-editor' const types = ['BigQuery'] as const const TypeEnum = z.enum(types) const FormSchema = z.object({ type: TypeEnum, name: z.string().min(1, 'Name is required'), projectId: z.string().min(1, 'Project id is required'), datasetId: z.string().min(1, 'Dataset id is required'), serviceAccountKey: z.string().min(1, 'Service account key is required'), publicationName: z.string().min(1, 'Publication is required'), maxSize: z.number().min(1, 'Max Size must be greater than 0').int(), maxFillSecs: z.number().min(1, 'Max Fill seconds should be greater than 0').int(), maxStalenessMins: z.number().nonnegative(), enabled: z.boolean(), }) const defaultValues = useMemo( () => ({ type: TypeEnum.enum.BigQuery, name: sinkData?.name ?? '', projectId: sinkData?.config?.big_query?.project_id ?? '', datasetId: sinkData?.config?.big_query?.dataset_id ?? '', serviceAccountKey: sinkData?.config?.big_query?.service_account_key ?? '', publicationName: pipelineData?.publication_name ?? '', maxSize: pipelineData?.config?.config?.max_size ?? 1000, maxFillSecs: pipelineData?.config?.config?.max_fill_secs ?? 10, maxStalenessMins: sinkData?.config?.big_query?.max_staleness_mins ?? 5, enabled: existingDestination?.enabled ?? true, }), [sinkData, pipelineData, existingDestination] ) const form = useForm>({ mode: 'onBlur', reValidateMode: 'onBlur', resolver: zodResolver(FormSchema), defaultValues, }) const onSubmit = async (data: z.infer) => { if (!projectRef) return console.error('Project ref is required') try { if (editMode && existingDestination) { if (!sourceId) { console.error('Source id is required') return } if (!existingDestination.pipelineId) { console.error('Pipeline id is required') return } // Update existing destination await updateSinkPipeline({ sinkId: existingDestination.sinkId, pipelineId: existingDestination.pipelineId, projectRef, sinkName: data.name, sinkConfig: { bigQuery: { projectId: data.projectId, datasetId: data.datasetId, serviceAccountKey: data.serviceAccountKey, maxStalenessMins: data.maxStalenessMins, }, }, pipelinConfig: { config: { maxSize: data.maxSize, maxFillSecs: data.maxFillSecs }, }, publicationName: data.publicationName, sourceId, }) if (data.enabled) { await startPipeline({ projectRef, pipelineId: existingDestination.pipelineId }) } else { await stopPipeline({ projectRef, pipelineId: existingDestination.pipelineId }) } toast.success('Successfully updated destination') } else { // Create new destination if (!sourceId) { console.error('Source id is required') return } const { pipeline_id: pipelineId } = await createSinkPipeline({ projectRef, sinkName: data.name, sinkConfig: { bigQuery: { projectId: data.projectId, datasetId: data.datasetId, serviceAccountKey: data.serviceAccountKey, maxStalenessMins: data.maxStalenessMins, }, }, sourceId, publicationName: data.publicationName, pipelinConfig: { config: { maxSize: data.maxSize, maxFillSecs: data.maxFillSecs }, }, }) if (data.enabled) { await startPipeline({ projectRef, pipelineId }) } toast.success('Successfully created destination') } onClose() } catch (error) { toast.error(`Failed to ${editMode ? 'update' : 'create'} destination`) } } const onEnableReplication = async () => { if (!projectRef) return console.error('Project ref is required') await createTenantSource({ projectRef }) } const { enabled } = form.watch() useEffect(() => { if (editMode && sinkData && pipelineData) { form.reset(defaultValues) } }, [sinkData, pipelineData, editMode, defaultValues, form]) return ( <> {sourceId ? ( <>
{editMode ? 'Edit Destination' : 'New Destination'} {editMode ? null : 'Send data to a new destination'}
{ form.setValue('enabled', checked) }} />
( )} />

What data to send

( pub.name) || []} loading={loadingPublications} field={field} onNewPublicationClick={() => setPublicationPanelVisible(true)} /> )} />

Where to send that data

( {field.value} BigQuery )} > ( )} /> ( )} /> ( )} /> Advanced Settings ( )} /> ( )} /> ( )} />
( )} />
setPublicationPanelVisible(false)} /> ) : ( <>
New Destination {/* Pricing to be decided yet */} Enabling replication will cost additional $xx.xx
)} ) } export default DestinationPanel