diff --git a/connectors/slack/README.md b/connectors/slack/README.md new file mode 100644 index 00000000..d6e5042a --- /dev/null +++ b/connectors/slack/README.md @@ -0,0 +1,54 @@ +# Slack Connector for Plot + +Follow Slack channels and DMs, reply in threads, and start new conversations. + +## What it does + +- OAuth 2.0 authentication with Slack — a user-token connection only, no bot + user installed in the workspace +- Direct messages and group DMs, each as one ongoing Plot thread +- Channel threads that mention you (directly, via a user group, or through + `@here`/`@channel`) or that you've starred — never whole channels +- Starred (saved) Slack items sync as Plot to-dos +- Real-time sync via the Slack Events API +- Reactions round-trip in both directions +- Replying in Plot posts back to Slack, including file attachments + +## OAuth scopes + +Required: `channels:history`, `channels:read`, `groups:history`, +`groups:read`, `users:read`, `users:read.email`, `chat:write`, `files:write`, +`stars:read`, `stars:write`, `reactions:read`, `reactions:write`. + +Optional (connect-time toggles): custom emoji in reactions (`emoji:read`), +@-mentions of a Slack user group you belong to (`usergroups:read`), and +direct/group DMs (`im:history`, `im:write`, `im:read`, `mpim:history`, +`mpim:write`, `mpim:read`). + +## Read state + +Read state is kept in step with Slack in both directions, within the limits of +what Slack's API exposes. + +**Slack → Plot.** Slack tracks two separate read cursors: one for a channel's +timeline, and one for each thread. A conversation or channel message you have +already read in Slack is marked read in Plot, and a thread you have opened in +Slack is marked read once its own cursor moves. A channel message is settled by +a once-daily reconciliation pass; a thread settles as soon as it sees another +reply, because the thread's cursor arrives with the messages. + +Because the two cursors are independent, catching up on a channel does **not** +mark its threads read — that matches Slack, where a thread you never opened +stays unread in your Threads view. A thread you read in Slack and that then +goes permanently quiet keeps its Plot unread until you open it in Plot. + +Marking something unread in Slack is not propagated to Plot. + +**Plot → Slack.** Reading a direct message in Plot marks that conversation read +in Slack. Channel threads do not write back: Slack's API offers no per-thread +mark, and the only available call moves the entire channel's cursor, which +would clear unread on every other message in it. + +## License + +MIT diff --git a/connectors/slack/src/slack-api.test.ts b/connectors/slack/src/slack-api.test.ts index 02714bcb..410199be 100644 --- a/connectors/slack/src/slack-api.test.ts +++ b/connectors/slack/src/slack-api.test.ts @@ -1,6 +1,8 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { + SlackApi, transformSlackThread, + syncSlackChannel, type SlackMessage, type SlackUserInfoMap, } from "./slack-api"; @@ -67,3 +69,95 @@ describe("transformSlackThread", () => { ).toBeUndefined(); }); }); + +describe("SlackApi.getConversationInfo", () => { + it("returns the caller's last_read cursor", async () => { + const api = new SlackApi("xoxp-test"); + const call = vi + .spyOn(api, "call") + .mockResolvedValue({ channel: { id: "C1", last_read: "1700000000.000001" } }); + + await expect(api.getConversationInfo("C1")).resolves.toEqual({ + lastRead: "1700000000.000001", + }); + expect(call).toHaveBeenCalledWith("conversations.info", { channel: "C1" }); + }); + + it("returns null when Slack omits last_read rather than inventing one", async () => { + const api = new SlackApi("xoxp-test"); + vi.spyOn(api, "call").mockResolvedValue({ channel: { id: "C1" } }); + + await expect(api.getConversationInfo("C1")).resolves.toEqual({ lastRead: null }); + }); +}); + +describe("SlackApi.markConversationRead", () => { + it("marks the conversation read at the given ts", async () => { + const api = new SlackApi("xoxp-test"); + const call = vi.spyOn(api, "call").mockResolvedValue({ ok: true }); + + await api.markConversationRead("D1", "1700000000.000001"); + + expect(call).toHaveBeenCalledWith("conversations.mark", { + channel: "D1", + ts: "1700000000.000001", + }); + }); +}); + +describe("syncSlackChannel thread parent", () => { + it("keeps the conversations.replies parent so its thread cursor survives", async () => { + const historyParent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + reply_count: 1, + }; + const repliesParent = { ...historyParent, unread_count: 0, subscribed: true }; + const reply = { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "reply", + }; + + const api = { + getConversationHistory: vi + .fn() + .mockResolvedValue({ messages: [historyParent], hasMore: false }), + getThread: vi.fn().mockResolvedValue([repliesParent, reply]), + getThreadReplies: vi.fn(), + }; + + const { threads } = await syncSlackChannel(api as never, { channelId: "C1" }); + + expect(threads).toHaveLength(1); + expect(threads[0]![0]!.unread_count).toBe(0); + expect(threads[0]![1]!.ts).toBe("1700000002.000000"); + expect(api.getThreadReplies).not.toHaveBeenCalled(); + }); + + it("falls back to the history parent when conversations.replies returns nothing", async () => { + const historyParent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + reply_count: 1, + }; + const api = { + getConversationHistory: vi + .fn() + .mockResolvedValue({ messages: [historyParent], hasMore: false }), + getThread: vi.fn().mockResolvedValue([]), + }; + + const { threads } = await syncSlackChannel(api as never, { channelId: "C1" }); + + expect(threads).toEqual([[historyParent]]); + }); +}); diff --git a/connectors/slack/src/slack-api.ts b/connectors/slack/src/slack-api.ts index 9758a225..1a1085da 100644 --- a/connectors/slack/src/slack-api.ts +++ b/connectors/slack/src/slack-api.ts @@ -55,6 +55,20 @@ export type SlackMessage = { }>; reply_count?: number; reply_users_count?: number; + latest_reply?: string; + /** + * Per-THREAD read state for the calling user, present on the parent message + * of a `conversations.replies` response. Slack tracks a thread's read state + * separately from its channel's: reading the channel does not advance these, + * and opening the thread does not advance the channel's `last_read`. + * + * Absent on messages from `conversations.history`, which is why the thread + * fetch in `syncSlackChannel` keeps the replies parent rather than the + * history one. + */ + subscribed?: boolean; + last_read?: string; + unread_count?: number; }; export type SlackUser = { @@ -372,6 +386,37 @@ export class SlackApi { return messages.slice(1); } + /** + * The calling user's read cursor for one conversation. + * + * Tier 3, and NOT subject to the 1 rpm non-Marketplace limit that applies to + * `conversations.history`/`conversations.replies` — so this is safe to call + * per conversation in a sweep. Returns `null` when Slack omits `last_read` + * (the caller must then abstain rather than assume a state). + */ + public async getConversationInfo( + channelId: string + ): Promise<{ lastRead: string | null }> { + const data = await this.call("conversations.info", { channel: channelId }); + const lastRead = data.channel?.last_read; + return { lastRead: typeof lastRead === "string" ? lastRead : null }; + } + + /** + * Move the calling user's read cursor for one conversation to `ts`. + * + * CONVERSATION-scoped: there is no per-thread equivalent in the public Web + * API, so this must only ever be called for a direct conversation, where the + * Plot link IS the whole conversation. Calling it for a channel would move + * that channel's cursor for every other message in it. + */ + public async markConversationRead( + channelId: string, + ts: string + ): Promise { + await this.call("conversations.mark", { channel: channelId, ts }); + } + public async postMessage( channelId: string, text: string, @@ -858,8 +903,21 @@ export async function syncSlackChannel( // that cursor indefinitely. Degrade to the parent-only path so the // rest of the channel still advances. try { - const replies = await api.getThreadReplies(state.channelId, threadTs); - threads.push([parentMessage, ...replies]); + // `getThread` (conversations.replies), not `getThreadReplies`: the + // parent it returns carries this thread's own read cursor + // (`unread_count`/`last_read`/`latest_reply`), which the + // `conversations.history` parent does not have. Same single API call + // either way — `getThreadReplies` just threw the parent away. + // + // Spread the history parent underneath so any field only that response + // carried survives, then let the replies parent's fields win. + const full = await api.getThread(state.channelId, threadTs); + const [repliesParent, ...replies] = full; + threads.push( + repliesParent + ? [{ ...parentMessage, ...repliesParent }, ...replies] + : [parentMessage] + ); } catch (error) { console.warn( `conversations.replies failed for ${state.channelId}/${threadTs}; falling back to parent-only`, diff --git a/connectors/slack/src/slack-read-state.test.ts b/connectors/slack/src/slack-read-state.test.ts new file mode 100644 index 00000000..c04dd1c4 --- /dev/null +++ b/connectors/slack/src/slack-read-state.test.ts @@ -0,0 +1,161 @@ +import { describe, expect, it } from "vitest"; +import { + channelReadVerdict, + compareSlackTs, + deriveReadAnchor, + threadReadVerdict, +} from "./slack-read-state"; +import type { SlackMessage } from "./slack-api"; + +function msg(over: Partial & { ts: string }): SlackMessage { + return { type: "message", text: "", ...over }; +} + +describe("compareSlackTs", () => { + it("orders by seconds first", () => { + expect(compareSlackTs("1700000001.000000", "1700000000.999999")).toBe(1); + expect(compareSlackTs("1700000000.999999", "1700000001.000000")).toBe(-1); + }); + + it("orders by microseconds within the same second", () => { + expect(compareSlackTs("1700000000.000002", "1700000000.000001")).toBe(1); + expect(compareSlackTs("1700000000.000001", "1700000000.000002")).toBe(-1); + expect(compareSlackTs("1700000000.000001", "1700000000.000001")).toBe(0); + }); + + it("does not lose precision on 16-significant-digit timestamps", () => { + // parseFloat("1502126650.228446") rounds; a naive numeric compare calls + // these equal. The micro halves differ by one, so the result must be -1. + expect(compareSlackTs("1502126650.228446", "1502126650.228447")).toBe(-1); + }); + + it("treats Slack's never-read sentinel as older than everything", () => { + expect(compareSlackTs("0000000000.000000", "1700000000.000001")).toBe(-1); + }); + + it("returns NaN for a malformed timestamp rather than defaulting to a direction", () => { + // A naive `< 0 ? -1 : 1` shape resolves every NaN comparison to `1` — + // silently the UNSAFE "read" direction for a caller like + // channelReadVerdict. NaN must propagate so callers can detect it. + expect(Number.isNaN(compareSlackTs("not-a-timestamp", "1700000000.000001"))).toBe( + true + ); + expect(Number.isNaN(compareSlackTs("1700000000.000001", "not-a-timestamp"))).toBe( + true + ); + expect(Number.isNaN(compareSlackTs("1700000000.abc", "1700000000.000001"))).toBe( + true + ); + }); +}); + +describe("channelReadVerdict", () => { + it("is read when the cursor is at or past the newest message", () => { + expect(channelReadVerdict("1700000000.000001", "1700000000.000001")).toBe("read"); + expect(channelReadVerdict("1700000001.000000", "1700000000.000001")).toBe("read"); + }); + + it("is unread when the cursor is behind the newest message", () => { + expect(channelReadVerdict("1700000000.000000", "1700000000.000001")).toBe("unread"); + }); + + it("abstains when Slack did not return a cursor", () => { + expect(channelReadVerdict(null, "1700000000.000001")).toBe("unknown"); + expect(channelReadVerdict(undefined, "1700000000.000001")).toBe("unknown"); + expect(channelReadVerdict("", "1700000000.000001")).toBe("unknown"); + }); + + it("abstains rather than asserting read on a malformed cursor", () => { + // Regression: compareSlackTs used to fail toward "read" (the unsafe + // direction) on a NaN half; this must abstain instead. + expect(channelReadVerdict("not-a-timestamp", "1700000000.000001")).toBe( + "unknown" + ); + }); +}); + +describe("threadReadVerdict", () => { + it("prefers unread_count when present", () => { + expect(threadReadVerdict(msg({ ts: "1.0", unread_count: 0 }))).toBe("read"); + expect(threadReadVerdict(msg({ ts: "1.0", unread_count: 3 }))).toBe("unread"); + }); + + it("falls back to last_read vs latest_reply", () => { + expect( + threadReadVerdict( + msg({ ts: "1.0", last_read: "1700000002.000000", latest_reply: "1700000002.000000" }) + ) + ).toBe("read"); + expect( + threadReadVerdict( + msg({ ts: "1.0", last_read: "1700000001.000000", latest_reply: "1700000002.000000" }) + ) + ).toBe("unread"); + }); + + it("abstains when the parent carries no thread cursor at all", () => { + expect(threadReadVerdict(msg({ ts: "1.0" }))).toBe("unknown"); + expect(threadReadVerdict(undefined)).toBe("unknown"); + }); + + it("abstains when only one half of the fallback pair is present", () => { + expect(threadReadVerdict(msg({ ts: "1.0", last_read: "1700000002.000000" }))).toBe( + "unknown" + ); + expect(threadReadVerdict(msg({ ts: "1.0", latest_reply: "1700000002.000000" }))).toBe( + "unknown" + ); + }); + + it("abstains rather than asserting read on a malformed fallback-pair cursor", () => { + expect( + threadReadVerdict( + msg({ ts: "1.0", last_read: "not-a-timestamp", latest_reply: "1700000002.000000" }) + ) + ).toBe("unknown"); + }); +}); + +describe("deriveReadAnchor", () => { + it("anchors on the newest message ts", () => { + const anchor = deriveReadAnchor( + [msg({ ts: "1700000000.000001" }), msg({ ts: "1700000002.000000" })], + { direct: false, at: 1000 } + ); + expect(anchor).toEqual({ newest: "1700000002.000000", threaded: false, at: 1000 }); + }); + + it("marks a channel link threaded when it holds a real thread reply", () => { + const anchor = deriveReadAnchor( + [ + msg({ ts: "1700000000.000001", thread_ts: "1700000000.000001" }), + msg({ ts: "1700000002.000000", thread_ts: "1700000000.000001" }), + ], + { direct: false, at: 1000 } + ); + expect(anchor?.threaded).toBe(true); + }); + + it("does not treat a lone parent as threaded", () => { + const anchor = deriveReadAnchor( + [msg({ ts: "1700000000.000001", thread_ts: "1700000000.000001" })], + { direct: false, at: 1000 } + ); + expect(anchor?.threaded).toBe(false); + }); + + it("never marks a direct conversation threaded — a DM uses the conversation cursor", () => { + const anchor = deriveReadAnchor( + [ + msg({ ts: "1700000000.000001", thread_ts: "1700000000.000001" }), + msg({ ts: "1700000002.000000", thread_ts: "1700000000.000001" }), + ], + { direct: true, at: 1000 } + ); + expect(anchor?.threaded).toBe(false); + }); + + it("returns null for an empty message set", () => { + expect(deriveReadAnchor([], { direct: false, at: 1000 })).toBeNull(); + }); +}); diff --git a/connectors/slack/src/slack-read-state.ts b/connectors/slack/src/slack-read-state.ts new file mode 100644 index 00000000..3b1109ee --- /dev/null +++ b/connectors/slack/src/slack-read-state.ts @@ -0,0 +1,134 @@ +import type { SlackMessage } from "./slack-api"; + +/** + * Whether Slack considers a Plot link read. + * + * `"unknown"` is distinct from `"unread"` on purpose: it means Slack did not + * give us a usable cursor, so the connector must leave Plot's state alone + * rather than assert anything. Guessing in either direction is worse than + * abstaining — a wrong `"read"` hides a message the user never saw. + */ +export type SlackReadVerdict = "read" | "unread" | "unknown"; + +/** + * Compare two Slack timestamps (`"1700000000.000001"`). + * + * Done on the two halves as integers rather than via `parseFloat`: a Slack ts + * carries 16 significant digits, which is at the edge of float64 precision, + * so `parseFloat` can round two genuinely different timestamps to the same + * value. Returns <0, 0, or >0 like a comparator — or `NaN` when either + * timestamp is malformed enough that neither half parses as a number, so a + * meaningful comparison isn't possible. + * + * Callers MUST check `Number.isNaN` on the result and treat it as "unknown", + * never fall through to a directional default: every JS comparison against + * `NaN` (`< 0`, `>= 0`, `=== 0`) evaluates to `false`, so an unguarded + * `compareSlackTs(...) >= 0 ? "read" : "unread"` silently resolves to + * "unread" — and the equally unguarded `< 0 ? -1 : 1` shape used to resolve + * to `1`, i.e. "read", the UNSAFE direction for a cursor comparison. + */ +export function compareSlackTs(a: string, b: string): number { + const [aSec = "0", aMicro = "0"] = a.split("."); + const [bSec = "0", bMicro = "0"] = b.split("."); + const secDiff = Number(aSec) - Number(bSec); + if (Number.isNaN(secDiff)) return NaN; + if (secDiff !== 0) return secDiff < 0 ? -1 : 1; + const microDiff = + Number(aMicro.padEnd(6, "0")) - Number(bMicro.padEnd(6, "0")); + if (Number.isNaN(microDiff)) return NaN; + return microDiff === 0 ? 0 : microDiff < 0 ? -1 : 1; +} + +/** + * Project a CHANNEL-level cursor (`conversations.info.last_read`) onto a link. + * + * Correct only for links whose every message sits in the channel timeline — + * a direct conversation, or a channel message with no thread replies. Reading + * a channel does not advance a thread's own cursor, so a threaded link must + * use {@link threadReadVerdict} instead. + */ +export function channelReadVerdict( + lastRead: string | null | undefined, + newestTs: string +): SlackReadVerdict { + if (!lastRead) return "unknown"; + const cmp = compareSlackTs(lastRead, newestTs); + if (Number.isNaN(cmp)) return "unknown"; + return cmp >= 0 ? "read" : "unread"; +} + +/** + * Project a THREAD's own cursor onto a link, from the thread parent returned + * by `conversations.replies`. + * + * `unread_count` is the direct answer and is preferred. The + * `last_read`/`latest_reply` pair is the fallback for responses that carry + * the cursor but not the count. A parent with neither means the caller is not + * subscribed to the thread (or Slack simply omitted the state), so abstain. + */ +export function threadReadVerdict( + parent: SlackMessage | undefined +): SlackReadVerdict { + if (!parent) return "unknown"; + if (typeof parent.unread_count === "number") { + return parent.unread_count === 0 ? "read" : "unread"; + } + if (parent.last_read && parent.latest_reply) { + const cmp = compareSlackTs(parent.last_read, parent.latest_reply); + if (Number.isNaN(cmp)) return "unknown"; + return cmp >= 0 ? "read" : "unread"; + } + return "unknown"; +} + +/** + * The reconciliation state one saved link needs. + * + * `newest` is what a cursor is compared against. `threaded` picks which cursor + * governs: a channel link holding real thread replies is settled by the thread + * cursor on the live path, everything else by the channel cursor in the daily + * sweep. `at` is the write time, for the retention drop. + * + * `title` rides along so `reconcileReadState`'s upsert can re-send it — + * `upsert_thread` takes an "archived" code path whenever the user's + * `thread_priority` row points at an archived priority (or is missing), and + * on that path the platform's title default is the literal string + * "Untitled", never null, so an upsert that omitted `title` there would + * destroy the thread's real one. `undefined`/`null` when the link that wrote + * this anchor had no title to carry (see `applyReadAnchor` in slack.ts, + * which falls back to a previously-stored anchor's title rather than losing + * it in that case). + */ +export type SlackReadAnchor = { + newest: string; + threaded: boolean; + at: number; + title?: string | null; +}; + +/** + * Derive the anchor for a set of messages about to be saved as one link. + * + * A direct conversation is never `threaded`: its link flattens Slack's reply + * threads into one running conversation, so the conversation cursor is the + * only cursor that describes it. + * + * `title` is accepted rather than inferred from `messages`: only the caller + * has the link (and, for the carry-forward case, the previously-stored + * anchor) to source it from. + */ +export function deriveReadAnchor( + messages: SlackMessage[], + opts: { direct: boolean; at: number; title?: string | null } +): SlackReadAnchor | null { + let newest: string | null = null; + let threaded = false; + for (const message of messages) { + if (!newest || compareSlackTs(message.ts, newest) > 0) newest = message.ts; + if (!opts.direct && message.thread_ts && message.thread_ts !== message.ts) { + threaded = true; + } + } + if (!newest) return null; + return { newest, threaded, at: opts.at, title: opts.title }; +} diff --git a/connectors/slack/src/slack.test.ts b/connectors/slack/src/slack.test.ts index 4749c8ac..65fc59fa 100644 --- a/connectors/slack/src/slack.test.ts +++ b/connectors/slack/src/slack.test.ts @@ -30,6 +30,12 @@ function makeStore(initial: Record = {}) { list: vi.fn(async (prefix: string) => [...map.keys()].filter((k) => k.startsWith(prefix)) ), + listEntries: vi.fn(async (prefix: string) => + [...map.entries()].filter(([k]) => k.startsWith(prefix)) + ), + clearMany: vi.fn(async (keys: string[]) => { + for (const key of keys) map.delete(key); + }), }; } @@ -3711,3 +3717,846 @@ describe("onNoteCreated reply targeting", () => { expect(postMessage).toHaveBeenCalledWith("C123", "ack", "100.0"); }); }); + +describe("buildConversationLink — thread read cursor", () => { + function buildSlack() { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + return { slack, store }; + } + + const parent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + }; + const reply = { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "reply", + }; + + async function build(messages: unknown[]) { + const { slack } = buildSlack(); + return (slack as unknown as { + buildConversationLink: (o: unknown) => Promise<{ unread?: boolean } | null>; + }).buildConversationLink({ + channelId: "C1", + messages, + initialSync: false, + }); + } + + it("marks the link read when the thread's own cursor says read", async () => { + const link = await build([{ ...parent, unread_count: 0 }, reply]); + expect(link?.unread).toBe(false); + }); + + it("leaves unread alone when the thread cursor says unread", async () => { + const link = await build([{ ...parent, unread_count: 2 }, reply]); + expect(link).not.toHaveProperty("unread"); + }); + + it("abstains when the parent carries no thread cursor", async () => { + const link = await build([parent, reply]); + expect(link).not.toHaveProperty("unread"); + }); + + it("abstains on a root-only link — the channel cursor is the sweep's job", async () => { + const link = await build([{ ...parent, unread_count: 0 }]); + expect(link).not.toHaveProperty("unread"); + }); + + it("never sets unread true, even when the thread is unread in Slack", async () => { + const link = await build([{ ...parent, unread_count: 5 }, reply]); + expect(link?.unread).not.toBe(true); + }); +}); + +describe("read anchors", () => { + const parent = { + type: "message", + ts: "1700000000.000001", + thread_ts: "1700000000.000001", + user: "U1", + text: "parent", + }; + const reply = { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "reply", + }; + + async function build(messages: unknown[]) { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + const link = await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise<{ unread?: boolean } | null>; + }).buildConversationLink({ channelId: "C1", messages, initialSync: false }); + return { store, link }; + } + + const KEY = "read_anchor:C1:1700000000.000001"; + + it("writes an anchor for a link it could not mark read", async () => { + const { store } = await build([parent]); + const anchor = store.map.get(KEY) as { newest: string; threaded: boolean }; + expect(anchor.newest).toBe("1700000000.000001"); + expect(anchor.threaded).toBe(false); + }); + + it("marks a threaded link's anchor threaded so the sweep skips it", async () => { + const { store } = await build([parent, reply]); + expect((store.map.get(KEY) as { threaded: boolean }).threaded).toBe(true); + }); + + it("deletes the anchor once the link is marked read", async () => { + const { store, link } = await build([{ ...parent, unread_count: 0 }, reply]); + expect(link?.unread).toBe(false); + expect(store.map.has(KEY)).toBe(false); + }); + + it("keys the anchor on the thread root, so a new reply replaces it", async () => { + const { store } = await build([parent, reply]); + expect([...store.map.keys()]).toEqual([KEY]); + expect((store.map.get(KEY) as { newest: string }).newest).toBe( + "1700000002.000000" + ); + }); + + it("does not write an anchor when there is nothing to save", async () => { + const { store } = await build([]); + expect([...store.map.keys()]).toEqual([]); + }); + + it("anchors a direct conversation on the conversation id, never threaded", async () => { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + vi.spyOn( + slack as unknown as { + dmCounterpartyUserId: (c: string) => Promise; + }, + "dmCounterpartyUserId" + ).mockResolvedValue("U2"); + + await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise; + }).buildConversationLink({ + channelId: "D1", + messages: [ + { type: "message", ts: "1700000000.000001", user: "U2", text: "hi" }, + { + type: "message", + ts: "1700000002.000000", + thread_ts: "1700000000.000001", + user: "U2", + text: "threaded", + }, + ], + initialSync: false, + }); + + const anchor = store.map.get("read_anchor:D1:D1") as { + newest: string; + threaded: boolean; + title?: string | null; + }; + expect(anchor.newest).toBe("1700000002.000000"); + expect(anchor.threaded).toBe(false); + // No `userInfos` was passed, so assembleSlackDmLink omits `title` (see + // its own comment on why — the raw Slack user id would otherwise + // permanently rename a real person's thread). Nothing else supplied one + // here, so the anchor genuinely has none to carry. + expect(anchor.title).toBeUndefined(); + }); + + it("carries a DM anchor's previously-stored title forward when the new link has none", async () => { + // assembleSlackDmLink omits `title` whenever `users.info` was + // unavailable — a real, recurring window, not an edge case. The anchor + // must not lose a title it already had just because a later sync landed + // in that window; reconcileReadState needs it to survive an + // archived-priority upsert (see SlackReadAnchor's doc comment). + const store = makeStore({ + "read_anchor:D1:D1": { + newest: "1699999999.000000", + threaded: false, + at: 1000, + title: "Alice Example", + }, + }); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + vi.spyOn( + slack as unknown as { + dmCounterpartyUserId: (c: string) => Promise; + }, + "dmCounterpartyUserId" + ).mockResolvedValue("U2"); + + await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise; + }).buildConversationLink({ + channelId: "D1", + // No userInfos passed — assembleSlackDmLink omits `title` on this link. + messages: [ + { type: "message", ts: "1700000000.000001", user: "U2", text: "hi again" }, + ], + initialSync: false, + }); + + const anchor = store.map.get("read_anchor:D1:D1") as { + newest: string; + title?: string | null; + }; + expect(anchor.newest).toBe("1700000000.000001"); + expect(anchor.title).toBe("Alice Example"); + }); + + it("leaves no anchor for a DM's initial sync — the link is already read", async () => { + const store = makeStore(); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + vi.spyOn( + slack as unknown as { + dmCounterpartyUserId: (c: string) => Promise; + }, + "dmCounterpartyUserId" + ).mockResolvedValue("U2"); + + const link = await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise<{ unread?: boolean } | null>; + }).buildConversationLink({ + channelId: "D1", + messages: [ + { type: "message", ts: "1700000000.000001", user: "U2", text: "hi" }, + ], + initialSync: true, + }); + + expect(link?.unread).toBe(false); + expect(store.map.has("read_anchor:D1:D1")).toBe(false); + }); + + it("does not recreate a channel-thread anchor on a reaction refresh (advanceConversationHead: false)", async () => { + // Regression test: `refreshSlackThread` re-fetches with + // `advanceConversationHead: false` for a reaction refresh. A root-only + // channel message has no thread cursor of its own, so the read verdict + // is "unknown" — that must NOT fall through to rewriting the anchor, + // since no new content actually arrived. + const store = makeStore({ + "read_anchor:C1:1700000000.000001": { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }, + }); + const slack = makeSlack({ + store, + integrationsGet: vi.fn(), + createWebhook: vi.fn(), + }); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + vi.spyOn( + slack as unknown as { + customEmojiContext: (c: string) => Promise<{ teamId?: string }>; + }, + "customEmojiContext" + ).mockResolvedValue({}); + + await (slack as unknown as { + buildConversationLink: (o: unknown) => Promise; + }).buildConversationLink({ + channelId: "C1", + messages: [parent], + initialSync: false, + advanceConversationHead: false, + }); + + const anchorSetCalls = store.set.mock.calls.filter( + (call) => call[0] === "read_anchor:C1:1700000000.000001" + ); + expect(anchorSetCalls).toHaveLength(0); + expect( + (store.map.get("read_anchor:C1:1700000000.000001") as { at: number }).at + ).toBe(1000); + }); +}); + +describe("reconcileReadState", () => { + const NOW = 1_700_000_000_000; + + function setup(anchors: Record, lastRead: string | null) { + const store = makeStore(anchors); + const saveLink = vi.fn().mockResolvedValue("thread-1"); + const markNeedsReauth = vi.fn(); + const tools = { + store, + integrations: { get: vi.fn(), saveLink, markNeedsReauth }, + network: { createWebhook: vi.fn() }, + files: {}, + }; + const slack = new Slack( + "twist-instance-1" as never, + { getTools: () => tools } as never + ); + const api = { getConversationInfo: vi.fn().mockResolvedValue({ lastRead }) }; + vi.spyOn( + slack as unknown as { getApi: (c: string) => Promise }, + "getApi" + ).mockResolvedValue(api); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(false); + // The `finally` re-arms the daily chain; neither the callbacks nor the + // tasks tool is in the test tool shed, so both are stubbed. + vi.spyOn( + slack as unknown as { callback: (...a: unknown[]) => Promise }, + "callback" + ).mockResolvedValue("cb-token"); + vi.spyOn( + slack as unknown as { scheduleRecurring: (...a: unknown[]) => Promise }, + "scheduleRecurring" + ).mockResolvedValue(undefined); + vi.spyOn(Date, "now").mockReturnValue(NOW); + return { slack, store, saveLink, api, markNeedsReauth }; + } + + const anchor = ( + over: Partial<{ + newest: string; + threaded: boolean; + at: number; + title: string | null; + }> = {} + ) => ({ + newest: "1700000000.000001", + threaded: false, + at: NOW, + ...over, + }); + + it("marks a root-only link read once the channel cursor passes it", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).toHaveBeenCalledTimes(1); + const saved = saveLink.mock.calls[0][0]; + expect(saved.unread).toBe(false); + expect(saved.channelId).toBe("C1"); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(false); + }); + + it("sends the anchor's title on the reconcile upsert, so an archived-priority thread keeps its real title", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor({ title: "Real title" }) }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink.mock.calls[0][0].title).toBe("Real title"); + }); + + it("omits `title` entirely when the anchor has none — never worse than today", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink.mock.calls[0][0]).not.toHaveProperty("title"); + }); + + it("never sends `created` on the reconcile upsert", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + // saveLink drops an `unread: false` save whose date predates the plan's + // sync-history limit, so a reconcile upsert must carry no date at all. + expect(saveLink.mock.calls[0][0]).not.toHaveProperty("created"); + expect(saveLink.mock.calls[0][0]).not.toHaveProperty("schedules"); + }); + + it("does not rewrite content on the reconcile upsert", async () => { + const { slack, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + const saved = saveLink.mock.calls[0][0]; + expect(saved).not.toHaveProperty("title"); + expect(saved).not.toHaveProperty("preview"); + expect(saved).not.toHaveProperty("notes"); + }); + + it("leaves a link alone while the channel cursor is still behind it", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000005.000000": anchor({ newest: "1700000005.000000" }) }, + "1700000000.000001" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000005.000000")).toBe(true); + }); + + it("skips threaded anchors — the channel cursor does not speak for them", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor({ threaded: true }) }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + }); + + it("abstains when Slack returns no cursor", async () => { + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor() }, + null + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + }); + + it("issues one conversations.info per conversation, not per anchor", async () => { + const { slack, api } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C1:1700000000.000002": anchor({ newest: "1700000000.000002" }), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(api.getConversationInfo).toHaveBeenCalledTimes(2); + }); + + it("drops anchors past the retention window without marking them read", async () => { + const stale = NOW - 31 * 24 * 60 * 60 * 1000; + const { slack, store, saveLink } = setup( + { "read_anchor:C1:1700000000.000001": anchor({ at: stale }) }, + "1700000000.000000" + ); + + await slack.reconcileReadState("C1"); + + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(false); + }); + + it("stops the pass and keeps every remaining anchor when Slack rate limits", async () => { + const { slack, store, api } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + api.getConversationInfo.mockRejectedValue( + new SlackRateLimitedError("conversations.info", 60_000) + ); + + await slack.reconcileReadState("C1"); + + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(true); + }); + + it("skips one unreachable conversation without abandoning the rest", async () => { + const { slack, store, api, saveLink } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + api.getConversationInfo + .mockRejectedValueOnce( + new SlackPermanentError("conversations.info", "channel_not_found") + ) + .mockResolvedValue({ lastRead: "1700000005.000000" }); + + await slack.reconcileReadState("C1"); + + expect(saveLink).toHaveBeenCalledTimes(1); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(false); + }); + + it("flags reauth and stops the pass on an auth-shaped permanent error", async () => { + const { slack, store, api, saveLink, markNeedsReauth } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + // invalid_auth is in SLACK_AUTH_ERRORS: the grant itself is bad, so this + // is not something a retry (or waiting for the user) will fix. + api.getConversationInfo.mockRejectedValue( + new SlackPermanentError("conversations.info", "invalid_auth") + ); + + await slack.reconcileReadState("C1"); + + expect(markNeedsReauth).toHaveBeenCalledWith("C1"); + expect(saveLink).not.toHaveBeenCalled(); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(true); + }); + + it("skips a conversation whose optional dms scope was declined, without abandoning the rest", async () => { + const { slack, store, api, saveLink, markNeedsReauth } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + // missing_scope is a member of SLACK_AUTH_ERRORS too, but a declined + // optional scope group is a user decision, not a broken connection — + // mirrors onThreadRead's handling of the same error. + api.getConversationInfo + .mockRejectedValueOnce( + new SlackPermanentError("conversations.info", "missing_scope") + ) + .mockResolvedValue({ lastRead: "1700000005.000000" }); + + await slack.reconcileReadState("C1"); + + expect(markNeedsReauth).not.toHaveBeenCalled(); + expect(saveLink).toHaveBeenCalledTimes(1); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(false); + }); + + it("skips a conversation the token can no longer reach (no_permission), without abandoning the rest", async () => { + const { slack, store, api, saveLink, markNeedsReauth } = setup( + { + "read_anchor:C1:1700000000.000001": anchor(), + "read_anchor:C2:1700000000.000003": anchor({ newest: "1700000000.000003" }), + }, + "1700000005.000000" + ); + // no_permission on THIS conversation means the user lost access to it + // specifically — not evidence the token itself is dead. + api.getConversationInfo + .mockRejectedValueOnce( + new SlackPermanentError("conversations.info", "no_permission") + ) + .mockResolvedValue({ lastRead: "1700000005.000000" }); + + await slack.reconcileReadState("C1"); + + expect(markNeedsReauth).not.toHaveBeenCalled(); + expect(saveLink).toHaveBeenCalledTimes(1); + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(true); + expect(store.map.has("read_anchor:C2:1700000000.000003")).toBe(false); + }); + + it("resolves a DM anchor to the DM identity source, never an app_redirect URL", async () => { + // The DM anchor key (`read_anchor:D1:D1`) parses to a threadTs of "D1" — + // a non-timestamp fed straight into conversationSource(). If + // isKnownDMChannel ever returned false here, the sweep would synthesise + // a bogus app_redirect URL and create a brand-new, empty thread instead + // of updating the real DM link. + const { slack, saveLink } = setup( + { "read_anchor:D1:D1": anchor() }, + "1700000005.000000" + ); + vi.spyOn( + slack as unknown as { isKnownDMChannel: (c: string) => Promise }, + "isKnownDMChannel" + ).mockResolvedValue(true); + + await slack.reconcileReadState("C1"); + + expect(saveLink).toHaveBeenCalledTimes(1); + const saved = saveLink.mock.calls[0][0]; + expect(saved.type).toBe("dm"); + expect(saved.source).not.toMatch(/^https:\/\/slack\.com\/app_redirect/); + expect(saved.source).toBe("slack:chat:D1"); + }); + + it("does no work on a second invocation within 24h of the last sweep", async () => { + // The anchor stays behind the channel cursor (verdict "unread"), so it + // survives the first pass — proving the second call's silence comes from + // the top-of-function guard, not from there being nothing left to sweep. + const { slack, api } = setup( + { "read_anchor:C1:1700000005.000000": anchor({ newest: "1700000005.000000" }) }, + "1700000000.000001" + ); + + await slack.reconcileReadState("C1"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(1); + + await slack.reconcileReadState("C1"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(1); + }); + + it("arms a fresh 24h chain for a second channel once the guard has passed", async () => { + // Sanity check on the other side of the same guard: once 24h have + // elapsed (simulated by advancing the mocked clock), a differently-keyed + // invocation is free to sweep again. + const { slack, api } = setup( + { "read_anchor:C1:1700000005.000000": anchor({ newest: "1700000005.000000" }) }, + "1700000000.000001" + ); + + await slack.reconcileReadState("C1"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(1); + + vi.spyOn(Date, "now").mockReturnValue(NOW + 24 * 60 * 60 * 1000 + 1); + await slack.reconcileReadState("C2"); + expect(api.getConversationInfo).toHaveBeenCalledTimes(2); + }); +}); + +describe("onThreadRead", () => { + function setup() { + const store = makeStore(); + const markNeedsReauth = vi.fn(); + const tools = { + store, + integrations: { get: vi.fn(), markNeedsReauth }, + network: { createWebhook: vi.fn() }, + files: {}, + }; + const slack = new Slack( + "twist-instance-1" as never, + { getTools: () => tools } as never + ); + const api = { markConversationRead: vi.fn().mockResolvedValue(undefined) }; + vi.spyOn( + slack as unknown as { getApi: (c: string) => Promise }, + "getApi" + ).mockResolvedValue(api); + return { slack, store, api, markNeedsReauth }; + } + + it("marks a direct conversation read in Slack", async () => { + const { slack, api } = setup(); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(api.markConversationRead).toHaveBeenCalledWith( + "D1", + "1700000000.000001" + ); + }); + + it("does nothing for a channel thread — conversations.mark is channel-wide", async () => { + const { slack, api } = setup(); + const thread = { meta: { channelId: "C1", threadTs: "1700000000.000001" } }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); + + it("does nothing when the thread is marked UNREAD — Slack has no un-mark", async () => { + const { slack, api } = setup(); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, true); + + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); + + it("no-ops when the meta carries no anchor message", async () => { + const { slack, api } = setup(); + const thread = { meta: { channelId: "D1", direct: true } }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); + + it("no-ops when the dms scope group was declined", async () => { + const { slack, api, markNeedsReauth } = setup(); + api.markConversationRead.mockRejectedValue( + new SlackPermanentError("conversations.mark", "missing_scope") + ); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await expect( + slack.onThreadRead(thread as never, {} as never, false) + ).resolves.toBeUndefined(); + // missing_scope is a member of SLACK_AUTH_ERRORS too, but a declined + // optional group is a user decision, not a broken connection — it must + // NOT be indistinguishable from a genuinely dead token below. + expect(markNeedsReauth).not.toHaveBeenCalled(); + }); + + it("flags reauth on a genuinely dead token — auth-shaped, not missing_scope", async () => { + const { slack, api, markNeedsReauth } = setup(); + api.markConversationRead.mockRejectedValue( + new SlackPermanentError("conversations.mark", "invalid_auth") + ); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await expect( + slack.onThreadRead(thread as never, {} as never, false) + ).resolves.toBeUndefined(); + expect(markNeedsReauth).toHaveBeenCalledWith("D1"); + }); + + it("clears a channel thread's read anchor when Plot marks it read", async () => { + const { slack, store } = setup(); + store.map.set("read_anchor:C1:1700000000.000001", { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }); + const thread = { meta: { channelId: "C1", threadTs: "1700000000.000001" } }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(store.map.has("read_anchor:C1:1700000000.000001")).toBe(false); + }); + + it("clears a DM's read anchor keyed on the conversation id, not meta.threadTs", async () => { + const { slack, store } = setup(); + // The DM anchor lives at read_anchor:D1:D1 — keyed on the conversation id + // in both positions, never on meta.threadTs (which names the + // conversation's latest message, not its anchor). + store.map.set("read_anchor:D1:D1", { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, false); + + expect(store.map.has("read_anchor:D1:D1")).toBe(false); + }); + + it("clears the anchor even when marking UNREAD, so a manual override can't be reverted by the next sweep", async () => { + const { slack, store, api } = setup(); + store.map.set("read_anchor:D1:D1", { + newest: "1700000000.000001", + threaded: false, + at: 1000, + }); + const thread = { + meta: { channelId: "D1", direct: true, threadTs: "1700000000.000001" }, + }; + + await slack.onThreadRead(thread as never, {} as never, true); + + expect(store.map.has("read_anchor:D1:D1")).toBe(false); + // Marking unread still writes back nothing to Slack — only the anchor + // bookkeeping changed. + expect(api.markConversationRead).not.toHaveBeenCalled(); + }); +}); diff --git a/connectors/slack/src/slack.ts b/connectors/slack/src/slack.ts index c721b529..3a7593fa 100644 --- a/connectors/slack/src/slack.ts +++ b/connectors/slack/src/slack.ts @@ -46,6 +46,12 @@ import { assembleSlackDmLink, slackConversationIdentity } from "./slack-dm"; import { unicodeToSlackName } from "./slack-emoji"; import { slackFacets } from "./slack-facets"; import { mentionsUser, type MentionContext } from "./slack-mentions"; +import { + channelReadVerdict, + deriveReadAnchor, + type SlackReadAnchor, + threadReadVerdict, +} from "./slack-read-state"; /** * Slack integration source. @@ -65,6 +71,9 @@ import { mentionsUser, type MentionContext } from "./slack-mentions"; * through `@here`/`@channel`) or that the user saved — never whole channels * - Star-based to-do sync against the user's saved items * - Reactions round-trip on everything that is synced + * - Read state synced from Slack: a conversation you have read in Slack is + * read in Plot, and reading a direct conversation in Plot marks it read in + * Slack * * **Required OAuth User Scopes** (each backs a shipped, user-visible feature — * kept in sync with the authoritative {@link Slack.SCOPES} array below): @@ -94,6 +103,13 @@ import { mentionsUser, type MentionContext } from "./slack-mentions"; * `im:history`/`mpim:history`, which only grant reading message content * within a conversation whose id is already known — they do not grant * enumeration, so `listDMChannels` needs both. + * + * Read state uses only scopes the connector already holds: `conversations.info` + * is covered by `channels:read`/`groups:read` (required) or `im:read`/`mpim:read` + * (the optional `dms` group), and `conversations.mark` for direct conversations + * by `im:write`/`mpim:write` (also `dms`). Channel threads are read-in only — + * Slack's public API exposes no per-thread mark, and `conversations.mark` would + * move the whole channel's cursor. */ /** @@ -103,6 +119,15 @@ import { mentionsUser, type MentionContext } from "./slack-mentions"; */ const INCREMENTAL_SYNC_COALESCE_MS = 10_000; +/** + * How long an unresolved read anchor is swept before it is dropped. + * + * Past this point Plot's own read state is the one that matters, and a link + * the user has left unread for a month is not going to be settled by Slack's + * cursor. Dropping it keeps the sweep bounded by live work. + */ +const READ_ANCHOR_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; + /** * How far back an event-triggered incremental sync reads. Wide enough to * absorb delayed webhook delivery and the coalescing delay above, narrow @@ -436,6 +461,7 @@ export class Slack extends Connector { await this.cancelScheduledTask(`custom-emoji-sync:${channel.id}`); await this.cancelScheduledTask(`user-groups-sync:${channel.id}`); await this.cancelScheduledTask(`dm-channels-sync:${channel.id}`); + await this.cancelScheduledTask(`read-state-sync:${channel.id}`); await this.cancelDrain(`incremental-sync:${channel.id}`); await this.cancelDrain(`reaction-refresh:${channel.id}`); await this.cancelDrain(`subscribed-thread:${channel.id}`); @@ -457,6 +483,18 @@ export class Slack extends Connector { // thread that already earned its place in Plot keeps updating. const starredKeys = await this.tools.store.list(`starred:${channel.id}:`); for (const key of starredKeys) await this.clear(key); + + // Read anchors are per-conversation reconciliation state, not a record of + // what earned a place in Plot, so unlike `sync_thread:` they are swept — + // but only a CHANNEL THREAD's anchor is keyed on this channel id + // (`read_anchor::`) and reachable this way. A + // direct conversation's anchor is keyed on the conversation id in both + // positions (`read_anchor::`), and a DM + // conversation id is never an enabled channel, so this sweep does not + // reach it. That's fine left alone: DM anchors are bounded by + // `reconcileReadState`'s 30-day retention regardless. + const anchorKeys = await this.tools.store.list(`read_anchor:${channel.id}:`); + for (const key of anchorKeys) await this.clear(key); } /** @@ -803,6 +841,31 @@ export class Slack extends Connector { // and rewinding what this one just wrote. await this.set(this.dmHeadKey(channelId), true); } + // Anchor the conversation for the daily sweep via the shared helper + // below (see `applyReadAnchor`). A direct conversation is never + // `threaded`: this one permanent link flattens Slack's reply threads + // into a running conversation, so the CONVERSATION cursor is the only + // cursor that describes it, and it is keyed on the conversation id in + // both positions — there is no thread root to key on. There is no + // live, cursor-informed "read" verdict for a DM (the channel cursor is + // settled only by the daily `reconcileReadState` sweep), so `read` is + // always `false` here; the helper's write gate + // (`advanceConversationHead && link.unread !== false`) still applies, + // covering both "these messages are not the conversation head" (a + // reaction or save naming an older message, which must not rewind the + // anchor) and "the link is already read" (`assembleSlackDmLink` sets + // `unread: false` for `initialSync`, so an anchor for it could only + // ever re-assert what is already true and would outlive a later manual + // "mark unread" in Plot). + await this.applyReadAnchor({ + channelId, + anchorId: channelId, + messages: kept, + direct: true, + link, + read: false, + advanceConversationHead, + }); return link; } @@ -829,9 +892,115 @@ export class Slack extends Connector { syncableId: channelId, }; if (messages[0]) link.facets = slackFacets(messages[0], channelId); + // Apply Slack's per-thread read cursor via the same shared helper the DM + // branch above uses (see `applyReadAnchor`). A root-only channel message + // has no thread cursor of its own — only the daily sweep's + // `conversations.info` can speak for it — so `threaded` and the read + // verdict below are both false, and the helper's write gate + // (`advanceConversationHead && link.unread !== false`) is what stops a + // stale anchor from being recreated when nothing new actually arrived + // (e.g. `refreshSlackThread`'s reaction refresh, which re-fetches with + // `advanceConversationHead: false`). + // + // Read from THIS response, never from a cache: the parent's + // `unread_count` is computed after the reply we are saving landed, so a + // genuine new-message unread can never be suppressed by a stale cursor. + // + // `initialSync` already asserts read (it sets `unread: false`), so those + // links need no anchor at all — the sweep would only re-assert what is + // already true. + const threadTs = + (link.meta?.threadTs as string | undefined) ?? messages[0]?.ts; + if (threadTs) { + const threaded = Boolean( + deriveReadAnchor(messages, { direct: false, at: Date.now() })?.threaded + ); + const read = + link.unread === false || + (threaded && threadReadVerdict(messages[0]) === "read"); + await this.applyReadAnchor({ + channelId, + anchorId: threadTs, + messages, + direct: false, + link, + read, + advanceConversationHead, + }); + } return link; } + /** + * Settle one link's read-anchor bookkeeping: mark it read (and drop the + * anchor) when the caller already has a cursor-backed verdict, or decide + * whether an anchor still needs to be (re)written to track it. + * + * The anchor's EXISTENCE is the transition gate for + * {@link reconcileReadState}: present means "not yet marked read for this + * `newest` ts". Deleting it on a successful mark-read is what makes a later + * manual "mark unread" in Plot stick — the sweep has nothing left to act on + * until new content recreates the anchor. + * + * Shared by both {@link buildConversationLink} branches (direct + * conversation and channel thread) so one gating rule governs both writes. + * They used to each inline their own version of this and drifted apart: the + * channel-thread copy wrote an anchor unconditionally whenever the verdict + * wasn't "read", even on a `refreshSlackThread` reaction refresh that asked + * `advanceConversationHead: false` — a root-only message has no thread + * cursor, so that verdict is always "unknown", and the anchor was + * recreated with the same `newest` even though no new content had arrived, + * letting the next sweep re-assert read over a user's manual unread. + * + * `advanceConversationHead` gates the WRITE only, never the read-clear: + * solid evidence that a cursor already says read is safe to act on no + * matter why the caller fetched these messages, but recreating a "not yet + * read" anchor is only valid when the caller is reporting the + * conversation's actual current head. + * + * Also carries the link's `title` onto the anchor (see `SlackReadAnchor` + * for why `reconcileReadState` needs it). `assembleSlackDmLink` deliberately + * omits `title` when `users.info` was unavailable, so a link with no title + * here does not mean "no title exists" — it falls back to whatever title + * is already on the stored anchor rather than dropping it for that window. + */ + private async applyReadAnchor(opts: { + channelId: string; + /** Thread root ts, or the conversation id for a direct conversation. */ + anchorId: string; + messages: SlackMessage[]; + direct: boolean; + link: { unread?: boolean; title?: string | null }; + /** Did a cursor say this link is read? */ + read: boolean; + advanceConversationHead: boolean; + }): Promise { + const { + channelId, + anchorId, + messages, + direct, + link, + read, + advanceConversationHead, + } = opts; + const anchorKey = this.readAnchorKey(channelId, anchorId); + if (read) { + link.unread = false; + await this.clear(anchorKey); + return; + } + if (advanceConversationHead && link.unread !== false) { + let title = link.title; + if (!title) { + const existing = await this.get(anchorKey); + title = existing?.title ?? undefined; + } + const anchor = deriveReadAnchor(messages, { direct, at: Date.now(), title }); + if (anchor) await this.set(anchorKey, anchor); + } + } + /** * The `source` the link for this conversation is keyed on — the same value * {@link buildConversationLink} produces, so any path addressing an @@ -1721,6 +1890,18 @@ export class Slack extends Connector { return `sync_thread:${channelId}:${threadTs}`; } + /** + * Key for one link's read anchor. Keyed on the thread root (the same id + * `subscriptionKey` uses) so a new reply REPLACES the anchor rather than + * accumulating one per message. + * + * A direct conversation is one permanent link, so it anchors on the + * conversation id in both positions. + */ + private readAnchorKey(channelId: string, threadTs: string): string { + return `read_anchor:${channelId}:${threadTs}`; + } + private async subscribeThread( channelId: string, threadTs: string @@ -1888,6 +2069,218 @@ export class Slack extends Connector { } } + /** + * Daily reconciliation of Slack's CHANNEL-level read cursor into Plot. + * + * Slack has no push signal for read state — `channel_marked`/`im_marked` are + * RTM-only — so this is the only way a link that has gone quiet ever learns + * the user read it. It uses `conversations.info` (Tier 3) exclusively and + * never touches `conversations.history`/`conversations.replies`, which are + * limited to 1 rpm and are what live message ingestion runs on: a sweep that + * spent that budget would starve inbound messages. + * + * Threaded anchors are skipped — a channel cursor says nothing about whether + * a thread was opened. Those are settled for free on the live path, where + * the thread's own cursor arrives with the messages. + * + * `channelId` is only used to resolve a token; Slack tokens are + * workspace-wide, and the conversations to sweep come from the anchors. + * + * Gated to run at most once per 24 hours per WORKSPACE (like `syncMembers`, + * placed the same way — before the `try`, so a redundant call skips the + * `finally` below too and never arms a chain of its own). The sweep itself + * is workspace-scoped (`listEntries("read_anchor:")`, no channel filter), + * but the caller, `queueWorkspaceDailyTasks`, dispatches it once per + * ENABLED CHANNEL, and Slack auto-observes new channels as the user + * composes into them. Without this guard, every newly-enabled channel would + * arm its own daily recurring chain (`read-state-sync:` is keyed + * per channel, so `scheduleRecurring` cannot dedupe them against each + * other) — N chains sweeping the same anchors once a day each means N times + * the `conversations.info` calls and N duplicate writes per anchor that + * actually resolves. + */ + async reconcileReadState(channelId: string): Promise { + const now = Date.now(); + const ONE_DAY_MS = 24 * 60 * 60 * 1000; + const lastReadSync = await this.get("readStateSyncedAt"); + if (lastReadSync && now - lastReadSync < ONE_DAY_MS) return; + + try { + const entries = await this.tools.store.listEntries( + "read_anchor:" + ); + if (entries.length === 0) return; + + // Drop anything past retention before spending an API call on it. + const expired = entries.filter( + ([, anchor]) => now - anchor.at >= READ_ANCHOR_RETENTION_MS + ); + if (expired.length > 0) { + await this.tools.store.clearMany(expired.map(([key]) => key)); + } + + // Threaded anchors are the live path's job; grouping by conversation is + // what keeps this to one `conversations.info` per conversation rather + // than one per link. + const byConversation = new Map(); + for (const [key, anchor] of entries) { + if (anchor.threaded) continue; + if (now - anchor.at >= READ_ANCHOR_RETENTION_MS) continue; + const rest = key.slice("read_anchor:".length); + const separator = rest.indexOf(":"); + if (separator <= 0) continue; + const conversationId = rest.slice(0, separator); + const existing = byConversation.get(conversationId); + if (existing) existing.push([key, anchor]); + else byConversation.set(conversationId, [[key, anchor]]); + } + if (byConversation.size === 0) return; + + let api: SlackApi; + try { + api = await this.getApi(channelId); + } catch (error) { + console.warn("reconcileReadState: Slack token unavailable", error); + return; + } + + const conversations = [...byConversation]; + for (let i = 0; i < conversations.length; i++) { + const [conversationId, anchors] = conversations[i]!; + let lastRead: string | null; + try { + ({ lastRead } = await api.getConversationInfo(conversationId)); + } catch (error) { + if (error instanceof SlackRateLimitedError) { + // Every remaining conversation needs the same method, so carrying + // on would just re-issue guaranteed-429s. Anchors are durable — + // tomorrow's pass (or the retry below) picks up exactly where this + // one stopped. `conversations.length - i` is what's actually left + // — this one included, since it never got resolved — not the + // conversation-count total the map started with. + const remaining = conversations.length - i; + console.log( + `reconcileReadState: rate limited on ${error.method}; ${remaining} conversation(s) left for a later pass` + ); + return; + } + if (error instanceof SlackPermanentError) { + if (error.slackError === "missing_scope") { + // The optional `dms` scope group was declined at connect time. + // That is a user decision, not a broken connection — mirrors + // `onThreadRead`'s handling of the same error, so a user who + // opted out of DM sync doesn't get a spurious reconnect prompt + // while the channels behind this DM in `byConversation` go + // unswept. + console.warn( + `reconcileReadState: skipping ${conversationId}: ${error.method} → missing_scope` + ); + continue; + } + if (error.slackError === "no_permission") { + // This conversation specifically is unreachable (e.g. the user + // was removed from it) — not evidence the token itself is dead. + console.warn( + `reconcileReadState: skipping ${conversationId}: ${error.method} → no_permission` + ); + continue; + } + if (SLACK_AUTH_ERRORS.has(error.slackError)) { + await this.tools.integrations.markNeedsReauth(channelId); + return; + } + // A single gone/forbidden conversation must not abandon the rest. + console.warn( + `reconcileReadState: skipping ${conversationId}: ${error.method} → ${error.slackError}` + ); + continue; + } + throw error; + } + + const resolved: string[] = []; + for (const [key, anchor] of anchors) { + if (channelReadVerdict(lastRead, anchor.newest) !== "read") continue; + const threadTs = key.slice( + `read_anchor:${conversationId}:`.length + ); + // Minimal upsert. `created` and `schedules` are omitted + // deliberately: `saveLink` reads `unread === false` as the + // initial-sync signal and DROPS the save outright when the item's + // date predates the plan's sync history limit. `preview` and + // `notes` are omitted too, so the upsert preserves whatever + // content is stored rather than rewriting it — `preview`'s + // platform default derives from the notes, so with none supplied + // it resolves to null and COALESCE falls through to the stored + // value. + // + // `title` is the one field that must be RE-SENT, not omitted: + // `upsert_thread` takes an "archived" code path whenever the + // user's `thread_priority` row points at an archived priority + // (they archived the focus this thread was filed under) or is + // missing, and on that path the platform's title default is the + // literal string "Untitled" — never null — so an omitted title + // there always loses to it, destroying the thread's real title. + // Sending `title` also re-writes it on the normal, non-archived + // path, which is harmless: a channel thread's title is derived + // deterministically from its root message and a DM's is the + // counterparty name, so both are re-sent unchanged by every + // ordinary sync anyway — this isn't a new source of truth, just + // re-asserting the same value. Sourced from the anchor (carried + // there by `applyReadAnchor`, since this upsert has no access to + // the live Slack message) and omitted here when the anchor has + // none — never worse than today's behaviour. + // + // `author: null` documents that this upsert is genuinely + // authorless — it only flips `unread` (and now `title`), never + // introduces new content — and silences `saveLink`'s + // development-time unattributed-link warning that would otherwise + // fire on every reconciled link. + await this.tools.integrations.saveLink({ + // Reuse the connector's own source helper — a reconcile upsert + // that guessed the key would create a second, empty thread + // instead of updating the one the save path wrote. + source: await this.conversationSource(conversationId, threadTs), + channelId: conversationId, + type: (await this.isKnownDMChannel(conversationId)) ? "dm" : "thread", + unread: false, + author: null, + ...(anchor.title ? { title: anchor.title } : {}), + }); + resolved.push(key); + } + if (resolved.length > 0) await this.tools.store.clearMany(resolved); + } + + // Stamped on completion, not entry — matches `membersSyncedAt` / + // `customEmojiSyncedAt`. A pass that returned early above (nothing to + // sweep, no token, rate limited) skips this: that's correct, since + // suppressing `queueWorkspaceDailyTasks`'s backstop for 24h on a pass + // that did no real work would leave anchors unreconciled with nothing + // to re-trigger it sooner. + await this.set("readStateSyncedAt", Date.now()); + } catch (error) { + // Unlike the rate-limit/permanent-error branches above (which return + // early and set nothing here), an unexpected error still lets the + // `finally` re-arm the daily chain below — mirrors `syncMembers`: a + // sweep that dies silently would leave read state unreconciled forever + // with no recovery signal, which is worse than retrying tomorrow. + console.error("reconcileReadState: unexpected error", error); + throw error; + } finally { + // Unconditional: nothing in this function suppresses the daily + // reschedule (unlike `syncMembers`, which stops its own chain on a + // permanent error) — even a dead-token pass is worth retrying + // tomorrow, since the user may have reauthed by then. The top-of- + // function guard above is what stops a REDUNDANT chain from ever being + // armed; this always re-arms the one chain that got past it. + const daily = await this.callback(this.reconcileReadState, channelId); + await this.scheduleRecurring(`read-state-sync:${channelId}`, daily, { + intervalMs: ONE_DAY_MS, + }); + } + } + async onThreadToDo( thread: Thread, _actor: Actor, @@ -1937,6 +2330,98 @@ export class Slack extends Connector { } } + /** + * Write a Plot read back to Slack — direct conversations only. + * + * `conversations.mark` is CONVERSATION-scoped and Slack's public Web API has + * no per-thread equivalent, so this is only coherent where the Plot link IS + * the whole conversation. For a channel thread the same call would move that + * channel's cursor: forward, clearing unread on every other message in it; + * backward, re-unreading messages the user had already read. Both are wrong, + * so channel threads write back nothing. + * + * Marking a thread UNREAD is likewise not propagated — Slack offers no + * un-mark, and re-pointing the cursor at an older message would un-read + * unrelated conversation history. + * + * Whichever direction Plot's read state changed, that's the user's final + * word — any outstanding {@link reconcileReadState} anchor for this link is + * cleared FIRST, before either the `unread` or `meta.direct` guard below. + * Skipping this for a DM read used to let the very `conversations.mark` call + * a few lines down move Slack's cursor to `anchor.newest` without ever + * clearing the anchor Slack was catching up to — so marking the thread + * unread again in Plot right after left a stale anchor in place, and the + * next sweep saw a cursor it had itself advanced and reverted the user's + * unread. Clearing here — for both link shapes and both directions — keeps + * a manual mark-unread from being undone the same way even without a + * write-back to trigger it. + */ + override async onThreadRead( + thread: Thread, + _actor: Actor, + unread: boolean + ): Promise { + const meta = thread.meta ?? {}; + const channelId = meta.channelId as string | undefined; + const threadTs = meta.threadTs as string | undefined; + if (channelId && threadTs) { + // A DM's `meta.threadTs` is the conversation's latest message, not its + // anchor key — `buildConversationLink` anchors a direct conversation on + // the conversation id in both positions (see `readAnchorKey`), so using + // `threadTs` directly here would clear nothing. + const anchorId = meta.direct === true ? channelId : threadTs; + await this.clear(this.readAnchorKey(channelId, anchorId)); + } + + if (unread) return; + if (meta.direct !== true) return; + if (!channelId || !threadTs) return; + + let api: SlackApi; + try { + api = await this.getApi(channelId); + } catch (error) { + // Read state already lives in Plot; a missing token is not worth failing + // the dispatch over. + console.warn("onThreadRead: Slack token unavailable", error); + return; + } + + try { + await api.markConversationRead(channelId, threadTs); + } catch (error) { + if (error instanceof SlackRateLimitedError) { + // The read is already recorded in Plot and the next inbound message + // re-establishes the cursor; a deferred write-back is not worth the + // bookkeeping for a marker the user cannot see. + console.log("onThreadRead: rate limited; skipping write-back"); + return; + } + if (error instanceof SlackPermanentError) { + if (error.slackError === "missing_scope") { + // The optional `dms` scope group was declined at connect time. + // That is a user decision, not a broken connection — degrade + // quietly rather than prompting a pointless reconnect. + console.warn( + `onThreadRead: ${error.method} → missing_scope; skipping write-back` + ); + return; + } + if (SLACK_AUTH_ERRORS.has(error.slackError)) { + // A genuinely dead token. Flag it the same way every other write-back + // path in this connector does, so the user is prompted to reconnect. + await this.tools.integrations.markNeedsReauth(channelId); + return; + } + console.warn( + `onThreadRead: ${error.method} → ${error.slackError}; skipping write-back` + ); + return; + } + throw error; + } + } + // ---- Compose new messages from Plot ---- /** @@ -2272,6 +2757,18 @@ export class Slack extends Connector { const dmListCallback = await this.callback(this.listDMChannels, channelId); await this.runTask(dmListCallback); } + + const lastReadSync = await this.get("readStateSyncedAt"); + const readClaimedAt = await this.get("readStateSyncClaimedAt"); + const readClaimed = + readClaimedAt !== null && + readClaimedAt !== undefined && + now - readClaimedAt < CLAIM_TTL_MS; + if ((!lastReadSync || now - lastReadSync >= ONE_DAY_MS) && !readClaimed) { + await this.set("readStateSyncClaimedAt", now); + const readCallback = await this.callback(this.reconcileReadState, channelId); + await this.runTask(readCallback); + } } /**