游标分页与中继连接：从 offset 到 cursor 的工程实践

# offset 分页：page = 2, size = 20
query {
  users(offset: 20, limit: 20) {
    id
    name
  }
}

query {
  users(first: 20, after: "YXJyYXljb25uZWN0aW9uOjE5") {
    edges { node { id name } }
    pageInfo { hasNextPage endCursor }
  }
}

type UserConnection {
  edges: [UserEdge!]!
  pageInfo: PageInfo!
  totalCount: Int!
}

type UserEdge {
  node: User!
  cursor: String!
}

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

type Query {
  users(first: Int, after: String, last: Int, before: String): UserConnection!
}

// 方案一：Base64 编码（Relay 默认，明文可读但可逆）
export function encodeCursor(offsetOrId: string | number): string {
  return Buffer.from(`arrayconnection:${offsetOrId}`).toString('base64');
}

export function decodeCursor(cursor: string): string | number {
  const raw = Buffer.from(cursor, 'base64').toString('utf8');
  return raw.replace(/^arrayconnection:/, '');
}

// 方案二：键集编码（多列排序时编码完整排序键）
export function encodeKeysetCursor(row: { createdAt: Date; id: string }): string {
  const payload = JSON.stringify([row.createdAt.toISOString(), row.id]);
  return Buffer.from(payload).toString('base64url');
}

async function usersConnection(args, ctx) {
  const { first = 20, after } = args;

  let where = '';
  const params: unknown[] = [];

  if (after) {
    const { createdAt, id } = decodeKeysetCursor(after);
    params.push(createdAt, id);
    // keyset 条件：created_at DESC, id DESC 的"在其后"
    where = `WHERE (created_at, id) < ($1::timestamptz, $2::text)`;
  }

  // 多取一条判断 hasNextPage
  const rows = await db.query(
    `SELECT id, name, created_at
     FROM users
     ${where}
     ORDER BY created_at DESC, id DESC
     LIMIT $${params.length + 1}`,
    [...params, first + 1],
  );

  const hasNextPage = rows.length > first;
  const pageRows = rows.slice(0, first);

  return {
    edges: pageRows.map((row) => ({
      node: row,
      cursor: encodeKeysetCursor(row),
    })),
    pageInfo: {
      hasNextPage,
      startCursor: pageRows[0] ? encodeKeysetCursor(pageRows[0]) : null,
      endCursor: pageRows[pageRows.length - 1] ? encodeKeysetCursor(pageRows[pageRows.length - 1]) : null,
    },
    totalCount: await countUsers(ctx), // 见第五节
  };
}

// 缓存 totalCount 的模板（写路径失效）
async function getCachedTotalCount(key: string, freshCount: () => Promise<number>) {
  const cached = await redis.get(key);
  if (cached != null) return Number(cached);
  const total = await freshCount();
  await redis.set(key, total, 'EX', 300); // 5 分钟
  return total;
}

// 任何写操作成功后
await redis.del(`list:users:total`);

import { useCallback, useState } from 'react';
import { gql, useQuery } from '@apollo/client';

const FEED_QUERY = gql`
  query Feed($first: Int!, $after: String) {
    feed(first: $first, after: $after) {
      edges { node { id title } cursor }
      pageInfo { hasNextPage endCursor }
    }
  }
`;

export function useInfiniteFeed() {
  const [items, setItems] = useState<FeedItem[]>([]);
  const [cursor, setCursor] = useState<string | null>(null);
  const { data, fetchMore, loading } = useQuery(FEED_QUERY, {
    variables: { first: 20, after: null },
  });

  const loadMore = useCallback(async () => {
    if (loading || !data?.feed.pageInfo.hasNextPage) return;
    const { data: more } = await fetchMore({
      variables: { first: 20, after: data.feed.pageInfo.endCursor },
    });
    setCursor(more.feed.pageInfo.endCursor);
  }, [data, fetchMore, loading]);

  return { items, loadMore, hasNextPage: data?.feed.pageInfo.hasNextPage };
}

import { InMemoryCache } from '@apollo/client';

const cache = new InMemoryCache({
  typePolicies: {
    Query: {
      fields: {
        feed: {
          keyArgs: false, // 忽略分页变量，按同一列表缓存
          merge(existing: any = { edges: [], pageInfo: {} }, incoming: any) {
            return {
              edges: [...existing.edges, ...incoming.edges],
              pageInfo: incoming.pageInfo,
            };
          },
        },
      },
    },
  },
});

// IntersectionObserver 触发 loadMore
const sentinelRef = useRef<HTMLDivElement>(null);
useEffect(() => {
  const observer = new IntersectionObserver((entries) => {
    if (entries[0].isIntersecting) loadMore();
  });
  if (sentinelRef.current) observer.observe(sentinelRef.current);
  return () => observer.disconnect();
}, [loadMore]);

// 游标分页 + Redis 排序集（ZSET）缓存
const KEY = `feed:${tenantId}:${userId}`;

// 写入时维护 ZSET：score = createdAt, member = postId
await redis.zadd(KEY, post.createdAtMs, post.id);
await redis.zremrangebyrank(KEY, 0, -5000); // 只保留最近 5000 条，防膨胀

// 翻页：ZREVRANGEBYSCORE 从游标 score 之后取
const nextPosts = await redis.zrevrangebyscore(KEY, cursorScore - 1, '-inf', 'LIMIT', 0, first);

type Query {
  # 面向页码 UI
  users(page: Int = 1, pageSize: Int = 20): UsersPage!
  # 面向滚动流
  usersConnection(first: Int, after: String): UserConnection!
}

type MessageConnection {
  edges: [MessageEdge!]!
  pageInfo: PageInfo!
}

type MessageEdge {
  node: Message!
  cursor: String!
}

type Message {
  id: ID!
  content: String!
  status: MessageStatus!
  createdAt: String!
}

enum MessageStatus {
  SENT
  DELIVERED
  READ
  RECALLED
}

type Query {
  messages(conversationId: ID!, first: Int = 20, after: String): MessageConnection!
}