diff --git a/docs/superpowers/plans/2026-08-25-mail.md b/docs/superpowers/plans/2026-08-25-mail.md index e346163..78a7bba 100644 --- a/docs/superpowers/plans/2026-08-25-mail.md +++ b/docs/superpowers/plans/2026-08-25-mail.md @@ -5115,9 +5115,9 @@ describe('pull', () => { }); const provider = createWecomProvider({ fetch }); const cursor = wecomCursor(Math.floor(now.getTime() / 1000) - 1000); - const result = await provider.pull(account, cursor, { limit: 10, now, backfillFrom: null, known: async () => new Set(['m2']) }); + const result = await provider.pull(account, cursor, { limit: 10, now, backfillFrom: null, known: async (refs) => new Set(refs.filter((r) => r === 'm2')) }); const listCall = calls.find((c) => c.url.includes('get_mail_list'))?.body; - expect(listCall).toMatchObject({ begin_time: cursor.lastTime - 600, end_time: Math.floor(now.getTime() / 1000), limit: 10 }); + expect(listCall).toMatchObject({ begin_time: cursor.lastTime - 600, end_time: Math.floor(now.getTime() / 1000) }); expect(result.messages.map((m) => m.providerRef)).toEqual(['m1', 'm3']); expect(new TextDecoder().decode(result.messages[1]?.eml)).toBe(EML); expect(calls.filter((c) => c.url.includes('read_mail')).map((c) => c.body['mail_id'])).toEqual(['m1', 'm3']); @@ -5131,9 +5131,12 @@ describe('pull', () => { 'get_mail_list': () => ({ errcode: 0, next_cursor: 'c2', has_more: 1, mail_list: [{ mail_id: 'm1' }, { mail_id: 'm2' }] }), 'read_mail': () => ({ errcode: 0, mail_data: EML }), }); - const result = await createWecomProvider({ fetch }).pull(account, wecomCursor(1), { limit: 2, now, backfillFrom: null, known }); + const cursor = wecomCursor(1); + const result = await createWecomProvider({ fetch }).pull(account, cursor, { limit: 2, now, backfillFrom: null, known }); expect(result.messages).toHaveLength(2); expect(result.more).toBe(true); + // **截断时游标原样退回**:没读到的那几封不能被跳过去(重叠窗兜不住一次积压) + expect(result.cursor).toEqual(cursor); }); }); @@ -5303,19 +5306,25 @@ export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => numbe async pull(account, cursor, opts) { const { beginTime, endTime } = wecomWindow(cursor, opts.backfillFrom, opts.now); - const ids: string[] = []; + /* + * **先滤库里已有的,再数够不够** —— 顺序反了会卡死:一个窗口里若前 limit 条恰好都已入库, + * 这一轮就交出 0 封,而引擎见空批即收工、游标又没动,下一轮还是这 0 封,永远吃不到后面的新信。 + * 滤在前,每一轮都实实在在往前吃 limit 封**新**的。 + */ + const fresh: string[] = []; let pageCursor: string | undefined; let exhausted = false; - // 取满 limit 就停,余下的下一轮接着(窗口重叠 + 唯一键保证不丢不重) - while (ids.length < opts.limit) { + while (fresh.length < opts.limit) { const page = await withToken(account, 'get_mail_list', '/cgi-bin/exmail/app/get_mail_list', { begin_time: beginTime, end_time: endTime, ...(pageCursor === undefined ? {} : { cursor: pageCursor }), - limit: Math.min(LIST_PAGE, opts.limit - ids.length), + limit: LIST_PAGE, }); const list = Array.isArray(page['mail_list']) ? (page['mail_list'] as Array<{ mail_id?: unknown }>) : []; - for (const item of list) if (typeof item.mail_id === 'string') ids.push(item.mail_id); + const ids = list.flatMap((item) => (typeof item.mail_id === 'string' ? [item.mail_id] : [])); + const known = await opts.known(ids); + for (const id of ids) if (!known.has(id)) fresh.push(id); const hasMore = page['has_more'] === 1 || page['has_more'] === true; pageCursor = typeof page['next_cursor'] === 'string' && page['next_cursor'] !== '' ? page['next_cursor'] : undefined; if (!hasMore || list.length === 0 || pageCursor === undefined) { @@ -5323,16 +5332,20 @@ export function createWecomProvider(deps: { fetch?: FetchLike; now?: () => numbe break; } } - const known = await opts.known(ids); + const wanted = fresh.slice(0, opts.limit); const messages: RawMessage[] = []; - for (const mailId of ids.slice(0, opts.limit)) { - if (known.has(mailId)) continue; + for (const mailId of wanted) { const mail = await withToken(account, 'read_mail', '/cgi-bin/exmail/app/read_mail', { mail_id: mailId }); const data = mail['mail_data']; if (typeof data !== 'string' || data === '') continue; messages.push({ providerRef: mailId, eml: emlBytesOf(data), receivedAt: opts.now }); } - return { messages, cursor: wecomCursor(endTime), more: !exhausted }; + /* + * 游标**只在把这个窗口翻完时**才推到窗口上沿;被 limit 截断时原样退回旧游标 —— + * 否则没读到的那几封就被跳过去了(重叠窗只兜 10 分钟,兜不住一次积压)。 + * 退回不会卡住:已入库的下一轮被 `known` 滤掉,每轮照旧往前吃一批。 + */ + return { messages, cursor: exhausted ? wecomCursor(endTime) : cursor, more: !exhausted || fresh.length > wanted.length }; }, async send(account, message) {