konnosDashboard
pg-meta-publications.ts252 lines · main
1import { z } from 'zod'
2
3import { ident, joinSqlFragments, literal, safeSql, type SafeSqlFragment } from './pg-format'
4import { PUBLICATIONS_SQL } from './sql/publications'
5
6const pgPublicationTableZod = z.object({
7 id: z.number().optional(),
8 name: z.string(),
9 schema: z.string(),
10})
11
12const pgPublicationZod = z.object({
13 id: z.number(),
14 name: z.string(),
15 owner: z.string(),
16 publish_insert: z.boolean(),
17 publish_update: z.boolean(),
18 publish_delete: z.boolean(),
19 publish_truncate: z.boolean(),
20 tables: z.array(pgPublicationTableZod).nullable(),
21})
22
23const pgPublicationArrayZod = z.array(pgPublicationZod)
24const pgPublicationOptionalZod = z.optional(pgPublicationZod)
25
26export type PGPublication = z.infer<typeof pgPublicationZod>
27
28function list({
29 limit,
30 offset,
31}: {
32 limit?: number
33 offset?: number
34} = {}): {
35 sql: SafeSqlFragment
36 zod: typeof pgPublicationArrayZod
37} {
38 let sql = safeSql`with publications as (${PUBLICATIONS_SQL}) select * from publications`
39 if (limit) {
40 sql = safeSql`${sql} limit ${literal(limit)}`
41 }
42 if (offset) {
43 sql = safeSql`${sql} offset ${literal(offset)}`
44 }
45 return {
46 sql,
47 zod: pgPublicationArrayZod,
48 }
49}
50
51type PublicationIdentifier = Pick<PGPublication, 'id'> | Pick<PGPublication, 'name'>
52
53function getIdentifierWhereClause(identifier: PublicationIdentifier): SafeSqlFragment {
54 if ('id' in identifier && identifier.id) {
55 return safeSql`${ident('id')} = ${literal(identifier.id)}`
56 } else if ('name' in identifier && identifier.name) {
57 return safeSql`${ident('name')} = ${literal(identifier.name)}`
58 }
59 throw new Error('Must provide either id or name')
60}
61
62function retrieve(identifier: PublicationIdentifier): {
63 sql: SafeSqlFragment
64 zod: typeof pgPublicationOptionalZod
65} {
66 const sql = safeSql`with publications as (${PUBLICATIONS_SQL}) select * from publications where ${getIdentifierWhereClause(identifier)};`
67 return {
68 sql,
69 zod: pgPublicationOptionalZod,
70 }
71}
72
73type PublicationCreateParams = {
74 name: string
75 publish_insert?: boolean
76 publish_update?: boolean
77 publish_delete?: boolean
78 publish_truncate?: boolean
79 tables?: string[] | null
80}
81
82function create({
83 name,
84 publish_insert = false,
85 publish_update = false,
86 publish_delete = false,
87 publish_truncate = false,
88 tables = null,
89}: PublicationCreateParams): { sql: SafeSqlFragment } {
90 let tableClause: SafeSqlFragment
91 if (tables === undefined || tables === null) {
92 tableClause = safeSql`FOR ALL TABLES`
93 } else if (tables.length === 0) {
94 tableClause = safeSql``
95 } else {
96 tableClause = safeSql`FOR TABLE ${joinSqlFragments(
97 tables.map((t) => {
98 if (!t.includes('.')) {
99 return ident(t)
100 }
101 const [schema, ...rest] = t.split('.')
102 const table = rest.join('.')
103 return safeSql`${ident(schema)}.${ident(table)}`
104 }),
105 ','
106 )}`
107 }
108
109 const publishOps: Array<string> = []
110 if (publish_insert) publishOps.push('insert')
111 if (publish_update) publishOps.push('update')
112 if (publish_delete) publishOps.push('delete')
113 if (publish_truncate) publishOps.push('truncate')
114
115 const sql = safeSql`
116CREATE PUBLICATION ${ident(name)} ${tableClause}
117 WITH (publish = ${literal(publishOps.join(','))});`
118
119 return { sql }
120}
121
122type PublicationUpdateParams = {
123 name?: string
124 owner?: string
125 publish_insert?: boolean
126 publish_update?: boolean
127 publish_delete?: boolean
128 publish_truncate?: boolean
129 tables?: string[] | null
130}
131
132function update(
133 id: number,
134 {
135 name,
136 owner,
137 publish_insert,
138 publish_update,
139 publish_delete,
140 publish_truncate,
141 tables,
142 }: PublicationUpdateParams
143): { sql: SafeSqlFragment } {
144 const sql = safeSql`
145do $$
146declare
147 id oid := ${literal(id)};
148 old record;
149 new_name text := ${name === undefined ? literal(null) : literal(name)};
150 new_owner text := ${owner === undefined ? literal(null) : literal(owner)};
151 new_publish_insert bool := ${literal(publish_insert ?? null)};
152 new_publish_update bool := ${literal(publish_update ?? null)};
153 new_publish_delete bool := ${literal(publish_delete ?? null)};
154 new_publish_truncate bool := ${literal(publish_truncate ?? null)};
155 new_tables text := ${
156 tables === undefined
157 ? literal(null)
158 : literal(
159 tables === null
160 ? 'all tables'
161 : tables
162 .map((t) => {
163 if (!t.includes('.')) {
164 return ident(t)
165 }
166
167 const [schema, ...rest] = t.split('.')
168 const table = rest.join('.')
169 return safeSql`${ident(schema)}.${ident(table)}`
170 })
171 .join(',')
172 )
173 };
174begin
175 select * into old from pg_publication where oid = id;
176 if old is null then
177 raise exception 'Cannot find publication with id %', id;
178 end if;
179
180 if new_tables is null then
181 null;
182 elsif new_tables = 'all tables' then
183 if old.puballtables then
184 null;
185 else
186 -- Need to recreate because going from list of tables <-> all tables with alter is not possible.
187 execute(format('drop publication %1$I; create publication %1$I for all tables;', old.pubname));
188 end if;
189 else
190 if old.puballtables then
191 -- Need to recreate because going from list of tables <-> all tables with alter is not possible.
192 execute(format('drop publication %1$I; create publication %1$I;', old.pubname));
193 elsif exists(select from pg_publication_rel where prpubid = id) then
194 execute(
195 format(
196 'alter publication %I drop table %s',
197 old.pubname,
198 (select string_agg(prrelid::regclass::text, ', ') from pg_publication_rel where prpubid = id)
199 )
200 );
201 end if;
202
203 -- At this point the publication must have no tables.
204
205 if new_tables != '' then
206 execute(format('alter publication %I add table %s', old.pubname, new_tables));
207 end if;
208 end if;
209
210 execute(
211 format(
212 'alter publication %I set (publish = %L);',
213 old.pubname,
214 concat_ws(
215 ', ',
216 case when coalesce(new_publish_insert, old.pubinsert) then 'insert' end,
217 case when coalesce(new_publish_update, old.pubupdate) then 'update' end,
218 case when coalesce(new_publish_delete, old.pubdelete) then 'delete' end,
219 case when coalesce(new_publish_truncate, old.pubtruncate) then 'truncate' end
220 )
221 )
222 );
223
224 execute(format('alter publication %I owner to %I;', old.pubname, coalesce(new_owner, old.pubowner::regrole::name)));
225
226 -- Using the same name in the rename clause gives an error, so only do it if the new name is different.
227 if new_name is not null and new_name != old.pubname then
228 execute(format('alter publication %I rename to %I;', old.pubname, coalesce(new_name, old.pubname)));
229 end if;
230
231 -- We need to retrieve the publication later, so we need a way to uniquely identify which publication this is.
232 -- We can't rely on id because it gets changed if it got recreated.
233 -- We use a temp table to store the unique name - DO blocks can't return a value.
234 create temp table pg_meta_publication_tmp (name) on commit drop as values (coalesce(new_name, old.pubname));
235end $$;
236`
237 return { sql }
238}
239
240function remove(publication: Pick<PGPublication, 'name'>): { sql: SafeSqlFragment } {
241 const sql = safeSql`DROP PUBLICATION IF EXISTS ${ident(publication.name)};`
242 return { sql }
243}
244
245export default {
246 list,
247 retrieve,
248 create,
249 update,
250 remove,
251 zod: pgPublicationZod,
252}