Como eliminei 24 horas de latência com Salesforce CDC, EventBridge e Lambda

# Secret do Firebird
aws secretsmanager create-secret \
  --name "firebird/prod/credentials" \
  --description "Credenciais do Firebird (ERP)" \
  --secret-string '{
    "host": "10.0.1.50",
    "port": "3050",
    "database": "/dados/ERP.FDB",
    "user": "SYSDBA",
    "password": "SUA_SENHA_FIREBIRD"
  }'

# Secret do Salesforce
aws secretsmanager create-secret \
  --name "salesforce/prod/credentials" \
  --description "Credenciais OAuth do Salesforce Connected App" \
  --secret-string '{
    "domain": "sua-instancia.my.salesforce.com",
    "consumer_key": "SEU_CONSUMER_KEY",
    "consumer_secret": "SEU_CONSUMER_SECRET",
    "username": "integracao@exemplo.com.br",
    "password": "SUA_SENHA_SF",
    "security_token": "SEU_TOKEN"
  }'

aws sqs create-queue \
  --queue-name "salesforce-cdc-dlq" \
  --attributes '{
    "MessageRetentionPeriod": "1209600",
    "VisibilityTimeout": "300"
  }'

# Anote o QueueUrl e o ARN retornados — usei nas próximas etapas

# Criar o Security Group
aws ec2 create-security-group \
  --group-name "sg-lambda-salesforce-cdc" \
  --description "SG para Lambda CDC - acesso ao Firebird e internet via NAT" \
  --vpc-id "vpc-XXXXXXXX"

# Permitir tráfego de entrada na porta 3050 (Firebird) dentro da VPC
aws ec2 authorize-security-group-ingress \
  --group-id "sg-XXXXXXXX" \
  --protocol tcp \
  --port 3050 \
  --cidr "10.0.0.0/16"

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "SecretsAccess",
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue"
      ],
      "Resource": [
        "arn:aws:secretsmanager:us-east-1:ACCOUNT_ID:secret:firebird/prod/*",
        "arn:aws:secretsmanager:us-east-1:ACCOUNT_ID:secret:salesforce/prod/*"
      ]
    },
    {
      "Sid": "DLQAccess",
      "Effect": "Allow",
      "Action": [
        "sqs:SendMessage"
      ],
      "Resource": "arn:aws:sqs:us-east-1:ACCOUNT_ID:salesforce-cdc-dlq"
    },
    {
      "Sid": "CloudWatchLogs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:us-east-1:ACCOUNT_ID:*"
    },
    {
      "Sid": "VPCNetworkInterfaces",
      "Effect": "Allow",
      "Action": [
        "ec2:CreateNetworkInterface",
        "ec2:DescribeNetworkInterfaces",
        "ec2:DeleteNetworkInterface"
      ],
      "Resource": "*"
    }
  ]
}

# Criar a role
aws iam create-role \
  --role-name "lambda-salesforce-cdc-role" \
  --assume-role-policy-document file://trust-policy.json

# Anexar a policy inline
aws iam put-role-policy \
  --role-name "lambda-salesforce-cdc-role" \
  --policy-name "salesforce-cdc-permissions" \
  --policy-document file://inline-policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "EventBridgePartnerAccess",
      "Effect": "Allow",
      "Action": [
        "events:PutPartnerEvents",
        "events:CreatePartnerEventSource",
        "events:ActivatePartnerEventSource"
      ],
      "Resource": "*"
    }
  ]
}

# Criar o usuário
aws iam create-user --user-name "salesforce-eventbridge-relay"

# Anexar a policy
aws iam put-user-policy \
  --user-name "salesforce-eventbridge-relay" \
  --policy-name "eventbridge-partner-access" \
  --policy-document file://eventbridge-relay-policy.json

# Gerar Access Key
aws iam create-access-key --user-name "salesforce-eventbridge-relay"

# ANOTE o AccessKeyId e SecretAccessKey — usei na Named Credential (Etapa 6)

# Criar a regra
aws events put-rule \
  --name "salesforce-cdc-to-lambda" \
  --event-bus-name "aws.partner/salesforce.com/XXXXXXXX/salesforce-cdc" \
  --event-pattern '{
    "source": ["aws.partner/salesforce.com/XXXXXXXX/salesforce-cdc"]
  }' \
  --state "ENABLED" \
  --description "Roteia eventos CDC do Salesforce para Lambda de sincronização"

# Adicionar Lambda como target com DLQ
aws events put-targets \
  --rule "salesforce-cdc-to-lambda" \
  --event-bus-name "aws.partner/salesforce.com/XXXXXXXX/salesforce-cdc" \
  --targets '[
    {
      "Id": "lambda-sync-firebird",
      "Arn": "arn:aws:lambda:us-east-1:ACCOUNT_ID:function:salesforce-cdc-sync",
      "DeadLetterConfig": {
        "Arn": "arn:aws:sqs:us-east-1:ACCOUNT_ID:salesforce-cdc-dlq"
      },
      "RetryPolicy": {
        "MaximumRetryAttempts": 3,
        "MaximumEventAgeInSeconds": 3600
      }
    }
  ]'

# Criar a função Lambda
aws lambda create-function \
  --function-name "salesforce-cdc-sync" \
  --runtime "python3.12" \
  --role "arn:aws:iam::ACCOUNT_ID:role/lambda-salesforce-cdc-role" \
  --handler "handler.lambda_handler" \
  --timeout 60 \
  --memory-size 256 \
  --vpc-config '{
    "SubnetIds": ["subnet-XXXXXXXX"],
    "SecurityGroupIds": ["sg-XXXXXXXX"]
  }' \
  --environment '{
    "Variables": {
      "FIREBIRD_SECRET_NAME": "firebird/prod/credentials",
      "SALESFORCE_SECRET_NAME": "salesforce/prod/credentials",
      "DLQ_URL": "https://sqs.us-east-1.amazonaws.com/ACCOUNT_ID/salesforce-cdc-dlq"
    }
  }' \
  --zip-file fileb://deployment-package.zip

# Permitir que o EventBridge invoque a Lambda
aws lambda add-permission \
  --function-name "salesforce-cdc-sync" \
  --statement-id "eventbridge-invoke" \
  --action "lambda:InvokeFunction" \
  --principal "events.amazonaws.com" \
  --source-arn "arn:aws:events:us-east-1:ACCOUNT_ID:rule/aws.partner/salesforce.com/XXXXXXXX/salesforce-cdc/salesforce-cdc-to-lambda"

"""
Lambda: Salesforce CDC → Firebird sync
Recebe eventos CDC do Salesforce via EventBridge e atualiza
TABCAVALETE e TABBLOCOCHAPAS no Firebird.
"""

import json
import os
import logging
import boto3
import firebirdsql
from simple_salesforce import Salesforce

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# ---------------------------------------------------------------------------
# Mapeamento de status Salesforce → Firebird
# ---------------------------------------------------------------------------
STATUS_MAP = {
    "Disponível":    "D",
    "Reservado":     "R",
    "Vendido":       "V",
    "Faturado":      "F",
    "Embarcado":     "E",
    "Cancelado":     "C",
    "Direcionado":   "R",   # caso especial — tratado em resolve_status()
}

# ---------------------------------------------------------------------------
# Cache de secrets (reutilizado entre invocações do mesmo container)
# ---------------------------------------------------------------------------
_secrets_cache = {}

sqs = boto3.client("sqs")


def get_secret(secret_name: str) -> dict:
    """Busca secret no Secrets Manager com cache em memória."""
    if secret_name in _secrets_cache:
        return _secrets_cache[secret_name]

    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId=secret_name)
    secret = json.loads(response["SecretString"])
    _secrets_cache[secret_name] = secret
    logger.info(f"Secret '{secret_name}' carregado com sucesso")
    return secret


def get_salesforce_record(sf: Salesforce, record_id: str) -> dict:
    """
    Consulta o registro completo no Salesforce via SOQL.
    O evento CDC traz apenas os campos alterados — precisamos do registro
    inteiro para garantir consistência na atualização do Firebird.
    """
    query = f"""
        SELECT Id, Name, Status__c, Cavalete__c, Bloco__c,
               NumeroPeca__c, Tipo__c, Observacao__c
        FROM Embalagem__c
        WHERE Id = '{record_id}'
    """
    result = sf.query(query)

    if result["totalSize"] == 0:
        raise ValueError(f"Registro {record_id} não encontrado no Salesforce")

    return result["records"][0]


def resolve_status(record: dict) -> str:
    """
    Resolve o status para gravação no Firebird.

    Caso especial: 'Direcionado' no Salesforce significa que o material
    foi pré-alocado para um cliente, mas ainda não foi vendido formalmente.
    No Firebird, isso é tratado como 'R' (Reservado), porém com o campo
    OBSERVACAO preenchido com 'DIRECIONADO - [cliente]'.
    """
    status_sf = record.get("Status__c", "")
    status_fb = STATUS_MAP.get(status_sf)

    if status_fb is None:
        logger.warning(f"Status desconhecido: '{status_sf}'. Usando 'D' (Disponível).")
        return "D"

    return status_fb


def update_cavalete(cursor, record: dict, status_fb: str):
    """
    Atualiza a tabela TABCAVALETE no Firebird.
    TABCAVALETE armazena dados do cavalete/pallet como unidade.
    """
    cavalete = record.get("Cavalete__c")
    if not cavalete:
        logger.info("Sem Cavalete__c no registro — pulando TABCAVALETE")
        return

    sql = """
        UPDATE TABCAVALETE
        SET STATUS = ?,
            OBSERVACAO = ?,
            DTALTERACAO = CURRENT_TIMESTAMP
        WHERE CODCAVALETE = ?
    """

    obs = record.get("Observacao__c", "") or ""
    if record.get("Status__c") == "Direcionado":
        obs = f"DIRECIONADO - {obs}".strip(" -")

    cursor.execute(sql, (status_fb, obs, cavalete))
    logger.info(f"TABCAVALETE atualizado: CODCAVALETE={cavalete}, STATUS={status_fb}")


def update_chapa(cursor, record: dict, status_fb: str):
    """
    Atualiza a tabela TABBLOCOCHAPAS no Firebird.
    TABBLOCOCHAPAS armazena dados de cada chapa individual cortada de um bloco.
    """
    bloco = record.get("Bloco__c")
    numero_peca = record.get("NumeroPeca__c")

    if not bloco or not numero_peca:
        logger.info("Sem Bloco__c ou NumeroPeca__c — pulando TABBLOCOCHAPAS")
        return

    sql = """
        UPDATE TABBLOCOCHAPAS
        SET STATUS = ?,
            OBSERVACAO = ?,
            DTALTERACAO = CURRENT_TIMESTAMP
        WHERE CODBLOCO = ?
          AND NUMEROPECA = ?
    """

    obs = record.get("Observacao__c", "") or ""
    if record.get("Status__c") == "Direcionado":
        obs = f"DIRECIONADO - {obs}".strip(" -")

    cursor.execute(sql, (status_fb, obs, bloco, int(numero_peca)))
    logger.info(
        f"TABBLOCOCHAPAS atualizado: CODBLOCO={bloco}, "
        f"NUMEROPECA={numero_peca}, STATUS={status_fb}"
    )


def lambda_handler(event, context):
    """
    Entry point — recebe evento do EventBridge (Salesforce CDC).

    Estrutura do evento:
      event["detail"]["payload"]["ChangeEventHeader"]["entityName"]
      event["detail"]["payload"]["ChangeEventHeader"]["recordIds"]
      event["detail"]["payload"]["ChangeEventHeader"]["changeType"]
    """
    logger.info(f"Evento recebido: {json.dumps(event, default=str)}")

    try:
        # ----- Extrair header do CDC -----
        payload = event.get("detail", {}).get("payload", {})
        header = payload.get("ChangeEventHeader", {})

        entity_name = header.get("entityName", "")
        change_type = header.get("changeType", "")
        record_ids = header.get("recordIds", [])

        # Filtrar: só processamos Embalagem__c
        if entity_name != "Embalagem__c":
            logger.info(f"Ignorando evento de '{entity_name}' — não é Embalagem__c")
            return {"statusCode": 200, "body": "Ignored — wrong entity"}

        # Filtrar: só processamos UPDATE e CREATE
        if change_type not in ("UPDATE", "CREATE"):
            logger.info(f"Ignorando changeType '{change_type}'")
            return {"statusCode": 200, "body": f"Ignored — {change_type}"}

        if not record_ids:
            logger.warning("Evento sem recordIds — nada a processar")
            return {"statusCode": 200, "body": "No record IDs"}

        # ----- Conectar ao Salesforce -----
        sf_creds = get_secret(os.environ["SALESFORCE_SECRET_NAME"])
        sf = Salesforce(
            username=sf_creds["username"],
            password=sf_creds["password"],
            security_token=sf_creds["security_token"],
            consumer_key=sf_creds["consumer_key"],
            consumer_secret=sf_creds["consumer_secret"],
            domain=sf_creds["domain"].replace(".my.salesforce.com", ""),
        )

        # ----- Conectar ao Firebird -----
        fb_creds = get_secret(os.environ["FIREBIRD_SECRET_NAME"])
        conn = firebirdsql.connect(
            host=fb_creds["host"],
            port=int(fb_creds["port"]),
            database=fb_creds["database"],
            user=fb_creds["user"],
            password=fb_creds["password"],
            charset="UTF8",
        )
        cursor = conn.cursor()

        # ----- Processar cada recordId -----
        processed = 0
        for record_id in record_ids:
            logger.info(f"Processando recordId: {record_id}")

            # Buscar registro completo no Salesforce
            record = get_salesforce_record(sf, record_id)
            status_fb = resolve_status(record)

            # Atualizar Firebird
            update_cavalete(cursor, record, status_fb)
            update_chapa(cursor, record, status_fb)

            processed += 1

        conn.commit()
        cursor.close()
        conn.close()

        logger.info(f"Concluído: {processed} registro(s) processado(s)")
        return {
            "statusCode": 200,
            "body": f"Processed {processed} record(s)"
        }

    except Exception as e:
        logger.error(f"Erro ao processar evento: {str(e)}", exc_info=True)

        # Enviar evento para DLQ para reprocessamento
        try:
            sqs.send_message(
                QueueUrl=os.environ["DLQ_URL"],
                MessageBody=json.dumps({
                    "error": str(e),
                    "event": event
                }, default=str),
            )
            logger.info("Evento enviado para DLQ")
        except Exception as dlq_err:
            logger.error(f"Falha ao enviar para DLQ: {str(dlq_err)}")

        raise

# Criar diretório de build
mkdir -p build && cp handler.py build/

# Instalar dependências (separar pip install para evitar PyO3 ImportModuleError)
pip install simple-salesforce -t build/
pip install firebirdsql -t build/

# Empacotar
cd build && zip -r ../deployment-package.zip . && cd ..

# Fazer deploy
aws lambda update-function-code \
  --function-name "salesforce-cdc-sync" \
  --zip-file fileb://deployment-package.zip