experimenting with dynamodb streams

.editorconfig

@@ -0,0 +1,12 @@
+root = true
+
+[*]
+indent_style = space
+indent_size = 2
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
+
+[*.md]
+trim_trailing_whitespace = false

.github/workflows/dynamodb-table-pull.yaml

@@ -0,0 +1,29 @@
+name: dynamodb-table-pull
+on:
+  pull_request:
+    paths:
+      - dynamodb-table/**
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          sparse-checkout: dynamodb-table
+      - name: Setup Node.js
+        uses: actions/setup-node@v3
+        with:
+          node-version: 18.x
+          registry-url: https://registry.npmjs.org
+      - name: Install winglang
+        run: npm i -g winglang
+      - name: Install dependencies
+        run: npm install --include=dev
+        working-directory: dynamodb-table
+      - name: Test
+        run: wing test
+        working-directory: dynamodb-table
+      - name: Pack
+        run: wing pack
+        working-directory: dynamodb-table

.github/workflows/dynamodb-table-release.yaml

@@ -0,0 +1,37 @@
+name: dynamodb-table-release
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - dynamodb-table/**
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          sparse-checkout: dynamodb-table
+      - name: Setup Node.js
+        uses: actions/setup-node@v3
+        with:
+          node-version: 18.x
+          registry-url: https://registry.npmjs.org
+      - name: Install winglang
+        run: npm i -g winglang
+      - name: Install dependencies
+        run: npm install --include=dev
+        working-directory: dynamodb-table
+      - name: Test
+        run: wing test
+        working-directory: dynamodb-table
+      - name: Pack
+        run: wing pack
+        working-directory: dynamodb-table
+      - name: Publish
+        run: npm publish --access=public --registry https://registry.npmjs.org --tag
+          latest *.tgz
+        working-directory: dynamodb-table
+        env:
+          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}

dynamodb-table/.gitignore

@@ -0,0 +1,2 @@
+target/
+node_modules/

dynamodb-table/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Wing
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

dynamodb-table/README.md

@@ -0,0 +1,23 @@
+# dynamodb-table
+
+## Prerequisites
+
+* [winglang](https://winglang.io).
+
+## Installation
+
+`sh
+npm i @winglibs/dynamodb-table
+`
+
+## Usage
+
+`js
+bring dynamodb-table;
+
+let adder = new dynamodb-table.Adder();
+`
+
+## License
+
+This library is licensed under the [MIT License](./LICENSE).

dynamodb-table/lib.mjs

@@ -0,0 +1,178 @@
+export { default as getPort } from "get-port";
+
+import * as child_process from "node:child_process";
+
+// export interface SpawnOptions {
+//   command: string;
+//   arguments: string[];
+//   cwd?: string;
+//   env?: Record<string, string>;
+//   onData?: (data: string) => void;
+// }
+
+export const spawn = async (options) => {
+  let child = child_process.spawn(options.command, options.arguments, {
+    cwd: options.cwd,
+    env: options.env,
+  });
+
+  child.stdout.on("data", (data) => options.onData?.(data.toString()));
+  child.stderr.on("data", (data) => options.onData?.(data.toString()));
+
+  return {
+    kill() {
+      child.kill("SIGINT");
+    },
+  };
+};
+
+import * as dynamodb from "@aws-sdk/client-dynamodb";
+import { DynamoDBDocument } from "@aws-sdk/lib-dynamodb";
+
+// export interface CreateClientOptions {
+//   endpoint: string;
+// }
+
+export const createClient = (endpoint) => {
+  return new dynamodb.DynamoDB({
+    region: "local",
+    credentials: {
+      accessKeyId: "local",
+      secretAccessKey: "local",
+    },
+    endpoint,
+  });
+};
+
+export const createDocumentClient = (endpoint) => {
+  const client = createClient(endpoint);
+  return DynamoDBDocument.from(client, {
+    marshallOptions: {
+      removeUndefinedValues: true,
+      convertEmptyValues: true,
+    },
+    unmarshallOptions: {
+      wrapNumbers: true,
+    },
+  });
+};
+
+// const x =  DynamoDBDocument.from();
+// x.transactWrite({
+//   TransactItems: [
+//     {
+//       Put: {}
+//     }
+//   ]
+// })
+
+import * as streams from "@aws-sdk/client-dynamodb-streams";
+
+// export const createStreamsClient = (endpoint) => {
+//   return new streams.DynamoDBStreams({
+//     region: "local",
+//     credentials: {
+//       accessKeyId: "local",
+//       secretAccessKey: "local",
+//     },
+//     endpoint,
+//   });
+// };
+
+/**
+ * @param {import("@aws-sdk/client-dynamodb-streams").DynamoDBStreams} client
+ * @param {string} StreamArn
+ * @param {(record: any) => void|Promise<void>} handler
+ */
+const processStreamRecords = async (client, StreamArn, handler) => {
+  // while (true) {
+  // try {
+  const { StreamDescription } = await client.describeStream({ StreamArn });
+
+  for (const { ShardId } of StreamDescription.Shards) {
+    const shardIteratorData = await client.getShardIterator({
+      StreamArn,
+      ShardId,
+      ShardIteratorType: "TRIM_HORIZON",
+    });
+
+    let shardIterator = shardIteratorData.ShardIterator;
+    while (shardIterator) {
+      const recordsData = await client.getRecords({
+        ShardIterator: shardIterator,
+      });
+
+      for (const record of recordsData.Records) {
+        try {
+          await handler(record);
+        } catch (error) {
+          console.error("Error processing stream record:", error, record);
+        }
+      }
+
+      shardIterator = recordsData.NextShardIterator;
+
+      // AWS DynamoDB fetches records 4 times a second.
+      await new Promise((resolve) => setTimeout(resolve, 250));
+    }
+  }
+  // } catch (error) {
+  //   console.error("Error processing stream records:", error);
+  //   console.error("error name", error.name);
+  //   // if (error.name === "TrimmedDataAccessException") {
+  //   //   // Ignore...
+  //   // } else {
+  //   //   throw error;
+  //   // }
+  // }
+
+  //   console.log("something went wrong");
+  //   await new Promise((resolve) => setTimeout(resolve, 250));
+  // }
+};
+
+const processRecords = async (endpoint, tableName, handler) => {
+  while (true) {
+    const client = new streams.DynamoDBStreams({
+      region: "local",
+      credentials: {
+        accessKeyId: "local",
+        secretAccessKey: "local",
+      },
+      endpoint,
+    });
+
+    try {
+      const { Streams } = await client.listStreams({ TableName: tableName });
+      await Promise.all(
+        Streams.map(({ StreamArn }) =>
+          processStreamRecords(client, StreamArn, handler)
+        )
+      );
+    } catch (error) {
+      if (error.message.includes("ECONNREFUSED")) {
+        // Stop processing records when the host is down.
+        throw error;
+      } else if (error.name === "TrimmedDataAccessException") {
+        // Ignore. This error seems to be a bug in DynamoDB Local.
+        // The desired behavior is to retry processing the records.
+      } else if (error.name === "ExpiredIteratorException") {
+        // Ignore. This error happens after the computer wakes up from sleep.
+      } else {
+        throw error;
+      }
+      await new Promise((resolve) => setTimeout(resolve, 250));
+    }
+  }
+};
+
+export const processRecordsAsync = async (endpoint, tableName, handler) => {
+  processRecords(endpoint, tableName, handler).catch((error) => {
+    if (error.message.includes("ECONNREFUSED")) {
+      // Ignore. This error happens when reloading the console.
+      // We can safely end the execution here.
+    } else {
+      console.error(error);
+    }
+  });
+};