☁️ Google Cloud Datastream Stream Terraform Module

module "orders_cdc_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "orders-mysql-to-gcs"
  stream_id    = "orders-mysql-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      include_objects = {
        mysql_databases = [
          { database = "orders_db" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/orders/"
      avro_file_format = true
    }
  }

  backfill_none = true
}

module "customers_cdc_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "customers-mysql-to-gcs"
  stream_id    = "customers-mysql-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      include_objects = {
        mysql_databases = [
          { database = "customers_db" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/customers/"
      avro_file_format = true
    }
  }

  backfill_all = {
    mysql_excluded_objects = {
      mysql_databases = [
        {
          database = "customers_db"
          mysql_tables = [
            { table = "audit_log" }
          ]
        }
      ]
    }
  }
}

module "pg_orders_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "pg-orders-to-bq"
  stream_id    = "pg-orders-to-bq"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.postgresql_source_profile.id
    postgresql_source_config = {
      replication_slot = "datastream_orders_slot"
      publication      = "datastream_orders_pub"
      include_objects = {
        postgresql_schemas = [
          { schema = "public" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.bq_destination_profile.id
    bigquery_destination_config = {
      single_target_dataset = {
        dataset_id = module.orders_dataset.id
      }
      merge = true
    }
  }

  backfill_none = true
}

module "pg_inventory_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "pg-inventory-to-gcs"
  stream_id    = "pg-inventory-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.postgresql_source_profile.id
    postgresql_source_config = {
      replication_slot = "datastream_inventory_slot"
      publication      = "datastream_inventory_pub"
      include_objects = {
        postgresql_schemas = [
          { schema = "inventory" }
        ]
      }
      exclude_objects = {
        postgresql_schemas = [
          {
            schema = "inventory"
            postgresql_tables = [
              { table = "staging_temp" }
            ]
          }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path = "cdc/inventory/"
      json_file_format = {
        schema_file_format = "NO_SCHEMA_FILE"
        compression        = "GZIP"
      }
    }
  }

  backfill_none = true
}

module "oracle_finance_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "oracle-finance-to-bq"
  stream_id    = "oracle-finance-to-bq"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.oracle_source_profile.id
    oracle_source_config = {
      include_objects = {
        oracle_schemas = [
          { schema = "FINANCE" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.bq_destination_profile.id
    bigquery_destination_config = {
      single_target_dataset = {
        dataset_id = module.finance_dataset.id
      }
      merge = true
    }
  }

  backfill_none = true
}

module "sqlserver_hr_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "sqlserver-hr-to-gcs"
  stream_id    = "sqlserver-hr-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.sqlserver_source_profile.id
    sql_server_source_config = {
      change_tables = true
      include_objects = {
        schemas = [
          { schema = "hr" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/hr/"
      avro_file_format = true
    }
  }

  backfill_none = true
}

module "multi_schema_bq_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "oracle-multi-schema-to-bq"
  stream_id    = "oracle-multi-schema-to-bq"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.oracle_source_profile.id
    oracle_source_config = {
      include_objects = {
        oracle_schemas = [
          { schema = "SALES" },
          { schema = "MARKETING" },
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.bq_destination_profile.id
    bigquery_destination_config = {
      source_hierarchy_datasets = {
        dataset_template = {
          location          = "us-east1"
          dataset_id_prefix = "oracle_cdc"
          # A SEPARATE, independently-optional CMEK reference from the stream-level
          # customer_managed_encryption_key — scoped only to datasets created by this template.
          kms_key_name = module.kms_keyring.crypto_key_ids["bq-dataset-key"]
        }
      }
      merge = true
    }
  }

  backfill_none = true
}

module "staged_mysql_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "staged-mysql-to-gcs"
  stream_id    = "staged-mysql-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      include_objects = {
        mysql_databases = [
          { database = "staging_db" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/staging/"
      avro_file_format = true
    }
  }

  backfill_none = true

  # Explicit and deliberate — fully configured, but replicates nothing yet.
  desired_state = "NOT_STARTED"
}

desired_state = "RUNNING"

module "encrypted_mysql_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "encrypted-mysql-to-gcs"
  stream_id    = "encrypted-mysql-to-gcs"
  location     = "us-east1"

  customer_managed_encryption_key = module.kms_keyring.crypto_key_ids["datastream-key"]

  source_config = {
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      include_objects = {
        mysql_databases = [
          { database = "sensitive_db" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/sensitive/"
      avro_file_format = true
    }
  }

  backfill_none = true
}

# Variant A: JSON output, no embedded schema, gzip-compressed.
module "json_output_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "orders-mysql-to-gcs-json"
  stream_id    = "orders-mysql-to-gcs-json"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      include_objects = {
        mysql_databases = [{ database = "orders_db" }]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path = "cdc/orders-json/"
      json_file_format = {
        schema_file_format = "NO_SCHEMA_FILE"
        compression        = "GZIP"
      }
    }
  }

  backfill_none = true
}

# Variant B: AVRO output (self-describing schema, this module's Quick Start default).
module "avro_output_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "orders-mysql-to-gcs-avro"
  stream_id    = "orders-mysql-to-gcs-avro"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      include_objects = {
        mysql_databases = [{ database = "orders_db" }]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/orders-avro/"
      avro_file_format = true
    }
  }

  backfill_none = true
}

module "gtid_mysql_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "gtid-mysql-to-gcs"
  stream_id    = "gtid-mysql-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      gtid = true
      include_objects = {
        mysql_databases = [
          { database = "orders_db" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/orders-gtid/"
      avro_file_format = true
    }
  }

  backfill_none = true
}

module "oracle_docs_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "oracle-docs-to-gcs"
  stream_id    = "oracle-docs-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.oracle_source_profile.id
    oracle_source_config = {
      drop_large_objects = true
      include_objects = {
        oracle_schemas = [
          { schema = "DOCS" }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/docs/"
      avro_file_format = true
    }
  }

  backfill_none = true
}

module "sqlserver_payroll_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "sqlserver-payroll-to-gcs"
  stream_id    = "sqlserver-payroll-to-gcs"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.sqlserver_source_profile.id
    sql_server_source_config = {
      transaction_logs = true
      include_objects = {
        schemas = [
          {
            schema = "payroll"
            tables = [
              {
                table = "employee_pay"
                columns = [
                  { column = "employee_id" },
                  { column = "pay_period" },
                  { column = "gross_pay" },
                ]
              }
            ]
          }
        ]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.gcs_destination_profile.id
    gcs_destination_config = {
      path             = "cdc/payroll/"
      avro_file_format = true
    }
  }

  backfill_none = true
}

# Variant A: merge (BigQuery's own default write mode — reflects current source state, no history).
module "merge_mode_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "orders-pg-to-bq-merge"
  stream_id    = "orders-pg-to-bq-merge"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.postgresql_source_profile.id
    postgresql_source_config = {
      replication_slot = "datastream_orders_slot"
      publication      = "datastream_orders_pub"
      include_objects = {
        postgresql_schemas = [{ schema = "public" }]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.bq_destination_profile.id
    bigquery_destination_config = {
      single_target_dataset = { dataset_id = module.orders_dataset.id }
      merge                 = true
    }
  }

  backfill_none = true
}

# Variant B: append_only (retains full historical change events — INSERT/UPDATE-INSERT/
# UPDATE-DELETE/DELETE — as distinct rows).
module "append_only_mode_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "orders-pg-to-bq-append"
  stream_id    = "orders-pg-to-bq-append"
  location     = "us-east1"

  source_config = {
    source_connection_profile = module.postgresql_source_profile.id
    postgresql_source_config = {
      replication_slot = "datastream_orders_audit_slot"
      publication      = "datastream_orders_audit_pub"
      include_objects = {
        postgresql_schemas = [{ schema = "public" }]
      }
    }
  }

  destination_config = {
    destination_connection_profile = module.bq_destination_profile.id
    bigquery_destination_config = {
      single_target_dataset = { dataset_id = module.orders_audit_dataset.id }
      append_only           = true
    }
  }

  backfill_none = true
}

# Source: an existing MySQL server, referenced by a source-type connection profile.
module "mysql_source_profile" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-connection-profile.git?ref=v1.0.0"

  display_name          = "orders-mysql-source"
  connection_profile_id = "orders-mysql-source"
  location              = "us-east1"

  mysql_profile = {
    hostname                       = "10.20.0.10"
    username                       = "datastream_reader"
    secret_manager_stored_password = module.mysql_reader_secret.secret_version_id
  }
}

# Destination: a BigQuery-type connection profile.
module "bq_destination_profile" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-connection-profile.git?ref=v1.0.0"

  display_name          = "orders-bq-destination"
  connection_profile_id = "orders-bq-destination"
  location              = "us-east1"

  bigquery_profile = {}
}

# The BigQuery dataset this stream's single_target_dataset targets.
module "orders_dataset" {
  source = "git::https://github.com/microsoftexpert/terraform-google-bigquery-dataset.git?ref=v1.0.0"

  dataset_id = "casey_orders_cdc"
  location   = "us-east1"
}

# The stream itself — this module.
module "orders_cdc_stream" {
  source = "git::https://github.com/microsoftexpert/terraform-google-datastream-stream.git?ref=v1.0.0"

  display_name = "orders-mysql-to-bq"
  stream_id    = "orders-mysql-to-bq"
  location     = "us-east1"

  source_config = {
    # Consumes terraform-google-datastream-connection-profile's id output (source instance).
    source_connection_profile = module.mysql_source_profile.id
    mysql_source_config = {
      include_objects = {
        mysql_databases = [
          { database = "orders_db" }
        ]
      }
    }
  }

  destination_config = {
    # Consumes terraform-google-datastream-connection-profile's id output (destination instance —
    # a SEPARATE instance from the source above).
    destination_connection_profile = module.bq_destination_profile.id
    bigquery_destination_config = {
      # Consumes terraform-google-bigquery-dataset's id output.
      single_target_dataset = {
        dataset_id = module.orders_dataset.id
      }
      merge = true
    }
  }

  backfill_all = {
    mysql_excluded_objects = {
      mysql_databases = [
        {
          database = "orders_db"
          mysql_tables = [
            { table = "session_cache" }
          ]
        }
      ]
    }
  }

  desired_state = "RUNNING"
}

variable "source_config" {
 type = object({
 source_connection_profile = string

 mysql_source_config = optional(object({
 max_concurrent_cdc_tasks = optional(number)
 max_concurrent_backfill_tasks = optional(number)
 binary_log_position = optional(bool, false)
 gtid = optional(bool, false)
 include_objects = optional(object({
 mysql_databases = list(object({
 database = string
 mysql_tables = optional(list(object({
 table = string
 mysql_columns = optional(list(object({
 column = optional(string)
 data_type = optional(string)
 collation = optional(string)
 primary_key = optional(bool)
 nullable = optional(bool)
 ordinal_position = optional(number)
 })), [])
 })), [])
 }))
 }))
 exclude_objects = optional(object({ mysql_databases = list(object({... })) }))
 }))

 oracle_source_config = optional(object({
 max_concurrent_cdc_tasks = optional(number)
 max_concurrent_backfill_tasks = optional(number)
 drop_large_objects = optional(bool, false)
 stream_large_objects = optional(bool, false)
 include_objects = optional(object({
 oracle_schemas = list(object({
 schema = string
 oracle_tables = optional(list(object({
 table = string
 oracle_columns = optional(list(object({
 column = optional(string)
 data_type = optional(string)
 })), [])
 })), [])
 }))
 }))
 exclude_objects = optional(object({ oracle_schemas = list(object({... })) }))
 }))

 postgresql_source_config = optional(object({
 replication_slot = string # REQUIRED once this block is set
 publication = string # REQUIRED once this block is set
 max_concurrent_backfill_tasks = optional(number)
 # No max_concurrent_cdc_tasks — confirmed absent for this engine.
 include_objects = optional(object({
 postgresql_schemas = list(object({
 schema = string
 postgresql_tables = optional(list(object({
 table = string
 postgresql_columns = optional(list(object({
 column = optional(string)
 data_type = optional(string)
 primary_key = optional(bool)
 nullable = optional(bool)
 ordinal_position = optional(number)
 })), [])
 })), [])
 }))
 }))
 exclude_objects = optional(object({ postgresql_schemas = list(object({... })) }))
 }))

 sql_server_source_config = optional(object({
 max_concurrent_cdc_tasks = optional(number)
 max_concurrent_backfill_tasks = optional(number)
 transaction_logs = optional(bool, false)
 change_tables = optional(bool, false)
 include_objects = optional(object({
 schemas = list(object({
 schema = string
 tables = optional(list(object({
 table = string
 columns = optional(list(object({
 column = optional(string)
 data_type = optional(string)
 })), [])
 })), [])
 }))
 }))
 exclude_objects = optional(object({ schemas = list(object({... })) }))
 }))
 })
}

variable "destination_config" {
  type = object({
    destination_connection_profile = string

    gcs_destination_config = optional(object({
      path                   = optional(string)
      file_rotation_mb       = optional(number)
      file_rotation_interval = optional(string)
      avro_file_format       = optional(bool, false)
      json_file_format = optional(object({
        schema_file_format = optional(string) # NO_SCHEMA_FILE | AVRO_SCHEMA_FILE
        compression        = optional(string) # NO_COMPRESSION | GZIP
      }))
    }))

    bigquery_destination_config = optional(object({
      data_freshness = optional(string)
      single_target_dataset = optional(object({
        dataset_id = string
      }))
      source_hierarchy_datasets = optional(object({
        project_id = optional(string)
        dataset_template = object({
          dataset_id_prefix = optional(string)
          kms_key_name      = optional(string)
          location          = string
        })
      }))
      merge       = optional(bool, false)
      append_only = optional(bool, false)
      blmt_config = optional(object({
        bucket          = string
        connection_name = string
        file_format     = string
        table_format    = string
        root_path       = optional(string)
      }))
    }))
  })
}