Set updatedAt to an updated value for all records in all tables in an AppSync environment

Re-index AWS AppSync ElasticSearch.md

What

When using AWS AppSync with ElasticSearch, sometimes, the data on elastic search gets corrupted or gets missing, what you need to do to get it fixed is to update each items on all the table, which you don't wanna do manually.

This script will update each items on all dynamodb table in your AppSync by setting the updatedAt to the current timestamp.

THIS SCRIPT WILL UPDATE ALL ITEMS OF YOUR DYNAMODB IN YOUR APPSYNC APP

Usage

1. Run yarn cross-fetch/polyfill @aws-sdk/util-dynamodb aws-sdk
2. Copy-paste the reindex.js file to your computer.
3. Add the necessary constants on the configuration section of the file.
4. Run node reindex.js and wait for it to complete.

The amount of time this script takes is highly proportional to how many tables you have and how many items you have on each of those table. Because unlike MySQL where you can run update table set field = value and it will update ALL RECORDS IN THAT TABLE, in DynamoDB you have to know specifically the unique Key of each of those items and then you have to update them one by one.

reindex.js

require('cross-fetch/polyfill');

const { differenceInHours, startOfYesterday } = require('date-fns');
const { marshall, unmarshall } = require('@aws-sdk/util-dynamodb');
const AWS = require('aws-sdk');

/**
 * Configuration
 * make sure to add the values for the following constants
 */

// aws region where the appsync lives
const region = 'ap-southeast-2';

// aws profile that you want to use, defined in ~/.aws/config
const profile = 'default';

// api id from AppSync
const apiId = '';

// if you want to only fetch data less than a date, set to null if not.
const toDate = startOfYesterday().toISOString();

//////////// END CONFIGURATION STUFF

const credentials = new AWS.SharedIniFileCredentials({ profile });
const updatedAt = new Date().toISOString();

const dynamodb = new AWS.DynamoDB({
  apiVersion: '2012-08-10',
  region,
  credentials
});

async function getAllItems(tableName, limit, nextToken) {
  return new Promise((resolve, reject) => {
    const params = {
      TableName: tableName,
      Limit: limit,
      ExclusiveStartKey: nextToken
    };

    if (toDate) {
      params.FilterExpression = 'updatedAt < :a';

      params.ExpressionAttributeValues = {
        ':a': {
          S: toDate
        }
      };
    }

    dynamodb.scan(params, (error, data) => {
      if (error) {
        reject(error);
      } else {
        resolve({
          items: data.Items.map(item => unmarshall(item)),
          nextToken: data.LastEvaluatedKey
        });
      }
    });
  });
}

async function updateItem(tableName, key, data) {
  return new Promise((resolve, reject) => {
    let updateExpression = '';
    const expressionAttributeValues = {};

    Object.keys(data).forEach(field => {
      const value = data[field];
      const expressionAttributeValue = `:${field}`;

      updateExpression += `${field} = ${expressionAttributeValue}`;
      expressionAttributeValues[expressionAttributeValue] = value;
    });

    dynamodb.updateItem(
      {
        ExpressionAttributeValues: marshall(
          expressionAttributeValues
        ),
        UpdateExpression: `SET ${updateExpression}`,
        Key: marshall(key),
        TableName: tableName,
        ReturnValues: 'ALL_NEW'
      },
      (error, data) => {
        if (error) reject(error);
        else resolve(unmarshall(data.Attributes));
      }
    );
  });
}

async function getDynamoDBTables() {
  const appsync = new AWS.AppSync({
    apiVersion: '2017-07-25',
    region,
    credentials
  });

  let nextToken = null;

  const tables = [];

  do {
    const result = await new Promise((resolve, reject) => {
      appsync.listDataSources(
        {
          apiId,
          maxResults: 25,
          nextToken
        },
        (error, data) => {
          if (error) reject(error);
          else resolve(data);
        }
      );
    });

    result.dataSources.forEach(({ dynamodbConfig }) => {
      if (!dynamodbConfig) return;
      tables.push(dynamodbConfig.tableName);
    });

    nextToken = result.nextToken;
  } while (nextToken);

  return tables;
}

async function updateTable(tableName) {
  let nextToken = null;
  let numUpdated = 0;

  do {
    const result = await getAllItems(tableName);

    await Promise.all(
      result.items.map(item =>
        updateItem(tableName, { id: item.id }, { updatedAt })
      )
    );

    numUpdated += result.items.length;

    nextToken = result.nextToken;
  } while (nextToken);

  console.log(
    `Updated ${numUpdated.toLocaleString()} item/s for ${tableName}`
  );
}

(async () => {
  const start = new Date();
  const tables = await getDynamoDBTables();
  console.log('Will update', tables.length, 'table/s');
  await Promise.all(tables.map(table => updateTable(table)));
  console.log(
    `Done in ${differenceInHours(start, new Date())} hours`
  );
})();