konnosDashboard
index.tsx453 lines · main
1import { useParams } from 'common'
2import { uniq } from 'lodash'
3import { Loader2 } from 'lucide-react'
4import Link from 'next/link'
5import { useRouter } from 'next/router'
6import { parseAsBoolean, useQueryState } from 'nuqs'
7import { useEffect, useMemo, useState } from 'react'
8import { Button, Card, CardContent } from 'ui'
9import { EmptyStatePresentational } from 'ui-patterns'
10import { Admonition } from 'ui-patterns/admonition'
11import { GenericTableLoader } from 'ui-patterns/ShimmeringLoader'
12
13import { DeleteAnalyticsBucketModal } from '../DeleteAnalyticsBucketModal'
14import { useSelectedAnalyticsBucket } from '../useSelectedAnalyticsBucket'
15import { HIDE_REPLICATION_USER_FLOW } from './AnalyticsBucketDetails.constants'
16import { BucketHeader } from './BucketHeader'
17import { CreateTableInstructions } from './CreateTable/CreateTableInstructions'
18import { NamespaceWithTables } from './NamespaceWithTables'
19import { SimpleConfigurationDetails } from './SimpleConfigurationDetails'
20import { useAnalyticsBucketAssociatedEntities } from './useAnalyticsBucketAssociatedEntities'
21import { useIcebergWrapperExtension } from './useIcebergWrapper'
22import { INTEGRATIONS } from '@/components/interfaces/Integrations/Landing/Integrations.constants'
23import { WrapperMeta } from '@/components/interfaces/Integrations/Wrappers/Wrappers.types'
24import {
25 convertKVStringArrayToJson,
26 formatWrapperTables,
27} from '@/components/interfaces/Integrations/Wrappers/Wrappers.utils'
28import {
29 ScaffoldContainer,
30 ScaffoldSection,
31 ScaffoldSectionTitle,
32} from '@/components/layouts/Scaffold'
33import AlertError from '@/components/ui/AlertError'
34import { InlineLink } from '@/components/ui/InlineLink'
35import {
36 DatabaseExtension,
37 useDatabaseExtensionsQuery,
38} from '@/data/database-extensions/database-extensions-query'
39import { useReplicationPipelineStatusQuery } from '@/data/replication/pipeline-status-query'
40import { useStartPipelineMutation } from '@/data/replication/start-pipeline-mutation'
41import { useIcebergNamespacesQuery } from '@/data/storage/iceberg-namespaces-query'
42import { useIcebergWrapperCreateMutation } from '@/data/storage/iceberg-wrapper-create-mutation'
43import { useSelectedProjectQuery } from '@/hooks/misc/useSelectedProject'
44import { DOCS_URL } from '@/lib/constants'
45
46export const AnalyticBucketDetails = () => {
47 const router = useRouter()
48 const { ref: projectRef } = useParams()
49 const { data: project } = useSelectedProjectQuery()
50 const { state: extensionState } = useIcebergWrapperExtension()
51 const {
52 data: bucket,
53 error: bucketError,
54 isSuccess: isSuccessBucket,
55 isError: isErrorBucket,
56 } = useSelectedAnalyticsBucket()
57
58 const [showDeleteModal, setShowDeleteModal] = useQueryState(
59 'delete',
60 parseAsBoolean.withDefault(false).withOptions({ history: 'push', clearOnDefault: true })
61 )
62 // [Joshen] Namespaces are now created asynchronously when the pipeline is started, so long poll after
63 // updating connected tables until namespaces are updated
64 // Namespace would just be the schema (Which is currently limited to public)
65 // Wrapper table would be {schema}_{table}_changelog
66 const [pollIntervalNamespaces, setPollIntervalNamespaces] = useState(0)
67 const [pollIntervalNamespaceTables, setPollIntervalNamespaceTables] = useState(0)
68
69 const { mutateAsync: startPipeline, isPending: isStartingPipeline } = useStartPipelineMutation()
70
71 const {
72 publication,
73 pipeline,
74 icebergWrapper: wrapperInstance,
75 isLoadingWrapperInstance,
76 } = useAnalyticsBucketAssociatedEntities({
77 projectRef,
78 bucketId: bucket?.name,
79 })
80 const { data, isSuccess: isSuccessPipelineStatus } = useReplicationPipelineStatusQuery(
81 { projectRef, pipelineId: pipeline?.id },
82 {
83 refetchInterval: (query) => {
84 const data = query.state.data
85 if (data?.status.name !== 'started') return 4000
86 else return false
87 },
88 }
89 )
90 const pipelineStatus = data?.status.name
91 const isPipelineRunning = pipelineStatus === 'started'
92 const isPipelineStopped = ['failed', 'stopped'].includes(pipelineStatus ?? '')
93
94 const wrapperValues = convertKVStringArrayToJson(wrapperInstance?.server_options ?? [])
95 const integration = INTEGRATIONS.find((i) => i.id === 'iceberg_wrapper' && i.type === 'wrapper')
96 const wrapperMeta = (integration?.type === 'wrapper' && integration.meta) as WrapperMeta
97 const state = isLoadingWrapperInstance
98 ? 'loading'
99 : extensionState === 'installed'
100 ? wrapperInstance
101 ? 'added'
102 : 'missing'
103 : extensionState
104
105 const wrapperTables = useMemo(() => {
106 if (!wrapperInstance) return []
107 return formatWrapperTables(wrapperInstance, wrapperMeta!)
108 }, [wrapperInstance, wrapperMeta])
109
110 const { data: extensionsData } = useDatabaseExtensionsQuery({
111 projectRef: project?.ref,
112 connectionString: project?.connectionString,
113 })
114 const wrappersExtension = extensionsData?.find((ext) => ext.name === 'wrappers')
115
116 const {
117 data: namespacesData = [],
118 isPending: isLoadingNamespaces,
119 isSuccess: isSuccessNamespaces,
120 } = useIcebergNamespacesQuery(
121 {
122 projectRef,
123 warehouse: wrapperValues.warehouse,
124 },
125 {
126 refetchInterval: (query) => {
127 const data = query.state.data
128 if (pollIntervalNamespaces === 0) return false
129
130 const publicationTableSchemas = publication?.tables.map((x) => x.schema) ?? []
131 const isSynced = !publicationTableSchemas.some((x) => !data?.includes(x))
132 if (isSynced) {
133 setPollIntervalNamespaces(0)
134 return false
135 }
136
137 return pollIntervalNamespaces
138 },
139 }
140 )
141
142 const publicationTableSchemas = (publication?.tables ?? []).map((x) => x.schema)
143 const isSyncedPublicationTableSchemasAndNamespaces = !publicationTableSchemas.some(
144 (x) => !namespacesData.includes(x)
145 )
146 const isPollingForData = pollIntervalNamespaces > 0 || pollIntervalNamespaceTables > 0
147
148 const namespaces = useMemo(() => {
149 const fdwNamespaces = wrapperTables.map((t) => t.table.split('.')[0]) as string[]
150 const namespaces = uniq([...fdwNamespaces, ...(namespacesData ?? [])])
151
152 return namespaces.map((namespace) => {
153 const tables = wrapperTables.filter((t) => t.table.split('.')[0] === namespace)
154 const schema = tables[0]?.schema
155
156 return {
157 namespace: namespace,
158 schema: schema,
159 tables: tables,
160 }
161 })
162 }, [wrapperTables, namespacesData])
163
164 useEffect(() => {
165 if (isSuccessNamespaces && !isSyncedPublicationTableSchemasAndNamespaces) {
166 setPollIntervalNamespaces(4000)
167 }
168 }, [isSuccessNamespaces, isSyncedPublicationTableSchemasAndNamespaces])
169
170 return (
171 <>
172 {isErrorBucket ? (
173 <ScaffoldContainer bottomPadding>
174 <ScaffoldSection isFullWidth>
175 <AlertError subject="Failed to fetch analytics buckets" error={bucketError} />
176 </ScaffoldSection>
177 </ScaffoldContainer>
178 ) : (
179 <ScaffoldContainer bottomPadding>
180 {state === 'loading' ? (
181 <ScaffoldSection isFullWidth>
182 <BucketHeader showActions={false} />
183 <GenericTableLoader />
184 </ScaffoldSection>
185 ) : state === 'not-installed' ? (
186 <ExtensionNotInstalled
187 bucketName={bucket?.name}
188 projectRef={project?.ref!}
189 wrapperMeta={wrapperMeta}
190 wrappersExtension={wrappersExtension!}
191 />
192 ) : state === 'needs-upgrade' ? (
193 <ExtensionNeedsUpgrade
194 bucketName={bucket?.name}
195 projectRef={project?.ref!}
196 wrapperMeta={wrapperMeta}
197 wrappersExtension={wrappersExtension!}
198 />
199 ) : state === 'missing' ? (
200 <WrapperMissing bucketName={bucket?.name} />
201 ) : state === 'added' && wrapperInstance ? (
202 <>
203 <ScaffoldSection isFullWidth>
204 <BucketHeader />
205
206 {isLoadingNamespaces || isLoadingWrapperInstance ? (
207 <GenericTableLoader headers={['Name']} />
208 ) : namespaces.length === 0 ? (
209 <>
210 {HIDE_REPLICATION_USER_FLOW ? (
211 <CreateTableInstructions />
212 ) : isPollingForData ? (
213 <EmptyStatePresentational
214 icon={
215 <Loader2
216 size={24}
217 strokeWidth={1.5}
218 className="animate-spin text-foreground-muted"
219 />
220 }
221 title="Connecting table(s) to bucket"
222 description="Tables will be shown here once the connection is complete"
223 />
224 ) : null}
225 </>
226 ) : (
227 <>
228 {!!pipeline && !!isSuccessPipelineStatus && !isPipelineRunning && (
229 <Admonition
230 type="note"
231 layout="horizontal"
232 className="[&>div]:pl-10 [&>div]:translate-y-[-3px]"
233 childProps={{ title: { className: 'block capitalize-sentence' } }}
234 showIcon={isPipelineStopped}
235 title={
236 isPipelineStopped
237 ? `Replication on the bucket has ${pipelineStatus}`
238 : `${pipelineStatus} replication on the bucket...`
239 }
240 description={
241 isPipelineStopped
242 ? 'Data changes from Postgres tables is currently not streaming to their corresponding analytics bucket table'
243 : 'Data changes from Postgres tables will resume streaming once pipeline has started'
244 }
245 actions={
246 <div className="flex items-center gap-x-2">
247 <Button asChild type="default">
248 <Link
249 href={`/project/${projectRef}/database/replication/${pipeline.replicator_id}`}
250 >
251 View replication
252 </Link>
253 </Button>
254 {isPipelineStopped && (
255 <Button
256 type="default"
257 loading={isStartingPipeline}
258 onClick={async () => {
259 if (projectRef) {
260 await startPipeline({ projectRef, pipelineId: pipeline.id })
261 }
262 }}
263 >
264 Restart
265 </Button>
266 )}
267 </div>
268 }
269 >
270 {!isPipelineStopped && (
271 <Loader2 size={18} className="absolute top-1.5 left-[3px] animate-spin" />
272 )}
273 </Admonition>
274 )}
275 <div className="flex flex-col gap-y-10">
276 {namespaces.map(({ namespace, schema, tables }) => (
277 <NamespaceWithTables
278 key={namespace}
279 namespace={namespace}
280 sourceType="direct"
281 schema={schema}
282 tables={tables as any}
283 wrapperValues={wrapperValues}
284 pollIntervalNamespaceTables={pollIntervalNamespaceTables}
285 setPollIntervalNamespaceTables={setPollIntervalNamespaceTables}
286 />
287 ))}
288 </div>
289 </>
290 )}
291 </ScaffoldSection>
292
293 <SimpleConfigurationDetails bucketName={bucket?.name} />
294 </>
295 ) : null}
296
297 <ScaffoldSection isFullWidth className="flex flex-col gap-y-4">
298 <header>
299 <ScaffoldSectionTitle>Manage</ScaffoldSectionTitle>
300 </header>
301 <Card>
302 <CardContent className="flex flex-col md:flex-row md:justify-between gap-y-4 gap-x-8 md:items-center">
303 <div className="flex flex-col">
304 <h3>Delete bucket</h3>
305 <p className="text-sm text-foreground-lighter">
306 This will also delete any data in your bucket. Make sure you have a backup if
307 you want to keep your data.
308 </p>
309 </div>
310 <Button
311 type="danger"
312 disabled={!bucket?.name || !isSuccessBucket}
313 onClick={() => setShowDeleteModal(true)}
314 >
315 Delete bucket
316 </Button>
317 </CardContent>
318 </Card>
319 </ScaffoldSection>
320 </ScaffoldContainer>
321 )}
322
323 <DeleteAnalyticsBucketModal
324 visible={showDeleteModal}
325 bucketId={bucket?.name}
326 onClose={() => setShowDeleteModal(false)}
327 onSuccess={() => router.push(`/project/${projectRef}/storage/analytics`)}
328 />
329 </>
330 )
331}
332
333const ExtensionNotInstalled = ({
334 bucketName,
335 projectRef,
336 wrapperMeta,
337 wrappersExtension,
338}: {
339 bucketName?: string
340 projectRef: string
341 wrapperMeta: WrapperMeta
342 wrappersExtension: DatabaseExtension
343}) => {
344 const databaseNeedsUpgrading =
345 (wrappersExtension?.default_version ?? '') < (wrapperMeta?.minimumExtensionVersion ?? '')
346
347 return (
348 <>
349 <ScaffoldSection isFullWidth>
350 <Admonition type="warning" title="Missing required extension">
351 <p>
352 The Wrappers extension is required in order to query analytics tables.{' '}
353 {databaseNeedsUpgrading &&
354 'Please first upgrade your database and then install the extension.'}{' '}
355 <InlineLink
356 href={`${DOCS_URL}/guides/database/extensions/wrappers/iceberg`}
357 target="_blank"
358 rel="noreferrer"
359 className="text-foreground-lighter hover:text-foreground transition-colors"
360 >
361 Learn more
362 </InlineLink>
363 </p>
364 <Button type="default" asChild className="mt-2" onClick={() => {}}>
365 <Link
366 href={
367 databaseNeedsUpgrading
368 ? `/project/${projectRef}/settings/infrastructure`
369 : `/project/${projectRef}/database/extensions?filter=wrappers`
370 }
371 >
372 {databaseNeedsUpgrading ? 'Upgrade database' : 'Install extension'}
373 </Link>
374 </Button>
375 </Admonition>
376 </ScaffoldSection>
377 <SimpleConfigurationDetails bucketName={bucketName} />
378 </>
379 )
380}
381
382const ExtensionNeedsUpgrade = ({
383 bucketName,
384 projectRef,
385 wrapperMeta,
386 wrappersExtension,
387}: {
388 bucketName?: string
389 projectRef: string
390 wrapperMeta: WrapperMeta
391 wrappersExtension: DatabaseExtension
392}) => {
393 // [Joshen] Default version is what's on the DB, so if the installed version is already the default version
394 // but still doesnt meet the minimum extension version, then DB upgrade is required
395 const databaseNeedsUpgrading =
396 wrappersExtension?.installed_version === wrappersExtension?.default_version
397
398 return (
399 <>
400 <ScaffoldSection isFullWidth>
401 <Admonition type="warning" title="Outdated extension version">
402 <p>
403 The {wrapperMeta.label} wrapper requires a minimum extension version of{' '}
404 {wrapperMeta.minimumExtensionVersion}. You have version{' '}
405 {wrappersExtension?.installed_version} installed. Please{' '}
406 {databaseNeedsUpgrading && 'first upgrade your database, and then '}update the extension
407 by disabling and enabling the Wrappers extension.
408 </p>
409 <p>
410 Before reinstalling the wrapper extension, you must first remove all existing wrappers.
411 Afterward, you can recreate the wrappers.
412 </p>
413 <Button asChild type="default">
414 <Link
415 href={
416 databaseNeedsUpgrading
417 ? `/project/${projectRef}/settings/infrastructure`
418 : `/project/${projectRef}/database/extensions?filter=wrappers`
419 }
420 >
421 {databaseNeedsUpgrading ? 'Upgrade database' : 'Extensions'}
422 </Link>
423 </Button>
424 </Admonition>
425 </ScaffoldSection>
426 <SimpleConfigurationDetails bucketName={bucketName} />
427 </>
428 )
429}
430
431const WrapperMissing = ({ bucketName }: { bucketName?: string }) => {
432 const { mutateAsync: createIcebergWrapper, isPending: isCreatingIcebergWrapper } =
433 useIcebergWrapperCreateMutation()
434
435 const onSetupWrapper = async () => {
436 if (!bucketName) return console.error('Bucket name is required')
437 await createIcebergWrapper({ bucketName })
438 }
439
440 return (
441 <>
442 <ScaffoldSection isFullWidth>
443 <Admonition type="warning" title="Missing integration">
444 <p>The Iceberg Wrapper integration is required in order to query analytics tables.</p>
445 <Button type="default" loading={isCreatingIcebergWrapper} onClick={onSetupWrapper}>
446 Install wrapper
447 </Button>
448 </Admonition>
449 </ScaffoldSection>
450 <SimpleConfigurationDetails bucketName={bucketName} />
451 </>
452 )
453}