[Feature Request] Add support for S3 events sent by cloudtrail

When setting up s3 event notifications, you can either choose to do it directly through s3 or you can do it through eventbridge. You cannot use a combination for the same bucket. Using s3 event notifications through eventbridge offers some advantages as it offers more native target AWS services compared to normal s3 events. The problem however is that the events generated by eventbridge are different from the native s3 events:

s3 event:
```
{
"Records": [
{
"eventVersion": "2.1",
"eventSource": "aws:s3",
"awsRegion": "eu-central-1",
"eventTime": "2025-08-21T01:42:56.748Z",
"eventName": "ObjectCreated:CompleteMultipartUpload",
"userIdentity": {
"principalId":
},
"requestParameters": {
"sourceIPAddress":
},
"responseElements": {
"x-amz-request-id": ,
"x-amz-id-2":
},
"s3": {
"s3SchemaVersion": "1.0",
"configurationId": ,
"bucket": {
"name": ,
"ownerIdentity": {
"principalId":
},
"arn":
},
"object": {
"key": ,
"size": 1997,
"eTag": ,
"versionId": ,
"sequencer":
}
}
}
]
```

eventbridge event:
```
{
"version": "0",
"id": ,
"detail-type": "Object Created",
"source": "aws.s3",
"account": ,
"time": "2025-08-21T08:06:59Z",
"region": "eu-central-1",
"resources": [

],
"detail": {
"version": "0",
"bucket": {
"name":
},
"object": {
"key": ,
"size": 3091,
"etag": ,
"version-id": ,
"sequencer":
},
"request-id": ,
"requester": ,
"source-ip-address": ,
"reason": "PutObject"
}
}
```

As you can see the structure is completely different and does not even include the same information such as the event name. This information is used to parse or throw away records:
```
private def parseSqsMessage(message: Message): Option[FileMetadata[T]] = {
implicit val formats: DefaultFormats.type = DefaultFormats
try {
val messageReceiptHandle = message.receiptHandle
val messageBody = message.body
logDebug(s"parse SQS message body: ${messageBody}")

val messageJson = parse(messageBody).extract[JValue]
val bucketName = (
messageJson \ "Records" \ "s3" \ "bucket" \ "name").extract[Array[String]].head
val eventName = (messageJson \ "Records" \ "eventName").extract[Array[String]].head

if (eventName.contains("ObjectCreated")) {
val timestamp = (messageJson \ "Records" \ "eventTime").extract[Array[String]].head
val timestampMills = convertTimestampToMills(timestamp)
val path = "s3://" +
bucketName + "/" +
(messageJson \ "Records" \ "s3" \ "object" \ "key").extract[Array[String]].head
logDebug("Successfully parsed sqs message")
metrics.parseMessageCounter.inc()

Some(FileMetadata(URLDecoder.decode(path, StandardCharsets.UTF_8.name()),
timestampMills, Some(messageReceiptHandle.asInstanceOf[T])))
}
else {
logDebug(s"Discarded event ${eventName}")
metrics.discardedMessageCounter.inc()
None
}
} catch {
case me: MappingException =>
logWarning(s"Error in parsing SQS message ${message.receiptHandle}", me)
metrics.parseMessageFailedCounter.inc()
None
case NonFatal(e) =>
logWarning(s"Unexpected error while parsing SQS message ${message.receiptHandle}", e)
metrics.parseMessageFailedCounter.inc()
None
}
}
```

If for whatever reason you need to configure your s3 events to use eventbridge, it should also be possible for the s3 connector to handle these type of events.