Refactor polling logic for Analytics Buckets while syncing with namespace (supabase#40381)

* Refactor long polling logic for analytics bucket to automatically long poll when landing back on page and data has yet to be synced

* Nit clean up refactor

* Clean up

* Nit

Author: joshenlim

apps/studio/components/interfaces/Storage/AnalyticsBuckets/AnalyticsBucketDetails/constants.ts renamed to apps/studio/components/interfaces/Storage/AnalyticsBuckets/AnalyticsBucketDetails/AnalyticsBucketDetails.constants.ts

@@ -15,3 +15,10 @@ export const getAnalyticsBucketFDWName = (bucketId: string) => {
 export const getAnalyticsBucketFDWServerName = (bucketId: string) => {
   return `${snakeCase(bucketId)}_fdw_server`
 }
+
+export const getNamespaceTableNameFromPostgresTableName = (table: {
+  name: string
+  schema: string
+}) => {
+  return `${snakeCase(`${table.schema}.${table.name}`)}_changelog`
+}

apps/studio/components/interfaces/Storage/AnalyticsBuckets/AnalyticsBucketDetails/AnalyticsBucketDetails.utils.ts

@@ -0,0 +1,62 @@
+import { noop } from 'lodash'
+import Link from 'next/link'
+
+import { useParams } from 'common'
+import { FormattedWrapperTable } from 'components/interfaces/Integrations/Wrappers/Wrappers.utils'
+import {
+  ScaffoldHeader,
+  ScaffoldSectionDescription,
+  ScaffoldSectionTitle,
+} from 'components/layouts/Scaffold'
+import { useReplicationPipelineStatusQuery } from 'data/etl/pipeline-status-query'
+import { Button } from 'ui'
+import { ConnectTablesDialog } from './ConnectTablesDialog'
+import { useAnalyticsBucketAssociatedEntities } from './useAnalyticsBucketAssociatedEntities'
+
+interface BucketHeaderProps {
+  showActions?: boolean
+  namespaces?: {
+    namespace: string
+    schema: string
+    tables: FormattedWrapperTable[]
+  }[]
+  onSuccessConnectTables?: () => void
+}
+
+export const BucketHeader = ({
+  showActions = true,
+  namespaces = [],
+  onSuccessConnectTables = noop,
+}: BucketHeaderProps) => {
+  const { ref: projectRef, bucketId } = useParams()
+
+  const { pipeline } = useAnalyticsBucketAssociatedEntities({ projectRef, bucketId })
+  const { data } = useReplicationPipelineStatusQuery({ projectRef, pipelineId: pipeline?.id })
+  const pipelineStatus = data?.status.name
+  const isPipelineRunning = pipelineStatus === 'started'
+
+  return (
+    <ScaffoldHeader className="pt-0 flex flex-row justify-between items-end gap-x-8">
+      <div>
+        <ScaffoldSectionTitle>Tables</ScaffoldSectionTitle>
+        <ScaffoldSectionDescription>
+          Analytics tables stored in this bucket
+        </ScaffoldSectionDescription>
+      </div>
+      {showActions && (
+        <div className="flex items-center gap-x-2">
+          {!!pipeline && isPipelineRunning && (
+            <Button asChild type="default">
+              <Link href={`/project/${projectRef}/database/etl/${pipeline.replicator_id}`}>
+                View replication
+              </Link>
+            </Button>
+          )}
+          {namespaces.length > 0 && (
+            <ConnectTablesDialog onSuccessConnectTables={onSuccessConnectTables} />
+          )}
+        </div>
+      )}
+    </ScaffoldHeader>
+  )
+}

apps/studio/components/interfaces/Storage/AnalyticsBuckets/AnalyticsBucketDetails/BucketHeader.tsx

@@ -92,7 +92,7 @@ const PROGRESS_INDICATORS = {
 }
 
 interface ConnectTablesDialogProps {
-  onSuccessConnectTables: (tables: { schema: string; name: string }[]) => void
+  onSuccessConnectTables: () => void
 }
 
 export const ConnectTablesDialog = ({ onSuccessConnectTables }: ConnectTablesDialogProps) => {
@@ -245,7 +245,7 @@ export const ConnectTablesDialogContent = ({
       setConnectingStep(PROGRESS_STAGE.START_PIPELINE)
       await startPipeline({ projectRef, pipelineId })
 
-      onSuccessConnectTables?.(publicationTables)
+      onSuccessConnectTables?.()
       toast.success(`Connected ${values.tables.length} tables to Analytics bucket!`)
       form.reset()
       onClose()
@@ -287,7 +287,7 @@ export const ConnectTablesDialogContent = ({
       setConnectingStep(PROGRESS_STAGE.START_PIPELINE)
       await startPipeline({ projectRef, pipelineId: pipeline.id })
 
-      onSuccessConnectTables?.(tablesToBeAdded)
+      onSuccessConnectTables?.()
       toast.success('Successfully updated connected tables! Pipeline is being restarted')
       onClose()
     } catch (error: any) {

apps/studio/components/interfaces/Storage/AnalyticsBuckets/AnalyticsBucketDetails/ConnectTablesDialog.tsx

@@ -1,4 +1,4 @@
-import { snakeCase, uniq } from 'lodash'
+import { uniq } from 'lodash'
 import { Loader2, MoreVertical, Pause, Play, Trash } from 'lucide-react'
 import Link from 'next/link'
 import { useMemo, useState } from 'react'
@@ -34,7 +34,10 @@ import {
   TooltipTrigger,
 } from 'ui'
 import { ConfirmationModal } from 'ui-patterns/Dialogs/ConfirmationModal'
-import { getAnalyticsBucketFDWServerName } from '../AnalyticsBucketDetails.utils'
+import {
+  getAnalyticsBucketFDWServerName,
+  getNamespaceTableNameFromPostgresTableName,
+} from '../AnalyticsBucketDetails.utils'
 import { useAnalyticsBucketAssociatedEntities } from '../useAnalyticsBucketAssociatedEntities'
 import { useAnalyticsBucketWrapperInstance } from '../useAnalyticsBucketWrapperInstance'
 import { inferPostgresTableFromNamespaceTable } from './NamespaceWithTables.utils'
@@ -116,7 +119,7 @@ export const TableRowComponent = ({
       // [Joshen ALPHA] Assumption here is that all the namespace tables have _changelog as suffix
       // May need to update if that assumption falls short (e.g for those dealing with iceberg APIs directly)
       const updatedTables = publication.tables.filter(
-        (x) => table.name !== snakeCase(`${x.schema}.${x.name}_changelog`)
+        (x) => table.name !== getNamespaceTableNameFromPostgresTableName(x)
       )
       await updatePublication({
         projectRef,
@@ -142,7 +145,9 @@ export const TableRowComponent = ({
     if (!pipeline) return toast.error('Unable to find existing pipeline')
 
     // [Joshen ALPHA] This has potential to be flaky - we should see how we can get the table name and schema better
-    const pgTable = tables?.find((t) => snakeCase(`${t.schema}.${t.name}_changelog`) === table.name)
+    const pgTable = tables?.find(
+      (t) => getNamespaceTableNameFromPostgresTableName(t) === table.name
+    )
     if (!pgTable) return toast.error('Unable to find corresponding Postgres table')
 
     try {

apps/studio/components/interfaces/Storage/AnalyticsBuckets/AnalyticsBucketDetails/NamespaceWithTables/TableRowComponent.tsx

@@ -1,5 +1,5 @@
 import { ChevronRight, Info, Loader2, Plus, RefreshCw } from 'lucide-react'
-import { useMemo, useState } from 'react'
+import { useEffect, useMemo, useState } from 'react'
 
 import { useParams } from 'common'
 import type { WrapperMeta } from 'components/interfaces/Integrations/Wrappers/Wrappers.types'
@@ -25,6 +25,7 @@ import {
   TooltipContent,
   TooltipTrigger,
 } from 'ui'
+import { getNamespaceTableNameFromPostgresTableName } from '../AnalyticsBucketDetails.utils'
 import { useAnalyticsBucketAssociatedEntities } from '../useAnalyticsBucketAssociatedEntities'
 import { TableRowComponent } from './TableRowComponent'
 
@@ -37,7 +38,6 @@ type NamespaceWithTablesProps = {
   wrapperInstance: FDW
   wrapperValues: Record<string, string>
   wrapperMeta: WrapperMeta
-  tablesToPoll: { schema: string; name: string }[]
   pollIntervalNamespaceTables: number
   setPollIntervalNamespaceTables: (value: number) => void
 }
@@ -51,7 +51,6 @@ export const NamespaceWithTables = ({
   wrapperInstance,
   wrapperValues,
   wrapperMeta,
-  tablesToPoll,
   pollIntervalNamespaceTables,
   setPollIntervalNamespaceTables,
 }: NamespaceWithTablesProps) => {
@@ -61,31 +60,42 @@ export const NamespaceWithTables = ({
 
   const { publication } = useAnalyticsBucketAssociatedEntities({ projectRef, bucketId })
 
-  const { data: tablesData, isLoading: isLoadingNamespaceTables } = useIcebergNamespaceTablesQuery(
+  const {
+    data: tablesData = [],
+    isLoading: isLoadingNamespaceTables,
+    isSuccess: isSuccessNamespaceTables,
+  } = useIcebergNamespaceTablesQuery(
     {
       catalogUri: wrapperValues.catalog_uri,
       warehouse: wrapperValues.warehouse,
       namespace: namespace,
       projectRef,
     },
     {
-      refetchInterval: (data) => {
+      refetchInterval: (data = []) => {
         if (pollIntervalNamespaceTables === 0) return false
-        if (tablesToPoll.length > 0) {
-          const hasMissingTables =
-            (data ?? []).filter((t) => !tables.find((table) => table.table.split('.')[1] === t))
-              .length > 0
 
-          if (hasMissingTables) {
-            setPollIntervalNamespaceTables(0)
-            return false
-          }
+        const publicationTables = publication?.tables ?? []
+        const isSynced = !publicationTables.some(
+          (x) => !data.includes(getNamespaceTableNameFromPostgresTableName(x))
+        )
+        if (isSynced) {
+          setPollIntervalNamespaceTables(0)
+          return false
         }
+
         return pollIntervalNamespaceTables
       },
     }
   )
 
+  const publicationTables = publication?.tables ?? []
+  const publicationTablesNotSyncedToNamespaceTables = publicationTables.filter(
+    (x) => !tablesData.includes(getNamespaceTableNameFromPostgresTableName(x))
+  )
+  const isSyncedPublicationTablesAndNamespaceTables =
+    publicationTablesNotSyncedToNamespaceTables.length === 0
+
   const { mutateAsync: importForeignSchema, isLoading: isImportingForeignSchema } =
     useFDWImportForeignSchemaMutation()
 
@@ -140,6 +150,13 @@ export const NamespaceWithTables = ({
     return `fdw_analytics_${namespace.replaceAll('-', '_')}`
   }, [schema, namespace])
 
+  useEffect(() => {
+    if (isSuccessNamespaceTables && !isSyncedPublicationTablesAndNamespaceTables) {
+      setPollIntervalNamespaceTables(4000)
+    }
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [isSuccessNamespaceTables, isSyncedPublicationTablesAndNamespaceTables])
+
   return (
     <Card>
       <CardHeader className="flex flex-row justify-between items-center px-4 py-5 space-y-0">
@@ -179,14 +196,28 @@ export const NamespaceWithTables = ({
           )}
         </CardTitle>
 
-        <div className="flex flex-row gap-x-2">
-          {tablesToPoll.length > 0 && (
-            <div className="flex items-center gap-x-2 ml-6 text-foreground-lighter">
-              <Loader2 size={14} className="animate-spin" />
-              <p className="text-sm">
-                Connecting {tablesToPoll.length} table{tablesToPoll.length > 1 ? 's' : ''}
-              </p>
-            </div>
+        <div className="flex flex-row gap-x-6">
+          {pollIntervalNamespaceTables > 0 && (
+            <Tooltip>
+              <TooltipTrigger>
+                <div className="flex items-center gap-x-2 text-foreground-lighter">
+                  <Loader2 size={14} className="animate-spin" />
+                  <p className="text-sm">
+                    Connecting {publicationTablesNotSyncedToNamespaceTables.length} table
+                    {publicationTablesNotSyncedToNamespaceTables.length > 1 ? 's' : ''}
+                  </p>
+                </div>
+              </TooltipTrigger>
+              <TooltipContent side="bottom" align="end">
+                <p className="mb-1">Waiting for namespace table to be created for:</p>
+                <ul className="list-disc pl-6">
+                  {publicationTablesNotSyncedToNamespaceTables.map((x) => {
+                    const value = `${x.schema}.${x.name}`
+                    return <li key={value}>{value}</li>
+                  })}
+                </ul>
+              </TooltipContent>
+            </Tooltip>
           )}
           {missingTables.length > 0 && (
             <Button
@@ -202,7 +233,7 @@ export const NamespaceWithTables = ({
         </div>
       </CardHeader>
 
-      {tablesToPoll.length > 0 && <LoadingLine loading />}
+      {pollIntervalNamespaceTables > 0 && <LoadingLine loading />}
 
       <Table>
         <TableHeader>

apps/studio/components/interfaces/Storage/AnalyticsBuckets/AnalyticsBucketDetails/NamespaceWithTables/index.tsx

@@ -10,7 +10,7 @@ import {
 import { InlineLink } from 'components/ui/InlineLink'
 import { DOCS_URL } from 'lib/constants'
 import { Card } from 'ui'
-import { DESCRIPTIONS, LABELS, OPTION_ORDER } from './constants'
+import { DESCRIPTIONS, LABELS, OPTION_ORDER } from './AnalyticsBucketDetails.constants'
 import { CopyEnvButton } from './CopyEnvButton'
 import { DecryptedReadOnlyInput } from './DecryptedReadOnlyInput'
 import { useAnalyticsBucketWrapperInstance } from './useAnalyticsBucketWrapperInstance'
@@ -27,7 +27,7 @@ export const SimpleConfigurationDetails = ({ bucketName }: { bucketName?: string
 
   return (
     <ScaffoldSection isFullWidth>
-      <ScaffoldHeader className="flex flex-row justify-between items-end gap-x-8">
+      <ScaffoldHeader className="flex flex-row justify-between items-end gap-x-8 pt-0">
         <div>
           <ScaffoldSectionTitle>Connection details</ScaffoldSectionTitle>
           <ScaffoldSectionDescription>