fix: create etl publication as postgres (#45043)

## What kind of change does this PR introduce?

Bug fix

## What is the current behavior?

Creating a schema only branch fails because ETL publication is owned by
`supabase_etl_admin` which users have no access.

## What is the new behavior?

Since ETL supports user managed publications, create them through pgmeta
so it's owned by `postgres` role instead.

## Additional context

mirrors [upstream
etl](https://github.com/supabase/etl/blob/main/etl-api/src/db/publications.rs#L22-L51)
implementation


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Added guards to prevent creating publications when project or
connection info is missing, with clearer error logging.
* Ensure the project connection string is explicitly passed so
publications target the correct database.

* **Refactor**
* Publication creation now executes generated SQL directly against the
database, with correct handling of empty or selected table lists and
proper identifier quoting for reliability.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Joshen Lim <joshenlimek@gmail.com>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: CodeRabbit <noreply@coderabbit.ai>

apps/studio/components/interfaces/Database/Replication/DestinationPanel/DestinationForm/NewPublicationPanel.tsx

@@ -22,6 +22,7 @@ import { z } from 'zod'
 
 import { useCreatePublicationMutation } from '@/data/replication/publication-create-mutation'
 import { useReplicationTablesQuery } from '@/data/replication/tables-query'
+import { useSelectedProjectQuery } from '@/hooks/misc/useSelectedProject'
 
 interface NewPublicationPanelProps {
   visible: boolean
@@ -31,6 +32,7 @@ interface NewPublicationPanelProps {
 
 export const NewPublicationPanel = ({ visible, sourceId, onClose }: NewPublicationPanelProps) => {
   const { ref: projectRef } = useParams()
+  const { data: project } = useSelectedProjectQuery()
   const { mutateAsync: createPublication, isPending: creatingPublication } =
     useCreatePublicationMutation()
   const { data: tables } = useReplicationTablesQuery({
@@ -55,6 +57,7 @@ export const NewPublicationPanel = ({ visible, sourceId, onClose }: NewPublicati
 
   const onSubmit = async (data: z.infer<typeof FormSchema>) => {
     if (!projectRef) return console.error('Project ref is required')
+    if (!project) return console.error('Project is required')
     if (!sourceId) return console.error('Source id is required')
     try {
       await createPublication({
@@ -65,6 +68,7 @@ export const NewPublicationPanel = ({ visible, sourceId, onClose }: NewPublicati
           const [schema, name] = table.split('.')
           return { schema, name }
         }),
+        connectionString: project.connectionString,
       })
       toast.success('Successfully created publication')
       onClose()

apps/studio/data/replication/publication-create-mutation.ts

@@ -1,8 +1,9 @@
+import { getCreatePublicationSQL } from '@supabase/pg-meta'
 import { useMutation, useQueryClient } from '@tanstack/react-query'
 import { toast } from 'sonner'
 
+import { executeSql } from '../sql/execute-sql-query'
 import { replicationKeys } from './keys'
-import { handleError, post } from '@/data/fetchers'
 import type { ResponseError, UseCustomMutationOptions } from '@/types'
 
 export type CreatePublicationParams = {
@@ -10,27 +11,19 @@ export type CreatePublicationParams = {
   sourceId: number
   name: string
   tables: { schema: string; name: string }[]
+  connectionString?: string | null
 }
 
 async function createPublication(
-  { projectRef, sourceId, name, tables }: CreatePublicationParams,
+  { projectRef, connectionString, name, tables }: CreatePublicationParams,
   signal?: AbortSignal
 ) {
   if (!projectRef) throw new Error('projectRef is required')
 
-  const { data, error } = await post(
-    '/platform/replication/{ref}/sources/{source_id}/publications',
-    {
-      params: { path: { ref: projectRef, source_id: sourceId } },
-      body: { name, tables },
-      signal,
-    }
-  )
-  if (error) {
-    handleError(error)
-  }
+  const sql = getCreatePublicationSQL({ name, tables })
+  const { result } = await executeSql({ projectRef, connectionString, sql }, signal)
 
-  return data
+  return result
 }
 
 type CreatePublicationData = Awaited<ReturnType<typeof createPublication>>

packages/pg-meta/src/sql/studio/database/index.ts

@@ -16,3 +16,4 @@ export * from './triggers'
 export * from './types'
 export * from './vault'
 export * from './views'
+export * from './publications'

packages/pg-meta/src/sql/studio/database/publications.ts

@@ -0,0 +1,16 @@
+import { ident } from '../../../pg-format'
+
+export const getCreatePublicationSQL = ({
+  name,
+  tables,
+}: {
+  name: string
+  tables: { schema: string; name: string }[]
+}) => {
+  const query =
+    tables.length > 0
+      ? `FOR TABLE ONLY ${tables.map(({ schema, name }) => `${ident(schema)}.${ident(name)}`).join(', ')} `
+      : ''
+
+  return `CREATE PUBLICATION ${ident(name)} ${query}WITH (publish_via_partition_root = true)`
+}