feat: refactor certificate repository for DynamoDB single-table

README.md

@@ -21,6 +21,37 @@ Este projeto é uma função AWS Lambda responsável por processar notificaçõe
 - **Deploy**: o workflow `.github/workflows/workflow_build.yaml` gera um ZIP e atualiza a função `tech-floripa-certificates-notification-dev`.
 - **Infra**: a fila SQS continua ligada pela infraestrutura Terraform; apenas o artefato de código mudou de imagem para ZIP.
 
+## DynamoDB Single-Table Design
+
+Este projeto utiliza o padrão DynamoDB Single-Table Design. O CertificateRepository opera na mesma tabela que a API.
+
+### Estrutura da Tabela (compartilhada)
+
+| Atributo | Tipo | Descrição |
+|----------|------|-----------|
+| `PK` | String | Chave de Partição principal |
+| `SK` | String | Chave de Ordenação principal |
+| `GSI1PK`, `GSI1SK` | String | GSI1 para Certificate por UUID |
+| `GSI2PK`, `GSI2SK` | String | GSI2 para acesso por email |
+| `GSI3PK`, `GSI3SK` | String | GSI3 para acesso por produto |
+| `GSI4PK`, `GSI4SK` | String | GSI4 para certificados por status |
+| `EntityType` | String | Tipo da entidade |
+
+### GSI4 - Acesso por Status de Sucesso
+
+| GSI4PK | GSI4SK | Uso |
+|--------|--------|-----|
+| `SUCCESS#true` | `CERTIFICATE#<uuid>` | Certificados bem-sucedidos |
+| `SUCCESS#false` | `CERTIFICATE#<uuid>` | Certificados falhados |
+
+### Repositório de Certificados
+
+O `CertificateRepositoryImpl` utiliza:
+- **GSI1**: Busca por UUID do certificado
+- **GSI2**: Busca por email do participante
+- **GSI3**: Busca por product_id
+- **GSI4**: Busca por status de sucesso
+
 ## Estrutura do Evento
 
 ### Entrada (SQS)

src/infrastructure/aws/dynamodb_keys.py

@@ -0,0 +1,88 @@
+from enum import Enum
+from typing import Optional
+
+
+class EntityType(str, Enum):
+    ORDER = "ORDER"
+    PRODUCT = "PRODUCT"
+    PARTICIPANT = "PARTICIPANT"
+    CERTIFICATE = "CERTIFICATE"
+
+
+def pk(entity_type: EntityType, entity_id: str | int) -> str:
+    return f"{entity_type.value}#{entity_id}"
+
+
+def sk(entity_type: EntityType, entity_id: str | int) -> str:
+    return f"{entity_type.value}#{entity_id}"
+
+
+def sk_order(order_id: int) -> str:
+    return sk(EntityType.ORDER, order_id)
+
+
+def sk_product(product_id: int) -> str:
+    return sk(EntityType.PRODUCT, product_id)
+
+
+def sk_participant(participant_id: str) -> str:
+    return sk(EntityType.PARTICIPANT, participant_id)
+
+
+def sk_certificate(order_id: int) -> str:
+    return sk(EntityType.CERTIFICATE, order_id)
+
+
+def gsi1pk_email(email: str) -> str:
+    return f"EMAIL#{email.lower().strip()}"
+
+
+def gsi1sk_participant(participant_id: str) -> str:
+    return f"PARTICIPANT#{participant_id}"
+
+
+def gsi1sk_order(order_id: int) -> str:
+    return f"ORDER#{order_id}"
+
+
+def gsi1sk_certificate(order_id: int) -> str:
+    return f"CERTIFICATE#{order_id}"
+
+
+def gsi2pk_product(product_id: int) -> str:
+    return f"PRODUCT#{product_id}"
+
+
+def gsi2sk_order(order_id: int) -> str:
+    return f"ORDER#{order_id}"
+
+
+def gsi2sk_certificate(order_id: int) -> str:
+    return f"CERTIFICATE#{order_id}"
+
+
+def gsi3pk_success(success: bool) -> str:
+    return f"SUCCESS#{1 if success else 0}"
+
+
+def gsi3sk_certificate(order_id: int) -> str:
+    return f"CERTIFICATE#{order_id}"
+
+
+def gsi1pk_cert_id(cert_id: str) -> str:
+    return f"CERT#{cert_id}"
+
+
+def gsi1sk_cert(cert_id: str) -> str:
+    return f"CERTIFICATE#{cert_id}"
+
+
+def parse_pk(pk: str) -> tuple[EntityType, str]:
+    parts = pk.split("#", 1)
+    if len(parts) != 2:
+        raise ValueError(f"Invalid PK format: {pk}")
+    return EntityType(parts[0]), parts[1]
+
+
+def parse_sk(sk: str) -> tuple[EntityType, str]:
+    return parse_pk(sk)

src/infrastructure/config/config.py

@@ -18,64 +18,24 @@ class Config:
         env_file_encoding = "utf-8"
 
     @property
-    def dynamodb_tables(self) -> Dict[str, Dict[str, str]]:
-        """
-        Retorna as configurações das tabelas do DynamoDB baseadas no ambiente.
-        Segue o padrão da infraestrutura Terraform criada.
-        """
+    def dynamodb_table(self) -> Dict[str, str]:
         base_name = f"{self.PROJECT_NAME}"
         environment = self.ENVIRONMENT
 
         return {
-            "certificates": {
-                "name": f"{base_name}-certificates-{environment}",
-                "arn": f"arn:aws:dynamodb:{self.REGION}:*:table/{base_name}-certificates-{environment}"
-            },
-            "orders": {
-                "name": f"{base_name}-orders-{environment}",
-                "arn": f"arn:aws:dynamodb:{self.REGION}:*:table/{base_name}-orders-{environment}"
-            },
-            "participants": {
-                "name": f"{base_name}-participants-{environment}",
-                "arn": f"arn:aws:dynamodb:{self.REGION}:*:table/{base_name}-participants-{environment}"
-            },
-            "products": {
-                "name": f"{base_name}-products-{environment}",
-                "arn": f"arn:aws:dynamodb:{self.REGION}:*:table/{base_name}-products-{environment}"
-            }
+            "name": f"{base_name}-{environment}",
+            "arn": f"arn:aws:dynamodb:{self.REGION}:*:table/{base_name}-{environment}"
         }
 
-    def get_table_name(self, entity: str) -> str:
-        """
-        Retorna o nome da tabela para uma entidade específica.
-
-        Args:
-            entity: Nome da entidade (certificates, orders, participants, products)
-
-        Returns:
-            str: Nome da tabela no DynamoDB
-        """
-        tables = self.dynamodb_tables
-        if entity not in tables:
-            raise ValueError(f"Entidade '{entity}' não encontrada. Entidades disponíveis: {list(tables.keys())}")
-
-        return tables[entity]["name"]
+    @property
+    def dynamodb_tables(self) -> Dict[str, Dict[str, str]]:
+        return {"single": self.dynamodb_table}
 
-    def get_table_arn(self, entity: str) -> str:
-        """
-        Retorna o ARN da tabela para uma entidade específica.
-
-        Args:
-            entity: Nome da entidade (certificates, orders, participants, products)
-
-        Returns:
-            str: ARN da tabela no DynamoDB
-        """
-        tables = self.dynamodb_tables
-        if entity not in tables:
-            raise ValueError(f"Entidade '{entity}' não encontrada. Entidades disponíveis: {list(tables.keys())}")
-
-        return tables[entity]["arn"]
+    def get_table_name(self, entity: str = None) -> str:
+        return self.dynamodb_table["name"]
+
+    def get_table_arn(self, entity: str = None) -> str:
+        return self.dynamodb_table["arn"]
 
 
 config = Config()