Class CpEventSource

CloudKitect Event Sourcing Pattern

If the event sent to domain x8EventBus has a EventType DOMAIN and DomainName that matches x8DomainName then those events are processed by this pipeline. First it will persist the event in the event source Domain Event Source Table then apply those events to the domain and store the updated domain state in the Domains Table.

Note: Currently only available for Python Based Lambda Functions

Input: Domain Event

{
     "DomainId": "UUID",
     "EventTime": "ISO_FORMAT",
     "EventId": "UUID",
     "CorrelationId": "UUID",
     "UserId": "UUID",
     "EventType": "DOMAIN/PROCESS",
     "EventName": "FileScanned",
     "DomainName": "File",
     "IpAddress": "1.1.1.1",
     "Payload": {
         {Any Valid JSON},
     },
     "Version": 1
}
Copy

Payload could be any valid json format, however it is required to provide a domain lambda layer with the following format

Domain layer should have a file with name domain_handler.py The domain handler must contain a Domain class and another data class such as FileDomain, UserDomain, AccountDomain etc

Infrastructure Diagram

@dataclass
class FileDomain:
    DomainId: str
    Name: str
    OwnerId: str
    More Fields

    def apply(self, domain_payload, event_time, event_type):
        self.Name = domain_payload["Name"]["S"]
        More Fields to match the Domain

class Domain:

    @staticmethod
    def from_dynamo_item(item):
        return FileDomain(
                DomainId=item.DomainId,
                OwnerId=item.OwnerId,
                Name=item.Name
                More Fields to match the Domain
        )


    @staticmethod
    def create_new(domain_id, user_id) -> FileDomain:
        now = datetime.utcnow()
        return FileDomain(
                DomainId=domain_id,
                OwnerId=user_id,
                Name="",
                More Fields to match the Domain
            )

    @staticmethod
    def to_dynamo_item(domain: FileDomain):
        return {
            'DomainId': domain.DomainId,
            'OwnerId': domain.OwnerId,
            'Name': domain.Name
             More Fields to match the Domain
        }
Copy

Output: Events persisted in Domain Events Table and Domain updated in Domain Table

Default Configuration

Default Event Pattern to match

detail: {
            EventType: "DOMAIN",
            DomainName: props.x8DomainName
        }
Copy

Default Alarms

Note that the default alarm uses the WaAlarm construct, which sets up an alarm action to notify the SNS Topic AlarmEventsTopic by default.

Examples

Default Usage

new CpEventSource(this, "LogicalId", {
    cpDomainLogicLayer: DomainPythonLayer,
    cpDomainName: "File"
});
Copy

Custom Configuration

new CpEventSource(this, "LogicalId", {
    cpDomainLogicLayer: DomainPythonLayer,
    cpDomainName: "File",
    cpEventBus: CustomEventBus
});
Copy

Compliance

It addresses the following compliance requirements