feat(miners): add cursor pagination to pulls and issues endpoints

packages/das/src/api/miners/miners.controller.ts

@@ -15,6 +15,7 @@ import {
   ApiTags,
 } from "@nestjs/swagger";
 import { MinersService } from "./miners.service";
+import { parsePaginationQuery } from "./pagination";
 
 // GitHub owner/repo pattern: alphanum + `.`, `_`, `-`, reasonable length.
 const REPO_FULL_NAME_PATTERN = /^[\w.-]{1,100}\/[\w.-]{1,100}$/;
@@ -122,13 +123,29 @@ export class MinersController {
     description:
       "ISO timestamp. Defaults to 35 days ago (midnight UTC) if omitted.",
   })
+  @ApiQuery({
+    name: "cursor",
+    required: false,
+    description:
+      "Opaque pagination cursor from a previous response's next_cursor field.",
+  })
+  @ApiQuery({
+    name: "limit",
+    required: false,
+    description: "Page size (default 50, max 200).",
+  })
   async getPullRequests(
     @Param("githubId") githubId: string,
     @Query("since") since?: string,
+    @Query("cursor") cursor?: string,
+    @Query("limit") limit?: string,
   ): Promise<unknown> {
+    const pagination = parsePaginationQuery(limit, cursor);
     return this.miners.getPullRequests(
       githubId,
       MinersService.resolveSince(since),
+      pagination.limit,
+      pagination.cursor,
     );
   }
 
@@ -169,11 +186,30 @@ export class MinersController {
       "ISO timestamp. When omitted, the response contains all currently-" +
       "OPEN issues with no time bound and no CLOSED history.",
   })
+  @ApiQuery({
+    name: "cursor",
+    required: false,
+    description:
+      "Opaque pagination cursor from a previous response's next_cursor field.",
+  })
+  @ApiQuery({
+    name: "limit",
+    required: false,
+    description: "Page size (default 50, max 200).",
+  })
   async getIssues(
     @Param("githubId") githubId: string,
     @Query("since") since?: string,
+    @Query("cursor") cursor?: string,
+    @Query("limit") limit?: string,
   ): Promise<unknown> {
-    return this.miners.getIssues(githubId, since ?? null);
+    const pagination = parsePaginationQuery(limit, cursor);
+    return this.miners.getIssues(
+      githubId,
+      since ?? null,
+      pagination.limit,
+      pagination.cursor,
+    );
   }
 
   @Post(":githubId/issues")

packages/das/src/api/miners/miners.service.ts

@@ -1,6 +1,12 @@
 /* eslint-disable @typescript-eslint/no-unsafe-return, @typescript-eslint/no-unsafe-assignment */
 import { Injectable } from "@nestjs/common";
 import { DataSource } from "typeorm";
+import {
+  DecodedCursor,
+  buildPaginatedResponse,
+  keysetParams,
+  keysetSql,
+} from "./pagination";
 
 const DEFAULT_SINCE_DAYS = 35;
 
@@ -167,12 +173,25 @@ export class MinersService {
   async getPullRequests(
     githubId: string,
     since: string,
+    limit: number,
+    cursor: DecodedCursor | null,
   ): Promise<{
     github_id: string;
     since: string;
     generated_at: string;
     pull_requests: unknown[];
+    next_cursor: string | null;
   }> {
+    const params: unknown[] = [githubId, since];
+    let keysetClause = "";
+    if (cursor) {
+      const startIdx = params.length + 1;
+      keysetClause = `AND ${keysetSql("p", "pr_number", startIdx)}`;
+      params.push(...keysetParams(cursor));
+    }
+    const limitIdx = params.length + 1;
+    params.push(limit + 1);
+
     const rows = await this.dataSource.query(
       `
       SELECT${PR_SELECT_COLUMNS}
@@ -188,16 +207,29 @@ export class MinersService {
           OR (p.state = 'MERGED' AND p.merged_at >= $2)
           OR (p.state = 'CLOSED' AND p.created_at >= $2)
         )
-      ORDER BY p.created_at DESC
+        ${keysetClause}
+      ORDER BY p.created_at DESC, LOWER(p.repo_full_name) DESC, p.pr_number DESC
+      LIMIT $${limitIdx}
       `,
-      [githubId, since],
+      params,
+    );
+
+    const page = buildPaginatedResponse(
+      rows as Record<string, unknown>[],
+      limit,
+      (row) => ({
+        created_at: row.created_at as string,
+        repo_full_name: row.repo_full_name as string,
+        pr_number: row.pr_number as number,
+      }),
     );
 
     return {
       github_id: githubId,
       since,
       generated_at: new Date().toISOString(),
-      pull_requests: rows,
+      pull_requests: page.items,
+      next_cursor: page.nextCursor,
     };
   }
 
@@ -253,12 +285,25 @@ export class MinersService {
   async getIssues(
     githubId: string,
     since: string | null,
+    limit: number,
+    cursor: DecodedCursor | null,
   ): Promise<{
     github_id: string;
     since: string | null;
     generated_at: string;
     issues: unknown[];
+    next_cursor: string | null;
   }> {
+    const params: unknown[] = [githubId, since];
+    let keysetClause = "";
+    if (cursor) {
+      const startIdx = params.length + 1;
+      keysetClause = `AND ${keysetSql("i", "issue_number", startIdx)}`;
+      params.push(...keysetParams(cursor));
+    }
+    const limitIdx = params.length + 1;
+    params.push(limit + 1);
+
     const rows = await this.dataSource.query(
       `
       SELECT${ISSUE_SELECT_COLUMNS}
@@ -268,16 +313,29 @@ export class MinersService {
           (i.state = 'OPEN' AND ($2::timestamptz IS NULL OR i.created_at >= $2))
           OR (i.state = 'CLOSED' AND i.closed_at >= $2)
         )
-      ORDER BY i.created_at DESC
+        ${keysetClause}
+      ORDER BY i.created_at DESC, LOWER(i.repo_full_name) DESC, i.issue_number DESC
+      LIMIT $${limitIdx}
       `,
-      [githubId, since],
+      params,
+    );
+
+    const page = buildPaginatedResponse(
+      rows as Record<string, unknown>[],
+      limit,
+      (row) => ({
+        created_at: row.created_at as string,
+        repo_full_name: row.repo_full_name as string,
+        issue_number: row.issue_number as number,
+      }),
     );
 
     return {
       github_id: githubId,
       since,
       generated_at: new Date().toISOString(),
-      issues: rows,
+      issues: page.items,
+      next_cursor: page.nextCursor,
     };
   }
 

packages/das/src/api/miners/pagination.ts

@@ -0,0 +1,142 @@
+import { BadRequestException } from "@nestjs/common";
+
+export const DEFAULT_PAGE_LIMIT = 50;
+export const MAX_PAGE_LIMIT = 200;
+
+export interface PaginationParams {
+  limit: number;
+  cursor: DecodedCursor | null;
+}
+
+export interface DecodedCursor {
+  createdAt: string;
+  repoFullName: string;
+  number: number;
+}
+
+export interface PaginatedResult<T> {
+  items: T[];
+  nextCursor: string | null;
+}
+
+interface CursorPayload {
+  created_at: string;
+  repo_full_name: string;
+  number: number;
+}
+
+export function parsePaginationQuery(
+  limitRaw?: string,
+  cursorRaw?: string,
+): PaginationParams {
+  let limit = DEFAULT_PAGE_LIMIT;
+  if (limitRaw !== undefined && limitRaw !== "") {
+    const parsed = Number.parseInt(limitRaw, 10);
+    if (!Number.isFinite(parsed) || parsed <= 0) {
+      throw new BadRequestException("limit must be a positive integer");
+    }
+    limit = Math.min(parsed, MAX_PAGE_LIMIT);
+  }
+
+  let cursor: DecodedCursor | null = null;
+  if (cursorRaw !== undefined && cursorRaw !== "") {
+    cursor = decodeCursor(cursorRaw);
+  }
+
+  return { limit, cursor };
+}
+
+export function decodeCursor(raw: string): DecodedCursor {
+  let payload: CursorPayload;
+  try {
+    const json = Buffer.from(raw, "base64url").toString("utf8");
+    payload = JSON.parse(json) as CursorPayload;
+  } catch {
+    throw new BadRequestException("cursor is invalid");
+  }
+
+  if (
+    typeof payload.created_at !== "string" ||
+    typeof payload.repo_full_name !== "string" ||
+    typeof payload.number !== "number" ||
+    !Number.isFinite(payload.number)
+  ) {
+    throw new BadRequestException("cursor is malformed");
+  }
+
+  return {
+    createdAt: payload.created_at,
+    repoFullName: payload.repo_full_name.toLowerCase(),
+    number: payload.number,
+  };
+}
+
+export function encodeCursor(row: {
+  created_at: string | Date;
+  repo_full_name: string;
+  pr_number?: number;
+  issue_number?: number;
+}): string {
+  const createdAt =
+    row.created_at instanceof Date
+      ? row.created_at.toISOString()
+      : String(row.created_at);
+  const number = row.pr_number ?? row.issue_number;
+  if (number === undefined) {
+    throw new Error("encodeCursor requires pr_number or issue_number");
+  }
+
+  const payload: CursorPayload = {
+    created_at: createdAt,
+    repo_full_name: String(row.repo_full_name).toLowerCase(),
+    number,
+  };
+
+  return Buffer.from(JSON.stringify(payload), "utf8").toString("base64url");
+}
+
+/**
+ * Keyset predicate for ORDER BY created_at DESC, repo_full_name DESC, number DESC.
+ * Placeholders are $startIdx, $startIdx+1, $startIdx+2 (timestamptz, text, int).
+ */
+export function keysetSql(
+  alias: string,
+  numberColumn: string,
+  startIdx: number,
+): string {
+  const created = `${alias}.created_at`;
+  const repo = `LOWER(${alias}.repo_full_name)`;
+  const num = `${alias}.${numberColumn}`;
+  const at = `$${startIdx}`;
+  const repoParam = `$${startIdx + 1}`;
+  const numParam = `$${startIdx + 2}`;
+  return `(
+    (${created} < ${at}::timestamptz)
+    OR (${created} = ${at}::timestamptz AND ${repo} < ${repoParam})
+    OR (${created} = ${at}::timestamptz AND ${repo} = ${repoParam} AND ${num} < ${numParam})
+  )`;
+}
+
+export function keysetParams(cursor: DecodedCursor): [string, string, number] {
+  return [cursor.createdAt, cursor.repoFullName, cursor.number];
+}
+
+export function buildPaginatedResponse<T extends Record<string, unknown>>(
+  rows: T[],
+  limit: number,
+  pickCursorRow: (row: T) => {
+    created_at: string | Date;
+    repo_full_name: string;
+    pr_number?: number;
+    issue_number?: number;
+  },
+): PaginatedResult<T> {
+  const hasMore = rows.length > limit;
+  const items = hasMore ? rows.slice(0, limit) : rows;
+  const nextCursor =
+    hasMore && items.length > 0
+      ? encodeCursor(pickCursorRow(items[items.length - 1]))
+      : null;
+
+  return { items, nextCursor };
+}