aaronrene / knowtation public
calendar-blob-store.mjs
345 lines 10.2 KB
Raw
sha256:b5f647cb9c409f563d4671fe3fc05ddea01fabfed9b41fc11cb923588e1c1baf mirror: GitHub Phase A durable MCP OAuth (#270) Human minor ⚠ breaking 11 days ago
1 /**
2 * Hosted bridge: persist calendar store + encrypted OAuth token blobs in Netlify Blobs.
3 *
4 * Self-hosted bridge uses DATA_DIR files only. On Netlify, DATA_DIR is ephemeral;
5 * this module hydrates from Blobs before calendar handlers and persists after writes.
6 *
7 * Calendar OAuth begin → Google → callback typically lands on a **different** Lambda
8 * instance within seconds. Netlify Blobs default to **eventual** consistency (up to ~60s
9 * edge drift). Hydrate therefore reads the calendar store with **strong** consistency so
10 * pending `oauth_pending.state` is visible to the callback (avoids `state_invalid`).
11 *
12 * @see hub/bridge/delegation-blob-store.mjs — same hydrate/persist pattern (7C-L1c)
13 */
14
15 import fs from 'fs';
16 import path from 'path';
17 import {
18 CALENDAR_OAUTH_DIR,
19 CALENDAR_STORE_FILENAME,
20 getCalendarStorePath,
21 loadCalendarStore,
22 } from '../../lib/calendar/event-store.mjs';
23
24 /**
25 * @typedef {{
26 * get: (key: string, opts?: { type?: string, consistency?: 'eventual' | 'strong' }) => Promise<string|ArrayBuffer|null>,
27 * set: (key: string, value: string) => Promise<void>
28 * }} BlobStore
29 */
30
31 export const CALENDAR_STORE_BLOB_KEY = `calendar/${CALENDAR_STORE_FILENAME}`;
32
33 /**
34 * @param {string} connectorId
35 * @returns {string}
36 */
37 export function calendarOAuthBlobKey(connectorId) {
38 return `calendar/oauth/${connectorId}.enc`;
39 }
40
41 /**
42 * @param {string} dataDir
43 * @returns {string}
44 */
45 export function calendarOAuthDir(dataDir) {
46 return path.join(dataDir, CALENDAR_OAUTH_DIR);
47 }
48
49 /**
50 * Status preference for connector merge (higher wins when comparing fresh writes).
51 *
52 * @param {Record<string, unknown>} connector
53 * @returns {number}
54 */
55 function connectorStatusScore(connector) {
56 const status = typeof connector.status === 'string' ? connector.status : '';
57 if (status === 'pending' && connector.oauth_pending) return 4;
58 if (status === 'connected') return 3;
59 if (status === 'pending') return 2;
60 if (status === 'revoked') return 1;
61 return 0;
62 }
63
64 /**
65 * @param {Record<string, unknown>} connector
66 * @returns {number}
67 */
68 function connectorRecencyMs(connector) {
69 const pending = connector.oauth_pending;
70 if (pending && typeof pending === 'object' && typeof pending.expires_at === 'string') {
71 const exp = Date.parse(pending.expires_at);
72 if (Number.isFinite(exp)) return exp;
73 }
74 for (const key of ['last_sync_at', 'revoked_at']) {
75 const raw = connector[key];
76 if (typeof raw === 'string') {
77 const ms = Date.parse(raw);
78 if (Number.isFinite(ms)) return ms;
79 }
80 }
81 return 0;
82 }
83
84 /**
85 * @param {Record<string, unknown>} a
86 * @param {Record<string, unknown>} b
87 * @returns {Record<string, unknown>}
88 */
89 function pickConnector(a, b) {
90 const scoreA = connectorStatusScore(a);
91 const scoreB = connectorStatusScore(b);
92 if (scoreA !== scoreB) return scoreA > scoreB ? a : b;
93 const timeA = connectorRecencyMs(a);
94 const timeB = connectorRecencyMs(b);
95 return timeB >= timeA ? b : a;
96 }
97
98 /**
99 * Merge local + blob calendar stores so warm-Lambda pending OAuth is not wiped by a
100 * stale eventual blob read (same failure mode as delegation grants on Netlify).
101 *
102 * @param {string} localRaw
103 * @param {string} blobRaw
104 * @returns {string}
105 */
106 export function mergeCalendarStoreJson(localRaw, blobRaw) {
107 const parse = (raw) => {
108 if (typeof raw !== 'string' || !raw.trim()) return null;
109 try {
110 return JSON.parse(raw);
111 } catch {
112 return null;
113 }
114 };
115
116 const local = parse(localRaw);
117 const blob = parse(blobRaw);
118 if (!local && !blob) return blobRaw || localRaw || '';
119 if (!local) return blobRaw;
120 if (!blob) return localRaw;
121
122 if (!local.vaults || typeof local.vaults !== 'object') local.vaults = {};
123 if (!blob.vaults || typeof blob.vaults !== 'object') blob.vaults = {};
124
125 const vaultIds = new Set([
126 ...Object.keys(local.vaults),
127 ...Object.keys(blob.vaults),
128 ]);
129
130 /** @type {Record<string, unknown>} */
131 const mergedVaults = {};
132
133 for (const vaultId of vaultIds) {
134 const localVault = local.vaults[vaultId];
135 const blobVault = blob.vaults[vaultId];
136
137 if (!localVault) {
138 mergedVaults[vaultId] = blobVault;
139 continue;
140 }
141 if (!blobVault) {
142 mergedVaults[vaultId] = localVault;
143 continue;
144 }
145
146 /** @type {Map<string, Record<string, unknown>>} */
147 const byId = new Map();
148 for (const connector of [
149 ...(Array.isArray(blobVault.connectors) ? blobVault.connectors : []),
150 ...(Array.isArray(localVault.connectors) ? localVault.connectors : []),
151 ]) {
152 if (!connector || typeof connector !== 'object') continue;
153 const id = typeof connector.connector_id === 'string' ? connector.connector_id.trim() : '';
154 if (!id) continue;
155 const existing = byId.get(id);
156 byId.set(id, existing ? pickConnector(existing, connector) : connector);
157 }
158
159 /** @type {Map<string, Record<string, unknown>>} */
160 const calendarsById = new Map();
161 for (const row of [
162 ...(Array.isArray(blobVault.source_calendars) ? blobVault.source_calendars : []),
163 ...(Array.isArray(localVault.source_calendars) ? localVault.source_calendars : []),
164 ]) {
165 if (!row || typeof row !== 'object') continue;
166 const id = typeof row.source_calendar_id === 'string' ? row.source_calendar_id.trim() : '';
167 if (!id) continue;
168 calendarsById.set(id, row);
169 }
170
171 /** @type {Map<string, Record<string, unknown>>} */
172 const eventsByKey = new Map();
173 for (const event of [
174 ...(Array.isArray(blobVault.events) ? blobVault.events : []),
175 ...(Array.isArray(localVault.events) ? localVault.events : []),
176 ]) {
177 if (!event || typeof event !== 'object') continue;
178 const id = typeof event.event_id === 'string' ? event.event_id.trim() : '';
179 const connectorId = typeof event.connector_id === 'string' ? event.connector_id.trim() : '';
180 const key = id && connectorId ? `${connectorId}:${id}` : id || JSON.stringify(event);
181 eventsByKey.set(key, event);
182 }
183
184 mergedVaults[vaultId] = {
185 ...blobVault,
186 ...localVault,
187 connectors: [...byId.values()],
188 source_calendars: [...calendarsById.values()],
189 events: [...eventsByKey.values()],
190 };
191 }
192
193 return JSON.stringify({
194 ...blob,
195 ...local,
196 vaults: mergedVaults,
197 });
198 }
199
200 /**
201 * Connector ids that may have encrypted token blobs on disk.
202 *
203 * @param {string} dataDir
204 * @returns {string[]}
205 */
206 export function listCalendarConnectorIdsForBlobSync(dataDir) {
207 const store = loadCalendarStore(dataDir);
208 /** @type {Set<string>} */
209 const ids = new Set();
210 for (const vault of Object.values(store.vaults ?? {})) {
211 for (const connector of vault.connectors ?? []) {
212 if (
213 connector.status === 'connected'
214 && typeof connector.connector_id === 'string'
215 && connector.connector_id.trim()
216 ) {
217 ids.add(connector.connector_id.trim());
218 }
219 }
220 }
221 const oauthDir = calendarOAuthDir(dataDir);
222 if (fs.existsSync(oauthDir)) {
223 for (const name of fs.readdirSync(oauthDir)) {
224 if (name.endsWith('.enc')) {
225 ids.add(name.slice(0, -4));
226 }
227 }
228 }
229 return [...ids];
230 }
231
232 /**
233 * Load calendar store + OAuth token blobs from Blobs into DATA_DIR (hosted cold-start hydration).
234 *
235 * Calendar store reads use **strong** consistency so OAuth callback can find pending state
236 * written by begin on another Lambda within seconds.
237 *
238 * @param {BlobStore|null|undefined} blobStore
239 * @param {string} dataDir
240 */
241 export async function hydrateCalendarStoresFromBlob(blobStore, dataDir) {
242 if (!blobStore || typeof blobStore.get !== 'function') return;
243 fs.mkdirSync(dataDir, { recursive: true });
244 fs.mkdirSync(calendarOAuthDir(dataDir), { recursive: true });
245
246 const storePath = getCalendarStorePath(dataDir);
247 let localRaw = '';
248 if (fs.existsSync(storePath)) {
249 try {
250 localRaw = fs.readFileSync(storePath, 'utf8');
251 } catch {
252 localRaw = '';
253 }
254 }
255
256 try {
257 const storeRaw = await blobStore.get(CALENDAR_STORE_BLOB_KEY, {
258 type: 'text',
259 consistency: 'strong',
260 });
261 if (typeof storeRaw === 'string' && storeRaw.trim()) {
262 const merged = mergeCalendarStoreJson(localRaw, storeRaw);
263 if (merged.trim()) {
264 fs.writeFileSync(storePath, merged, 'utf8');
265 }
266 }
267 } catch {
268 /* keep existing file or empty */
269 }
270
271 const connectorIds = listCalendarConnectorIdsForBlobSync(dataDir);
272 for (const connectorId of connectorIds) {
273 try {
274 const raw = await blobStore.get(calendarOAuthBlobKey(connectorId), {
275 type: 'text',
276 consistency: 'strong',
277 });
278 if (typeof raw === 'string' && raw.trim()) {
279 fs.writeFileSync(path.join(calendarOAuthDir(dataDir), `${connectorId}.enc`), raw, 'utf8');
280 }
281 } catch {
282 /* keep existing file or skip */
283 }
284 }
285 }
286
287 /**
288 * Write calendar store + OAuth token blobs from DATA_DIR to Blobs after a mutation.
289 *
290 * @param {BlobStore|null|undefined} blobStore
291 * @param {string} dataDir
292 */
293 export async function persistCalendarStoresToBlob(blobStore, dataDir) {
294 if (!blobStore || typeof blobStore.set !== 'function') return;
295
296 const storePath = getCalendarStorePath(dataDir);
297 if (fs.existsSync(storePath)) {
298 try {
299 const raw = fs.readFileSync(storePath, 'utf8');
300 if (raw.trim()) {
301 await blobStore.set(CALENDAR_STORE_BLOB_KEY, raw);
302 }
303 } catch {
304 /* non-fatal */
305 }
306 }
307
308 const oauthDir = calendarOAuthDir(dataDir);
309 if (!fs.existsSync(oauthDir)) return;
310
311 for (const name of fs.readdirSync(oauthDir)) {
312 if (!name.endsWith('.enc')) continue;
313 const connectorId = name.slice(0, -4);
314 const fp = path.join(oauthDir, name);
315 try {
316 const raw = fs.readFileSync(fp, 'utf8');
317 if (raw.trim()) {
318 await blobStore.set(calendarOAuthBlobKey(connectorId), raw);
319 }
320 } catch {
321 /* non-fatal */
322 }
323 }
324 }
325
326 /**
327 * Run a calendar handler with hosted Blob hydrate/persist when available.
328 *
329 * @template T
330 * @param {{
331 * blobStore: BlobStore|null|undefined,
332 * dataDir: string,
333 * persist?: boolean,
334 * run: () => T | Promise<T>,
335 * }} opts
336 * @returns {Promise<T>}
337 */
338 export async function withCalendarBlobSync(opts) {
339 await hydrateCalendarStoresFromBlob(opts.blobStore, opts.dataDir);
340 const result = await opts.run();
341 if (opts.persist !== false) {
342 await persistCalendarStoresToBlob(opts.blobStore, opts.dataDir);
343 }
344 return result;
345 }
File History 1 commit
sha256:b5f647cb9c409f563d4671fe3fc05ddea01fabfed9b41fc11cb923588e1c1baf mirror: GitHub Phase A durable MCP OAuth (#270) Human minor ⚠ 11 days ago