如何在AWS AppSync JS解析器实现DynamoDB双向分页？

背景场景 #

我正在用AWS AppSync构建GraphQL API，查询DynamoDB中的分页数据，目标是在AppSync JS解析器中实现完整的前后分页系统。

GraphQL Schema #

type ConnectionPageInfo {
    endCursor: String
    hasNextPage: Boolean!
    hasPreviousPage: Boolean!
    startCursor: String
}

type Item {
    id: ID!
    name: String!
}

type ItemEdge {
    node: Item!
}

type ItemPaginatedResponse {
    edges: [ItemEdge!]!
    pageInfo: ConnectionPageInfo!
}

type Query {
    listItems(
        first: Int,
        last: Int,
        after: String,
        before: String
    ): ItemPaginatedResponse
}

分页规则：

• first + after：向前分页（获取after之后的first条数据）
• last + before：向后分页（获取before之前的last条数据）

当前JS解析器代码 #

import { Context, util } from '@aws-appsync/utils';

export function request(ctx: Context) {
  const organisationId = 'organisation-1'; // 后续会改为参数传入
  const { first, last, after, before } = ctx.arguments;

  const limit = first ?? last ?? 10;
  const nextToken = after ?? before ?? undefined;
  const scanIndexForward = !!first;

  return {
    version: '2017-02-28',
    operation: 'Query',
    query: {
      expression: '#key = :value',
      expressionNames: { '#key': 'organisationId' },
      expressionValues: { ':value': util.dynamodb.toDynamoDB(organisationId) },
    },
    limit,
    nextToken: nextToken ? nextToken : undefined,
    scanIndexForward,
  };
}

export function response(ctx: Context) {
  if (ctx.result?.error) {
    return util.error(ctx.result.error.message, ctx.result.error.type);
  }
  console.log('obtained result:', ctx.result);

  const { items } = ctx.result;
  return {
    pageInfo: {
      endCursor: ctx.result.nextToken,
      hasNextPage: !!ctx.result.nextToken,
      hasPreviousPage: false, // 待实现
      startCursor: '', // 待生成
    },
    edges: items.map((item: { id: string; name: string }) => {
      return { node: { id: item.id, name: item.name } };
    }),
  };
}

问题解答 #

1. 如何基于DynamoDB对象生成AppSync DynamoDB分页Token？ #

AppSync的DynamoDB分页Token并非简单的主键Base64编码，而是包含DynamoDB主键结构的序列化字符串。要生成符合要求的Token，需使用AppSync提供的util.dynamodb.toCursor()工具方法，传入对应数据的主键DynamoDB格式对象。

假设你的DynamoDB表使用organisationId作为分区键、id作为排序键（复合主键），生成startCursor的逻辑如下：

// 在response函数中
const { items } = ctx.result;
let startCursor = '';
if (items.length > 0) {
  const firstItem = items[0];
  // 构造主键的DynamoDB格式对象
  const primaryKey = {
    organisationId: util.dynamodb.toDynamoDB(firstItem.organisationId),
    id: util.dynamodb.toDynamoDB(firstItem.id)
  };
  // 生成符合AppSync要求的cursor
  startCursor = util.dynamodb.toCursor(primaryKey);
}

如果是单主键（仅分区键），则只传入分区键的DynamoDB格式即可。

2. 如何获取previousToken并实现向后分页？ #

DynamoDB原生Query/Scan API不会直接返回previousToken，需通过多查一条数据+结果反转的方式实现向后分页逻辑，同时自行生成previousToken：

步骤1：修改request函数，处理last参数时多查一条

当使用last参数向后分页时，需要查询last + 1条数据，用于判断是否存在前一页：

export function request(ctx: Context) {
  const organisationId = 'organisation-1';
  const { first, last, after, before } = ctx.arguments;

  // 处理last参数时，多查1条用于判断hasPreviousPage
  let limit = first ?? last ?? 10;
  if (last !== undefined) {
    limit = last + 1;
  }

  const nextToken = after ?? before ?? undefined;
  // 向前分页时正向扫描，向后分页时反向扫描
  const scanIndexForward = !!first;

  return {
    version: '2017-02-28',
    operation: 'Query',
    query: {
      expression: '#key = :value',
      expressionNames: { '#key': 'organisationId' },
      expressionValues: { ':value': util.dynamodb.toDynamoDB(organisationId) },
    },
    limit,
    nextToken: nextToken ? nextToken : undefined,
    scanIndexForward,
  };
}

步骤2：修改response函数，处理结果反转与previousToken生成

export function response(ctx: Context) {
  if (ctx.result?.error) {
    return util.error(ctx.result.error.message, ctx.result.error.type);
  }

  let { items } = ctx.result;
  const { first, last } = ctx.arguments;
  let hasNextPage = !!ctx.result.nextToken;
  let hasPreviousPage = false;
  let startCursor = '';
  let endCursor = '';

  // 处理向后分页（last + before）的情况
  if (last !== undefined) {
    // 反向扫描的结果是倒序的，反转回正序
    items = items.reverse();
    // 如果查询结果数量超过last，说明还有前一页
    if (items.length > last) {
      // 截取需要的last条数据
      items = items.slice(0, last);
      hasPreviousPage = true;
      // 生成previousToken（即当前结果最后一条的cursor）
      const lastItemInPage = items[items.length - 1];
      endCursor = util.dynamodb.toCursor({
        organisationId: util.dynamodb.toDynamoDB(lastItemInPage.organisationId),
        id: util.dynamodb.toDynamoDB(lastItemInPage.id)
      });
    } else {
      hasPreviousPage = false;
    }
    // 生成startCursor
    if (items.length > 0) {
      const firstItem = items[0];
      startCursor = util.dynamodb.toCursor({
        organisationId: util.dynamodb.toDynamoDB(firstItem.organisationId),
        id: util.dynamodb.toDynamoDB(firstItem.id)
      });
    }
  } else {
    // 处理向前分页（first + after）的情况
    hasNextPage = !!ctx.result.nextToken;
    // 生成startCursor
    if (items.length > 0) {
      const firstItem = items[0];
      startCursor = util.dynamodb.toCursor({
        organisationId: util.dynamodb.toDynamoDB(firstItem.organisationId),
        id: util.dynamodb.toDynamoDB(firstItem.id)
      });
    }
    endCursor = ctx.result.nextToken;
  }

  return {
    pageInfo: {
      endCursor,
      hasNextPage,
      hasPreviousPage,
      startCursor,
    },
    edges: items.map((item: { id: string; name: string }) => {
      return { node: { id: item.id, name: item.name } };
    }),
  };
}

补充说明 #

• 解码AppSync返回的Token：可以用util.dynamodb.fromCursor(token)方法解析出主键结构，便于调试。
• 确保DynamoDB表的主键配置正确，复合主键需包含分区键和排序键，单主键仅需分区键。

内容的提问来源于stack exchange，提问作者Fabian Dios