alice.pds.demo.boltless.dev
This repository has no description
1import type { XRPCQueries } from "@atcute/lexicons/ambient";
2import type {
3 InferInput,
4 InferOutput,
5 ObjectSchema,
6 XRPCLexBodyParam
7} from "@atcute/lexicons/validations";
8import { ok, type BobbinContext, type XrpcRequestInit } from "./client";
9
10export type QueryOutput<TName extends keyof XRPCQueries> =
11 XRPCQueries[TName]["output"] extends XRPCLexBodyParam
12 ? InferOutput<XRPCQueries[TName]["output"]["schema"]>
13 : never;
14
15export type QueryParams<TName extends keyof XRPCQueries> =
16 XRPCQueries[TName]["params"] extends ObjectSchema
17 ? InferInput<XRPCQueries[TName]["params"]>
18 : Record<string, never>;
19
20type CursorPage = { items: readonly unknown[]; cursor?: string };
21
22// registered queries whose output is an { items, cursor? } page.
23export type PaginatedQuery = {
24 [K in keyof XRPCQueries]: QueryOutput<K> extends CursorPage ? K : never;
25}[keyof XRPCQueries];
26
27export type PageItem<TName extends PaginatedQuery> =
28 QueryOutput<TName> extends { items: readonly (infer I)[] } ? I : never;
29
30export type PageParams<TName extends PaginatedQuery> = Omit<QueryParams<TName>, "cursor">;
31
32export interface PaginateOptions extends XrpcRequestInit {
33 /** stop after this many network round-trips. */
34 maxPages?: number;
35}
36
37const requestPage = <TName extends PaginatedQuery>(
38 ctx: BobbinContext,
39 name: TName,
40 params: PageParams<TName> & { cursor?: string },
41 init?: XrpcRequestInit
42): Promise<QueryOutput<TName>> =>
43 ok(
44 ctx.xrpc.get(name, {
45 params,
46 signal: init?.signal,
47 headers: init?.headers
48 } as never)
49 ) as Promise<QueryOutput<TName>>;
50export const fetchPage = <TName extends PaginatedQuery>(
51 ctx: BobbinContext,
52 name: TName,
53 params: PageParams<TName> & { cursor?: string },
54 init?: XrpcRequestInit
55): Promise<QueryOutput<TName>> => requestPage(ctx, name, params, init);
56
57export async function* pages<TName extends PaginatedQuery>(
58 ctx: BobbinContext,
59 name: TName,
60 params: PageParams<TName>,
61 options: PaginateOptions = {}
62): AsyncGenerator<QueryOutput<TName>, void, unknown> {
63 let cursor: string | undefined;
64 let seen = 0;
65 do {
66 const data = await requestPage(
67 ctx,
68 name,
69 { ...params, cursor } as PageParams<TName> & { cursor?: string },
70 options
71 );
72 yield data;
73 cursor = (data as CursorPage).cursor;
74 if (options.maxPages && ++seen >= options.maxPages) break;
75 } while (cursor);
76}
77
78export async function* items<TName extends PaginatedQuery>(
79 ctx: BobbinContext,
80 name: TName,
81 params: PageParams<TName>,
82 options: PaginateOptions = {}
83): AsyncGenerator<PageItem<TName>, void, unknown> {
84 for await (const data of pages(ctx, name, params, options)) {
85 for (const item of (data as CursorPage).items) yield item as PageItem<TName>;
86 }
87}
88
89export interface CollectOptions extends PaginateOptions {
90 /** stop once this many items are gathered. */
91 max?: number;
92}
93
94export const collect = async <TName extends PaginatedQuery>(
95 ctx: BobbinContext,
96 name: TName,
97 params: PageParams<TName>,
98 options: CollectOptions = {}
99): Promise<PageItem<TName>[]> => {
100 const out: PageItem<TName>[] = [];
101 for await (const item of items(ctx, name, params, options)) {
102 out.push(item);
103 if (options.max && out.length >= options.max) break;
104 }
105 return out;
106};
107
108// cursor walker for schema-less list endpoints.
109export async function* paginateBy<T>(
110 load: (cursor: string | undefined) => Promise<{ items: readonly T[]; cursor?: string | null }>,
111 options: { maxPages?: number } = {}
112): AsyncGenerator<T, void, unknown> {
113 let cursor: string | undefined;
114 let seen = 0;
115 do {
116 const page = await load(cursor);
117 for (const item of page.items) yield item;
118 cursor = page.cursor ?? undefined;
119 if (options.maxPages && ++seen >= options.maxPages) break;
120 } while (cursor);
121}