main.tf

###############################################################################
# Local derivations
#
# - replication_subnet_group_id resolves to the group this module creates, or
# the caller-supplied existing group.
# - endpoint_arns / endpoint_engine_names merge the two endpoint resource
# families (generic aws_dms_endpoint + dedicated aws_dms_s3_endpoint) into a
# single lookup keyed by the caller's own map keys, so replication_configs
# and replication_tasks can reference either family via one
# *_endpoint_key field. Callers must keep keys unique across both maps if
# they intend to reference across families.
# - certificate_arns resolves aws_dms_certificate ARNs for endpoint wiring by
# certificate_key.
###############################################################################

locals {
 replication_subnet_group_id = (var.create_replication_subnet_group
 ? aws_dms_replication_subnet_group.this["this"].id
: var.replication_subnet_group_id)

 certificate_arns = { for k, v in aws_dms_certificate.this: k => v.certificate_arn }

 endpoint_certificate_arn = {
 for k, v in var.endpoints:
 k => v.certificate_arn != null ? v.certificate_arn: (v.certificate_key != null ? local.certificate_arns[v.certificate_key]: null)
 }

 s3_endpoint_certificate_arn = {
 for k, v in var.s3_endpoints:
 k => v.certificate_arn != null ? v.certificate_arn: (v.certificate_key != null ? local.certificate_arns[v.certificate_key]: null)
 }

 endpoint_arns = merge({ for k, v in aws_dms_endpoint.this: k => v.endpoint_arn },
 { for k, v in aws_dms_s3_endpoint.this: k => v.endpoint_arn },)

 endpoint_engine_names = merge({ for k, v in aws_dms_endpoint.this: k => v.engine_name },
 { for k, v in aws_dms_s3_endpoint.this: k => "s3" },)

 replication_config_source_arn = {
 for k, v in var.replication_configs:
 k => v.source_endpoint_arn != null ? v.source_endpoint_arn: local.endpoint_arns[v.source_endpoint_key]
 }
 replication_config_target_arn = {
 for k, v in var.replication_configs:
 k => v.target_endpoint_arn != null ? v.target_endpoint_arn: local.endpoint_arns[v.target_endpoint_key]
 }

 replication_task_source_arn = {
 for k, v in var.replication_tasks:
 k => v.source_endpoint_arn != null ? v.source_endpoint_arn: local.endpoint_arns[v.source_endpoint_key]
 }
 replication_task_target_arn = {
 for k, v in var.replication_tasks:
 k => v.target_endpoint_arn != null ? v.target_endpoint_arn: local.endpoint_arns[v.target_endpoint_key]
 }
}

###############################################################################
# Replication subnet group (supporting)
#
# Created from caller-supplied subnets (>=2 AZs). Guarded via for_each so it
# materializes only when create_replication_subnet_group is true; otherwise
# the replication instance / serverless compute_config use an
# externally-supplied subnet group id. Requires the dms-vpc-role IAM role to
# already exist in the account (see SCOPE.md AWS Prerequisites).
###############################################################################

resource "aws_dms_replication_subnet_group" "this" {
 for_each = var.create_replication_subnet_group ? { this = true }: {}

 replication_subnet_group_id = var.replication_subnet_group_id
 replication_subnet_group_description = coalesce(var.replication_subnet_group_description, "Managed by Terraform (tf_mod_aws_dms)")
 subnet_ids = var.subnet_ids

 tags = var.tags
}

###############################################################################
# Replication instance (keystone — classic, instance-based compute)
#
# Always encrypted (no on/off toggle) — kms_key_arn null uses the AWS-managed
# aws/dms key. Secure baseline: private (publicly_accessible = false),
# auto_minor_version_upgrade = true.
###############################################################################

resource "aws_dms_replication_instance" "this" {
 replication_instance_id = var.replication_instance_id
 replication_instance_class = var.replication_instance_class

 allocated_storage = var.allocated_storage
 engine_version = var.engine_version
 kms_key_arn = var.kms_key_arn
 network_type = var.network_type
 dns_name_servers = length(var.dns_name_servers) > 0 ? join(",", var.dns_name_servers): null

 replication_subnet_group_id = local.replication_subnet_group_id
 vpc_security_group_ids = var.vpc_security_group_ids
 availability_zone = var.multi_az ? null: var.availability_zone
 multi_az = var.multi_az
 publicly_accessible = var.publicly_accessible

 preferred_maintenance_window = var.preferred_maintenance_window
 auto_minor_version_upgrade = var.auto_minor_version_upgrade
 allow_major_version_upgrade = var.allow_major_version_upgrade
 apply_immediately = var.apply_immediately

 dynamic "kerberos_authentication_settings" {
 for_each = var.kerberos_authentication_settings != null ? [var.kerberos_authentication_settings]: []
 content {
 key_cache_secret_iam_arn = kerberos_authentication_settings.value.key_cache_secret_iam_arn
 key_cache_secret_id = kerberos_authentication_settings.value.key_cache_secret_id
 krb5_file_contents = kerberos_authentication_settings.value.krb5_file_contents
 }
 }

 tags = var.tags

 dynamic "timeouts" {
 for_each = (var.timeouts.create != null || var.timeouts.update != null || var.timeouts.delete != null) ? [var.timeouts]: []
 content {
 create = try(timeouts.value.create, null)
 update = try(timeouts.value.update, null)
 delete = try(timeouts.value.delete, null)
 }
 }
}

###############################################################################
# Certificates (imported PEM / Oracle wallet)
###############################################################################

resource "aws_dms_certificate" "this" {
 for_each = var.certificates

 certificate_id = each.key
 certificate_pem = each.value.certificate_pem
 certificate_wallet = each.value.certificate_wallet

 tags = merge(var.tags, each.value.tags)
}

###############################################################################
# Endpoints (generic — every engine except S3)
#
# Polymorphic on engine_name: only the *_settings block matching the chosen
# engine should be populated by the caller. Each dynamic block below renders
# only when its corresponding *_settings object is non-null, so mismatched
# settings never reach the API.
###############################################################################

resource "aws_dms_endpoint" "this" {
 for_each = var.endpoints

 endpoint_id = each.key
 endpoint_type = each.value.endpoint_type
 engine_name = each.value.engine_name

 server_name = each.value.server_name
 port = each.value.port
 database_name = each.value.database_name

 username = each.value.username
 password = each.value.password

 secrets_manager_arn = each.value.secrets_manager_arn
 secrets_manager_access_role_arn = each.value.secrets_manager_access_role_arn

 ssl_mode = each.value.ssl_mode
 certificate_arn = local.endpoint_certificate_arn[each.key]
 kms_key_arn = each.value.kms_key_arn

 extra_connection_attributes = each.value.extra_connection_attributes
 service_access_role = each.value.service_access_role
 pause_replication_tasks = each.value.pause_replication_tasks

 dynamic "elasticsearch_settings" {
 for_each = each.value.elasticsearch_settings != null ? [each.value.elasticsearch_settings]: []
 content {
 endpoint_uri = elasticsearch_settings.value.endpoint_uri
 service_access_role_arn = elasticsearch_settings.value.service_access_role_arn
 error_retry_duration = try(elasticsearch_settings.value.error_retry_duration, null)
 full_load_error_percentage = try(elasticsearch_settings.value.full_load_error_percentage, null)
 use_new_mapping_type = try(elasticsearch_settings.value.use_new_mapping_type, null)
 }
 }

 dynamic "kafka_settings" {
 for_each = each.value.kafka_settings != null ? [each.value.kafka_settings]: []
 content {
 broker = kafka_settings.value.broker
 topic = try(kafka_settings.value.topic, null)
 message_format = try(kafka_settings.value.message_format, null)
 message_max_bytes = try(kafka_settings.value.message_max_bytes, null)
 include_control_details = try(kafka_settings.value.include_control_details, null)
 include_null_and_empty = try(kafka_settings.value.include_null_and_empty, null)
 include_partition_value = try(kafka_settings.value.include_partition_value, null)
 include_table_alter_operations = try(kafka_settings.value.include_table_alter_operations, null)
 include_transaction_details = try(kafka_settings.value.include_transaction_details, null)
 no_hex_prefix = try(kafka_settings.value.no_hex_prefix, null)
 partition_include_schema_table = try(kafka_settings.value.partition_include_schema_table, null)
 sasl_mechanism = try(kafka_settings.value.sasl_mechanism, null)
 sasl_username = try(kafka_settings.value.sasl_username, null)
 sasl_password = try(kafka_settings.value.sasl_password, null)
 security_protocol = try(kafka_settings.value.security_protocol, null)
 ssl_ca_certificate_arn = try(kafka_settings.value.ssl_ca_certificate_arn, null)
 ssl_client_certificate_arn = try(kafka_settings.value.ssl_client_certificate_arn, null)
 ssl_client_key_arn = try(kafka_settings.value.ssl_client_key_arn, null)
 ssl_client_key_password = try(kafka_settings.value.ssl_client_key_password, null)
 }
 }

 dynamic "kinesis_settings" {
 for_each = each.value.kinesis_settings != null ? [each.value.kinesis_settings]: []
 content {
 stream_arn = try(kinesis_settings.value.stream_arn, null)
 service_access_role_arn = try(kinesis_settings.value.service_access_role_arn, null)
 message_format = try(kinesis_settings.value.message_format, null)
 include_control_details = try(kinesis_settings.value.include_control_details, null)
 include_null_and_empty = try(kinesis_settings.value.include_null_and_empty, null)
 include_partition_value = try(kinesis_settings.value.include_partition_value, null)
 include_table_alter_operations = try(kinesis_settings.value.include_table_alter_operations, null)
 include_transaction_details = try(kinesis_settings.value.include_transaction_details, null)
 partition_include_schema_table = try(kinesis_settings.value.partition_include_schema_table, null)
 use_large_integer_value = try(kinesis_settings.value.use_large_integer_value, null)
 }
 }

 dynamic "mongodb_settings" {
 for_each = each.value.mongodb_settings != null ? [each.value.mongodb_settings]: []
 content {
 auth_mechanism = try(mongodb_settings.value.auth_mechanism, null)
 auth_source = try(mongodb_settings.value.auth_source, null)
 auth_type = try(mongodb_settings.value.auth_type, null)
 docs_to_investigate = try(mongodb_settings.value.docs_to_investigate, null)
 extract_doc_id = try(mongodb_settings.value.extract_doc_id, null)
 nesting_level = try(mongodb_settings.value.nesting_level, null)
 use_update_lookup = try(mongodb_settings.value.use_update_lookup, null)
 }
 }

 dynamic "mysql_settings" {
 for_each = each.value.mysql_settings != null ? [each.value.mysql_settings]: []
 content {
 after_connect_script = try(mysql_settings.value.after_connect_script, null)
 authentication_method = try(mysql_settings.value.authentication_method, null)
 clean_source_metadata_on_mismatch = try(mysql_settings.value.clean_source_metadata_on_mismatch, null)
 events_poll_interval = try(mysql_settings.value.events_poll_interval, null)
 execute_timeout = try(mysql_settings.value.execute_timeout, null)
 max_file_size = try(mysql_settings.value.max_file_size, null)
 parallel_load_threads = try(mysql_settings.value.parallel_load_threads, null)
 server_timezone = try(mysql_settings.value.server_timezone, null)
 service_access_role_arn = try(mysql_settings.value.service_access_role_arn, null)
 target_db_type = try(mysql_settings.value.target_db_type, null)
 }
 }

 dynamic "postgres_settings" {
 for_each = each.value.postgres_settings != null ? [each.value.postgres_settings]: []
 content {
 after_connect_script = try(postgres_settings.value.after_connect_script, null)
 authentication_method = try(postgres_settings.value.authentication_method, null)
 babelfish_database_name = try(postgres_settings.value.babelfish_database_name, null)
 capture_ddls = try(postgres_settings.value.capture_ddls, null)
 database_mode = try(postgres_settings.value.database_mode, null)
 ddl_artifacts_schema = try(postgres_settings.value.ddl_artifacts_schema, null)
 execute_timeout = try(postgres_settings.value.execute_timeout, null)
 fail_tasks_on_lob_truncation = try(postgres_settings.value.fail_tasks_on_lob_truncation, null)
 heartbeat_enable = try(postgres_settings.value.heartbeat_enable, null)
 heartbeat_frequency = try(postgres_settings.value.heartbeat_frequency, null)
 heartbeat_schema = try(postgres_settings.value.heartbeat_schema, null)
 map_boolean_as_boolean = try(postgres_settings.value.map_boolean_as_boolean, null)
 map_jsonb_as_clob = try(postgres_settings.value.map_jsonb_as_clob, null)
 map_long_varchar_as = try(postgres_settings.value.map_long_varchar_as, null)
 max_file_size = try(postgres_settings.value.max_file_size, null)
 plugin_name = try(postgres_settings.value.plugin_name, null)
 service_access_role_arn = try(postgres_settings.value.service_access_role_arn, null)
 slot_name = try(postgres_settings.value.slot_name, null)
 }
 }

 dynamic "redis_settings" {
 for_each = each.value.redis_settings != null ? [each.value.redis_settings]: []
 content {
 auth_type = redis_settings.value.auth_type
 server_name = redis_settings.value.server_name
 port = redis_settings.value.port
 auth_password = try(redis_settings.value.auth_password, null)
 auth_user_name = try(redis_settings.value.auth_user_name, null)
 ssl_ca_certificate_arn = try(redis_settings.value.ssl_ca_certificate_arn, null)
 ssl_security_protocol = try(redis_settings.value.ssl_security_protocol, null)
 }
 }

 dynamic "redshift_settings" {
 for_each = each.value.redshift_settings != null ? [each.value.redshift_settings]: []
 content {
 bucket_folder = try(redshift_settings.value.bucket_folder, null)
 bucket_name = try(redshift_settings.value.bucket_name, null)
 encryption_mode = try(redshift_settings.value.encryption_mode, null)
 server_side_encryption_kms_key_id = try(redshift_settings.value.server_side_encryption_kms_key_id, null)
 service_access_role_arn = try(redshift_settings.value.service_access_role_arn, null)
 }
 }

 dynamic "oracle_settings" {
 for_each = each.value.oracle_settings != null ? [each.value.oracle_settings]: []
 content {
 authentication_method = try(oracle_settings.value.authentication_method, null)
 use_logminer_reader = try(oracle_settings.value.use_logminer_reader, null)
 use_bfile = try(oracle_settings.value.use_bfile, null)
 access_alternate_directly = try(oracle_settings.value.access_alternate_directly, null)
 archived_log_dest_id = try(oracle_settings.value.archived_log_dest_id, null)
 additional_archived_log_dest_id = try(oracle_settings.value.additional_archived_log_dest_id, null)
 add_supplemental_logging = try(oracle_settings.value.add_supplemental_logging, null)
 char_length_semantics = try(oracle_settings.value.char_length_semantics, null)
 convert_timestamp_with_zone_to_utc = try(oracle_settings.value.convert_timestamp_with_zone_to_utc, null)
 fail_task_on_lob_truncation = try(oracle_settings.value.fail_task_on_lob_truncation, null)
 number_datatype_scale = try(oracle_settings.value.number_datatype_scale, null)
 retry_interval = try(oracle_settings.value.retry_interval, null)
 security_db_encryption = try(oracle_settings.value.security_db_encryption, null)
 security_db_encryption_name = try(oracle_settings.value.security_db_encryption_name, null)
 standby_delay_time = try(oracle_settings.value.standby_delay_time, null)
 trim_space_in_char = try(oracle_settings.value.trim_space_in_char, null)
 secrets_manager_oracle_asm_access_role_arn = try(oracle_settings.value.secrets_manager_oracle_asm_access_role_arn, null)
 secrets_manager_oracle_asm_secret_id = try(oracle_settings.value.secrets_manager_oracle_asm_secret_id, null)
 asm_server = try(oracle_settings.value.asm_server, null)
 asm_user = try(oracle_settings.value.asm_user, null)
 asm_password = try(oracle_settings.value.asm_password, null)
 }
 }

 tags = merge(var.tags, each.value.tags)
}

###############################################################################
# S3 endpoints (dedicated resource)
###############################################################################

resource "aws_dms_s3_endpoint" "this" {
 for_each = var.s3_endpoints

 endpoint_id = each.key
 endpoint_type = each.value.endpoint_type
 bucket_name = each.value.bucket_name
 service_access_role_arn = each.value.service_access_role_arn

 bucket_folder = each.value.bucket_folder
 certificate_arn = local.s3_endpoint_certificate_arn[each.key]
 ssl_mode = each.value.ssl_mode

 data_format = each.value.data_format
 compression_type = each.value.compression_type
 encryption_mode = each.value.encryption_mode
 server_side_encryption_kms_key_id = each.value.server_side_encryption_kms_key_id

 csv_delimiter = each.value.csv_delimiter
 csv_row_delimiter = each.value.csv_row_delimiter
 csv_null_value = each.value.csv_null_value

 date_partition_enabled = each.value.date_partition_enabled
 date_partition_sequence = each.value.date_partition_sequence
 date_partition_delimiter = each.value.date_partition_delimiter
 date_partition_timezone = each.value.date_partition_timezone

 cdc_path = each.value.cdc_path
 cdc_max_batch_interval = each.value.cdc_max_batch_interval
 cdc_min_file_size = each.value.cdc_min_file_size
 cdc_inserts_and_updates = each.value.cdc_inserts_and_updates
 cdc_inserts_only = each.value.cdc_inserts_only

 add_column_name = each.value.add_column_name
 include_op_for_full_load = each.value.include_op_for_full_load
 external_table_definition = each.value.external_table_definition
 glue_catalog_generation = each.value.glue_catalog_generation

 parquet_version = each.value.parquet_version
 row_group_length = each.value.row_group_length
 expected_bucket_owner = each.value.expected_bucket_owner
 rfc_4180 = each.value.rfc_4180
 ignore_header_rows = each.value.ignore_header_rows
 max_file_size = each.value.max_file_size
 timestamp_column_name = each.value.timestamp_column_name

 tags = merge(var.tags, each.value.tags)
}

###############################################################################
# Replication configs (DMS Serverless)
#
# Modern replacement for the classic replication_instance + replication_task
# pair — provisions compute on demand via compute_config, no standing
# instance required. compute_config falls back to this module's own subnet
# group / security groups / CMK when the caller leaves those fields null.
###############################################################################

resource "aws_dms_replication_config" "this" {
 for_each = var.replication_configs

 replication_config_identifier = each.key
 replication_type = each.value.replication_type
 source_endpoint_arn = local.replication_config_source_arn[each.key]
 target_endpoint_arn = local.replication_config_target_arn[each.key]
 table_mappings = each.value.table_mappings
 replication_settings = each.value.replication_settings
 supplemental_settings = each.value.supplemental_settings
 resource_identifier = each.value.resource_identifier
 start_replication = each.value.start_replication

 compute_config {
 max_capacity_units = each.value.compute_config.max_capacity_units
 min_capacity_units = try(each.value.compute_config.min_capacity_units, null)
 replication_subnet_group_id = coalesce(try(each.value.compute_config.replication_subnet_group_id, null), local.replication_subnet_group_id)
 vpc_security_group_ids = try(each.value.compute_config.vpc_security_group_ids, null) != null ? each.value.compute_config.vpc_security_group_ids: var.vpc_security_group_ids
 availability_zone = try(each.value.compute_config.availability_zone, null)
 dns_name_servers = try(each.value.compute_config.dns_name_servers, null) != null ? join(",", each.value.compute_config.dns_name_servers): null
 kms_key_id = try(each.value.compute_config.kms_key_id, null) != null ? each.value.compute_config.kms_key_id: var.kms_key_arn
 multi_az = try(each.value.compute_config.multi_az, false)
 preferred_maintenance_window = try(each.value.compute_config.preferred_maintenance_window, null)
 }

 tags = merge(var.tags, each.value.tags)
}

###############################################################################
# Replication tasks (classic, instance-based — runs on this module's
# replication instance)
###############################################################################

resource "aws_dms_replication_task" "this" {
 for_each = var.replication_tasks

 replication_task_id = each.key
 migration_type = each.value.migration_type
 replication_instance_arn = aws_dms_replication_instance.this.replication_instance_arn
 source_endpoint_arn = local.replication_task_source_arn[each.key]
 target_endpoint_arn = local.replication_task_target_arn[each.key]
 table_mappings = each.value.table_mappings
 replication_task_settings = each.value.replication_task_settings
 cdc_start_time = each.value.cdc_start_time
 cdc_start_position = each.value.cdc_start_position
 resource_identifier = each.value.resource_identifier
 start_replication_task = each.value.start_replication_task

 tags = merge(var.tags, each.value.tags)
}

###############################################################################
# Event subscriptions (opt-in)
###############################################################################

resource "aws_dms_event_subscription" "this" {
 for_each = var.event_subscriptions

 name = each.key
 sns_topic_arn = each.value.sns_topic_arn
 source_type = each.value.source_type
 source_ids = length(each.value.source_ids) > 0 ? each.value.source_ids: null
 event_categories = length(each.value.event_categories) > 0 ? each.value.event_categories: null
 enabled = each.value.enabled

 tags = merge(var.tags, each.value.tags)
}