add-firehose-support-2

package.json

@@ -1,6 +1,6 @@
 {
   "name": "aws-lambda-stream",
-  "version": "1.0.28",
+  "version": "1.0.29",
   "description": "Create stream processors with AWS Lambda functions.",
   "keywords": [
     "aws",

src/filters/event.js

@@ -19,7 +19,7 @@ export const filterOnEventType = (rule, uow) => {
 
 export const prefilterOnEventTypes = (rules) =>
   (uow) =>
-    rules.reduce((a, c) => a || filterOnEventType(c, uow), false);
+    rules.reduce((a, r) => a || filterOnEventType(r, uow), false);
 
 export const filterOnContent = (rule, uow) => {
   /* istanbul ignore else */
@@ -29,3 +29,7 @@ export const filterOnContent = (rule, uow) => {
     return true;
   }
 };
+
+export const prefilterOnContent = (rules) =>
+  (uow) =>
+    rules.reduce((a, r) => a || filterOnContent(r, uow), false);

src/flavors/firehose.js

@@ -0,0 +1,60 @@
+import {
+  printStartPipeline, printEndPipeline,
+  faulty,
+} from '../utils';
+import {
+  filterOnEventType, filterOnContent,
+  prefilterOnEventTypes, prefilterOnContent,
+} from '../filters';
+
+export const firehoseTransform = (rule) => (s) => s
+  .filter(onEventType(rule))
+  .tap(printStartPipeline)
+
+  .filter(onContent(rule))
+
+  .map(transform(rule))
+
+  .tap(printEndPipeline);
+
+export const firehoseDrop = (rules) => (opt) => (s) => s
+  .filter((uow) => !(prefilterOnEventTypes(rules)(uow) && prefilterOnContent(rules)(uow)))
+  .tap(printEndPipeline);
+
+const onEventType = (rule) => faulty((uow) => filterOnEventType(rule, uow));
+const onContent = (rule) => faulty((uow) => filterOnContent(rule, uow));
+
+export const spreadDateTime = (dt) => {
+  const date = new Date(dt);
+
+  return {
+    year: `${date.getUTCFullYear()}`,
+    month: `${date.getUTCMonth() + 1}`.padStart(2, '0'), // JavaScript months are 0-indexed
+    day: `${date.getUTCDate()}`.padStart(2, '0'),
+    hour: `${date.getUTCHours()}`.padStart(2, '0'),
+    minute: `${date.getUTCMinutes()}`.padStart(2, '0'),
+  };
+};
+
+const metadata = (rule, uow) => (rule.metadata
+  ? /* istanbul ignore next */ rule.metadata(uow, rule)
+  : {
+    partitionKeys: {
+      table: rule.tableName,
+      ...spreadDateTime(uow.event.timestamp),
+    },
+  });
+
+const transform = (rule) => faulty((uow) => {
+  const transformed = rule.transform
+    ? rule.transform(uow, rule)
+    : /* istanbul ignore next */ {};
+
+  return {
+    ...uow,
+    transformed,
+    result: 'Ok',
+    data: Buffer.from(JSON.stringify(transformed), 'utf-8').toString('base64'),
+    metadata: metadata(rule, uow),
+  };
+});

src/flavors/index.js

@@ -3,6 +3,7 @@ export * from './correlate';
 export * from './cdc';
 export * from './evaluate';
 export * from './expired';
+export * from './firehose';
 export * from './job';
 export * from './materialize';
 export * from './materializeS3';

src/from/firehose.js

@@ -0,0 +1,42 @@
+import _ from 'highland';
+
+import {
+  faulty, decompress, compress,
+} from '../utils';
+import { outSkip } from '../filters';
+import { redeemClaimcheck } from '../queries';
+
+export const fromFirehose = (event) =>
+
+  _(event.records)
+    .map((record) =>
+      // create a unit-of-work for each event
+      // so we can correlate related work for error handling
+      ({
+        record,
+        event: Buffer.from(record.data, 'base64').toString('utf8'),
+        recordId: record.recordId,
+        result: 'Dropped', // by default, set to 'Ok' in transform, 'ProcessingFailed' in fault processing ???
+      }))
+
+    .map(faulty((uow) => ({
+      ...uow,
+      event: JSON.parse(uow.event, decompress),
+    })))
+    .filter(outSkip)
+    .through(redeemClaimcheck());
+
+// test helper
+export const toFirehoseRecords = (events, approximateArrivalTimestamp) => ({
+  invocationId: 'invocationIdExample',
+  deliveryStreamArn: 'arn:aws:kinesis:TEST',
+  region: process.env.AWS_REGION || /* istanbul ignore next */ 'us-west-2',
+  records: events.map((e, i) =>
+    ({
+      recordId: `${i}`, // "49546986683135544286507457936321625675700192471156785154",
+      data: Buffer.from(JSON.stringify(e, compress())).toString('base64'),
+      approximateArrivalTimestamp, // format: 1495072949453
+    })),
+});
+
+export const UNKNOWN_FIREHOSE_EVENT_TYPE = toFirehoseRecords([{ type: 'unknown-type' }]);

src/utils/handler.js

@@ -39,3 +39,21 @@ export const toPromise = (s) => {
     });
   }
 };
+
+export const toFirehose = (s) => new Promise((resolve, reject) => {
+  const records = [];
+  s.consume((err, x, push, next) => {
+    /* istanbul ignore if */
+    if (err) {
+      reject(err);
+    } else if (x === _.nil) {
+      resolve({
+        records,
+      });
+    } else {
+      records.push(x);
+      next();
+    }
+  })
+    .resume();
+});

test/unit/flavors/firehose.test.js

@@ -0,0 +1,106 @@
+import 'mocha';
+import { expect } from 'chai';
+import sinon from 'sinon';
+
+import {
+  initialize, initializeFrom,
+} from '../../../src';
+
+import { toFirehoseRecords, fromFirehose } from '../../../src/from/firehose';
+import { firehoseTransform, firehoseDrop } from '../../../src/flavors/firehose';
+
+describe('flavors/firehose.js', () => {
+  beforeEach(() => {
+  });
+
+  afterEach(sinon.restore);
+
+  it('should execute', (done) => {
+    const events = toFirehoseRecords([
+      {
+        type: 't1',
+        timestamp: 1734754684001,
+        thing: {
+          id: '1',
+          name: 'Thing One',
+          description: 'This is thing one',
+        },
+      },
+      {
+        type: 't2',
+        timestamp: 1734754684001,
+        thing: {
+          id: '2',
+          name: 'Thing Two',
+          description: 'This is thing two',
+        },
+      },
+      {
+        type: 't3', // should be dropped
+        timestamp: 1734754684001,
+      },
+    ]);
+
+    initialize({
+      ...initializeFrom(rules),
+      drop: firehoseDrop(rules),
+    })
+      .assemble(fromFirehose(events), false)
+      .collect()
+      // .tap((records) => console.log(JSON.stringify(records, null, 2)))
+      .tap((records) => {
+        expect(records.length).to.equal(3);
+        expect(records[0].pipeline).to.equal('ft1');
+        expect(records[0].event.type).to.equal('t1');
+        expect(records[0].transformed).to.deep.equal({
+          ID: '1',
+          NM: 'Thing One',
+          DESC: 'This is thing one',
+        });
+        expect(records[0].metadata).to.deep.equal({
+          partitionKeys: {
+            table: 'T1',
+            year: '2024',
+            month: '12',
+            day: '21',
+            hour: '04',
+            minute: '18',
+          },
+        });
+        expect(records[0].data).to.equal('eyJJRCI6IjEiLCJOTSI6IlRoaW5nIE9uZSIsIkRFU0MiOiJUaGlzIGlzIHRoaW5nIG9uZSJ9');
+        expect(records[0].result).to.equal('Ok');
+        expect(records[1].result).to.equal('Ok');
+        expect(records[2].result).to.equal('Dropped');
+      })
+      .done(done);
+  });
+});
+
+const transform = (uow) => ({
+  ID: uow.event.thing.id,
+  NM: uow.event.thing.name,
+  DESC: uow.event.thing.description,
+});
+
+const rules = [
+  {
+    id: 'ft1',
+    flavor: firehoseTransform,
+    eventType: 't1',
+    tableName: 'T1',
+    transform,
+  },
+  {
+    id: 'ft2',
+    flavor: firehoseTransform,
+    eventType: 't2',
+    filters: [() => true],
+    tableName: 'T2',
+    transform,
+  },
+  {
+    id: 'other1',
+    eventType: 'x9',
+    flavor: firehoseTransform,
+  },
+];

test/unit/from/firehose.test.js

@@ -0,0 +1,48 @@
+import 'mocha';
+import { expect } from 'chai';
+
+import { fromFirehose, toFirehoseRecords } from '../../../src/from/firehose';
+
+describe('from/firehose.js', () => {
+  it('should parse records', (done) => {
+    const event = toFirehoseRecords([
+      {
+        id: '1',
+        type: 't1',
+        partitionKey: '1',
+      },
+      {
+        id: 'x',
+        type: 't1',
+        partitionKey: '1',
+        tags: {
+          skip: true,
+        },
+      },
+    ]);
+
+    // console.log(JSON.stringify({ event }, null, 2));
+
+    fromFirehose(event)
+      .collect()
+      // .tap((collected) => console.log(JSON.stringify(collected, null, 2)))
+      .tap((collected) => {
+        expect(collected.length).to.equal(1);
+        expect(collected[0]).to.deep.equal({
+          record: {
+            recordId: '0',
+            approximateArrivalTimestamp: undefined,
+            data: 'eyJpZCI6IjEiLCJ0eXBlIjoidDEiLCJwYXJ0aXRpb25LZXkiOiIxIn0=',
+          },
+          event: {
+            id: '1',
+            type: 't1',
+            partitionKey: '1',
+          },
+          recordId: '0',
+          result: 'Dropped',
+        });
+      })
+      .done(done);
+  });
+});

test/unit/utils/handler.test.js

@@ -5,7 +5,7 @@ import sinon from 'sinon';
 import _ from 'highland';
 import Promise from 'bluebird';
 
-import { mw, toPromise } from '../../../src/utils/handler';
+import { mw, toPromise, toFirehose } from '../../../src/utils/handler';
 
 describe('utils/handler.js', () => {
   afterEach(sinon.restore);
@@ -78,4 +78,18 @@ describe('utils/handler.js', () => {
       .then(() => Promise.reject(new Error('failed')))
       .catch((err) => ('Caught'));
   });
+
+  it('should return firehose payload', async () => {
+    const handlerWithPromise = async (event, context) =>
+      _(event.records)
+        .through(toFirehose);
+
+    const result = await handlerWithPromise({
+      records: [{ data: 'r11' }, { data: 'r12' }],
+    }, {});
+
+    expect(result).to.deep.equal({
+      records: [{ data: 'r11' }, { data: 'r12' }],
+    });
+  });
 });