feat(logs): support LogGroup as a Stream

packages/@punchcard/constructs/package.json

@@ -35,6 +35,7 @@
     "@aws-cdk/aws-kms": "^1.88.0",
     "@aws-cdk/aws-lambda": "^1.88.0",
     "@aws-cdk/aws-logs": "^1.88.0",
+    "@aws-cdk/aws-logs-destinations": "^1.88.0",
     "@aws-cdk/aws-s3": "^1.88.0",
     "@aws-cdk/core": "^1.88.0",
     "@punchcard/erasure": "^0.2.0"
@@ -46,6 +47,7 @@
     "@aws-cdk/aws-kms": "^1.88.0",
     "@aws-cdk/aws-lambda": "^1.88.0",
     "@aws-cdk/aws-logs": "^1.88.0",
+    "@aws-cdk/aws-logs-destinations": "^1.88.0",
     "@aws-cdk/aws-s3": "^1.88.0",
     "@aws-cdk/core": "^1.88.0",
     "@punchcard/erasure": "^0.2.0"

packages/@punchcard/constructs/src/index.ts

@@ -1,4 +1,5 @@
 export * from './delivery-stream';
+export * from './log-group-event-source';
 
 import erasure = require('@punchcard/erasure');
 

packages/@punchcard/constructs/src/log-group-event-source.ts

@@ -0,0 +1,25 @@
+import lambda = require('@aws-cdk/aws-lambda');
+import logs = require('@aws-cdk/aws-logs');
+import logsdests = require('@aws-cdk/aws-logs-destinations');
+import cdk = require('@aws-cdk/core');
+
+/**
+ * Properties forwarded to the Lambda Subscription.
+ */
+export interface LogGroupEventSourceProps {
+  filterPattern?: logs.IFilterPattern;
+}
+
+/**
+ * Use an CloudWatch Logs log group as an event source for AWS Lambda.
+ */
+export class LogGroupEventSource implements lambda.IEventSource {
+  constructor(private readonly logGroup: logs.LogGroup, private readonly props?: LogGroupEventSourceProps) {}
+
+  public bind(target: lambda.IFunction) {
+    this.logGroup.addSubscriptionFilter('Lambda', {
+      destination: new logsdests.LambdaDestination(target),
+      filterPattern: this.props?.filterPattern ?? logs.FilterPattern.allEvents()
+    });
+  }
+}

packages/punchcard/package.json

@@ -41,83 +41,85 @@
     "webpack": "^4.31.0"
   },
   "devDependencies": {
-    "aws-sdk": "^2.839.0",
-    "constructs": "^3.3.12",
     "@aws-cdk/assert": "^1.88.0",
     "@aws-cdk/aws-apigateway": "^1.88.0",
     "@aws-cdk/aws-appsync": "^1.88.0",
     "@aws-cdk/aws-cloudformation": "^1.88.0",
     "@aws-cdk/aws-cognito": "^1.88.0",
     "@aws-cdk/aws-dynamodb": "^1.88.0",
     "@aws-cdk/aws-elasticsearch": "^1.88.0",
-    "@aws-cdk/aws-events-targets": "^1.88.0",
     "@aws-cdk/aws-events": "^1.88.0",
+    "@aws-cdk/aws-events-targets": "^1.88.0",
     "@aws-cdk/aws-glue": "^1.88.0",
     "@aws-cdk/aws-iam": "^1.88.0",
     "@aws-cdk/aws-kinesis": "^1.88.0",
     "@aws-cdk/aws-kinesisfirehose": "^1.88.0",
     "@aws-cdk/aws-kms": "^1.88.0",
-    "@aws-cdk/aws-lambda-event-sources": "^1.88.0",
     "@aws-cdk/aws-lambda": "^1.88.0",
+    "@aws-cdk/aws-lambda-event-sources": "^1.88.0",
     "@aws-cdk/aws-logs": "^1.88.0",
-    "@aws-cdk/aws-s3-assets": "^1.88.0",
+    "@aws-cdk/aws-logs-destinations": "^1.88.0",
     "@aws-cdk/aws-s3": "^1.88.0",
-    "@aws-cdk/aws-sns-subscriptions": "^1.88.0",
+    "@aws-cdk/aws-s3-assets": "^1.88.0",
     "@aws-cdk/aws-sns": "^1.88.0",
+    "@aws-cdk/aws-sns-subscriptions": "^1.88.0",
     "@aws-cdk/aws-sqs": "^1.88.0",
-    "@aws-cdk/aws-stepfunctions-tasks": "^1.88.0",
     "@aws-cdk/aws-stepfunctions": "^1.88.0",
+    "@aws-cdk/aws-stepfunctions-tasks": "^1.88.0",
     "@aws-cdk/core": "^1.88.0",
     "@aws-cdk/custom-resources": "^1.88.0",
     "@punchcard/constructs": "^0.8.0",
     "@punchcard/erasure": "^0.2.0",
+    "@punchcard/shape": "^0.5.0",
     "@punchcard/shape-dynamodb": "^0.5.0",
     "@punchcard/shape-hive": "^0.5.0",
     "@punchcard/shape-json": "^0.5.0",
     "@punchcard/shape-jsonpath": "^0.5.0",
     "@punchcard/shape-jsonschema": "^0.5.0",
-    "@punchcard/shape": "^0.5.0",
     "@types/jest": "^26.0.19",
+    "aws-sdk": "^2.839.0",
+    "constructs": "^3.3.12",
     "jest": "^26.6.3",
     "ts-jest": "^26.4.4"
   },
   "peerDependencies": {
-    "aws-sdk": "^2.839.0",
-    "constructs": "^3.3.12",
     "@aws-cdk/assert": "^1.88.0",
     "@aws-cdk/aws-apigateway": "^1.88.0",
     "@aws-cdk/aws-appsync": "^1.88.0",
     "@aws-cdk/aws-cloudformation": "^1.88.0",
     "@aws-cdk/aws-cognito": "^1.88.0",
     "@aws-cdk/aws-dynamodb": "^1.88.0",
     "@aws-cdk/aws-elasticsearch": "^1.88.0",
-    "@aws-cdk/aws-events-targets": "^1.88.0",
     "@aws-cdk/aws-events": "^1.88.0",
+    "@aws-cdk/aws-events-targets": "^1.88.0",
     "@aws-cdk/aws-glue": "^1.88.0",
     "@aws-cdk/aws-iam": "^1.88.0",
     "@aws-cdk/aws-kinesis": "^1.88.0",
     "@aws-cdk/aws-kinesisfirehose": "^1.88.0",
     "@aws-cdk/aws-kms": "^1.88.0",
-    "@aws-cdk/aws-lambda-event-sources": "^1.88.0",
     "@aws-cdk/aws-lambda": "^1.88.0",
+    "@aws-cdk/aws-lambda-event-sources": "^1.88.0",
     "@aws-cdk/aws-logs": "^1.88.0",
-    "@aws-cdk/aws-s3-assets": "^1.88.0",
+    "@aws-cdk/aws-logs-destinations": "^1.88.0",
     "@aws-cdk/aws-s3": "^1.88.0",
-    "@aws-cdk/aws-sns-subscriptions": "^1.88.0",
+    "@aws-cdk/aws-s3-assets": "^1.88.0",
     "@aws-cdk/aws-sns": "^1.88.0",
+    "@aws-cdk/aws-sns-subscriptions": "^1.88.0",
     "@aws-cdk/aws-sqs": "^1.88.0",
-    "@aws-cdk/aws-stepfunctions-tasks": "^1.88.0",
     "@aws-cdk/aws-stepfunctions": "^1.88.0",
+    "@aws-cdk/aws-stepfunctions-tasks": "^1.88.0",
     "@aws-cdk/core": "^1.88.0",
     "@aws-cdk/custom-resources": "^1.88.0",
     "@punchcard/constructs": "^0.8.0",
     "@punchcard/erasure": "^0.2.0",
+    "@punchcard/shape": "^0.5.0",
     "@punchcard/shape-dynamodb": "^0.5.0",
     "@punchcard/shape-hive": "^0.5.0",
     "@punchcard/shape-json": "^0.5.0",
     "@punchcard/shape-jsonpath": "^0.5.0",
     "@punchcard/shape-jsonschema": "^0.5.0",
-    "@punchcard/shape": "^0.5.0"
+    "aws-sdk": "^2.839.0",
+    "constructs": "^3.3.12"
   },
   "jest": {
     "testPathIgnorePatterns": [

packages/punchcard/src/index.ts

@@ -16,6 +16,7 @@ export * as Firehose from './firehose';
 export * as Glue from './glue';
 export * as Kinesis from './kinesis';
 export * as Lambda from './lambda';
+export * as Logs from './logs';
 export * as S3 from './s3';
 export * as SNS from './sns';
 export * as SQS from './sqs';

packages/punchcard/src/logs/event.ts

@@ -0,0 +1,31 @@
+import { array, map, string, timestamp, Type } from '@punchcard/shape';
+
+/**
+ * @see https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchlogs.html
+ */
+export namespace Event {
+  export class LogEvent extends Type({
+    id: string,
+    timestamp: timestamp,
+    message: string,
+  }) {}
+
+  export class Data extends Type({
+    // true type 'DATA_MESSAGE' | 'CONTROL_MESSAGE', latter can be ignored
+    messageType: string,
+    // account
+    owner: string,
+    logGroup: string,
+    logStream: string,
+    subscriptionFilters: array(string),
+    logEvents: array(LogEvent),
+  }) {}
+
+  export class EncodedData extends Type({
+    data: string,
+  }) {}
+
+  export class Payload extends Type({
+    awslogs: EncodedData,
+  }) {}
+}

packages/punchcard/src/logs/events.ts

@@ -0,0 +1,36 @@
+import type { LogGroupEventSource } from '@punchcard/constructs';
+
+import { Build } from '../core/build';
+import { CDK } from '../core/cdk';
+import { Clients } from '../core/client';
+import { Stream } from '../util/stream';
+import { Event } from './event';
+import { LogGroup } from './log-group';
+
+import type * as lambda from '@aws-cdk/aws-lambda';
+
+/**
+ * A `Stream` of events from a CloudWatch Logs log group
+ */
+export class Events<T, D extends any[]> extends Stream<typeof Event.Payload, T, D, undefined>  {
+  constructor(public readonly logGroup: LogGroup<any>, previous: Events<any, any>, input: {
+    depends: D;
+    handle: (value: AsyncIterableIterator<any>, deps: Clients<D>) => AsyncIterableIterator<T>;
+  }) {
+    super(previous, input.handle, input.depends);
+  }
+
+  public eventSource(): Build<lambda.IEventSource> {
+    return this.logGroup.resource.map(logGroup => {
+      const c = (require('@punchcard/constructs') as typeof import('@punchcard/constructs'));
+      return new c.LogGroupEventSource(logGroup)
+    });
+  }
+
+  public chain<U, D2 extends any[]>(input: {
+    depends: D2;
+    handle: (value: AsyncIterableIterator<T>, deps: Clients<D2>) => AsyncIterableIterator<U>
+  }): Stream<typeof Event.Payload, U, D2, undefined> {
+    return new Events<U, D2>(this.logGroup, this, input);
+  }
+}

packages/punchcard/src/logs/index.ts

@@ -0,0 +1,3 @@
+export * from './event';
+export * from './events';
+export * from './log-group';