feat(data): manage Postgres Advanced logical replication

src/commands/data/pg/logical-replication/publications/create.ts

@@ -0,0 +1,64 @@
+import {flags as Flags} from '@heroku-cli/command'
+import {color} from '@heroku/heroku-cli-util'
+import {Args, ux} from '@oclif/core'
+
+import BaseCommand from '../../../../../lib/data/base-command.js'
+import {PublicationTarget, resolveAdvancedDatabase} from '../../../../../lib/data/logical-replication.js'
+
+export default class DataPgLogicalReplicationPublicationsCreate extends BaseCommand {
+  static aliases = ['data:pg:lr:publications:create']
+  static args = {
+    database: Args.string({
+      description: 'database name, database attachment name, or related config var on an app',
+      required: true,
+    }),
+  }
+  static description = 'create a logical replication publication on a Postgres Advanced database'
+  static examples = [
+    '<%= config.bin %> <%= command.id %> DATABASE --name orders --table public.orders --app example-app',
+    '<%= config.bin %> <%= command.id %> DATABASE --name application --schema public --app example-app',
+    '<%= config.bin %> <%= command.id %> DATABASE --name application --all-schemas --app example-app',
+  ]
+  static flags = {
+    'all-schemas': Flags.boolean({
+      description: 'include all current customer schemas',
+      exclusive: ['schema', 'table'],
+    }),
+    app: Flags.app({required: true}),
+    name: Flags.string({description: 'lowercase name for the publication', required: true}),
+    remote: Flags.remote(),
+    schema: Flags.string({description: 'schema to include, including new tables created in the schema', multiple: true}),
+    table: Flags.string({description: 'fully-qualified table to include', multiple: true}),
+  }
+
+  async run(): Promise<void> {
+    const {args, flags} = await this.parse(DataPgLogicalReplicationPublicationsCreate)
+    const addon = await resolveAdvancedDatabase(this, args.database, flags.app)
+    const target = this.publicationTarget(flags['all-schemas'], flags.table, flags.schema)
+
+    try {
+      ux.action.start(`Creating publication ${color.name(flags.name)} on ${color.datastore(addon.name)}`)
+      await this.dataApi.post(`/data/postgres/v1/${addon.id}/logical-replication/publications`, {body: {name: flags.name, target}})
+      ux.action.stop()
+      if (flags['all-schemas']) {
+        ux.stdout('The publication includes all current customer schemas. Tables created later and new schemas are not added automatically.')
+      }
+    } catch (error) {
+      ux.action.stop(color.red('!'))
+      throw error
+    }
+  }
+
+  private publicationTarget(allSchemas: boolean, tables?: string[], schemas?: string[]): PublicationTarget {
+    if (allSchemas) return {type: 'all_customer_schemas'}
+
+    if (tables && schemas) {
+      ux.error('Specify either --table or --schema, not both.')
+    }
+
+    if (tables) return {tables, type: 'tables'}
+    if (schemas) return {schemas, type: 'schemas'}
+
+    ux.error('Specify --all-schemas, at least one --table, or at least one --schema.')
+  }
+}

src/commands/data/pg/logical-replication/publications/destroy.ts

@@ -0,0 +1,41 @@
+import {flags as Flags} from '@heroku-cli/command'
+import {color, hux} from '@heroku/heroku-cli-util'
+import {Args, ux} from '@oclif/core'
+
+import BaseCommand from '../../../../../lib/data/base-command.js'
+import {resolveAdvancedDatabase} from '../../../../../lib/data/logical-replication.js'
+
+export default class DataPgLogicalReplicationPublicationsDestroy extends BaseCommand {
+  static aliases = ['data:pg:lr:publications:destroy']
+  static args = {
+    database: Args.string({
+      description: 'database name, database attachment name, or related config var on an app',
+      required: true,
+    }),
+  }
+  static description = 'destroy a logical replication publication'
+  static examples = [
+    '<%= config.bin %> <%= command.id %> DATABASE --name orders --app example-app --confirm example-app',
+  ]
+  static flags = {
+    app: Flags.app({required: true}),
+    confirm: Flags.string({char: 'c', description: 'pass in the app name to skip confirmation prompts'}),
+    name: Flags.string({description: 'name of the publication', required: true}),
+    remote: Flags.remote(),
+  }
+
+  async run(): Promise<void> {
+    const {args, flags} = await this.parse(DataPgLogicalReplicationPublicationsDestroy)
+    const addon = await resolveAdvancedDatabase(this, args.database, flags.app)
+    await hux.confirmCommand({comparison: flags.app, confirmation: flags.confirm})
+
+    try {
+      ux.action.start(`Destroying publication ${color.name(flags.name)} on ${color.datastore(addon.name)}`)
+      await this.dataApi.delete(`/data/postgres/v1/${addon.id}/logical-replication/publications/${encodeURIComponent(flags.name)}`)
+      ux.action.stop()
+    } catch (error) {
+      ux.action.stop(color.red('!'))
+      throw error
+    }
+  }
+}

src/commands/data/pg/logical-replication/publications/index.ts

@@ -0,0 +1,48 @@
+import {flags as Flags} from '@heroku-cli/command'
+import {hux} from '@heroku/heroku-cli-util'
+import {Args, ux} from '@oclif/core'
+
+import BaseCommand from '../../../../../lib/data/base-command.js'
+import {LogicalReplicationPublicationsResponse, resolveAdvancedDatabase} from '../../../../../lib/data/logical-replication.js'
+import {huxTableNoWrapOptions} from '../../../../../lib/utils/table-utils.js'
+
+export default class DataPgLogicalReplicationPublicationsIndex extends BaseCommand {
+  static aliases = ['data:pg:lr:publications']
+  static args = {
+    database: Args.string({
+      description: 'database name, database attachment name, or related config var on an app',
+      required: true,
+    }),
+  }
+  static description = 'list logical replication publications on a Postgres Advanced database'
+  static examples = [
+    '<%= config.bin %> <%= command.id %> DATABASE --app example-app',
+  ]
+  static flags = {
+    app: Flags.app({required: true}),
+    'no-wrap': Flags.noWrap(),
+    remote: Flags.remote(),
+  }
+
+  async run(): Promise<void> {
+    const {args, flags} = await this.parse(DataPgLogicalReplicationPublicationsIndex)
+    const addon = await resolveAdvancedDatabase(this, args.database, flags.app)
+    const {body: {items}} = await this.dataApi.get<LogicalReplicationPublicationsResponse>(`/data/postgres/v1/${addon.id}/logical-replication/publications`)
+
+    if (items.length === 0) {
+      ux.stdout(`No logical replication publications exist on ${addon.name}.`)
+      return
+    }
+
+    hux.table(items, {
+      Name: {get: publication => publication.name},
+      'New Tables': {get: publication => publication.target.automatically_includes_new_tables ? 'included' : 'not included'},
+      Owner: {get: publication => publication.owner},
+      Target: {
+        get: publication => publication.target.type === 'schemas'
+          ? `schemas: ${publication.target.schemas.join(', ')}`
+          : `tables: ${publication.current_tables.join(', ')}`,
+      },
+    }, huxTableNoWrapOptions(flags['no-wrap']))
+  }
+}

src/commands/data/pg/logical-replication/publications/info.ts

@@ -0,0 +1,40 @@
+import {flags as Flags} from '@heroku-cli/command'
+import {hux} from '@heroku/heroku-cli-util'
+import {Args} from '@oclif/core'
+
+import BaseCommand from '../../../../../lib/data/base-command.js'
+import {LogicalReplicationPublication, resolveAdvancedDatabase} from '../../../../../lib/data/logical-replication.js'
+
+export default class DataPgLogicalReplicationPublicationsInfo extends BaseCommand {
+  static aliases = ['data:pg:lr:publications:info']
+  static args = {
+    database: Args.string({
+      description: 'database name, database attachment name, or related config var on an app',
+      required: true,
+    }),
+  }
+  static description = 'show a logical replication publication on a Postgres Advanced database'
+  static examples = [
+    '<%= config.bin %> <%= command.id %> DATABASE --name orders --app example-app',
+  ]
+  static flags = {
+    app: Flags.app({required: true}),
+    name: Flags.string({description: 'name of the publication', required: true}),
+    remote: Flags.remote(),
+  }
+
+  async run(): Promise<void> {
+    const {args, flags} = await this.parse(DataPgLogicalReplicationPublicationsInfo)
+    const addon = await resolveAdvancedDatabase(this, args.database, flags.app)
+    const {body: publication} = await this.dataApi.get<LogicalReplicationPublication>(`/data/postgres/v1/${addon.id}/logical-replication/publications/${encodeURIComponent(flags.name)}`)
+    const target = publication.target.type === 'schemas' ? publication.target.schemas.join(', ') : publication.current_tables.join(', ')
+
+    hux.styledObject({
+      'Current Tables': publication.current_tables.join(', '),
+      'Includes New Tables': publication.target.automatically_includes_new_tables ? 'yes' : 'no',
+      Name: publication.name,
+      Owner: publication.owner,
+      Target: `${publication.target.type}: ${target}`,
+    }, ['Name', 'Owner', 'Target', 'Current Tables', 'Includes New Tables'])
+  }
+}

src/commands/data/pg/logical-replication/publications/update.ts

@@ -0,0 +1,54 @@
+import {flags as Flags} from '@heroku-cli/command'
+import {color} from '@heroku/heroku-cli-util'
+import {Args, ux} from '@oclif/core'
+
+import BaseCommand from '../../../../../lib/data/base-command.js'
+import {PublicationTarget, resolveAdvancedDatabase} from '../../../../../lib/data/logical-replication.js'
+
+export default class DataPgLogicalReplicationPublicationsUpdate extends BaseCommand {
+  static aliases = ['data:pg:lr:publications:update']
+  static args = {
+    database: Args.string({
+      description: 'database name, database attachment name, or related config var on an app',
+      required: true,
+    }),
+  }
+  static description = 'replace the target of a logical replication publication'
+  static examples = [
+    '<%= config.bin %> <%= command.id %> DATABASE --name orders --table public.orders --app example-app',
+    '<%= config.bin %> <%= command.id %> DATABASE --name application --schema public --app example-app',
+  ]
+  static flags = {
+    app: Flags.app({required: true}),
+    name: Flags.string({description: 'name of the publication', required: true}),
+    remote: Flags.remote(),
+    schema: Flags.string({description: 'schema to include, including new tables created in the schema', multiple: true}),
+    table: Flags.string({description: 'fully-qualified table to include', multiple: true}),
+  }
+
+  async run(): Promise<void> {
+    const {args, flags} = await this.parse(DataPgLogicalReplicationPublicationsUpdate)
+    const addon = await resolveAdvancedDatabase(this, args.database, flags.app)
+    const target = this.publicationTarget(flags.table, flags.schema)
+
+    try {
+      ux.action.start(`Updating publication ${color.name(flags.name)} on ${color.datastore(addon.name)}`)
+      await this.dataApi.put(`/data/postgres/v1/${addon.id}/logical-replication/publications/${encodeURIComponent(flags.name)}`, {body: {target}})
+      ux.action.stop()
+    } catch (error) {
+      ux.action.stop(color.red('!'))
+      throw error
+    }
+  }
+
+  private publicationTarget(tables?: string[], schemas?: string[]): PublicationTarget {
+    if (tables && schemas) {
+      ux.error('Specify either --table or --schema, not both.')
+    }
+
+    if (tables) return {tables, type: 'tables'}
+    if (schemas) return {schemas, type: 'schemas'}
+
+    ux.error('Specify at least one --table or --schema.')
+  }
+}

src/commands/data/pg/logical-replication/publishing/enable.ts

@@ -0,0 +1,39 @@
+import {flags as Flags} from '@heroku-cli/command'
+import {color} from '@heroku/heroku-cli-util'
+import {Args, ux} from '@oclif/core'
+
+import BaseCommand from '../../../../../lib/data/base-command.js'
+import {resolveAdvancedDatabase} from '../../../../../lib/data/logical-replication.js'
+
+export default class DataPgLogicalReplicationPublishingEnable extends BaseCommand {
+  static aliases = ['data:pg:lr:publishing:enable']
+  static args = {
+    database: Args.string({
+      description: 'database name, database attachment name, or related config var on an app',
+      required: true,
+    }),
+  }
+  static description = 'enable logical replication publishing for a Postgres Advanced database'
+  static examples = [
+    '<%= config.bin %> <%= command.id %> DATABASE --app example-app',
+  ]
+  static flags = {
+    app: Flags.app({required: true}),
+    remote: Flags.remote(),
+  }
+
+  async run(): Promise<void> {
+    const {args, flags} = await this.parse(DataPgLogicalReplicationPublishingEnable)
+    const addon = await resolveAdvancedDatabase(this, args.database, flags.app)
+
+    try {
+      ux.action.start(`Enabling logical replication publishing for ${color.datastore(addon.name)}`)
+      await this.dataApi.post(`/data/postgres/v1/${addon.id}/logical-replication/publishing/enable`)
+      ux.action.stop('requested')
+      ux.stdout(`Wait for ${color.datastore(addon.name)} to finish updating before creating publications. Use ${color.code(`heroku data:pg:info ${addon.name} --app ${flags.app}`)} to track progress.`)
+    } catch (error) {
+      ux.action.stop(color.red('!'))
+      throw error
+    }
+  }
+}

src/commands/data/pg/logical-replication/subscribing/enable.ts

@@ -0,0 +1,40 @@
+import {flags as Flags} from '@heroku-cli/command'
+import {color} from '@heroku/heroku-cli-util'
+import {Args, ux} from '@oclif/core'
+
+import BaseCommand from '../../../../../lib/data/base-command.js'
+import {resolveAdvancedDatabase} from '../../../../../lib/data/logical-replication.js'
+
+export default class DataPgLogicalReplicationSubscribingEnable extends BaseCommand {
+  static aliases = ['data:pg:lr:subscribing:enable']
+  static args = {
+    database: Args.string({
+      description: 'database name, database attachment name, or related config var on an app',
+      required: true,
+    }),
+  }
+  static description = 'enable logical replication subscribing for a Postgres Advanced database'
+  static examples = [
+    '<%= config.bin %> <%= command.id %> DATABASE --app example-app',
+  ]
+  static flags = {
+    app: Flags.app({required: true}),
+    remote: Flags.remote(),
+  }
+
+  async run(): Promise<void> {
+    const {args, flags} = await this.parse(DataPgLogicalReplicationSubscribingEnable)
+    const addon = await resolveAdvancedDatabase(this, args.database, flags.app)
+
+    try {
+      ux.action.start(`Enabling logical replication subscribing for ${color.datastore(addon.name)}`)
+      await this.dataApi.post(`/data/postgres/v1/${addon.id}/logical-replication/subscribing/enable`)
+      ux.action.stop('requested')
+      ux.stdout(`Wait for ${color.datastore(addon.name)} to finish updating before creating subscriptions. `
+        + `Use ${color.code(`heroku data:pg:info ${addon.name} --app ${flags.app}`)} to track progress.`)
+    } catch (error) {
+      ux.action.stop(color.red('!'))
+      throw error
+    }
+  }
+}

src/lib/data/logical-replication.ts

@@ -0,0 +1,49 @@
+import type {pg} from '@heroku/heroku-cli-util'
+
+import {color, utils} from '@heroku/heroku-cli-util'
+import {ux} from '@oclif/core'
+
+import type BaseCommand from './base-command.js'
+
+export type PublicationTarget
+  = | {schemas: string[], type: 'schemas'}
+    | {tables: string[], type: 'tables'}
+    | {type: 'all_customer_schemas'}
+
+type PublicationResponseTarget
+  = | {
+    automatically_includes_new_schemas: boolean
+    automatically_includes_new_tables: boolean
+    schemas: string[]
+    type: 'schemas'
+  }
+    | {
+      automatically_includes_new_schemas: boolean
+      automatically_includes_new_tables: boolean
+      tables: string[]
+      type: 'tables'
+    }
+
+export type LogicalReplicationPublication = {
+  current_tables: string[]
+  name: string
+  owner: string
+  target: PublicationResponseTarget
+}
+
+export type LogicalReplicationPublicationsResponse = {
+  count: number
+  items: LogicalReplicationPublication[]
+  limit: number
+}
+
+export async function resolveAdvancedDatabase(command: BaseCommand, database: string, app: string): Promise<pg.ExtendedAddon> {
+  const addonResolver = new utils.AddonResolver(command.heroku)
+  const addon = await addonResolver.resolve(database, app, utils.pg.addonService())
+
+  if (!utils.pg.isAdvancedDatabase(addon)) {
+    ux.error(`You can only use this command on Advanced-tier databases.\nUse ${color.code(`heroku data:pg:info ${database} --app ${app}`)} to inspect an Advanced database.`)
+  }
+
+  return addon
+}