Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
20 kB · 543 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544import fs from "node:fs/promises";import { afterEach, describe, expect, test } from "vitest";import { FastmailJmapClient, FastmailJmapConnector, FastmailJmapError, type FastmailJmapPollResult,} from "../src/connectors/fastmail-jmap.js";import type { JazzThoughtStore } from "../src/jazz/store.js";import { rebuildRootActivity } from "../src/projections/activity.js";import { buildSourceHealth } from "../src/projections/source-health.js";import { temporaryProject, testStore } from "./helpers.js";
const stores: JazzThoughtStore[] = [];const roots: string[] = [];
afterEach(async () => { await Promise.all(stores.splice(0).map((store) => store.close())); await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true })));});
describe("live Fastmail JMAP polling", () => { test("initializes at current state without replaying historical mail", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []);
const poll = await fixture.connector.poll(fixture.store);
expect(publicPoll(poll)).toEqual({ initialized: true, resnapshot: false, inserted: 0, unchanged: 0, created: 0, updated: 0, destroyed: 0, }); expect(await fixture.store.listEvents({ types: ["stream.thought.source.email.observed"] })).toEqual([]); expect((await buildSourceHealth(fixture.store)).find((source) => source.source === "fastmail:fixture")).toMatchObject({ kind: "fastmail", enabled: true, configured: true, control: { lane: "mail-ingress", runtime: { state: "ready", revision: "fastmail-jmap-v1", current: true }, }, }); expect((await fixture.store.getSourceCursor("cursor:fastmail:fixture"))?.cursor).toEqual({ revision: "fastmail-jmap-v1", accountIdHash: expect.stringMatching(/^[a-f0-9]{64}$/), emailState: "email-1", queryState: "query-1", }); expect(fixture.jmap.requests.map((request) => request.methods)).toEqual([ [], ["Email/query", "Email/get"], ]); expect(fixture.jmap.requests[1]?.arguments).toEqual([ expect.objectContaining({ limit: 0 }), expect.objectContaining({ ids: [] }), ]); expect(fixture.jmap.requests.every((request) => request.authorization === "Bearer synthetic-token")).toBe(true); });
test("paginates dual state changes, fetches metadata only, and settles one causal batch", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); fixture.jmap.enqueueEmailChanges({ oldState: "email-1", newState: "email-1b", hasMoreChanges: true, created: ["created"], updated: [], destroyed: [], }); fixture.jmap.enqueueEmailChanges({ oldState: "email-1b", newState: "email-2", hasMoreChanges: false, created: [], updated: ["updated"], destroyed: ["destroyed"], }); fixture.jmap.enqueueQueryChanges("query-1", "query-2"); fixture.jmap.enqueueGet("email-2", [ syntheticEmail("created", { bodyValues: { forbidden: { value: "must not survive" } } }), syntheticEmail("updated"), ]);
const poll = await fixture.connector.poll(fixture.store);
expect(publicPoll(poll)).toEqual({ initialized: false, resnapshot: false, inserted: 3, unchanged: 0, created: 1, updated: 1, destroyed: 1, }); const events = await fixture.store.listEvents({ types: ["stream.thought.source.email.observed"] }); expect(events.map((event) => event.payload.operation).sort()).toEqual(["created", "destroyed", "updated"]); const created = events.find((event) => event.payload.operation === "created"); const updated = events.find((event) => event.payload.operation === "updated"); expect(created).toMatchObject({ privacy: "sensitive", source: "fastmail:fixture", sourceKind: "fastmail", payload: { emailId: "created", subject: "Synthetic created", preview: "Metadata-only preview for created.", attachments: [{ blobId: "blob-created", name: "created.txt", size: 128 }], }, }); expect(created?.payload).not.toHaveProperty("bodyValues"); expect(created?.payload).not.toHaveProperty("accountId"); expect(created?.payload.accountIdHash).toMatch(/^[a-f0-9]{64}$/); expect(created?.occurredAt).toBe("2026-08-13T00:00:00Z"); expect(updated?.occurredAt).not.toBe("2026-08-13T00:00:00Z"); expect(Date.parse(updated!.occurredAt)).toBeLessThanOrEqual(Date.parse(updated!.observedAt)); const requests = fixture.jmap.requests.slice(2); expect(requests.map((request) => request.methods)).toEqual([ ["Email/changes"], ["Email/changes"], ["Email/queryChanges"], ["Email/get"], ]); expect(requests.flatMap((request) => request.properties)).not.toContain("bodyValues"); expect((await fixture.store.getSourceCursor("cursor:fastmail:fixture"))?.cursor).toMatchObject({ emailState: "email-2", queryState: "query-2", }); const activity = await rebuildRootActivity(fixture.store); expect(activity.items.find((item) => item.id === created?.id)?.presentation).toEqual({ renderer: "email-message", title: "New email", body: "Metadata-only preview for created.", objectLabel: "Synthetic created", email: { operation: "created", senderName: "Synthetic Sender", senderEmail: "sender@example.invalid", subject: "Synthetic created", attachmentCount: 1, }, }); });
test("preserves both prior states when one JMAP state machine fails", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); const before = await fixture.store.getSourceCursor("cursor:fastmail:fixture"); fixture.jmap.enqueueEmailChanges({ oldState: "email-1", newState: "email-2", hasMoreChanges: false, created: ["created"], updated: [], destroyed: [], }); fixture.jmap.enqueueError("query-changes", "rateLimit");
await expect(fixture.connector.poll(fixture.store)).rejects.toMatchObject({ code: "method-ratelimit" });
const after = await fixture.store.getSourceCursor("cursor:fastmail:fixture"); expect(after?.cursor).toEqual(before?.cursor); expect(after?.lastFailureAt).toBeDefined(); expect(await fixture.store.listEvents({ types: ["stream.thought.source.email.observed"] })).toEqual([]); });
test("absorbs a replayed metadata observation even when the aggregate JMAP state advances", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); fixture.jmap.enqueueEmailChanges({ oldState: "email-1", newState: "email-2", hasMoreChanges: false, created: [], updated: ["same-metadata"], destroyed: [], }); fixture.jmap.enqueueQueryChanges("query-1", "query-2"); fixture.jmap.enqueueGet("email-2", [syntheticEmail("same-metadata")]); expect(publicPoll(await fixture.connector.poll(fixture.store))).toMatchObject({ inserted: 1, unchanged: 0 });
fixture.jmap.enqueueEmailChanges({ oldState: "email-2", newState: "email-3", hasMoreChanges: false, created: [], updated: ["same-metadata"], destroyed: [], }); fixture.jmap.enqueueQueryChanges("query-2", "query-3"); fixture.jmap.enqueueGet("email-3", [syntheticEmail("same-metadata")]); const replay = await fixture.connector.poll(fixture.store);
expect(publicPoll(replay)).toMatchObject({ inserted: 0, unchanged: 1 }); expect(await fixture.store.listEvents({ types: ["stream.thought.source.email.observed"] })).toHaveLength(1); expect((await fixture.store.getSourceCursor("cursor:fastmail:fixture"))?.cursor).toMatchObject({ emailState: "email-3", queryState: "query-3", }); });
test("repairs cannotCalculateChanges with one bounded metadata-only resnapshot", async () => { const fixture = await setup({ resnapshotLimit: 2 }); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); fixture.jmap.enqueueError("email-changes-0", "cannotCalculateChanges"); fixture.jmap.enqueueSnapshot("email-9", "query-9", [ syntheticEmail("recent-1"), syntheticEmail("recent-2"), ]);
const poll = await fixture.connector.poll(fixture.store);
expect(publicPoll(poll)).toMatchObject({ resnapshot: true, inserted: 2, updated: 2 }); const events = await fixture.store.listEvents({ types: ["stream.thought.source.email.observed"] }); expect(events).toHaveLength(2); expect(events.every((event) => event.payload.resnapshot === true)).toBe(true); expect((await fixture.store.getSourceCursor("cursor:fastmail:fixture"))?.cursor).toMatchObject({ emailState: "email-9", queryState: "query-9", }); });
test("repairs query tooManyChanges without retaining the partial email state", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); fixture.jmap.enqueueEmailChanges({ oldState: "email-1", newState: "email-partial", hasMoreChanges: false, created: ["discarded-partial"], updated: [], destroyed: [], }); fixture.jmap.enqueueError("query-changes", "tooManyChanges"); fixture.jmap.enqueueSnapshot("email-9", "query-9", [syntheticEmail("resnapshot")]);
const poll = await fixture.connector.poll(fixture.store);
expect(publicPoll(poll)).toMatchObject({ resnapshot: true, inserted: 1, updated: 1 }); expect((await fixture.store.getSourceCursor("cursor:fastmail:fixture"))?.cursor).toMatchObject({ emailState: "email-9", queryState: "query-9", }); expect((await fixture.store.listEvents({ types: ["stream.thought.source.email.observed"] })) .some((event) => event.externalId === "discarded-partial")).toBe(false); });
test("validates query state, chunks metadata gets to the server capability, and converts notFound to destroyed", async () => { const fixture = await setup(); fixture.jmap.enqueueSnapshot("email-1", "query-1", []); await fixture.connector.poll(fixture.store); fixture.jmap.enqueueEmailChanges({ oldState: "email-1", newState: "email-1b", hasMoreChanges: true, created: ["one", "two"], updated: [], destroyed: [], }); fixture.jmap.enqueueEmailChanges({ oldState: "email-1b", newState: "email-2", hasMoreChanges: false, created: [], updated: ["missing"], destroyed: [], }); fixture.jmap.enqueueQueryChanges("query-1", "query-2"); fixture.jmap.enqueueGet("email-2", [syntheticEmail("one"), syntheticEmail("two")]); fixture.jmap.enqueueGet("email-2", [], ["missing"]);
const poll = await fixture.connector.poll(fixture.store);
expect(publicPoll(poll)).toMatchObject({ inserted: 3, created: 2, destroyed: 1 }); expect(fixture.jmap.requests.slice(2).map((request) => request.methods)).toEqual([ ["Email/changes"], ["Email/changes"], ["Email/queryChanges"], ["Email/get"], ["Email/get"], ]); expect(fixture.jmap.requests.slice(-2).map((request) => request.arguments[0]?.ids)).toEqual([ ["missing", "one"], ["two"], ]); expect((await fixture.store.listEvents({ types: ["stream.thought.source.email.observed"] })) .find((event) => event.externalId === "missing")?.payload.operation).toBe("destroyed"); });
test("rejects cross-origin API discovery and response-byte overflow without exposing the token", async () => { const crossOrigin = new FastmailJmapClient({ token: "token-cross-origin", fetchImpl: async () => jsonResponse(sessionResponse("https://attacker.invalid/jmap/api")), }); await expect(crossOrigin.discover()).rejects.toMatchObject({ code: "cross-origin-api" });
const oversized = new FastmailJmapClient({ token: "token-overflow", maxResponseBytes: 1_024, fetchImpl: async () => jsonResponse({ ...sessionResponse("https://api.fastmail.com/jmap/api"), padding: "x".repeat(2_000) }), }); let caught: unknown; try { await oversized.discover(); } catch (error) { caught = error; } expect(caught).toBeInstanceOf(FastmailJmapError); expect(caught).toMatchObject({ code: "response-too-large" }); expect(String(caught)).not.toContain("token-overflow"); });
test("does not project runtime readiness before durable cursor compatibility passes", async () => { const project = await temporaryProject("thoughtstream-fastmail-cursor-policy-"); roots.push(project); const store = await testStore(project); stores.push(store); const jmap = new SyntheticJmap(); const client = new FastmailJmapClient({ token: "synthetic-token", fetchImpl: jmap.fetch }); await store.upsertSourceCursor({ id: "cursor:fastmail:cursor-policy", source: "fastmail:cursor-policy", cursor: { revision: "fastmail-jmap-v1", accountIdHash: "0".repeat(64), emailState: "email-prior", queryState: "query-prior", }, lastSuccessAt: "2026-08-13T00:00:00.000Z", updatedAt: "2026-08-13T00:00:00.000Z", }); const connector = new FastmailJmapConnector({ id: "fastmail:cursor-policy", client, maxChanges: 2, maxPages: 3, resnapshotLimit: 10, credentialCustody: "dedicated-mail-ingress", });
await expect(connector.poll(store)).rejects.toMatchObject({ code: "cursor-account" });
const health = (await buildSourceHealth(store)).find((source) => source.source === "fastmail:cursor-policy"); expect(health).toMatchObject({ configured: true, status: "failing", lastError: "cursor-account", control: { lane: "mail-ingress" }, }); expect(health?.control).not.toHaveProperty("runtime"); expect(await store.listEvents({ types: ["stream.thought.source.email.observed"] })).toEqual([]); });});
async function setup(options: { resnapshotLimit?: number } = {}) { const project = await temporaryProject("thoughtstream-fastmail-jmap-"); roots.push(project); const store = await testStore(project); stores.push(store); const jmap = new SyntheticJmap(); const client = new FastmailJmapClient({ token: "synthetic-token", fetchImpl: jmap.fetch, timeoutMs: 5_000, maxResponseBytes: 1_000_000, }); const connector = new FastmailJmapConnector({ id: "fastmail:fixture", client, maxChanges: 2, maxPages: 3, resnapshotLimit: options.resnapshotLimit ?? 10, credentialCustody: "dedicated-mail-ingress", }); return { project, store, jmap, connector };}
class SyntheticJmap { readonly requests: Array<{ methods: string[]; properties: string[]; arguments: Array<Record<string, unknown>>; authorization: string | null; }> = []; private readonly responses: unknown[] = [];
readonly fetch: typeof globalThis.fetch = async (_input, init) => { const body = typeof init?.body === "string" ? JSON.parse(init.body) as { methodCalls?: Array<[string, Record<string, unknown>, string]>; } : undefined; const calls = body?.methodCalls ?? []; this.requests.push({ methods: calls.map(([name]) => name), properties: calls.flatMap(([, arguments_]) => Array.isArray(arguments_.properties) ? arguments_.properties.map(String) : []), arguments: calls.map(([, arguments_]) => arguments_), authorization: new Headers(init?.headers).get("authorization"), }); const response = this.responses.shift(); if (response === undefined) throw new Error("Synthetic JMAP response queue is empty"); return jsonResponse(response); };
constructor() { this.responses.push(sessionResponse("https://phl.api.fastmail.com/jmap/api/")); }
enqueueSnapshot(emailState: string, queryState: string, emails: unknown[]) { this.responses.push(jmapResponse([ ["Email/query", { accountId: "account-fixture", queryState, canCalculateChanges: true, position: 0, ids: emails.map((email) => (email as { id: string }).id), }, "snapshot-query"], ["Email/get", { accountId: "account-fixture", state: emailState, list: emails, notFound: [], }, "snapshot-get"], ])); }
enqueueEmailChanges(change: { oldState: string; newState: string; hasMoreChanges: boolean; created: string[]; updated: string[]; destroyed: string[]; }) { this.responses.push(jmapResponse([["Email/changes", { accountId: "account-fixture", ...change, }, `email-changes-${this.emailChangesQueued()}`]])); }
enqueueQueryChanges(oldQueryState: string, newQueryState: string) { this.responses.push(jmapResponse([["Email/queryChanges", { accountId: "account-fixture", oldQueryState, newQueryState, removed: [], added: [], }, "query-changes"]])); }
enqueueGet(state: string, emails: unknown[], notFound: string[] = []) { const id = `email-get-${this.emailGetQueued()}`; this.responses.push(jmapResponse([["Email/get", { accountId: "account-fixture", state, list: emails, notFound, }, id]])); }
enqueueError(callId: string, type: string) { this.responses.push(jmapResponse([["error", { type }, callId]])); }
private emailChangesQueued(): number { return this.responses.filter((response) => JSON.stringify(response).includes("Email/changes")).length; }
private emailGetQueued(): number { return this.responses.filter((response) => JSON.stringify(response).includes("email-get-")).length; }}
function syntheticEmail(id: string, extra: Record<string, unknown> = {}): Record<string, unknown> { return { id, threadId: `thread-${id}`, mailboxIds: { inbox: true }, keywords: { "$seen": false }, from: [{ name: "Synthetic Sender", email: "sender@example.invalid" }], to: [{ name: "Synthetic Recipient", email: "recipient@example.invalid" }], cc: null, bcc: null, replyTo: null, subject: `Synthetic ${id}`, receivedAt: "2026-08-13T00:00:00Z", preview: `Metadata-only preview for ${id}.`, hasAttachment: true, attachments: [{ blobId: `blob-${id}`, name: `${id}.txt`, type: "text/plain", size: 128, disposition: "attachment", cid: null, }], ...extra, };}
function sessionResponse(apiUrl: string) { return { capabilities: { "urn:ietf:params:jmap:core": { maxObjectsInGet: 2 }, "urn:ietf:params:jmap:mail": {}, }, accounts: { "account-fixture": { accountCapabilities: { "urn:ietf:params:jmap:mail": {} }, }, }, primaryAccounts: { "urn:ietf:params:jmap:mail": "account-fixture" }, apiUrl, };}
function jmapResponse(methodResponses: unknown[]) { return { methodResponses, sessionState: "session-fixture" };}
function jsonResponse(value: unknown): Response { return new Response(JSON.stringify(value), { status: 200, headers: { "content-type": "application/json" } });}
function publicPoll(poll: FastmailJmapPollResult) { return { initialized: poll.initialized, resnapshot: poll.resnapshot, inserted: poll.inserted, unchanged: poll.unchanged, created: poll.created, updated: poll.updated, destroyed: poll.destroyed, };}