Skip to content

Commit e8afda6

Browse files
KrisBraunKrisBraun
authored
Merge pull request #285 from plotday/slack-connection-reliability-and-dm-sync
Sync Slack DMs, fix false 'Reconnect' prompts across 8 connectors
2 parents c9fe52d + 33b389e commit e8afda6

35 files changed

Lines changed: 2749 additions & 43 deletions

‎connectors/airtable/package.json‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,16 @@
2525
"build": "tsc",
2626
"clean": "rm -rf dist",
2727
"deploy": "plot deploy",
28-
"lint": "plot lint"
28+
"lint": "plot lint",
29+
"test": "vitest run",
30+
"test:watch": "vitest"
2931
},
3032
"dependencies": {
3133
"@plotday/twister": "workspace:^"
3234
},
3335
"devDependencies": {
34-
"typescript": "^5.9.3"
36+
"typescript": "^5.9.3",
37+
"vitest": "^2.1.8"
3538
},
3639
"repository": {
3740
"type": "git",

‎connectors/airtable/src/airtable.test.ts‎

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,185 @@
1+
import { describe, expect, it, vi } from "vitest";
2+
import { Airtable } from "./airtable";
3+
import type { AirtableRecord, AirtableTable } from "./airtable-api";
4+
5+
function makeStore(initial: Record<string, unknown> = {}) {
6+
const map = new Map<string, unknown>(Object.entries(initial));
7+
return {
8+
map,
9+
get: vi.fn(async (k: string) => (map.has(k) ? map.get(k) : null)),
10+
set: vi.fn(async (k: string, v: unknown) => void map.set(k, v)),
11+
clear: vi.fn(async (k: string) => void map.delete(k)),
12+
list: vi.fn(async (p: string) => [...map.keys()].filter((k) => k.startsWith(p))),
13+
};
14+
}
15+
16+
function makeAirtable(
17+
opts: {
18+
store?: ReturnType<typeof makeStore>;
19+
integrations?: Record<string, unknown>;
20+
network?: Record<string, unknown>;
21+
} = {}
22+
): Airtable {
23+
const tools = {
24+
store: opts.store ?? makeStore(),
25+
integrations: {
26+
get: vi.fn().mockResolvedValue({ token: "tok" }),
27+
saveLink: vi.fn().mockResolvedValue("thread-1"),
28+
channelSyncCompleted: vi.fn().mockResolvedValue(undefined),
29+
...opts.integrations,
30+
},
31+
network: { createWebhook: vi.fn(), ...opts.network },
32+
};
33+
return new Airtable("twist-1" as never, { getTools: () => tools } as never);
34+
}
35+
36+
const baseId = "base1";
37+
38+
// A minimal detected task table with no assignee/status/notes fields — the
39+
// tests below only exercise pagination and completion signaling, not field
40+
// mapping, so the extra detail is intentionally omitted.
41+
const detectedTable = {
42+
tableId: "tbl1",
43+
tableName: "Tasks",
44+
primaryFieldId: "fld1",
45+
primaryFieldName: "Name",
46+
assigneeFieldId: null,
47+
assigneeFieldName: null,
48+
assigneeFieldType: null,
49+
dueDateFieldId: null,
50+
dueDateFieldName: null,
51+
statusFieldId: null,
52+
statusFieldName: null,
53+
statusFieldType: null,
54+
doneOptionName: null,
55+
todoOptionName: null,
56+
notesFieldId: null,
57+
notesFieldName: null,
58+
};
59+
60+
function record(id: string): AirtableRecord {
61+
return { id, createdTime: "2026-01-01T00:00:00Z", fields: { Name: `Task ${id}` } };
62+
}
63+
64+
describe("syncBatch", () => {
65+
it("signals channelSyncCompleted when the last page of the last table is reached (initial sync)", async () => {
66+
const store = makeStore({
67+
[`sync_enabled_${baseId}`]: true,
68+
[`sync_state_${baseId}`]: { tableIndex: 0, offset: null, initialSync: true },
69+
[`task_tables_${baseId}`]: [detectedTable],
70+
});
71+
const channelSyncCompleted = vi.fn().mockResolvedValue(undefined);
72+
const saveLink = vi.fn().mockResolvedValue("t1");
73+
const airtable = makeAirtable({ store, integrations: { channelSyncCompleted, saveLink } });
74+
const listRecords = vi
75+
.fn()
76+
.mockResolvedValue({ records: [record("rec1")], offset: undefined });
77+
(airtable as unknown as { getAPI: unknown }).getAPI = vi
78+
.fn()
79+
.mockResolvedValue({ listRecords });
80+
81+
await (airtable as unknown as { syncBatch: (b: string) => Promise<void> }).syncBatch(baseId);
82+
83+
expect(saveLink).toHaveBeenCalledTimes(1);
84+
expect(channelSyncCompleted).toHaveBeenCalledWith(baseId);
85+
// sync state is cleared once the chain has nothing left to schedule
86+
expect(store.map.has(`sync_state_${baseId}`)).toBe(false);
87+
});
88+
89+
it("does not signal channelSyncCompleted while more pages remain", async () => {
90+
const store = makeStore({
91+
[`sync_enabled_${baseId}`]: true,
92+
[`sync_state_${baseId}`]: { tableIndex: 0, offset: null, initialSync: true },
93+
[`task_tables_${baseId}`]: [detectedTable],
94+
});
95+
const channelSyncCompleted = vi.fn();
96+
const airtable = makeAirtable({ store, integrations: { channelSyncCompleted } });
97+
const listRecords = vi
98+
.fn()
99+
.mockResolvedValue({ records: [record("rec1")], offset: "cursor2" });
100+
(airtable as unknown as { getAPI: unknown }).getAPI = vi
101+
.fn()
102+
.mockResolvedValue({ listRecords });
103+
(airtable as unknown as { callback: unknown }).callback = vi.fn().mockResolvedValue("cb");
104+
(airtable as unknown as { runTask: unknown }).runTask = vi.fn().mockResolvedValue(undefined);
105+
106+
await (airtable as unknown as { syncBatch: (b: string) => Promise<void> }).syncBatch(baseId);
107+
108+
expect(channelSyncCompleted).not.toHaveBeenCalled();
109+
const state = store.map.get(`sync_state_${baseId}`) as { offset: string };
110+
expect(state.offset).toBe("cursor2");
111+
});
112+
113+
it("does not signal channelSyncCompleted when an incremental (non-initial) sync completes", async () => {
114+
const store = makeStore({
115+
[`sync_enabled_${baseId}`]: true,
116+
[`sync_state_${baseId}`]: { tableIndex: 0, offset: null, initialSync: false },
117+
[`task_tables_${baseId}`]: [detectedTable],
118+
});
119+
const channelSyncCompleted = vi.fn();
120+
const saveLink = vi.fn().mockResolvedValue("t1");
121+
const airtable = makeAirtable({ store, integrations: { channelSyncCompleted, saveLink } });
122+
const listRecords = vi
123+
.fn()
124+
.mockResolvedValue({ records: [record("rec1")], offset: undefined });
125+
(airtable as unknown as { getAPI: unknown }).getAPI = vi
126+
.fn()
127+
.mockResolvedValue({ listRecords });
128+
129+
await (airtable as unknown as { syncBatch: (b: string) => Promise<void> }).syncBatch(baseId);
130+
131+
expect(channelSyncCompleted).not.toHaveBeenCalled();
132+
});
133+
});
134+
135+
describe("detectAndSync", () => {
136+
it("signals channelSyncCompleted when the base has no task-shaped tables (initial sync)", async () => {
137+
const store = makeStore({ [`sync_enabled_${baseId}`]: true });
138+
const channelSyncCompleted = vi.fn().mockResolvedValue(undefined);
139+
const airtable = makeAirtable({ store, integrations: { channelSyncCompleted } });
140+
// No collaborator/status field on the only table → scoreTable filters it
141+
// out, so refreshTaskTables resolves an empty detected list and the
142+
// batch chain (syncBatch) never starts.
143+
const listTables: AirtableTable[] = [
144+
{
145+
id: "tbl1",
146+
name: "Notes",
147+
primaryFieldId: "fld1",
148+
fields: [{ id: "fld1", name: "Name", type: "singleLineText" }],
149+
},
150+
];
151+
(airtable as unknown as { getAPI: unknown }).getAPI = vi.fn().mockResolvedValue({
152+
listTables: vi.fn().mockResolvedValue(listTables),
153+
});
154+
155+
await (
156+
airtable as unknown as { detectAndSync: (b: string, i: boolean) => Promise<void> }
157+
).detectAndSync(baseId, true);
158+
159+
expect(channelSyncCompleted).toHaveBeenCalledWith(baseId);
160+
expect(store.map.has(`sync_state_${baseId}`)).toBe(false);
161+
});
162+
163+
it("does not signal channelSyncCompleted for a no-op incremental re-detect", async () => {
164+
const store = makeStore({ [`sync_enabled_${baseId}`]: true });
165+
const channelSyncCompleted = vi.fn();
166+
const airtable = makeAirtable({ store, integrations: { channelSyncCompleted } });
167+
const listTables: AirtableTable[] = [
168+
{
169+
id: "tbl1",
170+
name: "Notes",
171+
primaryFieldId: "fld1",
172+
fields: [{ id: "fld1", name: "Name", type: "singleLineText" }],
173+
},
174+
];
175+
(airtable as unknown as { getAPI: unknown }).getAPI = vi.fn().mockResolvedValue({
176+
listTables: vi.fn().mockResolvedValue(listTables),
177+
});
178+
179+
await (
180+
airtable as unknown as { detectAndSync: (b: string, i: boolean) => Promise<void> }
181+
).detectAndSync(baseId, false);
182+
183+
expect(channelSyncCompleted).not.toHaveBeenCalled();
184+
});
185+
});

‎connectors/airtable/src/airtable.ts‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -330,7 +330,15 @@ export class Airtable extends Connector<Airtable> {
330330

331331
const detected = await this.refreshTaskTables(baseId, api);
332332

333-
if (detected.length === 0) return;
333+
if (detected.length === 0) {
334+
// No task-shaped tables in this base — the batch chain below never
335+
// starts, so this IS the completion of the initial sync (trivially,
336+
// with nothing to sync). Without this, initial_sync_completed_at
337+
// would never be set for such bases and the stuck-sync watchdog would
338+
// eventually force-flag a healthy connection as needing reconnection.
339+
if (initialSync) await this.tools.integrations.channelSyncCompleted(baseId);
340+
return;
341+
}
334342

335343
await this.set<SyncState>(`sync_state_${baseId}`, {
336344
tableIndex: 0,
@@ -493,6 +501,15 @@ export class Airtable extends Connector<Airtable> {
493501
initialSync: state.initialSync,
494502
});
495503
} else {
504+
// No more tables or pages left to schedule — the initial-sync chain
505+
// (started in detectAndSync) is done. Signal completion so the
506+
// platform stamps initial_sync_completed_at; otherwise the
507+
// stuck-sync watchdog eventually force-flags a healthy connection as
508+
// needing reconnection. Idempotent/channel-scoped, so gating on
509+
// initialSync just avoids a pointless call on incremental re-syncs.
510+
if (state.initialSync) {
511+
await this.tools.integrations.channelSyncCompleted(baseId);
512+
}
496513
await this.clear(`sync_state_${baseId}`);
497514
return;
498515
}

‎connectors/airtable/vitest.config.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
import { defineConfig } from "vitest/config";
2+
3+
export default defineConfig({
4+
resolve: {
5+
// Resolve workspace connector packages from their TypeScript source
6+
// using the @plotday/connector export condition (same as the build path).
7+
conditions: ["@plotday/connector", "default"],
8+
},
9+
test: {},
10+
});

‎connectors/attio/package.json‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,16 @@
2525
"build": "tsc",
2626
"clean": "rm -rf dist",
2727
"deploy": "plot deploy",
28-
"lint": "plot lint"
28+
"lint": "plot lint",
29+
"test": "vitest run",
30+
"test:watch": "vitest"
2931
},
3032
"dependencies": {
3133
"@plotday/twister": "workspace:^"
3234
},
3335
"devDependencies": {
34-
"typescript": "^5.9.3"
36+
"typescript": "^5.9.3",
37+
"vitest": "^2.1.8"
3538
},
3639
"repository": {
3740
"type": "git",

0 commit comments

Comments
 (0)