# main.tf - Batch Feature Store Infrastructure
terraform {
  required_version = ">= 1.5.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

# EMR Cluster for feature computation
resource "aws_emr_cluster" "feature_store" {
  name          = "feature-store-cluster"
  release_label = "emr-7.0.0"
  applications  = ["Spark", "Hive", "Livy"]
  
  ec2_attributes {
    subnet_id                         = aws_subnet.private.id
    emr_managed_master_security_group = aws_security_group.emr_master.id
    emr_managed_slave_security_group  = aws_security_group.emr_slave.id
    instance_profile                  = aws_iam_instance_profile.emr_ec2_profile.arn
  }
  
  master_instance_group {
    instance_type = "m5.2xlarge"
    instance_count = 1
  }
  
  core_instance_group {
    instance_type  = "m5.4xlarge"
    instance_count = 4
    ebs_config {
      size                 = 256
      type                 = "gp3"
      volumes_per_instance = 1
    }
  }
  
  configurations_json = jsonencode([
    {
      "Classification" : "spark-defaults",
      "Properties" : {
        "spark.sql.adaptive.enabled" : "true",
        "spark.sql.adaptive.coalescePartitions.enabled" : "true",
        "spark.sql.adaptive.skewJoin.enabled" : "true",
        "spark.dynamicAllocation.enabled" : "true",
        "spark.serializer" : "org.apache.spark.serializer.KryoSerializer",
        "spark.sql.catalogImplementation" : "hive",
        "spark.hadoop.hive.metastore.client.factory.class" : "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory"
      }
    }
  ])
  
  service_role = aws_iam_role.emr_service_role.arn
  autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
  
  tags = {
    Project     = "feature-store"
    Environment = "production"
  }
}

# DynamoDB tables for feature serving
resource "aws_dynamodb_table" "feature_store" {
  name           = "feature-store"
  billing_mode   = "PAY_PER_REQUEST"
  hash_key       = "entity_id"
  range_key      = "feature_timestamp"
  
  attribute {
    name = "entity_id"
    type = "S"
  }
  
  attribute {
    name = "feature_timestamp"
    type = "N"
  }
  
  # GSI for feature type queries
  global_secondary_index {
    name               = "feature_type-index"
    hash_key           = "feature_type"
    range_key          = "feature_timestamp"
    projection_type    = "INCLUDE"
    non_key_attributes = ["entity_id", "feature_values", "feature_version"]
  }
  
  # GSI for feature version queries
  global_secondary_index {
    name               = "feature_version-index"
    hash_key           = "feature_version"
    range_key          = "feature_timestamp"
    projection_type    = "ALL"
  }
  
  ttl {
    attribute_name = "expiry_time"
    enabled        = true
  }
  
  point_in_time_recovery {
    enabled = true
  }
  
  tags = {
    Project     = "feature-store"
    Environment = "production"
  }
}

# S3 buckets for raw data and features
resource "aws_s3_bucket" "feature_store" {
  bucket = "feature-store-${var.environment}-${random_id.bucket_suffix.hex}"
  
  tags = {
    Project     = "feature-store"
    Environment = var.environment
  }
}

resource "aws_s3_bucket_versioning" "feature_store" {
  bucket = aws_s3_bucket.feature_store.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_lifecycle_configuration" "feature_store" {
  bucket = aws_s3_bucket.feature_store.id
  
  rule {
    id     = "raw-data-transition"
    status = "Enabled"
    
    filter {
      prefix = "raw/"
    }
    
    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }
    
    transition {
      days          = 90
      storage_class = "GLACIER"
    }
  }
  
  rule {
    id     = "feature-data-retention"
    status = "Enabled"
    
    filter {
      prefix = "features/"
    }
    
    expiration {
      days = 730  # 2 years retention
    }
  }
}

# Glue Data Catalog database
resource "aws_glue_catalog_database" "feature_store" {
  name = "feature_store"
  
  parameters = {
    description = "Feature store metadata database"
  }
}

# Step Functions for pipeline orchestration
resource "aws_sfn_state_machine" "feature_pipeline" {
  name     = "feature-pipeline"
  role_arn = aws_iam_role.step_functions.arn
  
  definition = jsonencode({
    "Comment" : "Batch Feature Computation Pipeline",
    "StartAt" : "ValidateInput",
    "States" : {
      "ValidateInput" : {
        "Type" : "Task",
        "Resource" : "arn:aws:states:::lambda:invoke",
        "Parameters" : {
          "FunctionName" : "${aws_lambda_function.validate_input.arn}",
          "Payload" : {
            "input.$" : "$"
          }
        },
        "Next" : "ComputeFeatures"
      },
      "ComputeFeatures" : {
        "Type" : "Task",
        "Resource" : "arn:aws:states:::elasticmapreduce:addStep.sync",
        "Parameters" : {
          "ClusterId" : aws_emr_cluster.feature_store.id,
          "Step" : {
            "Name" : "ComputeFeatures",
            "ActionOnFailure" : "TERMINATE_CLUSTER",
            "HadoopJarStep" : {
              "Jar" : "command-runner.jar",
              "Args" : [
                "spark-submit",
                "--deploy-mode",
                "cluster",
                "--class",
                "com.featurestore.BatchFeatureComputation",
                "s3://${aws_s3_bucket.feature_store.id}/jobs/feature-computation.jar",
                "--input-path",
                "s3://${aws_s3_bucket.feature_store.id}/raw/",
                "--output-path",
                "s3://${aws_s3_bucket.feature_store.id}/features/",
                "--feature-version",
                "v1.0"
              ]
            }
          }
        },
        "Next" : "LoadToDynamoDB"
      },
      "LoadToDynamoDB" : {
        "Type" : "Task",
        "Resource" : "arn:aws:states:::lambda:invoke",
        "Parameters" : {
          "FunctionName" : "${aws_lambda_function.load_to_dynamodb.arn}",
          "Payload" : {
            "feature_path.$" : "$.OutputPath",
            "feature_version.$" : "$.FeatureVersion"
          }
        },
        "End" : true
      }
    }
  })
  
  tags = {
    Project     = "feature-store"
    Environment = "production"
  }
}

  

# feature_computation.py - Advanced Spark Feature Computation
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql.window import Window
from datetime import datetime, timedelta
import json
from typing import Dict, List, Any

class BatchFeatureComputation:
    def __init__(self, spark_session: SparkSession):
        self.spark = spark_session
        self.feature_registry = {}
        
    def register_feature_definition(self, feature_name: str, 
                                  computation_func, 
                                  dependencies: List[str] = None):
        """Register feature computation logic with dependencies"""
        self.feature_registry[feature_name] = {
            'computation': computation_func,
            'dependencies': dependencies or []
        }
    
    def compute_user_features(self, users_df, transactions_df, 
                            feature_date: str, feature_version: str):
        """Compute comprehensive user features with point-in-time correctness"""
        
        # Filter data for point-in-time correctness
        transactions_pit = transactions_df.filter(
            col("transaction_timestamp") <= feature_date
        )
        
        users_pit = users_df.filter(
            col("created_at") <= feature_date
        )
        
        # User demographic features
        demographic_features = users_pit.select(
            col("user_id").alias("entity_id"),
            lit(feature_date).cast("timestamp").alias("feature_timestamp"),
            col("age"),
            col("gender"),
            col("location"),
            (year(current_date()) - year(col("created_at"))).alias("account_age_years"),
            when(col("premium_member") == True, 1).otherwise(0).alias("is_premium_member"),
            lit(feature_version).alias("feature_version"),
            lit("user_demographic").alias("feature_type")
        )
        
        # Transaction behavior features (last 30 days)
        thirty_days_ago = (datetime.strptime(feature_date, "%Y-%m-%d") - 
                          timedelta(days=30)).strftime("%Y-%m-%d")
        
        recent_transactions = transactions_pit.filter(
            col("transaction_timestamp") >= thirty_days_ago
        )
        
        # Window functions for sequential features
        user_window = Window.partitionBy("user_id").orderBy("transaction_timestamp")
        
        transaction_features = recent_transactions.groupBy("user_id").agg(
            count("*").alias("transaction_count_30d"),
            sum("amount").alias("total_spend_30d"),
            avg("amount").alias("avg_transaction_amount_30d"),
            stddev("amount").alias("std_transaction_amount_30d"),
            countDistinct("merchant_category").alias("unique_categories_30d"),
            sum(when(col("amount") > 100, 1).otherwise(0)).alias("large_transactions_30d"),
            
            # Time-based features
            datediff(
                lit(feature_date), 
                max("transaction_timestamp").cast("date")
            ).alias("days_since_last_transaction"),
            
            # Sequential features using window functions
            first("amount").over(user_window.rowsBetween(-10, -1)).alias("last_10_transactions_avg")
        ).withColumnRenamed("user_id", "entity_id")
        
        # Advanced feature: Spending patterns by day of week
        spending_patterns = recent_transactions.groupBy(
            "user_id", 
            dayofweek("transaction_timestamp").alias("day_of_week")
        ).agg(
            sum("amount").alias("daily_spend"),
            count("*").alias("daily_transactions")
        ).groupBy("user_id").pivot("day_of_week").agg(
            first("daily_spend").alias("spend"),
            first("daily_transactions").alias("transactions")
        ).fillna(0)
        
        # Feature: Transaction frequency changes
        current_period = recent_transactions.filter(
            col("transaction_timestamp") >= (datetime.strptime(feature_date, "%Y-%m-%d") - 
                                           timedelta(days=15)).strftime("%Y-%m-%d")
        ).groupBy("user_id").agg(
            count("*").alias("recent_transaction_count")
        )
        
        previous_period = transactions_pit.filter(
            (col("transaction_timestamp") >= (datetime.strptime(feature_date, "%Y-%m-%d") - 
                                            timedelta(days=30)).strftime("%Y-%m-%d")) &
            (col("transaction_timestamp") < (datetime.strptime(feature_date, "%Y-%m-%d") - 
                                           timedelta(days=15)).strftime("%Y-%m-%d"))
        ).groupBy("user_id").agg(
            count("*").alias("previous_transaction_count")
        )
        
        frequency_change = current_period.join(
            previous_period, "user_id", "left"
        ).fillna(0).withColumn(
            "transaction_frequency_change",
            when(col("previous_transaction_count") == 0, 0).otherwise(
                (col("recent_transaction_count") - col("previous_transaction_count")) / 
                col("previous_transaction_count")
            )
        ).select("user_id", "transaction_frequency_change")
        
        # Combine all features
        final_features = demographic_features \
            .join(transaction_features, "entity_id", "left") \
            .join(spending_patterns, "entity_id", "left") \
            .join(frequency_change.withColumnRenamed("user_id", "entity_id"), "entity_id", "left") \
            .fillna(0)
        
        return final_features
    
    def compute_rolling_window_features(self, df: DataFrame, entity_col: str, 
                                      timestamp_col: str, value_col: str,
                                      windows: List[int] = [7, 30, 90]):
        """Compute rolling window statistics for time-series features"""
        
        features = df
        
        for window_days in windows:
            window_spec = Window.partitionBy(entity_col) \
                              .orderBy(col(timestamp_col).cast("timestamp").cast("long")) \
                              .rangeBetween(-window_days * 86400, 0)
            
            features = features \
                .withColumn(f"rolling_avg_{window_days}d", 
                           avg(value_col).over(window_spec)) \
                .withColumn(f"rolling_std_{window_days}d", 
                           stddev(value_col).over(window_spec)) \
                .withColumn(f"rolling_sum_{window_days}d", 
                           sum(value_col).over(window_spec)) \
                .withColumn(f"rolling_count_{window_days}d", 
                           count(value_col).over(window_spec))
        
        return features
    
    def compute_cross_entity_features(self, primary_df: DataFrame, 
                                    secondary_df: DataFrame, 
                                    join_key: str, feature_prefix: str):
        """Compute features by joining with related entities"""
        
        # Aggregate secondary entity features
        secondary_agg = secondary_df.groupBy(join_key).agg(
            count("*").alias(f"{feature_prefix}_count"),
            sum("amount").alias(f"{feature_prefix}_total_amount"),
            avg("amount").alias(f"{feature_prefix}_avg_amount"),
            countDistinct("category").alias(f"{feature_prefix}_unique_categories")
        )
        
        # Join with primary entities
        cross_features = primary_df.join(secondary_agg, join_key, "left").fillna(0)
        
        return cross_features
    
    def save_features_to_s3(self, features_df: DataFrame, output_path: str, 
                          partition_cols: List[str] = None):
        """Save computed features to S3 with partitioning"""
        
        writer = features_df.write \
            .mode("overwrite") \
            .option("compression", "snappy")
        
        if partition_cols:
            writer = writer.partitionBy(*partition_cols)
        
        writer.parquet(output_path)
        
        # Write feature metadata
        feature_metadata = {
            "feature_count": features_df.count(),
            "computation_timestamp": datetime.now().isoformat(),
            "schema": features_df.schema.json(),
            "partition_columns": partition_cols or []
        }
        
        # Save metadata
        metadata_rdd = self.spark.sparkContext.parallelize([json.dumps(feature_metadata)])
        metadata_rdd.saveAsTextFile(f"{output_path}/_metadata/")
    
    def validate_features(self, features_df: DataFrame) -> Dict[str, Any]:
        """Validate feature quality and data integrity"""
        
        validation_results = {}
        
        # Check for null values
        null_counts = {}
        for column in features_df.columns:
            null_count = features_df.filter(col(column).isNull()).count()
            null_counts[column] = null_count
        
        validation_results["null_counts"] = null_counts
        
        # Check for data type consistency
        schema_validation = {}
        for field in features_df.schema.fields:
            schema_validation[field.name] = {
                "data_type": str(field.dataType),
                "nullable": field.nullable
            }
        
        validation_results["schema_validation"] = schema_validation
        
        # Statistical validation
        numeric_columns = [f.name for f in features_df.schema.fields 
                          if isinstance(f.dataType, (DoubleType, FloatType, IntegerType, LongType))]
        
        stats_validation = {}
        for column in numeric_columns:
            stats = features_df.select(
                mean(col(column)).alias("mean"),
                stddev(col(column)).alias("stddev"),
                min(col(column)).alias("min"),
                max(col(column)).alias("max")
            ).collect()[0]
            
            stats_validation[column] = {
                "mean": stats["mean"],
                "stddev": stats["stddev"],
                "min": stats["min"],
                "max": stats["max"]
            }
        
        validation_results["statistical_validation"] = stats_validation
        
        return validation_results

# Main execution
def main():
    spark = SparkSession.builder \
        .appName("BatchFeatureComputation") \
        .config("spark.sql.adaptive.enabled", "true") \
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
        .getOrCreate()
    
    # Initialize feature computation
    feature_engine = BatchFeatureComputation(spark)
    
    # Load source data
    users_df = spark.read.parquet("s3://feature-store/raw/users/")
    transactions_df = spark.read.parquet("s3://feature-store/raw/transactions/")
    
    # Compute features for specific date
    feature_date = "2025-01-25"
    feature_version = "v1.2"
    
    user_features = feature_engine.compute_user_features(
        users_df, transactions_df, feature_date, feature_version
    )
    
    # Validate features
    validation_results = feature_engine.validate_features(user_features)
    print("Feature validation results:", validation_results)
    
    # Save features
    feature_engine.save_features_to_s3(
        user_features, 
        "s3://feature-store/features/user_features/",
        partition_cols=["feature_timestamp"]
    )
    
    spark.stop()

if __name__ == "__main__":
    main()

  

# feature_serving.py - DynamoDB Feature Serving Layer
import boto3
from botocore.config import Config
from datetime import datetime, timedelta
import json
from typing import Dict, List, Any, Optional
import pandas as pd
from decimal import Decimal

class DynamoDBFeatureStore:
    def __init__(self, table_name: str, region: str = "us-east-1"):
        self.config = Config(
            retries={
                'max_attempts': 10,
                'mode': 'adaptive'
            },
            read_timeout=300,
            connect_timeout=300
        )
        
        self.dynamodb = boto3.resource('dynamodb', region_name=region, config=self.config)
        self.table = self.dynamodb.Table(table_name)
        self.client = boto3.client('dynamodb', region_name=region, config=self.config)
    
    def _convert_floats_to_decimals(self, obj):
        """Convert float values to Decimal for DynamoDB compatibility"""
        if isinstance(obj, list):
            return [self._convert_floats_to_decimals(item) for item in obj]
        elif isinstance(obj, dict):
            return {k: self._convert_floats_to_decimals(v) for k, v in obj.items()}
        elif isinstance(obj, float):
            return Decimal(str(obj))
        else:
            return obj
    
    def store_features(self, entity_id: str, features: Dict[str, Any], 
                      feature_timestamp: datetime, feature_version: str,
                      feature_type: str, ttl_days: int = 365):
        """Store computed features in DynamoDB with TTL"""
        
        # Prepare feature item
        feature_item = {
            'entity_id': entity_id,
            'feature_timestamp': int(feature_timestamp.timestamp()),
            'feature_type': feature_type,
            'feature_version': feature_version,
            'feature_values': self._convert_floats_to_decimals(features),
            'created_at': datetime.now().isoformat(),
            'expiry_time': int((datetime.now() + timedelta(days=ttl_days)).timestamp())
        }
        
        try:
            response = self.table.put_item(Item=feature_item)
            return True
        except Exception as e:
            print(f"Error storing features for {entity_id}: {e}")
            return False
    
    def batch_store_features(self, feature_batch: List[Dict[str, Any]]):
        """Batch store features for better performance"""
        
        with self.table.batch_writer() as batch:
            for feature_item in feature_batch:
                batch.put_item(Item=feature_item)
    
    def get_features(self, entity_id: str, feature_timestamp: datetime,
                    feature_type: str = None, feature_version: str = None) -> Optional[Dict[str, Any]]:
        """Retrieve features for a specific entity and timestamp"""
        
        key_conditions = {
            'entity_id': {
                'AttributeValueList': [{'S': entity_id}],
                'ComparisonOperator': 'EQ'
            },
            'feature_timestamp': {
                'AttributeValueList': [{'N': str(int(feature_timestamp.timestamp()))}],
                'ComparisonOperator': 'EQ'
            }
        }
        
        # Add filter conditions if provided
        query_kwargs = {
            'KeyConditions': key_conditions,
            'Limit': 1
        }
        
        if feature_type:
            query_kwargs['QueryFilter'] = {
                'feature_type': {
                    'AttributeValueList': [{'S': feature_type}],
                    'ComparisonOperator': 'EQ'
                }
            }
        
        try:
            response = self.client.query(
                TableName=self.table.name,
                **query_kwargs
            )
            
            if response['Items']:
                item = response['Items'][0]
                return self._convert_dynamo_to_python(item)
            else:
                return None
                
        except Exception as e:
            print(f"Error retrieving features for {entity_id}: {e}")
            return None
    
    def get_latest_features(self, entity_id: str, feature_type: str = None,
                          max_lookback_days: int = 30) -> Optional[Dict[str, Any]]:
        """Get the most recent features for an entity within lookback window"""
        
        lookback_timestamp = int((datetime.now() - timedelta(days=max_lookback_days)).timestamp())
        
        key_conditions = {
            'entity_id': {
                'AttributeValueList': [{'S': entity_id}],
                'ComparisonOperator': 'EQ'
            },
            'feature_timestamp': {
                'AttributeValueList': [{'N': str(lookback_timestamp)}],
                'ComparisonOperator': 'GT'
            }
        }
        
        query_kwargs = {
            'KeyConditions': key_conditions,
            'ScanIndexForward': False,  # Most recent first
            'Limit': 1
        }
        
        if feature_type:
            query_kwargs['QueryFilter'] = {
                'feature_type': {
                    'AttributeValueList': [{'S': feature_type}],
                    'ComparisonOperator': 'EQ'
                }
            }
        
        try:
            response = self.client.query(
                TableName=self.table.name,
                **query_kwargs
            )
            
            if response['Items']:
                item = response['Items'][0]
                return self._convert_dynamo_to_python(item)
            else:
                return None
                
        except Exception as e:
            print(f"Error retrieving latest features for {entity_id}: {e}")
            return None
    
    def get_feature_history(self, entity_id: str, start_time: datetime,
                          end_time: datetime, feature_type: str = None) -> List[Dict[str, Any]]:
        """Get feature history for an entity within a time range"""
        
        key_conditions = {
            'entity_id': {
                'AttributeValueList': [{'S': entity_id}],
                'ComparisonOperator': 'EQ'
            },
            'feature_timestamp': {
                'AttributeValueList': [
                    {'N': str(int(start_time.timestamp()))},
                    {'N': str(int(end_time.timestamp()))}
                ],
                'ComparisonOperator': 'BETWEEN'
            }
        }
        
        query_kwargs = {
            'KeyConditions': key_conditions
        }
        
        if feature_type:
            query_kwargs['QueryFilter'] = {
                'feature_type': {
                    'AttributeValueList': [{'S': feature_type}],
                    'ComparisonOperator': 'EQ'
                }
            }
        
        try:
            response = self.client.query(
                TableName=self.table.name,
                **query_kwargs
            )
            
            features = []
            for item in response['Items']:
                features.append(self._convert_dynamo_to_python(item))
            
            return features
            
        except Exception as e:
            print(f"Error retrieving feature history for {entity_id}: {e}")
            return []
    
    def batch_get_features(self, entity_ids: List[str], feature_timestamp: datetime,
                          feature_type: str = None) -> Dict[str, Any]:
        """Batch retrieve features for multiple entities"""
        
        keys = []
        for entity_id in entity_ids:
            key = {
                'entity_id': {'S': entity_id},
                'feature_timestamp': {'N': str(int(feature_timestamp.timestamp()))}
            }
            keys.append(key)
        
        request_items = {
            self.table.name: {
                'Keys': keys
            }
        }
        
        if feature_type:
            request_items[self.table.name]['ExpressionAttributeNames'] = {
                '#ft': 'feature_type'
            }
            request_items[self.table.name]['ExpressionAttributeValues'] = {
                ':ft': {'S': feature_type}
            }
            request_items[self.table.name]['FilterExpression'] = '#ft = :ft'
        
        try:
            response = self.client.batch_get_item(RequestItems=request_items)
            features = {}
            
            for item in response['Responses'][self.table.name]:
                entity_id = item['entity_id']['S']
                features[entity_id] = self._convert_dynamo_to_python(item)
            
            return features
            
        except Exception as e:
            print(f"Error in batch get features: {e}")
            return {}
    
    def _convert_dynamo_to_python(self, dynamo_item: Dict) -> Dict[str, Any]:
        """Convert DynamoDB item to Python native types"""
        result = {}
        
        for key, value in dynamo_item.items():
            if 'S' in value:
                result[key] = value['S']
            elif 'N' in value:
                # Try to convert to int first, then float
                num_str = value['N']
                if '.' in num_str:
                    result[key] = float(num_str)
                else:
                    result[key] = int(num_str)
            elif 'M' in value:
                result[key] = self._convert_dynamo_to_python(value['M'])
            elif 'L' in value:
                result[key] = [self._convert_dynamo_to_python(item) for item in value['L']]
            elif 'BOOL' in value:
                result[key] = value['BOOL']
            elif 'NULL' in value:
                result[key] = None
            else:
                result[key] = value
        
        return result
    
    def get_feature_statistics(self, feature_type: str, 
                             start_time: datetime, end_time: datetime) -> Dict[str, Any]:
        """Get statistics about stored features for monitoring"""
        
        # Use GSI for feature type queries
        response = self.client.query(
            TableName=self.table.name,
            IndexName='feature_type-index',
            KeyConditions={
                'feature_type': {
                    'AttributeValueList': [{'S': feature_type}],
                    'ComparisonOperator': 'EQ'
                },
                'feature_timestamp': {
                    'AttributeValueList': [
                        {'N': str(int(start_time.timestamp()))},
                        {'N': str(int(end_time.timestamp()))}
                    ],
                    'ComparisonOperator': 'BETWEEN'
                }
            },
            Select='COUNT'
        )
        
        stats = {
            'feature_count': response['Count'],
            'scanned_count': response['ScannedCount'],
            'feature_type': feature_type,
            'time_range': {
                'start': start_time.isoformat(),
                'end': end_time.isoformat()
            }
        }
        
        return stats

# Example usage
def example_usage():
    feature_store = DynamoDBFeatureStore("feature-store")
    
    # Store features
    features = {
        "transaction_count_30d": 45,
        "total_spend_30d": 1250.75,
        "avg_transaction_amount": 27.79,
        "is_premium_member": True
    }
    
    feature_store.store_features(
        entity_id="user_12345",
        features=features,
        feature_timestamp=datetime.now(),
        feature_version="v1.2",
        feature_type="user_behavior"
    )
    
    # Retrieve features
    latest_features = feature_store.get_latest_features("user_12345")
    print("Latest features:", latest_features)
    
    # Batch retrieval
    entity_ids = ["user_12345", "user_67890", "user_11111"]
    batch_features = feature_store.batch_get_features(
        entity_ids, 
        datetime.now() - timedelta(days=1)
    )
    print("Batch features:", batch_features)

if __name__ == "__main__":
    example_usage()

  

# monitoring.py - Feature Store Monitoring and Data Quality
import boto3
from datetime import datetime, timedelta
import json
import pandas as pd
from typing import Dict, List, Any
import logging
from dataclasses import dataclass

@dataclass
class DataQualityCheck:
    name: str
    check_type: str  # 'completeness', 'freshness', 'distribution', 'schema'
    threshold: float
    description: str

class FeatureStoreMonitor:
    def __init__(self, dynamodb_table: str, cloudwatch_namespace: str = "FeatureStore"):
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table(dynamodb_table)
        self.cloudwatch = boto3.client('cloudwatch')
        self.namespace = cloudwatch_namespace
        self.logger = logging.getLogger(__name__)
        
        # Define data quality checks
        self.quality_checks = [
            DataQualityCheck(
                name="feature_freshness",
                check_type="freshness",
                threshold=24,  # hours
                description="Features should be updated within 24 hours"
            ),
            DataQualityCheck(
                name="feature_completeness",
                check_type="completeness", 
                threshold=0.95,  # 95% completeness
                description="At least 95% of expected features should be available"
            ),
            DataQualityCheck(
                name="value_distribution",
                check_type="distribution",
                threshold=0.01,  # 1% outlier threshold
                description="Feature values should be within expected distribution"
            )
        ]
    
    def check_feature_freshness(self, feature_type: str, 
                              expected_update_frequency_hours: int = 24) -> Dict[str, Any]:
        """Check if features are being updated as expected"""
        
        # Get the most recent feature timestamp
        response = self.table.query(
            IndexName='feature_type-index',
            KeyConditionExpression='feature_type = :ft',
            ExpressionAttributeValues={':ft': feature_type},
            ScanIndexForward=False,  # Most recent first
            Limit=1
        )
        
        if not response['Items']:
            return {
                'check_name': 'feature_freshness',
                'status': 'FAILED',
                'message': f'No features found for type: {feature_type}',
                'last_update': None,
                'hours_since_update': None
            }
        
        latest_item = response['Items'][0]
        last_update_timestamp = latest_item['feature_timestamp']
        last_update_time = datetime.fromtimestamp(int(last_update_timestamp))
        hours_since_update = (datetime.now() - last_update_time).total_seconds() / 3600
        
        status = 'PASS' if hours_since_update <= expected_update_frequency_hours else 'FAIL'
        
        return {
            'check_name': 'feature_freshness',
            'status': status,
            'message': f'Last update: {hours_since_update:.1f} hours ago',
            'last_update': last_update_time.isoformat(),
            'hours_since_update': hours_since_update
        }
    
    def check_feature_completeness(self, feature_type: str, 
                                 expected_entity_count: int) -> Dict[str, Any]:
        """Check if all expected entities have features"""
        
        # Count distinct entities with features
        # Note: This is a simplified implementation
        # In production, you might need more sophisticated counting
        
        response = self.table.query(
            IndexName='feature_type-index',
            KeyConditionExpression='feature_type = :ft',
            ExpressionAttributeValues={':ft': feature_type},
            Select='COUNT'
        )
        
        feature_count = response['Count']
        completeness_ratio = feature_count / expected_entity_count
        
        status = 'PASS' if completeness_ratio >= 0.95 else 'FAIL'
        
        return {
            'check_name': 'feature_completeness',
            'status': status,
            'message': f'Completeness: {completeness_ratio:.2%} ({feature_count}/{expected_entity_count})',
            'completeness_ratio': completeness_ratio,
            'feature_count': feature_count,
            'expected_count': expected_entity_count
        }
    
    def run_data_quality_checks(self, feature_type: str, 
                              expected_entity_count: int = None) -> List[Dict[str, Any]]:
        """Run all data quality checks for a feature type"""
        
        results = []
        
        for check in self.quality_checks:
            if check.check_type == 'freshness':
                result = self.check_feature_freshness(feature_type)
            elif check.check_type == 'completeness' and expected_entity_count:
                result = self.check_feature_completeness(feature_type, expected_entity_count)
            else:
                continue
            
            results.append(result)
            
            # Publish to CloudWatch
            self._publish_metric(
                metric_name=f"{check.name}_{feature_type}",
                value=1 if result['status'] == 'PASS' else 0,
                dimensions={'FeatureType': feature_type, 'CheckName': check.name}
            )
        
        return results
    
    def monitor_feature_serving_latency(self):
        """Monitor feature retrieval latency"""
        # This would integrate with your serving layer
        # For example, you could use X-Ray or custom timing
        pass
    
    def track_feature_usage(self, feature_type: str, entity_count: int):
        """Track feature usage patterns"""
        
        self._publish_metric(
            metric_name="feature_usage_count",
            value=entity_count,
            dimensions={'FeatureType': feature_type}
        )
    
    def _publish_metric(self, metric_name: str, value: float, 
                       dimensions: Dict[str, str] = None):
        """Publish custom metric to CloudWatch"""
        
        metric_data = {
            'MetricName': metric_name,
            'Value': value,
            'Unit': 'Count',
            'Timestamp': datetime.now()
        }
        
        if dimensions:
            metric_data['Dimensions'] = [
                {'Name': k, 'Value': v} for k, v in dimensions.items()
            ]
        
        try:
            self.cloudwatch.put_metric_data(
                Namespace=self.namespace,
                MetricData=[metric_data]
            )
        except Exception as e:
            self.logger.error(f"Failed to publish metric {metric_name}: {e}")
    
    def create_dashboard(self, feature_types: List[str]):
        """Create CloudWatch dashboard for feature store monitoring"""
        
        dashboard_body = {
            "widgets": []
        }
        
        for feature_type in feature_types:
            # Add freshness widget
            dashboard_body["widgets"].append({
                "type": "metric",
                "properties": {
                    "metrics": [
                        [self.namespace, "feature_freshness_status", "FeatureType", feature_type]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "region": "us-east-1",
                    "title": f"{feature_type} - Freshness Status",
                    "yAxis": {
                        "left": {
                            "min": 0,
                            "max": 1
                        }
                    }
                }
            })
            
            # Add completeness widget
            dashboard_body["widgets"].append({
                "type": "metric", 
                "properties": {
                    "metrics": [
                        [self.namespace, "feature_completeness_ratio", "FeatureType", feature_type]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "region": "us-east-1",
                    "title": f"{feature_type} - Completeness Ratio"
                }
            })
        
        try:
            self.cloudwatch.put_dashboard(
                DashboardName="FeatureStore-Monitoring",
                DashboardBody=json.dumps(dashboard_body)
            )
            self.logger.info("CloudWatch dashboard created successfully")
        except Exception as e:
            self.logger.error(f"Failed to create dashboard: {e}")

# Example usage
def monitor_example():
    monitor = FeatureStoreMonitor("feature-store")
    
    # Run data quality checks
    results = monitor.run_data_quality_checks(
        feature_type="user_behavior",
        expected_entity_count=10000
    )
    
    for result in results:
        print(f"Check: {result['check_name']}, Status: {result['status']}")
    
    # Create monitoring dashboard
    monitor.create_dashboard(["user_behavior", "product_features", "transaction_features"])

if __name__ == "__main__":
    monitor_example()

  

# main.tf - Core ECS Fargate Infrastructure
terraform {
  required_version = ">= 1.5.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

# ECS Cluster (Fargate doesn't require EC2 instances)
resource "aws_ecs_cluster" "main" {
  name = "production-fargate-cluster"

  setting {
    name  = "containerInsights"
    value = "enabled"
  }

  configuration {
    execute_command_configuration {
      logging = "DEFAULT"
    }
  }

  tags = {
    Environment = "production"
    ManagedBy   = "terraform"
  }
}

# Fargate Task Definition with advanced features
resource "aws_ecs_task_definition" "web_app" {
  family                   = "web-app"
  requires_compatibilities = ["FARGATE"]
  network_mode             = "awsvpc"
  cpu                      = 1024
  memory                   = 2048
  execution_role_arn       = aws_iam_role.ecs_task_execution_role.arn
  task_role_arn            = aws_iam_role.ecs_task_role.arn
  
  runtime_platform {
    cpu_architecture        = "X86_64"
    operating_system_family = "LINUX"
  }

  container_definitions = jsonencode([{
    name      = "web-app"
    image     = "${aws_ecr_repository.web_app.repository_url}:latest"
    essential = true
    
    portMappings = [{
      containerPort = 8080
      hostPort      = 8080
      protocol      = "tcp"
    }]

    environment = [
      { name = "NODE_ENV", value = "production" },
      { name = "LOG_LEVEL", value = "info" }
    ]

    secrets = [
      {
        name      = "DATABASE_URL"
        valueFrom = "${aws_secretsmanager_secret.database_url.arn}"
      }
    ]

    logConfiguration = {
      logDriver = "awslogs"
      options = {
        "awslogs-group"         = "/ecs/web-app"
        "awslogs-region"        = var.region
        "awslogs-stream-prefix" = "ecs"
      }
    }

    healthCheck = {
      command     = ["CMD-SHELL", "curl -f http://localhost:8080/health || exit 1"]
      interval    = 30
      timeout     = 5
      retries     = 3
      startPeriod = 60
    }

    # Resource limits for Fargate
    resourceRequirements = [
      {
        type  = "InferenceAccelerator"
        value = "var.inference_accelerator_type"
      }
    ]
  }])

  ephemeral_storage {
    size_in_gib = 21
  }

  tags = {
    Application = "web-app"
    Environment = "production"
  }
}

  

# networking.tf - Secure Fargate Networking
# VPC with private subnets only for Fargate
resource "aws_vpc" "fargate_vpc" {
  cidr_block           = "10.1.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true

  tags = {
    Name = "fargate-vpc"
  }
}

# Private subnets for Fargate tasks
resource "aws_subnet" "private" {
  count             = 3
  vpc_id            = aws_vpc.fargate_vpc.id
  cidr_block        = cidrsubnet(aws_vpc.fargate_vpc.cidr_block, 8, count.index)
  availability_zone = data.aws_availability_zones.available.names[count.index]

  tags = {
    Name = "fargate-private-${count.index + 1}"
  }
}

# Security group for Fargate tasks
resource "aws_security_group" "fargate_tasks" {
  name_prefix = "fargate-tasks-"
  description = "Security group for Fargate tasks"
  vpc_id      = aws_vpc.fargate_vpc.id

  ingress {
    description     = "Application traffic from ALB"
    from_port       = 8080
    to_port         = 8080
    protocol        = "tcp"
    security_groups = [aws_security_group.alb.id]
  }

  ingress {
    description = "SSM Session Manager"
    from_port   = 22
    to_port     = 22
    protocol    = "tcp"
    self        = true
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }

  tags = {
    Name = "fargate-tasks-sg"
  }
}

# VPC endpoints for private ECS operation
resource "aws_vpc_endpoint" "ecr_api" {
  vpc_id              = aws_vpc.fargate_vpc.id
  service_name        = "com.amazonaws.${var.region}.ecr.api"
  vpc_endpoint_type   = "Interface"
  private_dns_enabled = true
  subnet_ids          = aws_subnet.private[*].id

  security_group_ids = [aws_security_group.vpc_endpoints.id]

  tags = {
    Name = "ecr-api-endpoint"
  }
}

resource "aws_vpc_endpoint" "ecr_dkr" {
  vpc_id              = aws_vpc.fargate_vpc.id
  service_name        = "com.amazonaws.${var.region}.ecr.dkr"
  vpc_endpoint_type   = "Interface"
  private_dns_enabled = true
  subnet_ids          = aws_subnet.private[*].id

  security_group_ids = [aws_security_group.vpc_endpoints.id]

  tags = {
    Name = "ecr-dkr-endpoint"
  }
}

# ECS Service discovery for internal communication
resource "aws_service_discovery_private_dns_namespace" "internal" {
  name        = "internal.ecs"
  description = "Internal service discovery namespace"
  vpc         = aws_vpc.fargate_vpc.id
}

resource "aws_service_discovery_service" "web_app" {
  name = "web-app"

  dns_config {
    namespace_id = aws_service_discovery_private_dns_namespace.internal.id

    dns_records {
      ttl  = 10
      type = "A"
    }

    routing_policy = "MULTIVALUE"
  }

  health_check_custom_config {
    failure_threshold = 1
  }
}

  

# service.tf - Advanced ECS Service Configuration
resource "aws_ecs_service" "web_app" {
  name            = "web-app"
  cluster         = aws_ecs_cluster.main.id
  task_definition = aws_ecs_task_definition.web_app.arn
  desired_count   = 2
  launch_type     = "FARGATE"

  network_configuration {
    subnets          = aws_subnet.private[*].id
    security_groups  = [aws_security_group.fargate_tasks.id]
    assign_public_ip = false
  }

  load_balancer {
    target_group_arn = aws_lb_target_group.web_app.arn
    container_name   = "web-app"
    container_port   = 8080
  }

  service_registries {
    registry_arn = aws_service_discovery_service.web_app.arn
  }

  # Blue-Green deployment configuration
  deployment_controller {
    type = "CODE_DEPLOY"
  }

  deployment_circuit_breaker {
    enable   = true
    rollback = true
  }

  # Advanced capacity provider strategy
  capacity_provider_strategy {
    capacity_provider = "FARGATE"
    weight            = 1
    base              = 1
  }

  capacity_provider_strategy {
    capacity_provider = "FARGATE_SPOT"
    weight            = 2
  }

  enable_ecs_managed_tags = true
  propagate_tags          = "SERVICE"

  # Wait for steady state before continuing
  wait_for_steady_state = true

  tags = {
    Environment = "production"
    Application = "web-app"
  }
}

# Application Auto Scaling for Fargate service
resource "aws_appautoscaling_target" "web_app" {
  max_capacity       = 10
  min_capacity       = 2
  resource_id        = "service/${aws_ecs_cluster.main.name}/${aws_ecs_service.web_app.name}"
  scalable_dimension = "ecs:service:DesiredCount"
  service_namespace  = "ecs"
}

# CPU-based scaling policy
resource "aws_appautoscaling_policy" "web_app_cpu" {
  name               = "web-app-cpu-scaling"
  policy_type        = "TargetTrackingScaling"
  resource_id        = aws_appautoscaling_target.web_app.resource_id
  scalable_dimension = aws_appautoscaling_target.web_app.scalable_dimension
  service_namespace  = aws_appautoscaling_target.web_app.service_namespace

  target_tracking_scaling_policy_configuration {
    predefined_metric_specification {
      predefined_metric_type = "ECSServiceAverageCPUUtilization"
    }

    target_value       = 70.0
    scale_in_cooldown  = 300
    scale_out_cooldown = 60
  }
}

# Memory-based scaling policy
resource "aws_appautoscaling_policy" "web_app_memory" {
  name               = "web-app-memory-scaling"
  policy_type        = "TargetTrackingScaling"
  resource_id        = aws_appautoscaling_target.web_app.resource_id
  scalable_dimension = aws_appautoscaling_target.web_app.scalable_dimension
  service_namespace  = aws_appautoscaling_target.web_app.service_namespace

  target_tracking_scaling_policy_configuration {
    predefined_metric_specification {
      predefined_metric_type = "ECSServiceAverageMemoryUtilization"
    }

    target_value       = 80.0
    scale_in_cooldown  = 300
    scale_out_cooldown = 60
  }
}

  

# iam.tf - Secure IAM Configuration for Fargate
# Task execution role for ECS to pull images and logs
resource "aws_iam_role" "ecs_task_execution_role" {
  name_prefix = "ecs-task-execution-"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "ecs-tasks.amazonaws.com"
        }
      }
    ]
  })

  tags = {
    Service = "ecs"
  }
}

# Attach managed policy for basic ECS operations
resource "aws_iam_role_policy_attachment" "ecs_task_execution_role_policy" {
  role       = aws_iam_role.ecs_task_execution_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy"
}

# Custom task execution role policy for additional permissions
resource "aws_iam_role_policy" "ecs_task_execution_custom" {
  name_prefix = "ecs-task-execution-custom-"
  role        = aws_iam_role.ecs_task_execution_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "ssm:GetParameters",
          "secretsmanager:GetSecretValue",
          "kms:Decrypt"
        ]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = [
          "logs:CreateLogStream",
          "logs:PutLogEvents",
          "logs:CreateLogGroup"
        ]
        Resource = "arn:aws:logs:*:*:*"
      }
    ]
  })
}

# Task role for application-specific permissions
resource "aws_iam_role" "ecs_task_role" {
  name_prefix = "ecs-task-role-"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "ecs-tasks.amazonaws.com"
        }
      }
    ]
  })

  tags = {
    Service = "ecs"
  }
}

# Application-specific permissions for the task
resource "aws_iam_role_policy" "ecs_task_policy" {
  name_prefix = "ecs-task-policy-"
  role        = aws_iam_role.ecs_task_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:GetObject",
          "s3:PutObject",
          "s3:ListBucket"
        ]
        Resource = [
          "arn:aws:s3:::my-app-bucket",
          "arn:aws:s3:::my-app-bucket/*"
        ]
      },
      {
        Effect = "Allow"
        Action = [
          "dynamodb:GetItem",
          "dynamodb:PutItem",
          "dynamodb:UpdateItem",
          "dynamodb:Query",
          "dynamodb:Scan"
        ]
        Resource = "arn:aws:dynamodb:*:*:table/my-app-table"
      },
      {
        Effect = "Allow"
        Action = [
          "ses:SendEmail",
          "ses:SendRawEmail"
        ]
        Resource = "*"
        Condition = {
          StringEquals = {
            "ses:FromAddress": "noreply@myapp.com"
          }
        }
      }
    ]
  })
}

  

# monitoring.tf - Comprehensive Observability Setup
# CloudWatch Log Group for ECS tasks
resource "aws_cloudwatch_log_group" "ecs_web_app" {
  name              = "/ecs/web-app"
  retention_in_days = 30

  tags = {
    Application = "web-app"
    Environment = "production"
  }
}

# Container Insights for enhanced ECS monitoring
resource "aws_cloudwatch_log_group" "container_insights" {
  name              = "/aws/ecs/containerinsights/${aws_ecs_cluster.main.name}/performance"
  retention_in_days = 7

  tags = {
    Application = "web-app"
    Environment = "production"
  }
}

# Custom CloudWatch metrics and alarms
resource "aws_cloudwatch_metric_alarm" "ecs_cpu_high" {
  alarm_name          = "ecs-web-app-cpu-utilization-high"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = "2"
  metric_name         = "CPUUtilization"
  namespace           = "AWS/ECS"
  period              = "120"
  statistic           = "Average"
  threshold           = "80"
  alarm_description   = "This metric monitors ECS CPU utilization"
  alarm_actions       = [aws_sns_topic.alerts.arn]

  dimensions = {
    ClusterName = aws_ecs_cluster.main.name
    ServiceName = aws_ecs_service.web_app.name
  }

  tags = {
    Application = "web-app"
  }
}

resource "aws_cloudwatch_metric_alarm" "ecs_memory_high" {
  alarm_name          = "ecs-web-app-memory-utilization-high"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = "2"
  metric_name         = "MemoryUtilization"
  namespace           = "AWS/ECS"
  period              = "120"
  statistic           = "Average"
  threshold           = "85"
  alarm_description   = "This metric monitors ECS memory utilization"
  alarm_actions       = [aws_sns_topic.alerts.arn]

  dimensions = {
    ClusterName = aws_ecs_cluster.main.name
    ServiceName = aws_ecs_service.web_app.name
  }
}

# ECS Exec logging for session management
resource "aws_cloudwatch_log_group" "ecs_exec_sessions" {
  name              = "/ecs/exec-sessions"
  retention_in_days = 7

  tags = {
    Service = "ecs-exec"
  }
}

# X-Ray for distributed tracing
resource "aws_iam_role_policy_attachment" "xray_write" {
  role       = aws_iam_role.ecs_task_role.name
  policy_arn = "arn:aws:iam::aws:policy/AWSXRayDaemonWriteAccess"
}

# Custom application metrics
resource "aws_cloudwatch_log_metric_filter" "application_errors" {
  name           = "WebAppErrorCount"
  pattern        = "ERROR"
  log_group_name = aws_cloudwatch_log_group.ecs_web_app.name

  metric_transformation {
    name      = "ErrorCount"
    namespace = "WebApp"
    value     = "1"
  }
}

resource "aws_cloudwatch_metric_alarm" "high_error_rate" {
  alarm_name          = "web-app-high-error-rate"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = "1"
  metric_name         = "ErrorCount"
  namespace           = "WebApp"
  period              = "300"
  statistic           = "Sum"
  threshold           = "10"
  alarm_description   = "Monitor application error rate"
  alarm_actions       = [aws_sns_topic.alerts.arn]

  tags = {
    Application = "web-app"
  }
}

  

# cost-optimization.tf - Fargate Cost Optimization
# Mixed capacity provider strategy for cost optimization
resource "aws_ecs_cluster_capacity_providers" "main" {
  cluster_name = aws_ecs_cluster.main.name

  capacity_providers = ["FARGATE", "FARGATE_SPOT"]

  default_capacity_provider_strategy {
    capacity_provider = "FARGATE_SPOT"
    weight            = 3
    base              = 1
  }
}

# Cost and usage reporting
resource "aws_cur_report_definition" "fargate_costs" {
  report_name                = "fargate-cost-report"
  time_unit                  = "HOURLY"
  format                     = "Parquet"
  compression                = "Parquet"
  additional_schema_elements = ["RESOURCES"]
  s3_bucket                  = aws_s3_bucket.cost_reports.bucket
  s3_prefix                  = "fargate"
  s3_region                  = var.region
  additional_artifacts       = ["REDSHIFT", "QUICKSIGHT"]

  report_versioning = "OVERWRITE_REPORT"
}

# Budget alerts for Fargate spending
resource "aws_budgets_budget" "fargate_monthly" {
  name              = "fargate-monthly-budget"
  budget_type       = "COST"
  limit_amount      = "1000"
  limit_unit        = "USD"
  time_unit         = "MONTHLY"
  time_period_start = "2025-01-01_00:00"

  cost_types {
    include_credit             = false
    include_discount           = true
    include_other_subscription = true
    include_recurring          = true
    include_refund             = false
    include_subscription       = true
    include_support            = true
    include_tax                = true
    include_upfront            = true
    use_amortized              = false
    use_blended                = false
  }

  notification {
    comparison_operator        = "GREATER_THAN"
    threshold                  = 80
    threshold_type             = "PERCENTAGE"
    notification_type          = "ACTUAL"
    subscriber_email_addresses = [var.budget_alert_email]
  }
}

  

# main.tf - Core Transit Gateway and VPC Configuration
terraform {
  required_version = ">= 1.5.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

# Transit Gateway for centralized routing
resource "aws_ec2_transit_gateway" "main" {
  description                     = "Central Transit Gateway for private cloud"
  amazon_side_asn                 = 64512
  auto_accept_shared_attachments  = "enable"
  default_route_table_association = "disable"
  default_route_table_propagation = "disable"
  
  tags = {
    Name = "main-tgw"
  }
}

# Application VPC with only private subnets
resource "aws_vpc" "app_vpc" {
  cidr_block           = "10.1.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true
  
  tags = {
    Name = "app-vpc-private"
  }
}

# Private subnets across multiple AZs
resource "aws_subnet" "app_private" {
  count             = 3
  vpc_id            = aws_vpc.app_vpc.id
  cidr_block        = cidrsubnet(aws_vpc.app_vpc.cidr_block, 8, count.index)
  availability_zone = data.aws_availability_zones.available.names[count.index]
  
  tags = {
    Name = "app-private-${count.index + 1}"
  }
}

# Transit Gateway VPC attachment
resource "aws_ec2_transit_gateway_vpc_attachment" "app_vpc" {
  subnet_ids         = aws_subnet.app_private[*].id
  transit_gateway_id = aws_ec2_transit_gateway.main.id
  vpc_id             = aws_vpc.app_vpc.id
  
  tags = {
    Name = "app-vpc-attachment"
  }
}

  

# vpc-endpoints.tf - PrivateLink Configuration for AWS Services
# SSM VPC Endpoint for Session Manager
resource "aws_vpc_endpoint" "ssm" {
  vpc_id              = aws_vpc.app_vpc.id
  service_name        = "com.amazonaws.${var.region}.ssm"
  vpc_endpoint_type   = "Interface"
  private_dns_enabled = true
  subnet_ids          = aws_subnet.app_private[*].id
  
  security_group_ids = [
    aws_security_group.vpc_endpoints.id
  ]
  
  tags = {
    Name = "ssm-endpoint"
  }
}

# Additional SSM endpoints for full functionality
resource "aws_vpc_endpoint" "ssm_messages" {
  vpc_id              = aws_vpc.app_vpc.id
  service_name        = "com.amazonaws.${var.region}.ssmmessages"
  vpc_endpoint_type   = "Interface"
  private_dns_enabled = true
  subnet_ids          = aws_subnet.app_private[*].id
  
  security_group_ids = [
    aws_security_group.vpc_endpoints.id
  ]
  
  tags = {
    Name = "ssm-messages-endpoint"
  }
}

resource "aws_vpc_endpoint" "ec2_messages" {
  vpc_id              = aws_vpc.app_vpc.id
  service_name        = "com.amazonaws.${var.region}.ec2messages"
  vpc_endpoint_type   = "Interface"
  private_dns_enabled = true
  subnet_ids          = aws_subnet.app_private[*].id
  
  security_group_ids = [
    aws_security_group.vpc_endpoints.id
  ]
  
  tags = {
    Name = "ec2-messages-endpoint"
  }
}

# S3 Gateway Endpoint for package downloads and logs
resource "aws_vpc_endpoint" "s3" {
  vpc_id            = aws_vpc.app_vpc.id
  service_name      = "com.amazonaws.${var.region}.s3"
  vpc_endpoint_type = "Gateway"
  route_table_ids   = aws_route_table.private[*].id
  
  tags = {
    Name = "s3-gateway-endpoint"
  }
}

# ECR endpoints for Docker image pulls
resource "aws_vpc_endpoint" "ecr_api" {
  vpc_id              = aws_vpc.app_vpc.id
  service_name        = "com.amazonaws.${var.region}.ecr.api"
  vpc_endpoint_type   = "Interface"
  private_dns_enabled = true
  subnet_ids          = aws_subnet.app_private[*].id
  
  security_group_ids = [
    aws_security_group.vpc_endpoints.id
  ]
  
  tags = {
    Name = "ecr-api-endpoint"
  }
}

  

# security-groups.tf - Zero-Trust Security Configuration
# VPC Endpoints Security Group
resource "aws_security_group" "vpc_endpoints" {
  name_prefix = "vpc-endpoints-"
  description = "Security group for VPC endpoints"
  vpc_id      = aws_vpc.app_vpc.id
  
  ingress {
    description = "HTTPS from private subnets"
    from_port   = 443
    to_port     = 443
    protocol    = "tcp"
    cidr_blocks = [aws_vpc.app_vpc.cidr_block]
  }
  
  ingress {
    description = "SSM from private subnets"
    from_port   = 443
    to_port     = 443
    protocol    = "tcp"
    cidr_blocks = [aws_vpc.app_vpc.cidr_block]
  }
  
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
  
  tags = {
    Name = "vpc-endpoints-sg"
  }
}

# Application instances security group
resource "aws_security_group" "app_instances" {
  name_prefix = "app-instances-"
  description = "Security group for application instances"
  vpc_id      = aws_vpc.app_vpc.id
  
  ingress {
    description = "SSH via Session Manager"
    from_port   = 22
    to_port     = 22
    protocol    = "tcp"
    cidr_blocks = [aws_vpc.app_vpc.cidr_block]
  }
  
  ingress {
    description = "Application traffic from internal"
    from_port   = 8080
    to_port     = 8080
    protocol    = "tcp"
    cidr_blocks = [aws_vpc.app_vpc.cidr_block]
  }
  
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
  
  tags = {
    Name = "app-instances-sg"
  }
}

# Database security group with strict rules
resource "aws_security_group" "database" {
  name_prefix = "database-"
  description = "Security group for database instances"
  vpc_id      = aws_vpc.app_vpc.id
  
  ingress {
    description = "PostgreSQL from app instances"
    from_port   = 5432
    to_port     = 5432
    protocol    = "tcp"
    security_groups = [aws_security_group.app_instances.id]
  }
  
  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
  
  tags = {
    Name = "database-sg"
  }
}

  

# iam-ssm.tf - IAM Roles and SSM Configuration
# SSM Instance Role
resource "aws_iam_role" "ssm_instance_role" {
  name_prefix = "SSMInstanceRole-"
  
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "ec2.amazonaws.com"
        }
      }
    ]
  })
  
  tags = {
    Name = "ssm-instance-role"
  }
}

# AmazonSSMManagedInstanceCore policy attachment
resource "aws_iam_role_policy_attachment" "ssm_core" {
  role       = aws_iam_role.ssm_instance_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore"
}

# Custom SSM policy for additional permissions
resource "aws_iam_role_policy" "ssm_custom" {
  name_prefix = "SSMCustomPolicy-"
  role        = aws_iam_role.ssm_instance_role.id
  
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:GetObject",
          "s3:PutObject",
          "s3:ListBucket"
        ]
        Resource = [
          "arn:aws:s3:::my-ssm-logs-bucket/*",
          "arn:aws:s3:::my-ssm-logs-bucket"
        ]
      },
      {
        Effect = "Allow"
        Action = [
          "logs:CreateLogStream",
          "logs:PutLogEvents",
          "logs:DescribeLogGroups",
          "logs:DescribeLogStreams"
        ]
        Resource = "*"
      }
    ]
  })
}

# Instance Profile for EC2 instances
resource "aws_iam_instance_profile" "ssm_instance" {
  name_prefix = "SSMInstanceProfile-"
  role        = aws_iam_role.ssm_instance_role.name
}

# SSM Document for session preferences
resource "aws_ssm_document" "session_preferences" {
  name          = "SSM-SessionManagerRunShell"
  document_type = "Session"
  
  content = jsonencode({
    schemaVersion = "1.0"
    description   = "Document to hold regional session settings"
    sessionType   = "Standard_Stream"
    inputs = {
      s3BucketName                = "my-ssm-logs-bucket"
      s3KeyPrefix                 = "ssm-sessions"
      s3EncryptionEnabled         = true
      cloudWatchLogGroupName      = "/aws/ssm/sessions"
      cloudWatchEncryptionEnabled = true
      cloudWatchStreamingEnabled  = true
      idleSessionTimeout          = "20"
      maxSessionDuration          = "60"
      shellProfile = {
        linux = "echo 'Welcome to Secure Session Manager'"
      }
    }
  })
  
  tags = {
    Name = "session-preferences"
  }
}

  

# tgw-routing.tf - Advanced Transit Gateway Configuration
# Segregated route tables for different environments
resource "aws_ec2_transit_gateway_route_table" "production" {
  transit_gateway_id = aws_ec2_transit_gateway.main.id
  
  tags = {
    Name = "production-rt"
  }
}

resource "aws_ec2_transit_gateway_route_table" "development" {
  transit_gateway_id = aws_ec2_transit_gateway.main.id
  
  tags = {
    Name = "development-rt"
  }
}

resource "aws_ec2_transit_gateway_route_table" "shared_services" {
  transit_gway_id = aws_ec2_transit_gateway.main.id
  
  tags = {
    Name = "shared-services-rt"
  }
}

# Route table associations
resource "aws_ec2_transit_gateway_route_table_association" "app_vpc_prod" {
  transit_gateway_attachment_id  = aws_ec2_transit_gateway_vpc_attachment.app_vpc.id
  transit_gateway_route_table_id = aws_ec2_transit_gateway_route_table.production.id
}

# Static routes for specific traffic patterns
resource "aws_ec2_transit_gateway_route" "to_inspection_vpc" {
  destination_cidr_block         = "0.0.0.0/0"
  transit_gateway_attachment_id  = aws_ec2_transit_gateway_vpc_attachment.inspection.id
  transit_gateway_route_table_id = aws_ec2_transit_gateway_route_table.production.id
}

# Route table propagations
resource "aws_ec2_transit_gateway_route_table_propagation" "prod_to_shared" {
  transit_gateway_attachment_id  = aws_ec2_transit_gateway_vpc_attachment.shared_services.id
  transit_gateway_route_table_id = aws_ec2_transit_gateway_route_table.production.id
}

  

# monitoring.tf - Comprehensive Observability Setup
# VPC Flow Logs for network traffic monitoring
resource "aws_cloudwatch_log_group" "vpc_flow_logs" {
  name              = "/aws/vpc/flow-logs"
  retention_in_days = 365
  
  tags = {
    Name = "vpc-flow-logs"
  }
}

resource "aws_iam_role" "vpc_flow_log_role" {
  name_prefix = "VPCFlowLogRole-"
  
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "vpc-flow-logs.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy" "vpc_flow_log_policy" {
  name_prefix = "VPCFlowLogPolicy-"
  role        = aws_iam_role.vpc_flow_log_role.id
  
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "logs:CreateLogGroup",
          "logs:CreateLogStream",
          "logs:PutLogEvents",
          "logs:DescribeLogGroups",
          "logs:DescribeLogStreams"
        ]
        Resource = "*"
      }
    ]
  })
}

resource "aws_flow_log" "app_vpc" {
  iam_role_arn    = aws_iam_role.vpc_flow_log_role.arn
  log_destination = aws_cloudwatch_log_group.vpc_flow_logs.arn
  traffic_type    = "ALL"
  vpc_id          = aws_vpc.app_vpc.id
  
  tags = {
    Name = "app-vpc-flow-logs"
  }
}

# Transit Gateway Flow Logs
resource "aws_ec2_transit_gateway_flow_log" "main" {
  transit_gateway_id          = aws_ec2_transit_gateway.main.id
  transit_gateway_attachment_id = aws_ec2_transit_gateway_vpc_attachment.app_vpc.id
  log_destination             = aws_cloudwatch_log_group.vpc_flow_logs.arn
  iam_role_arn                = aws_iam_role.vpc_flow_log_role.arn
  traffic_type                = "ALL"
  
  tags = {
    Name = "tgw-flow-logs"
  }
}

# CloudWatch Alarms for security monitoring
resource "aws_cloudwatch_log_metric_filter" "unauthorized_access" {
  name           = "UnauthorizedAccessAttempts"
  pattern        = "[version, account, eni, source, destination, srcport, destport, protocol, packets, bytes, windowstart, windowend, action = \"REJECT\"]"
  log_group_name = aws_cloudwatch_log_group.vpc_flow_logs.name
  
  metric_transformation {
    name      = "UnauthorizedAccessCount"
    namespace = "VPC/Security"
    value     = "1"
  }
}

resource "aws_cloudwatch_metric_alarm" "high_unauthorized_access" {
  alarm_name          = "HighUnauthorizedAccessAttempts"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = "2"
  metric_name         = "UnauthorizedAccessCount"
  namespace           = "VPC/Security"
  period              = "300"
  statistic           = "Sum"
  threshold           = "10"
  alarm_description   = "This metric monitors for high unauthorized access attempts"
  alarm_actions       = [aws_sns_topic.security_alerts.arn]
  
  tags = {
    Name = "unauthorized-access-alarm"
  }
}

  

# advanced-security.tf - Enterprise Security Controls
# AWS Network Firewall for deep packet inspection
resource "aws_networkfirewall_firewall" "inspection" {
  name                = "inspection-firewall"
  firewall_policy_arn = aws_networkfirewall_firewall_policy.inspection.arn
  vpc_id              = aws_vpc.inspection.id
  
  subnet_mapping {
    subnet_id = aws_subnet.firewall.id
  }
  
  tags = {
    Name = "inspection-firewall"
  }
}

resource "aws_networkfirewall_firewall_policy" "inspection" {
  name = "inspection-policy"
  
  firewall_policy {
    stateless_default_actions          = ["aws:forward_to_sfe"]
    stateless_fragment_default_actions = ["aws:forward_to_sfe"]
    
    stateful_rule_group_reference {
      resource_arn = aws_networkfirewall_rule_group.threat_prevention.arn
    }
    
    stateful_engine_options {
      rule_order = "STRICT_ORDER"
    }
  }
  
  tags = {
    Name = "inspection-firewall-policy"
  }
}

# Security Hub integration for compliance monitoring
resource "aws_securityhub_account" "main" {}

resource "aws_securityhub_standards_subscription" "cis" {
  depends_on    = [aws_securityhub_account.main]
  standards_arn = "arn:aws:securityhub:::ruleset/cis-aws-foundations-benchmark/v/1.2.0"
}

resource "aws_securityhub_standards_subscription" "pci" {
  depends_on    = [aws_securityhub_account.main]
  standards_arn = "arn:aws:securityhub:::ruleset/pci-dss/v/3.2.1"
}

# GuardDuty for threat detection
resource "aws_guardduty_detector" "main" {
  enable = true
  
  datasources {
    s3_logs {
      enable = true
    }
    kubernetes {
      audit_logs {
        enable = false
      }
    }
    malware_protection {
      scan_ec2_instance_with_findings {
        ebs_volumes {
          enable = true
        }
      }
    }
  }
}