template.yml

---
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Core Ocean Iceberg Metadata Function

Parameters:
  FunctionName:
    Description: The name of the Lambda function
    Type: String
    Default: core-ocean-metadata
  FunctionMemorySize:
    Description: MB of memory allocated to the function
    Type: Number
    Default: 512
    MinValue: 512
    MaxValue: 1024
  FunctionTimeout:
    Description: Number of seconds before the function is stopped
    Type: Number
    Default: 30
    MinValue: 20
    MaxValue: 120
  QueueName:
    Type: String
    Description: The name of the FIFO Queue.
    Default: ocean-metadata
  QueueBatchSize:
    Description: Size of message batches to send to Lambda invocation
    Type: Number
    Default: 10
  BucketName:
    Description: The S3 bucket name for the lake
    Type: String
    Default: mytiki-ocean
  DatabaseName:
    Description: The Glue database name
    Type: String
    Default: ocean
  UtilsName:
    Default: core-ocean-iceberg-utils
    Type: String
    Description: Enter the name of the utils layer.
  UtilsVersion:
    Description: Enter the version number of the utils layer.
    Type: Number
    Default: 1

Resources:
  Queue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Sub "${QueueName}.fifo"
      ContentBasedDeduplication: true
      DeduplicationScope: messageGroup
      FifoQueue: true
      FifoThroughputLimit: perMessageGroupId
      MessageRetentionPeriod: 604800
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt DeadLetterQueue.Arn
        maxReceiveCount: 5

  QueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref Queue
      PolicyDocument:
        Statement:
          - Effect: Allow
            Principal:
              AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"
            Action:
              - sqs:ChangeMessageVisibility
              - sqs:DeleteMessage
              - sqs:GetQueueAttributes
              - sqs:GetQueueUrl
              - sqs:ListQueueTags
              - sqs:ReceiveMessage
              - sqs:SendMessage
            Resource:
              - !GetAtt Queue.Arn

  DeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Sub "${QueueName}-DLQ.fifo"
      ContentBasedDeduplication: true
      DeduplicationScope: messageGroup
      FifoQueue: true
      FifoThroughputLimit: perMessageGroupId
      MessageRetentionPeriod: 604800
      RedriveAllowPolicy:
        redrivePermission: byQueue
        sourceQueueArns:
          - !Sub "arn:aws:sqs:${AWS::Region}:${AWS::AccountId}:${QueueName}.fifo"

  DeadLetterQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref DeadLetterQueue
      PolicyDocument:
        Statement:
          - Effect: Allow
            Principal:
              AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"
            Action:
              - sqs:ChangeMessageVisibility
              - sqs:DeleteMessage
              - sqs:GetQueueAttributes
              - sqs:GetQueueUrl
              - sqs:ListQueueTags
              - sqs:ReceiveMessage
              - sqs:SendMessage
            Resource:
              - !GetAtt DeadLetterQueue.Arn

  FunctionDeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: !Sub "${FunctionName}-function-DLQ"
      MessageRetentionPeriod: 604800

  Function:
    Type: AWS::Serverless::Function
    Properties:
      Description: Writes Parquet metadata to Iceberg tables
      FunctionName: !Ref FunctionName
      CodeUri: .
      Handler: com.mytiki.core.iceberg.metadata.App::handleRequest
      # TODO: Look into upgraded to Java 21
      # Need to bump the version for core-ocean-iceberg-utils and its layer
      # and runtime compatibility setting.
      Runtime: java17
      # TODO: Look into switching this to arm64
      Architectures:
        - x86_64
      MemorySize: !Ref FunctionMemorySize
      Timeout: !Ref FunctionTimeout
      Tracing: Active
      # AutoPublishAlias: latest
      # AutoPublishAliasAllProperties: false
      # SnapStart:
      #   ApplyOn: PublishedVersions
      LoggingConfig:
        LogFormat: JSON
        ApplicationLogLevel: TRACE
        SystemLogLevel: DEBUG
      DeadLetterQueue:
        Type: SQS
        TargetArn: !GetAtt FunctionDeadLetterQueue.Arn
      Environment:
        Variables:
          AWS_LAMBDA_EXEC_WRAPPER: "/opt/otel-sqs-handler"
          ICEBERG_WAREHOUSE: !Sub "s3://${BucketName}"
          ICEBERG_DATABASE_NAME: !Ref DatabaseName
      # TODO: Investigate concurrency config. I'm not sure we can have multiple
      # concurrent writers to the same table. Limit for now.
      ReservedConcurrentExecutions: 1
      # TODO: Add tags
      Tags: {}
      PropagateTags: true
      Events:
        SQSEvent:
          Type: SQS
          Properties:
            Enabled: true
            FunctionResponseTypes:
              - ReportBatchItemFailures
            BatchSize: !Ref QueueBatchSize
            Queue: !GetAtt Queue.Arn
      Policies:
        # TODO: Pair these permissions down
        - S3CrudPolicy:
            BucketName: !Ref BucketName
        - Version: "2012-10-17"
          Statement:
            - Effect: Allow
              Action:
                - glue:GetTable
                - glue:GetTables
                - glue:UpdateTable
              Resource:
                - !Sub "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog"
                - !Sub "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/${DatabaseName}"
                - !Sub "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${DatabaseName}/*"
            - Effect: Allow
              Action:
                - xray:PutTraceSegments
                - xray:PutTelemetryRecords
              Resource: "*"
            - Effect: Allow
              Action:
                - lakeformation:GetDataAccess
              Resource: "*"
      Layers:
        - !Sub "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:layer:${UtilsName}:${UtilsVersion}"
        - !Sub "arn:aws:lambda:${AWS::Region}:901920570463:layer:aws-otel-java-wrapper-amd64-ver-1-30-0:1"

Outputs:
  Function:
    Description: ARN of the Lambda function
    Value: !GetAtt Function.Arn

  FunctionArn:
    Description: ARN of the Lambda function
    Value: !GetAtt Function.Arn

  Queue:
    Description: ARN of the SQS queue for metadata events
    Value: !GetAtt Queue.Arn

  QueueArn:
    Description: ARN of the SQS queue for metadata events
    Value: !GetAtt Queue.Arn

  QueueName:
    Description: Name of the SQS queue for metadata events
    Value: !GetAtt Queue.QueueName

  QueueUrl:
    Description: URL of the SQS queue for metadata events
    Value: !GetAtt Queue.QueueUrl

  DeadLetterQueueArn:
    Description: ARN of the DLQ for metadata events
    Value: !GetAtt DeadLetterQueue.Arn

  DeadLetterQueueName:
    Description: Name of the DLQ for metadata events
    Value: !GetAtt DeadLetterQueue.QueueName

  DeadLetterQueueUrl:
    Description: URL of the DLQ for metadata events
    Value: !GetAtt DeadLetterQueue.QueueUrl

  FunctionDeadLetterQueueArn:
    Description: ARN of the DLQ for Lambda function
    Value: !GetAtt FunctionDeadLetterQueue.Arn

  FunctionDeadLetterQueueName:
    Description: Name of the DLQ for Lambda function
    Value: !GetAtt FunctionDeadLetterQueue.QueueName

  FunctionDeadLetterQueueUrl:
    Description: URL of the DLQ for Lambda function
    Value: !GetAtt FunctionDeadLetterQueue.QueueUrl