Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 65 additions & 23 deletions src/commands/emails/receiving/listen.ts
Original file line number Diff line number Diff line change
Expand Up @@ -47,8 +47,11 @@ type PageResult = {
const extractNewEmails = (
emails: ReadonlyArray<ListReceivingEmail>,
seenIds: BoundedSet<string>,
completedHeadId: string | undefined,
): PageResult => {
const idx = emails.findIndex((e) => seenIds.has(e.id));
const idx = emails.findIndex(
(e) => e.id === completedHeadId || seenIds.has(e.id),
);
return idx === -1
? { newEmails: emails, foundSeen: false }
: { newEmails: emails.slice(0, idx), foundSeen: true };
Expand All @@ -58,6 +61,8 @@ type FetchResult = {
readonly emails: ReadonlyArray<ListReceivingEmail>;
readonly error?: string;
readonly hasMore: boolean;
readonly completed: boolean;
readonly nextCursor?: string;
};

export const listenReceivingCommand = new Command('listen')
Expand Down Expand Up @@ -107,6 +112,10 @@ Ctrl+C exits cleanly.`,
);

const seenIds = createBoundedSet<string>();
// A completed range boundary must survive eviction from the ID cache.
let completedHeadId: string | undefined;
let pendingHeadId: string | undefined;
let resumeCursor: string | undefined;
let consecutiveErrors = 0;

try {
Expand All @@ -125,6 +134,7 @@ Ctrl+C exits cleanly.`,
for (const email of data.data) {
seenIds.add(email.id);
}
completedHeadId = data.data[0]?.id;

spinner.stop('Ready');
} catch (err) {
Expand Down Expand Up @@ -170,38 +180,65 @@ Ctrl+C exits cleanly.`,
pagesLeft: number,
): Promise<FetchResult> => {
if (pagesLeft <= 0) {
return { emails: accumulated, hasMore: true };
return {
emails: accumulated,
hasMore: true,
completed: false,
nextCursor: cursor,
};
}

const params = cursor
? { limit: PAGE_SIZE, after: cursor }
: { limit: PAGE_SIZE };
const { data, error } = await withRetry(() =>
resend.emails.receiving.list(params),
);
try {
const { data, error } = await withRetry(() =>
resend.emails.receiving.list(params),
);

if (error || !data) {
if (error || !data) {
return {
emails: accumulated,
error: error?.message ?? 'Empty response',
hasMore: false,
completed: false,
nextCursor: error?.name === 'not_found' ? undefined : cursor,
};
}

const { newEmails, foundSeen } = extractNewEmails(
data.data,
seenIds,
completedHeadId,
);
const allEmails = [...accumulated, ...newEmails];
if (cursor === undefined && pendingHeadId === undefined) {
pendingHeadId = data.data[0]?.id;
}

if (foundSeen || !data.has_more) {
return { emails: allEmails, hasMore: false, completed: true };
}

const nextCursor = data.data[data.data.length - 1]?.id;
if (!nextCursor) {
return {
emails: allEmails,
hasMore: true,
completed: false,
nextCursor: cursor,
};
}
return fetchPages(nextCursor, allEmails, pagesLeft - 1);
} catch (err) {
return {
emails: accumulated,
error: error?.message ?? 'Empty response',
error: errorMessage(err, 'Unknown error'),
hasMore: false,
completed: false,
nextCursor: cursor,
};
}

const { newEmails, foundSeen } = extractNewEmails(data.data, seenIds);
const allEmails = [...accumulated, ...newEmails];

if (foundSeen || !data.has_more) {
return { emails: allEmails, hasMore: false };
}

const nextCursor = data.data[data.data.length - 1]?.id;
if (!nextCursor) {
// Empty page with has_more: true would cause the next call to omit
// `after` and re-fetch from the head. Bail instead.
return { emails: allEmails, hasMore: true };
}
return fetchPages(nextCursor, allEmails, pagesLeft - 1);
};

let timeoutHandle: ReturnType<typeof setTimeout>;
Expand All @@ -219,7 +256,12 @@ Ctrl+C exits cleanly.`,

const poll = async (): Promise<void> => {
try {
const result = await fetchPages(undefined, [], MAX_PAGES_PER_POLL);
const result = await fetchPages(resumeCursor, [], MAX_PAGES_PER_POLL);
resumeCursor = result.nextCursor;
if (result.completed) {
completedHeadId = pendingHeadId ?? completedHeadId;
pendingHeadId = undefined;
}

if (result.error) {
handlePollError(result.error);
Expand Down
Loading