data "aws_iam_policy_document" "assume_lambda_role" {
  statement {
    actions = ["sts:AssumeRole"]

    principals {
      type        = "Service"
      identifiers = ["lambda.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "lambda_dynamo_to_s3_role" {
  name               = "lambda-${var.lambda_name}"
  assume_role_policy = "${data.aws_iam_policy_document.assume_lambda_role.json}"
}

data "aws_iam_policy_document" "lambda_basic_policy" {
  statement {
    effect = "Allow"

    actions = [
      "logs:*",
    ]

    resources = [
      "arn:aws:logs:*:*:*",
    ]
  }

  statement {
    effect = "Allow"

    actions = [
      "ec2:CreateNetworkInterface",
      "ec2:DescribeNetworkInterfaces",
      "ec2:DeleteNetworkInterface",
      "logs:CreateLogGroup",
      "logs:CreateLogStream",
      "logs:PutLogEvents",
    ]

    resources = [
      "*",
    ]
  }
}

resource "aws_iam_policy" "lambda_basic_policy" {
  name   = "lambda-${var.lambda_name}"
  policy = "${data.aws_iam_policy_document.lambda_basic_policy.json}"
}

resource "aws_iam_role_policy_attachment" "lambda_basic_policy_attachment" {
  role       = "${aws_iam_role.lambda_dynamo_to_s3_role.name}"
  policy_arn = "${aws_iam_policy.lambda_basic_policy.arn}"
}

# This effectively creates an empty S3 folder.
resource "aws_s3_bucket_object" "lambda_dynamo_to_s3_folder" {
  bucket = "${var.s3_bucket}"
  acl    = "private"
  key    = "${var.s3_path}/"
  source = "/dev/null"
}

data "aws_iam_policy_document" "lambda_dynamo_to_s3_policy_data" {
  statement {
    effect = "Allow"

    actions = [
      "s3:AbortMultipartUpload",
      "s3:DeleteObject",
      "s3:DeleteObjectVersion",
      "s3:GetBucketLocation",
      "s3:GetBucketVersioning",
      "s3:GetLifecycleConfiguration",
      "s3:GetObject",
      "s3:GetObjectAcl",
      "s3:GetObjectTorrent",
      "s3:GetObjectVersion",
      "s3:GetObjectVersionAcl",
      "s3:GetObjectVersionTorrent",
      "s3:GetReplicationConfiguration",
      "s3:ListBucket",
      "s3:ListBucketMultipartUploads",
      "s3:ListBucketVersions",
      "s3:ListMultipartUploadParts",
      "s3:PutObject",
      "s3:PutObjectAcl",
      "s3:PutObjectVersionAcl",
      "s3:ReplicateDelete",
      "s3:ReplicateObject",
      "s3:RestoreObject",
    ]

    resources = [
      "arn:aws:s3:::${var.s3_bucket}/${var.s3_path}",
      "arn:aws:s3:::${var.s3_bucket}/${var.s3_path}/*",
    ]
  }

  statement {
    effect = "Allow"

    actions = [
      "s3:GetBucketLocation",
      "s3:ListBucket",
    ]

    resources = [
      "arn:aws:s3:::${var.s3_bucket}}",
    ]
  }
}

resource "aws_iam_policy" "lambda_dynamo_to_s3_policy" {
  name   = "S3-${var.s3_bucket}-${var.s3_path}-RW"
  policy = "${data.aws_iam_policy_document.lambda_dynamo_to_s3_policy_data.json}"
}

resource "aws_iam_role_policy_attachment" "lambda_policy_attachment" {
  role       = "${aws_iam_role.lambda_dynamo_to_s3_role.name}"
  policy_arn = "${aws_iam_policy.lambda_dynamo_to_s3_policy.arn}"
}

resource "aws_lambda_function" "dynamo_to_s3" {
  filename      = "${var.lambda_function_package_path}"
  function_name = "${var.lambda_name}"
  description   = "${var.lambda_description}"
  runtime       = "python2.7"
  role          = "${aws_iam_role.lambda_dynamo_to_s3_role.arn}"
  handler       = "${var.lambda_function_handler}"
  publish       = true
  timeout       = "${var.lambda_function_timeout}"

  vpc_config {
    subnet_ids         = "${var.vpc_subnet_ids}"
    security_group_ids = "${var.vpc_security_group_ids}"
  }

  environment {
    variables = "${var.lambda_function_environment_variables}"
  }

  tags = {
    terraformed = "true"
  }
}

data "aws_iam_policy_document" "lambda_dynamo_stream_policy_data" {
  statement {
    effect = "Allow"

    actions = [
      "dynamodb:BatchGetItem",
      "dynamodb:BatchWriteItem",
      "dynamodb:DeleteItem",
      "dynamodb:DescribeLimits",
      "dynamodb:DescribeTable",
      "dynamodb:GetItem",
      "dynamodb:GetRecords",
      "dynamodb:GetShardIterator",
      "dynamodb:ListStreams",
      "dynamodb:PutItem",
      "dynamodb:Query",
      "dynamodb:Scan",
      "dynamodb:UpdateItem",
      "dynamodb:UpdateTable",
      "dynamodb:DescribeStream"
    ]

    resources = [
      "arn:aws:dynamodb:${var.aws_dynamo_region}:${data.aws_caller_identity.current.account_id}:table/${var.dynamo_table_name}/*"
    ]
  }
}

resource "aws_iam_policy" "lambda_dynamo_stream_policy" {
  name   = "Dynamo-${var.dynamo_table_name}-stream-RW"
  policy = "${data.aws_iam_policy_document.lambda_dynamo_stream_policy_data.json}"
}

resource "aws_iam_role_policy_attachment" "lambda_dynamo_stream_policy_attachment" {
  role       = "${aws_iam_role.lambda_dynamo_to_s3_role.name}"
  policy_arn = "${aws_iam_policy.lambda_dynamo_stream_policy.arn}"
}

data "aws_caller_identity" "current" {}

resource "aws_lambda_event_source_mapping" "event_source_mapping" {
  batch_size        = 500
  event_source_arn  = "arn:aws:dynamodb:${var.aws_dynamo_region}:${data.aws_caller_identity.current.account_id}:table/${var.dynamo_table_name}"
  enabled           = true
  function_name     = "${aws_lambda_function.dynamo_to_s3.arn}"
  starting_position = "TRIM_HORIZON"
}