# MSK Operational Dashboard
# ============================

locals {
  # Common legend configuration
  standard_legend_columns = ["avg", "min", "max", "value", "sum"]
  
  # Common timeseries style
  standard_style = {
    palette     = "dog_classic"
    order_by    = "values"
    line_type   = "solid"
    line_width  = "normal"
  }

  # Red palette for errors
  error_style = {
    palette     = "red"
    order_by    = "values"
    line_type   = "solid"
    line_width  = "normal"
  }

  # Widget size constants
  widget_full_width    = 12
  widget_third_width   = 4
  widget_header_height = 1
  widget_small_height  = 2
  widget_medium_height = 3
  widget_large_height  = 4

  # Dynamic topic category queries
  # Creates a list of objects with query names and topic patterns
  topic_category_queries = [
    for key, category in var.topic_categories : {
      query_name = "query${index(keys(var.topic_categories), key) + 1}"
      alias      = category.name
      pattern    = category.pattern
    }
  ]
}

# ========================================
# Main Dashboard Resource
# ========================================

resource "datadog_dashboard" "msk_operational_dashboard" {
  title            = "MSK: ${var.environment}-${var.cluster_name} operational dashboard"
  description      = "MSK Operational dashboard created by Terraform"
  restricted_roles = [data.datadog_role.datadog_admin_role.id]
  layout_type      = "ordered"
  reflow_type      = "fixed" 
  tags             = local.dashboard_tags

  # ========================================
  # Group 1: Cluster Health
  # ========================================
  widget {
    group_definition {
      title       = "Cluster Health"
      layout_type = "ordered"
      show_title  = true

      # Section Header: Resource Utilisation
      widget {
        free_text_definition {
          text       = "Resource Utilisation"
          color      = "#4d4d4d"
          font_size  = "36"
          text_align = "center"
        }
        widget_layout {
          x      = 0
          y      = 0
          width  = local.widget_full_width
          height = local.widget_header_height
        }
      }

      # CPU Usage
      widget {
        timeseries_definition {
          title       = "CPU Usage"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias   = "total_cpu"
              formula_expression = "query1 + query2"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.cpu_system{${local.cluster_filter}} by {broker_id}"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "max:aws.kafka.cpu_user{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = false
          }
        }
        widget_layout {
          x      = 0
          y      = 0
          width  = local.widget_third_width
          height = local.widget_medium_height
        }
      }

      # Used Memory
      widget {
        timeseries_definition {
          title       = "Used Memory"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.heap_memory_after_gc{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = false
          }

          marker {
            label        = "y = 60"
            value        = "y = 60"
            display_type = "warning dashed"
          }
        }
        widget_layout {
          x      = 4
          y      = 0
          width  = local.widget_third_width
          height = local.widget_medium_height
        }
      }

      # Data Disk Usage
      widget {
        timeseries_definition {
          title       = "Data Disk Usage"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Data logs"
              formula_expression = "query1"
            }

            formula {
              alias              = "App logs"
              formula_expression = "query2"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.kafka_data_logs_disk_used{${local.cluster_filter}}"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "max:aws.kafka.kafka_app_logs_disk_used{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = false
          }
        }
        widget_layout {
          x      = 8
          y      = 0
          width  = local.widget_third_width
          height = local.widget_medium_height
        }
      }

      # Section Header: Network
      widget {
        free_text_definition {
          text       = "Network"
          color      = "#4d4d4d"
          font_size  = "36"
          text_align = "center"
        }
        widget_layout {
          x      = 0
          y      = 3
          width  = local.widget_full_width
          height = local.widget_header_height
        }
      }

      # Received/Transmitted packets
      widget {
        timeseries_definition {
          title       = "Received/Transmitted packets"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Received Packets"
              formula_expression = "query1"
            }

            formula {
              alias              = "Transmitted Packets"
              formula_expression = "query2"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "sum:aws.kafka.network_rx_packets{${local.cluster_filter}}.as_count()"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "sum:aws.kafka.network_tx_packets{${local.cluster_filter}}.as_count()"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = false
          }
        }
        widget_layout {
          x      = 0
          y      = 4
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Packets Drop/Errors
      widget {
        timeseries_definition {
          title       = "Packets Drop/Errors"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Receive Errors/Dropped"
              formula_expression = "query1 + query2"
            }

            formula {
              alias              = "Transmit Errors/Dropped"
              formula_expression = "query3 + query4"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "sum:aws.kafka.network_rx_dropped{${local.cluster_filter}}.as_count()"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "sum:aws.kafka.network_rx_errors{${local.cluster_filter}}.as_count()"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query3"
                query       = "sum:aws.kafka.network_tx_dropped{${local.cluster_filter}}.as_count()"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query4"
                query       = "sum:aws.kafka.network_tx_errors{${local.cluster_filter}}.as_count()"
              }
            }

            style {
              palette    = local.error_style.palette
              line_type  = local.error_style.line_type
              line_width = local.error_style.line_width
            }

            display_type = "line"
          }
        }
         widget_layout {
          x      = 4
          y      = 4
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Traffic Shaping
      widget {
        timeseries_definition {
          title       = "Traffic Shaping"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.traffic_shaping{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
         widget_layout {
          x      = 8
          y      = 4
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }
    }
     widget_layout {
          x      = 0
          y      = 0
          width  = local.widget_full_width
          height = 7
        }
  }

  # ========================================
  # Group 2: Partition Usage
  # ========================================
  widget {
    group_definition {
      title       = "Partition usage"
      layout_type = "ordered"
      show_title  = true

      # Global Count
      widget {
        query_value_definition {
          title       = "Global Count"
          title_size  = "16"
          title_align = "left"
          autoscale   = false
          precision   = 0

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.global_partition_count{${local.cluster_filter}}"
                aggregator  = "last"
              }
            }
          }

          timeseries_background {
            type = "area"
          }
        }
        widget_layout {
          x      = 0
          y      = 0
          width  = local.widget_third_width
          height = local.widget_header_height
        }
      }

      # Partitions per Broker
      widget {
        timeseries_definition {
          title         = "Partitions per Broker"
          title_size    = "16"
          title_align   = "left"
          show_legend   = false
          legend_layout = "horizontal"
          
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "used"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.partition_count{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = false
          }

          marker {
            label        = " broker limit "
            value        = "y = ${local.broker_partition_limit}"
            display_type = "error bold"
          }
        }
        widget_layout {
          x      = 4
          y      = 0
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Under-replicated partitions
      widget {
        timeseries_definition {
          title       = "Under-replicated partitions"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.under_replicated_partitions{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
         widget_layout {
          x      = 8
          y      = 0
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Offline Partitions
      widget {
        query_value_definition {
          title       = "Offline Partitions"
          title_size  = "16"
          title_align = "left"
          autoscale   = true
          precision   = 2

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.offline_partitions_count{${local.cluster_filter}}"
                aggregator  = "last"
              }
            }
          }

          timeseries_background {
            type = "area"
          }
        }
         widget_layout {
          x      = 0
          y      = 1
          width  = local.widget_third_width
          height = local.widget_header_height
        }
      }

      # Global Topic Count
      widget {
        query_value_definition {
          title       = "Global Topic Count"
          title_size  = "16"
          title_align = "left"
          autoscale   = true
          precision   = 2

          request {
            formula {
              formula_expression = "query1 * 3"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.global_topic_count{${local.cluster_filter}}"
                aggregator  = "last"
              }
            }
          }

          timeseries_background {
            type = "area"
          }
        }
         widget_layout {
          x      = 0
          y      = 2
          width  = local.widget_third_width
          height = local.widget_header_height
        }
      }

      # Leader Partitions per Broker
      widget {
        timeseries_definition {
          title       = "Leader Partitions per Broker"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "sum:aws.kafka.leader_count{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = false
          }
        }
         widget_layout {
          x      = 4
          y      = 2
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Partitions under Min InSync Replicas
      widget {
        timeseries_definition {
          title       = "Partitions under Min InSync Replicas"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.under_min_isr_partition_count{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
         widget_layout {
          x      = 8
          y      = 2
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }
    }
    widget_layout {
          x      = 0
          y      = 7
          width  = local.widget_full_width
          height = 5
    }
  }

  # ========================================
  # Group 3: Application Performance
  # ========================================
  widget {
    group_definition {
      title       = "Application Performance"
      layout_type = "ordered"
      show_title  = true

      # Section Header: Throughput
      widget {
        free_text_definition {
          text       = "Throughput"
          color      = "#4d4d4d"
          font_size  = "36"
          text_align = "center"
        }
        widget_layout {
          x      = 0
          y      = 0
          width  = local.widget_full_width
          height = local.widget_header_height
        }
      }

      # Bytes IN per second
      widget {
        timeseries_definition {
          title       = "Bytes IN per second"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.bytes_in_per_sec{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 0
          y      = 1
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Bytes OUT per second
      widget {
        timeseries_definition {
          title       = "Bytes OUT per second"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "max:aws.kafka.bytes_out_per_sec{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 4
          y      = 1
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Messages IN per second
      widget {
        timeseries_definition {
          title       = "Messages IN per second"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "sum:aws.kafka.messages_in_per_sec{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 8
          y      = 1
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Section Header: Producer Latency
      widget {
        free_text_definition {
          text       = "Producer Latency"
          color      = "#4d4d4d"
          font_size  = "36"
          text_align = "center"
        }
        widget_layout {
          x      = 0
          y      = 3
          width  = local.widget_full_width
          height = local.widget_header_height
        }
      }

      # Mean Produce time
      widget {
        timeseries_definition {
          title       = "Mean Produce time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Produce Time"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.produce_total_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = true
          }
        }
        widget_layout {
          x      = 0
          y      = 4
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Produce Request vs Response Queue mean time
      widget {
        timeseries_definition {
          title       = "Produce Request vs Response Queue mean time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Request Time"
              formula_expression = "query1"
            }

            formula {
              alias              = "Response Time"
              formula_expression = "query2"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.produce_request_queue_time_ms_mean{${local.cluster_filter}}"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "avg:aws.kafka.produce_response_queue_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 4
          y      = 4
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Produce Response Send Mean time
      widget {
        timeseries_definition {
          title       = "Produce Response Send Mean time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Send Time"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.produce_response_send_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 8
          y      = 4
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Section Header: Consumer Latency
      widget {
        free_text_definition {
          text       = "Consumer Latency"
          color      = "#4d4d4d"
          font_size  = "36"
          text_align = "center"
        }
        widget_layout {
          x      = 0
          y      = 6
          width  = local.widget_full_width
          height = local.widget_header_height
        }
      }

      # Mean Consume time
      widget {
        timeseries_definition {
          title       = "Mean Consume time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Consume Time"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.fetch_consumer_total_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }

          yaxis {
            include_zero = true
          }
        }
        widget_layout {
          x      = 0
          y      = 7
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Consume Request vs Response Queue mean time
      widget {
        timeseries_definition {
          title       = "Consume Request vs Response Queue mean time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Request Time"
              formula_expression = "query1"
            }

            formula {
              alias              = "Response Time"
              formula_expression = "query2"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.fetch_consumer_request_queue_time_ms_mean{${local.cluster_filter}}"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "avg:aws.kafka.fetch_consumer_response_queue_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 4
          y      = 7
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Consume Response Send Mean time
      widget {
        timeseries_definition {
          title       = "Consume Response Send Mean time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Send Time"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.fetch_consumer_response_send_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 8
          y      = 7
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }
    }
    widget_layout {
          x      = 0
          y      = 12
          width  = local.widget_full_width
          height = 10
    }
  }

  # ========================================
  # Group 4: Internal / Replication
  # ========================================
  widget {
    group_definition {
      title       = "Internal / Replication"
      layout_type = "ordered"
      show_title  = true

      # Replication Bytes IN per second
      widget {
        timeseries_definition {
          title       = "Replication Bytes IN per second"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.replication_bytes_in_per_sec{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 0
          y      = 0
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Replication Bytes OUT per second
      widget {
        timeseries_definition {
          title       = "Replication Bytes OUT per second"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.replication_bytes_out_per_sec{${local.cluster_filter}} by {broker_id}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 4
          y      = 0
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Follower mean Fetch time
      widget {
        timeseries_definition {
          title       = "Follower mean Fetch time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Fetch time"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.fetch_follower_total_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 8
          y      = 0
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Follower Fetch Request/Response mean time
      widget {
        timeseries_definition {
          title       = "Follower Fetch Request/Response mean time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Request Time"
              formula_expression = "query1"
            }

            formula {
              alias              = "Response Time"
              formula_expression = "query2"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.fetch_follower_request_queue_time_ms_mean{${local.cluster_filter}}"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "avg:aws.kafka.fetch_follower_response_queue_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 0
          y      = 2
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Follower mean Send time
      widget {
        timeseries_definition {
          title       = "Follower mean Send time"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Send time"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "avg:aws.kafka.fetch_follower_response_send_time_ms_mean{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 4
          y      = 2
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }

      # Message Conversions
      widget {
        timeseries_definition {
          title       = "Message Conversions"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            formula {
              alias              = "Conversions"
              formula_expression = "query1"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "sum:aws.kafka.produce_message_conversions_per_sec{${local.cluster_filter}}"
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "line"
          }
        }
        widget_layout {
          x      = 8
          y      = 2
          width  = local.widget_third_width
          height = local.widget_small_height
        }
      }
    }
    widget_layout {
          x      = 0
          y      = 22
          width  = local.widget_full_width
          height = 5
    }
  }

  # ========================================
  # Group 5: Topics
  # ========================================
  widget {
    group_definition {
      title       = "Topics"
      layout_type = "ordered"
      show_title  = true

      # Low usage topics table
      widget {
        query_table_definition {
          title       = "Low usage topics"
          title_size  = "16"
          title_align = "left"

          request {
            formula {
              alias                = "Bytes Out"
              cell_display_mode    = "bar"
              formula_expression   = "query1"
            }

            formula {
              alias                = "Bytes In"
              cell_display_mode    = "bar"
              formula_expression   = "query2"
            }

            formula {
              alias                = "Offset Lag"
              cell_display_mode    = "bar"
              formula_expression   = "query3"
            }

            formula {
              alias                = "Messages In"
              cell_display_mode    = "bar"
              formula_expression   = "query4"
            }

            formula {
              alias                = "Total"
              formula_expression   = "query1 + query2 + query3 + query4"
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query1"
                query       = "sum:aws.kafka.bytes_out_per_sec{${local.cluster_filter}} by {topic}"
                aggregator  = "sum"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query2"
                query       = "sum:aws.kafka.bytes_in_per_sec{${local.cluster_filter}} by {topic}"
                aggregator  = "sum"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query3"
                query       = "sum:aws.kafka.sum_offset_lag{${local.cluster_filter}} by {topic}"
                aggregator  = "sum"
              }
            }

            query {
              metric_query {
                data_source = "metrics"
                name        = "query4"
                query       = "sum:aws.kafka.messages_in_per_sec{${local.cluster_filter}} by {topic}"
                aggregator  = "sum"
              }
            }
          }

          has_search_bar = "auto"
        }
        widget_layout {
          x      = 0
          y      = 0
          width  = local.widget_full_width
          height = local.widget_medium_height
        }
      }

      # Topic Usage by category
      widget {
        timeseries_definition {
          title       = "Topic Usage by category"
          title_size  = "16"
          title_align = "left"
          show_legend = true
          
          legend_layout  = "auto"
          legend_columns = local.standard_legend_columns

          request {
            # Dynamic formulas based on topic_categories variable
            dynamic "formula" {
              for_each = local.topic_category_queries
              content {
                alias              = formula.value.alias
                formula_expression = "count_not_null(${formula.value.query_name})"
              }
            }

            # Dynamic queries based on topic_categories variable
            dynamic "query" {
              for_each = local.topic_category_queries
              content {
                metric_query {
                  data_source = "metrics"
                  name        = query.value.query_name
                  query       = "avg:aws.kafka.messages_in_per_sec{${local.cluster_filter},topic:${query.value.pattern}} by {topic}"
                }
              }
            }

            style {
              palette    = local.standard_style.palette
              line_type  = local.standard_style.line_type
              line_width = local.standard_style.line_width
            }

            display_type = "area"
          }
        }
        widget_layout {
          x      = 0
          y      = 3
          width  = local.widget_full_width
          height = local.widget_large_height
        }
      }
    }
    widget_layout {
          x      = 0
          y      = 27
          width  = local.widget_full_width
          height = 8
    }
  }
}