source-objects.tf

// Datastream cannot create anything on the source database; it can only read a publication and a replication
// slot that already exist. These resources create them (and the user that reads them) through the postgres
// block's db admin function.
//
// Every resource adopts an object that already exists (`useExisting`), so an environment whose objects were
// created by hand keeps applying cleanly.

locals {
  // A schema with no tables listed replicates every table in the schema; an empty list replicates everything.
  // The publication is derived from the same variable that shapes the stream, so the two cannot drift.
  publication_all_tables = length(var.replication_objects) < 1
  publication_schemas    = [for obj in var.replication_objects : obj.schema if length(coalesce(obj.tables, [])) < 1]
  publication_tables = flatten([
    for obj in var.replication_objects : [
      for table in coalesce(obj.tables, []) : { schema = obj.schema, name = table }
    ]
  ])

  // `*` grants on every non-system schema, which is what an empty replication_objects replicates
  grant_schemas = local.publication_all_tables ? ["*"] : [for obj in var.replication_objects : obj.schema]
}

// Creating a logical replication slot is denied unless the role doing it has the REPLICATION attribute.
// Cloud SQL permits members of `cloudsqlsuperuser`, which includes the admin user the db admin function connects as, to grant the attribute, so the function can grant it to itself.
resource "restapi_object" "admin_replication" {
  path         = "/roles"
  id_attribute = "name"
  object_id    = local.db_admin_username
  force_new    = [local.db_admin_username]
  destroy_path = "/skip"

  data = jsonencode({
    name        = local.db_admin_username
    useExisting = true
    attributes = {
      replication = true
    }
  })
}

// Creates a postgres role that datastream uses to perform replication
resource "restapi_object" "role" {
  path         = "/roles"
  id_attribute = "name"
  object_id    = local.postgres_username
  force_new    = [local.postgres_username]
  destroy_path = "/skip"

  data = jsonencode({
    name        = local.postgres_username
    password    = local.postgres_password
    useExisting = true
    attributes = {
      replication = true
    }
  })

  depends_on = [restapi_object.admin_replication]
}

// Grants read-only access to the tables in replication scope
resource "restapi_object" "table_privileges" {
  for_each = toset(local.grant_schemas)

  path         = "/databases/${var.postgres_database}/table_privileges"
  id_attribute = "id"
  object_id    = "${each.value}::${local.postgres_username}"
  force_new    = [var.postgres_database, each.value, local.postgres_username]
  read_path    = "/databases/${var.postgres_database}/table_privileges/${each.value}::${local.postgres_username}"
  update_path  = "/databases/${var.postgres_database}/table_privileges/${each.value}::${local.postgres_username}"
  destroy_path = "/skip"

  data = jsonencode({
    database   = var.postgres_database
    schema     = each.value
    role       = local.postgres_username
    privileges = ["SELECT"]
    // Default privileges only cover objects created by a specific role, so this covers every role that
    // currently owns a table in the schema -- typically the application roles that run migrations
    includeFuture         = true
    futureFromTableOwners = true
    grantConnect          = true
  })

  depends_on = [restapi_object.role]
}

// Create a publication for the datastream
// This is destroyed with the module: `destroy_path` is intentionally left at its default (a real DELETE)
resource "restapi_object" "publication" {
  path         = "/databases/${var.postgres_database}/publications"
  id_attribute = "name"
  object_id    = var.replication_publication
  force_new    = [var.postgres_database, var.replication_publication]

  data = jsonencode({
    name        = var.replication_publication
    database    = var.postgres_database
    allTables   = local.publication_all_tables
    schemas     = local.publication_schemas
    tables      = local.publication_tables
    useExisting = true
  })

  depends_on = [restapi_object.table_privileges]
}

// Create a replication slot for datastream
// Unlike the other source objects, this one is really dropped on destroy: an abandoned logical slot retains
// WAL indefinitely and will eventually fill the instance's disk.
resource "restapi_object" "replication_slot" {
  path         = "/databases/${var.postgres_database}/replication_slots"
  id_attribute = "name"
  object_id    = var.replication_slot
  force_new    = [var.postgres_database, var.replication_slot]

  data = jsonencode({
    name        = var.replication_slot
    database    = var.postgres_database
    plugin      = "pgoutput"
    useExisting = true
  })

  lifecycle {
    precondition {
      condition     = local.logical_decoding_enabled
      error_message = <<EOF
The source database cannot support logical replication: wal_level is "${local.wal_level}", but a replication slot requires "logical".

Set the `cloudsql.logical_decoding` database flag to `on` on the postgres block this module connects to
(`gcp-cloudsql-postgres`, `var.db_flags`), then apply that block. The flag requires an instance restart,
so let the instance come back up before launching this block.
EOF
    }
  }

  depends_on = [
    restapi_object.admin_replication,
    restapi_object.publication,
  ]
}