Event-Driven Workflows: Replacing Manual Operations with Serverless Automation

import json
import boto3
import logging

dynamodb = boto3.resource('dynamodb')
accounts_table = dynamodb.Table('Accounts')
logger = logging.getLogger()

def lambda_handler(event, context):
    """
    Triggered by API Gateway when account closure requested.
    Validates request and starts Step Function execution.
    """
    
    try:
        account_id = event['pathParameters']['account_id']
        user_id = event['requestContext']['authorizer']['claims']['sub']
        
        # Verify permission: user must be account owner
        account = accounts_table.get_item(Key={'account_id': account_id})
        if not account:
            return {
                'statusCode': 404,
                'body': json.dumps({'error': 'Account not found'})
            }
        
        account_data = account['Item']
        if account_data['owner_id'] != user_id:
            return {
                'statusCode': 403,
                'body': json.dumps({'error': 'Unauthorized'})
            }
        
        # Check for pending transactions
        if account_data['status'] == 'pending_deletion':
            return {
                'statusCode': 409,
                'body': json.dumps({'error': 'Account closure already in progress'})
            }
        
        # Check for active subscriptions
        if account_data.get('active_subscriptions', 0) > 0:
            return {
                'statusCode': 400,
                'body': json.dumps({
                    'error': 'Account has active subscriptions',
                    'detail': 'Cancel all subscriptions before deleting account'
                })
            }
        
        # All validations passed - start Step Function
        sfn = boto3.client('stepfunctions')
        
        execution = sfn.start_execution(
            stateMachineArn='arn:aws:states:region:account:stateMachine:AccountClosure',
            name=f'closure-{account_id}-{int(time.time())}',
            input=json.dumps({
                'account_id': account_id,
                'user_id': user_id,
                'user_email': account_data['email'],
                'timestamp': int(time.time())
            })
        )
        
        # Update account status immediately
        accounts_table.update_item(
            Key={'account_id': account_id},
            UpdateExpression='SET #status = :status, closure_request_id = :request_id',
            ExpressionAttributeNames={'#status': 'status'},
            ExpressionAttributeValues={
                ':status': 'pending_deletion',
                ':request_id': execution['executionArn']
            }
        )
        
        logger.info(f'Account closure started: {account_id}')
        
        return {
            'statusCode': 202,
            'body': json.dumps({
                'message': 'Account closure initiated',
                'execution_id': execution['executionArn']
            })
        }
        
    except Exception as e:
        logger.error(f'Error initiating closure: {str(e)}', exc_info=True)
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Internal server error'})
        }

import boto3
import hashlib
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
accounts_table = dynamodb.Table('Accounts')
users_table = dynamodb.Table('Users')
logger = logging.getLogger()

def anonymize_user_data(account_id):
    """
    Remove PII from all user records associated with account.
    Uses hashing instead of deletion to maintain referential integrity.
    """
    
    try:
        # Get all users in account
        response = users_table.scan(
            FilterExpression='account_id = :account_id',
            ExpressionAttributeValues={':account_id': account_id}
        )
        
        users = response['Items']
        
        # Anonymize each user
        with users_table.batch_writer(
            overwrite_by_pkeys=['account_id', 'user_id']
        ) as batch:
            for user in users:
                # Hash email to maintain uniqueness without storing PII
                anonymized_email = hashlib.sha256(
                    f"{user['user_id']}{datetime.utcnow().isoformat()}".encode()
                ).hexdigest()[:20]
                
                batch.put_item(Item={
                    'account_id': account_id,
                    'user_id': user['user_id'],
                    'name': 'DELETED_USER',
                    'email': f'deleted+{anonymized_email}@deleted.local',
                    'phone': None,
                    'address': None,
                    'anonymized_at': datetime.utcnow().isoformat(),
                    'anonymized': True,
                    # Keep these for audit trail
                    'original_user_id': user['user_id'],
                    'deleted_account_id': account_id
                })
        
        logger.info(f'Anonymized {len(users)} users for account {account_id}')
        return {'success': True, 'users_anonymized': len(users)}
        
    except Exception as e:
        logger.error(f'Anonymization failed: {str(e)}', exc_info=True)
        raise

import boto3
import stripe
from decimal import Decimal

dynamodb = boto3.resource('dynamodb')
subscriptions_table = dynamodb.Table('Subscriptions')
stripe.api_key = os.environ['STRIPE_API_KEY']
logger = logging.getLogger()

def handle_billing(account_id):
    """
    Cancel subscriptions and issue refunds for unused time.
    Integrates with Stripe for payment processing.
    """
    
    try:
        # Get all active subscriptions for account
        response = subscriptions_table.scan(
            FilterExpression='account_id = :account_id AND #status = :status',
            ExpressionAttributeNames={'#status': 'status'},
            ExpressionAttributeValues={
                ':account_id': account_id,
                ':status': 'active'
            }
        )
        
        subscriptions = response['Items']
        total_refunded = Decimal('0')
        
        for subscription in subscriptions:
            try:
                stripe_sub_id = subscription['stripe_subscription_id']
                
                # Get subscription details from Stripe
                stripe_subscription = stripe.Subscription.retrieve(stripe_sub_id)
                
                # Calculate refund amount (unused days)
                current_period_end = stripe_subscription['current_period_end']
                amount_paid = stripe_subscription['plan']['amount']
                period_days = (current_period_end - stripe_subscription['current_period_start']) / 86400
                days_used = (datetime.utcnow().timestamp() - stripe_subscription['current_period_start']) / 86400
                days_remaining = period_days - days_used
                
                refund_amount = int((amount_paid / period_days) * days_remaining)
                
                # Issue refund
                if refund_amount > 0:
                    refund = stripe.Refund.create(
                        charge=stripe_subscription['default_payment_method'],
                        amount=refund_amount
                    )
                    logger.info(f'Refund issued: {refund_amount} cents')
                    total_refunded += Decimal(refund_amount) / 100
                
                # Cancel subscription
                stripe.Subscription.delete(stripe_sub_id)
                
                # Update local database
                subscriptions_table.update_item(
                    Key={
                        'account_id': account_id,
                        'subscription_id': subscription['subscription_id']
                    },
                    UpdateExpression='SET #status = :status, cancelled_at = :now, refund_amount = :refund',
                    ExpressionAttributeNames={'#status': 'status'},
                    ExpressionAttributeValues={
                        ':status': 'cancelled',
                        ':now': datetime.utcnow().isoformat(),
                        ':refund': refund_amount
                    }
                )
                
            except stripe.error.StripeError as e:
                logger.error(f'Stripe error: {str(e)}')
                raise
        
        logger.info(f'Billing handled: {len(subscriptions)} subscriptions cancelled, ${total_refunded} refunded')
        return {
            'success': True,
            'subscriptions_cancelled': len(subscriptions),
            'total_refunded': float(total_refunded)
        }
        
    except Exception as e:
        logger.error(f'Billing handling failed: {str(e)}', exc_info=True)
        raise

idempotency_key = f"refund-{account_id}-{subscription_id}"
refund = stripe.Refund.create(
    charge=charge_id,
    amount=refund_amount,
    idempotency_key=idempotency_key  # Stripe won't process twice
)

logger.info(f'Step: anonymization', extra={
    'account_id': account_id,
    'users_processed': len(users),
    'duration_ms': (end - start) * 1000,
    'step': 'anonymize_data'
})