اشیاء S3 را با توابع مرحله ای با استفاده از CDK + TS پردازش کنید

const userNotificationTopic = Topic.fromTopicArn(
  this,
  'userNotificationTopic',
  Fn.importValue('infrastructure::sns::user-notification-topic::arn'),
)

const bucket = new Bucket(this, 'bucket', {
  bucketName: 'automation-configuration',
  removalPolicy: RemovalPolicy.DESTROY,
})

const etlLambda = new NodejsFunction(this, 'etl-lambda', {
  entry: 'src/lambda/etl/etl.handler.ts',
  functionName: 'etl',
  timeout: Duration.seconds(30),
  architecture: Architecture.ARM_64,
})

bucket.grantReadWrite(
  new Alias(this, id.concat('alias'), {
    aliasName: 'current',
    version: etlLambda.currentVersion,
  }),
)

const categoriesLambda = new NodejsFunction(this, 'insert-categories-lambda', {
  entry: 'src/lambda/categories/categories.handler.ts',
  functionName: 'insert-categories',
  timeout: Duration.seconds(30),
  architecture: Architecture.ARM_64,
})

const hospitalsLambda = new NodejsFunction(this, 'hospitals-categories-lambda', {
  entry: 'src/lambda/categories/categories.handler.ts',
  functionName: 'insert-categories',
  timeout: Duration.seconds(30),
  architecture: Architecture.ARM_64,
})

private mountLambdaInvokes(
    lambdasInvoke: Array<{
      function: IFunction
      name?: string
      output?: string
    }>,
  ) {
    return lambdasInvoke.map(lambdaInvoke => {
      return new LambdaInvoke(this, `${lambdaInvoke.name || lambdaInvoke.function.functionName}-task`, {
        lambdaFunction: lambdaInvoke.function,
        inputPath: '$',
        outputPath: lambdaInvoke?.output || '$',
      })
    })
  }

const [etlTask, categoriesTask, hospitalsTask] = this.mountLambdaInvokes([
  { function: etlLambda, output: '$.Payload' },
  { function: categoriesLambda },
  { function: hospitalsLambda },
])

const errorTopicConfig = {
  topic: userNotificationTopic,
  subject: 'Automation Config Failed 😥',
  message: TaskInput.fromObject({
    message: 'Automation Config Failed due to an unexpected error.',
    cause: 'Unexpected Error',
    channel: 'email',
    destination: { ToAddresses: ['suporte@somemail.com'] },
  }),
}

const publishFailed = (publishFailedId: string) =>
  new SnsPublish(this, `automation-config-sns-failed-${publishFailedId}`, errorTopicConfig)

const jobFailed = new Fail(this, 'automation-config-job-failed', {
  cause: 'Unexpected Error',
})

const hospitalsCategoriesParallel = new Parallel(this, 'auto-config-exams-parallel-map')
  .branch(categoriesTask)
  .branch(hospitalsTask)
  .addRetry({ errors: ['States.ALL'], interval: Duration.seconds(5), maxAttempts: 1 })
  .addCatch(publishFailed('exams').next(jobFailed), {
    errors: ['States.ALL'],
  })

const definition = etlTask.next(hospitalsCategoriesParallel)

const logGroup = new LogGroup(this, 'automation-configuration-log-group', {
  retention: RetentionDays.ONE_WEEK,
  removalPolicy: RemovalPolicy.DESTROY,
})

const stateMachine = new StateMachine(this, `${id}-state-machine`, {
  definition,
  timeout: Duration.minutes(5),
  stateMachineName: 'automation-configuration',
  stateMachineType: StateMachineType.EXPRESS,
  logs: {
    destination: logGroup,
    includeExecutionData: true,
    level: LogLevel.ALL,
  },
})

const s3EventRule = new Rule(this, 'automation-config-s3-event-rule', {
  ruleName: 'automation-config-s3-event-rule',
})

const eventRole = new Role(this, 'eventRole', {
  assumedBy: new ServicePrincipal('events.amazonaws.com'),
})

stateMachine.grantStartExecution(eventRole)
s3EventRule.addTarget(
  new SfnStateMachine(stateMachine, {
    input: RuleTargetInput.fromObject({
      detail: EventField.fromPath('$.detail'),
    }),
    role: eventRole,
  }),
)
s3EventRule.addEventPattern({
  source: ['aws.s3'],
  detailType: ['Object Created'],
  detail: {
    bucket: {
      name: [bucket.bucketName],
    },
    object: {
      key: [
        {
          wildcard: 'csv/automation-configuration/*.csv',
        },
      ],
    },
  },
})

 const unexpectedFailRule = new Rule(this, 'exam-automation-config-unexpected-fail-rule', {
      ruleName: 'exam-automation-config-unexpected-fail-rule',
    })
    unexpectedFailRule.addTarget(
      new SnsTopic(userNotificationTopic, {
        message: RuleTargetInput.fromObject({
          subject: 'Exam Automation Config Failed 😥',
          message: 'Exam Automation Config Failed due to an unexpected error.',
          cause: 'Unexpected Error',
          channel: 'email',
          destination: { ToAddresses: ['it@somemail.com'] },
        }),
      }),
    )

    unexpectedFailRule.addEventPattern({
      source: ['aws.states'],
      detailType: ['Step Functions Execution Status Change'],
      detail: {
        stateMachineArn: [stateMachine.stateMachineArn],
        status: ['FAILED', 'TIMED_OUT', 'ABORTED'],
      },
    })

import { Duration, Fn, RemovalPolicy } from 'aws-cdk-lib'
import { Rule, RuleTargetInput, EventField } from 'aws-cdk-lib/aws-events'
import { SfnStateMachine, SnsTopic } from 'aws-cdk-lib/aws-events-targets'
import { Role, ServicePrincipal } from 'aws-cdk-lib/aws-iam'
import { Alias, Architecture, IFunction } from 'aws-cdk-lib/aws-lambda'
import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs'
import { LogGroup, RetentionDays } from 'aws-cdk-lib/aws-logs'
import { Bucket } from 'aws-cdk-lib/aws-s3'
import { Topic } from 'aws-cdk-lib/aws-sns'
import { Fail, LogLevel, Parallel, StateMachine, StateMachineType, TaskInput } from 'aws-cdk-lib/aws-stepfunctions'
import { LambdaInvoke, SnsPublish } from 'aws-cdk-lib/aws-stepfunctions-tasks'
import { Construct } from 'constructs'

export default class AutomationConfiguration extends Construct {
  constructor(scope: Construct, id: string) {
    super(scope, id)

    const userNotificationTopic = Topic.fromTopicArn(
      this,
      'userNotificationTopic',
      Fn.importValue('infrastructure::sns::user-notification-topic::arn'),
    )

    const bucket = new Bucket(this, 'bucket', {
      bucketName: 'automation-configuration',
      removalPolicy: RemovalPolicy.DESTROY,
    })

    const etlLambda = new NodejsFunction(this, 'etl-lambda', {
      entry: 'src/lambda/etl/etl.handler.ts',
      functionName: 'etl',
      timeout: Duration.seconds(30),
      architecture: Architecture.ARM_64,
    })

    bucket.grantRead(
      new Alias(this, id.concat('alias'), {
        aliasName: 'current',
        version: etlLambda.currentVersion,
      }),
    )

    const categoriesLambda = new NodejsFunction(this, 'insert-categories-lambda', {
      entry: 'src/lambda/categories/categories.handler.ts',
      functionName: 'insert-categories',
      timeout: Duration.seconds(30),
      architecture: Architecture.ARM_64,
    })

    const hospitalsLambda = new NodejsFunction(this, 'hospitals-categories-lambda', {
      entry: 'src/lambda/categories/categories.handler.ts',
      functionName: 'insert-categories',
      timeout: Duration.seconds(30),
      architecture: Architecture.ARM_64,
    })

    const [etlTask, categoriesTask, hospitalsTask] = this.mountLambdaInvokes([
      { function: etlLambda, output: '$.Payload' },
      { function: categoriesLambda },
      { function: hospitalsLambda },
    ])

    const errorTopicConfig = {
      topic: userNotificationTopic,
      subject: 'Automation Config Failed 😥',
      message: TaskInput.fromObject({
        message: 'Automation Config Failed due to an unexpected error.',
        cause: 'Unexpected Error',
        channel: 'email',
        destination: { ToAddresses: ['suporte@somemail.com'] },
      }),
    }

    const publishFailed = (publishFailedId: string) =>
      new SnsPublish(this, `automation-config-sns-failed-${publishFailedId}`, errorTopicConfig)

    const jobFailed = new Fail(this, 'automation-config-job-failed', {
      cause: 'Unexpected Error',
    })

    const hospitalsCategoriesParallel = new Parallel(this, 'auto-config-exams-parallel-map')
      .branch(categoriesTask)
      .branch(hospitalsTask)
      .addRetry({ errors: ['States.ALL'], interval: Duration.seconds(5), maxAttempts: 1 })
      .addCatch(publishFailed('exams').next(jobFailed), {
        errors: ['States.ALL'],
      })
    const definition = etlTask.next(hospitalsCategoriesParallel)

    const logGroup = new LogGroup(this, 'automation-configuration-log-group', {
      retention: RetentionDays.ONE_WEEK,
      removalPolicy: RemovalPolicy.DESTROY,
    })

    const stateMachine = new StateMachine(this, `${id}-state-machine`, {
      definition,
      timeout: Duration.minutes(5),
      stateMachineName: 'automation-configuration',
      stateMachineType: StateMachineType.EXPRESS,
      logs: {
        destination: logGroup,
        includeExecutionData: true,
        level: LogLevel.ALL,
      },
    })

    const s3EventRule = new Rule(this, 'automation-config-s3-event-rule', {
      ruleName: 'automation-config-s3-event-rule',
    })

    const eventRole = new Role(this, 'eventRole', {
      assumedBy: new ServicePrincipal('events.amazonaws.com'),
    })

    stateMachine.grantStartExecution(eventRole)
    s3EventRule.addTarget(
      new SfnStateMachine(stateMachine, {
        input: RuleTargetInput.fromObject({
          detail: EventField.fromPath('$.detail'),
        }),
        role: eventRole,
      }),
    )
    s3EventRule.addEventPattern({
      source: ['aws.s3'],
      detailType: ['Object Created'],
      detail: {
        bucket: {
          name: [bucket.bucketName],
        },
        object: {
          key: [
            {
              wildcard: 'csv/automation-configuration/*.csv',
            },
          ],
        },
      },
    })

    const unexpectedFailRule = new Rule(this, 'exam-automation-config-unexpected-fail-rule', {
      ruleName: 'exam-automation-config-unexpected-fail-rule',
    })
    unexpectedFailRule.addTarget(
      new SnsTopic(userNotificationTopic, {
        message: RuleTargetInput.fromObject({
          subject: 'Exam Automation Config Failed 😥',
          message: 'Exam Automation Config Failed due to an unexpected error.',
          cause: 'Unexpected Error',
          channel: 'email',
          destination: { ToAddresses: ['it@somemail.com'] },
        }),
      }),
    )

    unexpectedFailRule.addEventPattern({
      source: ['aws.states'],
      detailType: ['Step Functions Execution Status Change'],
      detail: {
        stateMachineArn: [stateMachine.stateMachineArn],
        status: ['FAILED', 'TIMED_OUT', 'ABORTED'],
      },
    })
  }

  private mountLambdaInvokes(
    lambdasInvoke: Array<{
      function: IFunction
      name?: string
      output?: string
    }>,
  ) {
    return lambdasInvoke.map(lambdaInvoke => {
      return new LambdaInvoke(this, `${lambdaInvoke.name || lambdaInvoke.function.functionName}-task`, {
        lambdaFunction: lambdaInvoke.function,
        inputPath: '$',
        outputPath: lambdaInvoke?.output || '$',
      })
    })
  }
}

const userNotificationTopic = Topic.fromTopicArn(
  this,
  'userNotificationTopic',
  Fn.importValue('infrastructure::sns::user-notification-topic::arn'),
)

const bucket = new Bucket(this, 'bucket', {
  bucketName: 'automation-configuration',
  removalPolicy: RemovalPolicy.DESTROY,
})

const etlLambda = new NodejsFunction(this, 'etl-lambda', {
  entry: 'src/lambda/etl/etl.handler.ts',
  functionName: 'etl',
  timeout: Duration.seconds(30),
  architecture: Architecture.ARM_64,
})

bucket.grantReadWrite(
  new Alias(this, id.concat('alias'), {
    aliasName: 'current',
    version: etlLambda.currentVersion,
  }),
)

const categoriesLambda = new NodejsFunction(this, 'insert-categories-lambda', {
  entry: 'src/lambda/categories/categories.handler.ts',
  functionName: 'insert-categories',
  timeout: Duration.seconds(30),
  architecture: Architecture.ARM_64,
})

const hospitalsLambda = new NodejsFunction(this, 'hospitals-categories-lambda', {
  entry: 'src/lambda/categories/categories.handler.ts',
  functionName: 'insert-categories',
  timeout: Duration.seconds(30),
  architecture: Architecture.ARM_64,
})

private mountLambdaInvokes(
    lambdasInvoke: Array<{
      function: IFunction
      name?: string
      output?: string
    }>,
  ) {
    return lambdasInvoke.map(lambdaInvoke => {
      return new LambdaInvoke(this, `${lambdaInvoke.name || lambdaInvoke.function.functionName}-task`, {
        lambdaFunction: lambdaInvoke.function,
        inputPath: '$',
        outputPath: lambdaInvoke?.output || '$',
      })
    })
  }

const [etlTask, categoriesTask, hospitalsTask] = this.mountLambdaInvokes([
  { function: etlLambda, output: '$.Payload' },
  { function: categoriesLambda },
  { function: hospitalsLambda },
])

const errorTopicConfig = {
  topic: userNotificationTopic,
  subject: 'Automation Config Failed 😥',
  message: TaskInput.fromObject({
    message: 'Automation Config Failed due to an unexpected error.',
    cause: 'Unexpected Error',
    channel: 'email',
    destination: { ToAddresses: ['suporte@somemail.com'] },
  }),
}

const publishFailed = (publishFailedId: string) =>
  new SnsPublish(this, `automation-config-sns-failed-${publishFailedId}`, errorTopicConfig)

const jobFailed = new Fail(this, 'automation-config-job-failed', {
  cause: 'Unexpected Error',
})

const hospitalsCategoriesParallel = new Parallel(this, 'auto-config-exams-parallel-map')
  .branch(categoriesTask)
  .branch(hospitalsTask)
  .addRetry({ errors: ['States.ALL'], interval: Duration.seconds(5), maxAttempts: 1 })
  .addCatch(publishFailed('exams').next(jobFailed), {
    errors: ['States.ALL'],
  })

const definition = etlTask.next(hospitalsCategoriesParallel)

const logGroup = new LogGroup(this, 'automation-configuration-log-group', {
  retention: RetentionDays.ONE_WEEK,
  removalPolicy: RemovalPolicy.DESTROY,
})

const stateMachine = new StateMachine(this, `${id}-state-machine`, {
  definition,
  timeout: Duration.minutes(5),
  stateMachineName: 'automation-configuration',
  stateMachineType: StateMachineType.EXPRESS,
  logs: {
    destination: logGroup,
    includeExecutionData: true,
    level: LogLevel.ALL,
  },
})

const s3EventRule = new Rule(this, 'automation-config-s3-event-rule', {
  ruleName: 'automation-config-s3-event-rule',
})

const eventRole = new Role(this, 'eventRole', {
  assumedBy: new ServicePrincipal('events.amazonaws.com'),
})

stateMachine.grantStartExecution(eventRole)
s3EventRule.addTarget(
  new SfnStateMachine(stateMachine, {
    input: RuleTargetInput.fromObject({
      detail: EventField.fromPath('$.detail'),
    }),
    role: eventRole,
  }),
)
s3EventRule.addEventPattern({
  source: ['aws.s3'],
  detailType: ['Object Created'],
  detail: {
    bucket: {
      name: [bucket.bucketName],
    },
    object: {
      key: [
        {
          wildcard: 'csv/automation-configuration/*.csv',
        },
      ],
    },
  },
})

 const unexpectedFailRule = new Rule(this, 'exam-automation-config-unexpected-fail-rule', {
      ruleName: 'exam-automation-config-unexpected-fail-rule',
    })
    unexpectedFailRule.addTarget(
      new SnsTopic(userNotificationTopic, {
        message: RuleTargetInput.fromObject({
          subject: 'Exam Automation Config Failed 😥',
          message: 'Exam Automation Config Failed due to an unexpected error.',
          cause: 'Unexpected Error',
          channel: 'email',
          destination: { ToAddresses: ['it@somemail.com'] },
        }),
      }),
    )

    unexpectedFailRule.addEventPattern({
      source: ['aws.states'],
      detailType: ['Step Functions Execution Status Change'],
      detail: {
        stateMachineArn: [stateMachine.stateMachineArn],
        status: ['FAILED', 'TIMED_OUT', 'ABORTED'],
      },
    })

import { Duration, Fn, RemovalPolicy } from 'aws-cdk-lib'
import { Rule, RuleTargetInput, EventField } from 'aws-cdk-lib/aws-events'
import { SfnStateMachine, SnsTopic } from 'aws-cdk-lib/aws-events-targets'
import { Role, ServicePrincipal } from 'aws-cdk-lib/aws-iam'
import { Alias, Architecture, IFunction } from 'aws-cdk-lib/aws-lambda'
import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs'
import { LogGroup, RetentionDays } from 'aws-cdk-lib/aws-logs'
import { Bucket } from 'aws-cdk-lib/aws-s3'
import { Topic } from 'aws-cdk-lib/aws-sns'
import { Fail, LogLevel, Parallel, StateMachine, StateMachineType, TaskInput } from 'aws-cdk-lib/aws-stepfunctions'
import { LambdaInvoke, SnsPublish } from 'aws-cdk-lib/aws-stepfunctions-tasks'
import { Construct } from 'constructs'

export default class AutomationConfiguration extends Construct {
  constructor(scope: Construct, id: string) {
    super(scope, id)

    const userNotificationTopic = Topic.fromTopicArn(
      this,
      'userNotificationTopic',
      Fn.importValue('infrastructure::sns::user-notification-topic::arn'),
    )

    const bucket = new Bucket(this, 'bucket', {
      bucketName: 'automation-configuration',
      removalPolicy: RemovalPolicy.DESTROY,
    })

    const etlLambda = new NodejsFunction(this, 'etl-lambda', {
      entry: 'src/lambda/etl/etl.handler.ts',
      functionName: 'etl',
      timeout: Duration.seconds(30),
      architecture: Architecture.ARM_64,
    })

    bucket.grantRead(
      new Alias(this, id.concat('alias'), {
        aliasName: 'current',
        version: etlLambda.currentVersion,
      }),
    )

    const categoriesLambda = new NodejsFunction(this, 'insert-categories-lambda', {
      entry: 'src/lambda/categories/categories.handler.ts',
      functionName: 'insert-categories',
      timeout: Duration.seconds(30),
      architecture: Architecture.ARM_64,
    })

    const hospitalsLambda = new NodejsFunction(this, 'hospitals-categories-lambda', {
      entry: 'src/lambda/categories/categories.handler.ts',
      functionName: 'insert-categories',
      timeout: Duration.seconds(30),
      architecture: Architecture.ARM_64,
    })

    const [etlTask, categoriesTask, hospitalsTask] = this.mountLambdaInvokes([
      { function: etlLambda, output: '$.Payload' },
      { function: categoriesLambda },
      { function: hospitalsLambda },
    ])

    const errorTopicConfig = {
      topic: userNotificationTopic,
      subject: 'Automation Config Failed 😥',
      message: TaskInput.fromObject({
        message: 'Automation Config Failed due to an unexpected error.',
        cause: 'Unexpected Error',
        channel: 'email',
        destination: { ToAddresses: ['suporte@somemail.com'] },
      }),
    }

    const publishFailed = (publishFailedId: string) =>
      new SnsPublish(this, `automation-config-sns-failed-${publishFailedId}`, errorTopicConfig)

    const jobFailed = new Fail(this, 'automation-config-job-failed', {
      cause: 'Unexpected Error',
    })

    const hospitalsCategoriesParallel = new Parallel(this, 'auto-config-exams-parallel-map')
      .branch(categoriesTask)
      .branch(hospitalsTask)
      .addRetry({ errors: ['States.ALL'], interval: Duration.seconds(5), maxAttempts: 1 })
      .addCatch(publishFailed('exams').next(jobFailed), {
        errors: ['States.ALL'],
      })
    const definition = etlTask.next(hospitalsCategoriesParallel)

    const logGroup = new LogGroup(this, 'automation-configuration-log-group', {
      retention: RetentionDays.ONE_WEEK,
      removalPolicy: RemovalPolicy.DESTROY,
    })

    const stateMachine = new StateMachine(this, `${id}-state-machine`, {
      definition,
      timeout: Duration.minutes(5),
      stateMachineName: 'automation-configuration',
      stateMachineType: StateMachineType.EXPRESS,
      logs: {
        destination: logGroup,
        includeExecutionData: true,
        level: LogLevel.ALL,
      },
    })

    const s3EventRule = new Rule(this, 'automation-config-s3-event-rule', {
      ruleName: 'automation-config-s3-event-rule',
    })

    const eventRole = new Role(this, 'eventRole', {
      assumedBy: new ServicePrincipal('events.amazonaws.com'),
    })

    stateMachine.grantStartExecution(eventRole)
    s3EventRule.addTarget(
      new SfnStateMachine(stateMachine, {
        input: RuleTargetInput.fromObject({
          detail: EventField.fromPath('$.detail'),
        }),
        role: eventRole,
      }),
    )
    s3EventRule.addEventPattern({
      source: ['aws.s3'],
      detailType: ['Object Created'],
      detail: {
        bucket: {
          name: [bucket.bucketName],
        },
        object: {
          key: [
            {
              wildcard: 'csv/automation-configuration/*.csv',
            },
          ],
        },
      },
    })

    const unexpectedFailRule = new Rule(this, 'exam-automation-config-unexpected-fail-rule', {
      ruleName: 'exam-automation-config-unexpected-fail-rule',
    })
    unexpectedFailRule.addTarget(
      new SnsTopic(userNotificationTopic, {
        message: RuleTargetInput.fromObject({
          subject: 'Exam Automation Config Failed 😥',
          message: 'Exam Automation Config Failed due to an unexpected error.',
          cause: 'Unexpected Error',
          channel: 'email',
          destination: { ToAddresses: ['it@somemail.com'] },
        }),
      }),
    )

    unexpectedFailRule.addEventPattern({
      source: ['aws.states'],
      detailType: ['Step Functions Execution Status Change'],
      detail: {
        stateMachineArn: [stateMachine.stateMachineArn],
        status: ['FAILED', 'TIMED_OUT', 'ABORTED'],
      },
    })
  }

  private mountLambdaInvokes(
    lambdasInvoke: Array<{
      function: IFunction
      name?: string
      output?: string
    }>,
  ) {
    return lambdasInvoke.map(lambdaInvoke => {
      return new LambdaInvoke(this, `${lambdaInvoke.name || lambdaInvoke.function.functionName}-task`, {
        lambdaFunction: lambdaInvoke.function,
        inputPath: '$',
        outputPath: lambdaInvoke?.output || '$',
      })
    })
  }
}