Schema-as-Configuration
for Data & Streaming

# main.kl - stream CDC from MSSQL into Snowflake (end-to-end)
provider "mssql" {
  host     = "mssql.internal"
  database = "orders"
  username = var.mssql_user
  password = var.mssql_password
}

provider "snowflake" {
  account   = var.snowflake_account
  database  = "ANALYTICS"
  schema    = "CDC"
  warehouse = "COMPUTE_WH"
  role      = "CDC_SYNC"
}

# Source table with PII column
create "mssql_table" "users" {
  name   = "users"
  schema = "dbo"

  columns = {
    id = {
      type        = "uniqueidentifier"
      default     = "NEWSEQUENTIALID()"
      primary_key = true
    }
    email = {
      type     = "varchar(255)"
      not_null = true
      unique   = true
    }
    created_at = {
      type    = "datetime2"
      default = "SYSUTCDATETIME()"
    }
  }
}

# Snowflake roles/users and masking for PII
create "snowflake_role" "cdc_consumer" {
  name = "CDC_CONSUMER"
}

create "snowflake_user" "cdc_sync_user" {
  name         = "CDC_SYNC_USER"
  default_role = snowflake_role.cdc_consumer.name
  password     = var.snowflake_password
}

create "snowflake_masking_policy" "mask_email" {
  name = "MASK_EMAIL"
  body = "CASE WHEN CURRENT_ROLE() = 'CDC_CONSUMER' THEN email ELSE REGEXP_REPLACE(email, '(.).+(@.*)', '\\\\1***\\\\2') END"
}

# Snowflake landing table with masking applied
create "snowflake_table" "users_replica" {
  database = "ANALYTICS"
  schema   = "CDC"
  name     = "USERS"

  columns = {
    id = {
      type        = "BINARY(16)"
      primary_key = true
    }
    email = {
      type           = "VARCHAR"
      masking_policy = snowflake_masking_policy.mask_email.name
    }
    created_at = {
      type = "TIMESTAMP_NTZ"
    }
  }
}

create "stream_sink" "snowflake_replica" {
  type = "snowflake"
  connection_info = {
    account   = var.snowflake_account
    user      = "CDC_SYNC_USER"
    password  = var.snowflake_password
    database  = "ANALYTICS"
    schema    = "CDC"
    warehouse = "COMPUTE_WH"
  }
  batch_mode    = "merge"
  buffer_size   = 1000
  create_tables = true
}

create "stream_route" "mssql_orders_to_snowflake" {
  source = {
    type        = "mssql_cdc"
    database    = "orders"
    slot_name   = "cdc_orders_slot"
    publication = "cdc_orders_publication"
  }

  sink = stream_sink.snowflake_replica.id

  filter {
    event_types = ["DML"]
    conditions  = ["schema NOT IN ('cdc', 'sys')"]
  }

  throughput_targets {
    target_tps   = 1000
    worker_count = 1
  }

  # Optional: allow DDL table creates from CDC (if desired)
  # filter {
  #   event_types = ["DDL", "DML"]
  #   conditions  = ["operation IN ('CREATE TABLE','ALTER TABLE','DROP TABLE')"]
  # }
}