aaronrene / knowtation public
docs-blob-store.mjs
298 lines 8.2 KB
Raw
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 11 days ago
1 /**
2 * Hosted bridge: persist docs connector store + encrypted OAuth token blobs in Netlify Blobs.
3 *
4 * Same strong-consistency hydrate rule as calendar-blob-store: OAuth begin and callback
5 * often land on different Lambdas within seconds. Eventual consistency would yield
6 * state_invalid on callback.
7 *
8 * Blob keys use docs/… — never calendar/oauth/….
9 *
10 * @see docs/KN-DOCS-SYNC-FREEZE.md D17
11 * @see hub/bridge/calendar-blob-store.mjs
12 */
13
14 import fs from 'fs';
15 import path from 'path';
16 import {
17 DOCS_STORE_FILENAME,
18 loadDocsStore,
19 } from '../../lib/docs/docs-connector-store.mjs';
20 import { oauthTokenVaultPath } from '../../lib/docs/oauth-token-vault.mjs';
21
22 /**
23 * @typedef {{
24 * get: (key: string, opts?: { type?: string, consistency?: 'eventual' | 'strong' }) => Promise<string|ArrayBuffer|null>,
25 * set: (key: string, value: string) => Promise<void>
26 * }} BlobStore
27 */
28
29 export const DOCS_STORE_BLOB_KEY = `docs/${DOCS_STORE_FILENAME}`;
30
31 /**
32 * @param {string} connectorId
33 * @returns {string}
34 */
35 export function docsOAuthBlobKey(connectorId) {
36 return `docs/oauth/${connectorId}.enc`;
37 }
38
39 /**
40 * @param {string} dataDir
41 * @returns {string}
42 */
43 export function docsOAuthDir(dataDir) {
44 return path.join(dataDir, 'docs_oauth');
45 }
46
47 /**
48 * @param {string} dataDir
49 * @returns {string}
50 */
51 export function getDocsStorePath(dataDir) {
52 return path.join(dataDir, DOCS_STORE_FILENAME);
53 }
54
55 /**
56 * Status preference for connector merge (higher wins).
57 * @param {Record<string, unknown>} connector
58 * @returns {number}
59 */
60 function connectorStatusScore(connector) {
61 const status = typeof connector.status === 'string' ? connector.status : '';
62 if (status === 'pending' && connector.oauth_pending) return 4;
63 if (status === 'connected') return 3;
64 if (status === 'pending') return 2;
65 if (status === 'needs_reauth') return 1;
66 if (status === 'revoked') return 0;
67 return 0;
68 }
69
70 /**
71 * @param {Record<string, unknown>} connector
72 * @returns {number}
73 */
74 function connectorRecencyMs(connector) {
75 const pending = connector.oauth_pending;
76 if (pending && typeof pending === 'object' && typeof pending.expires_at === 'string') {
77 const exp = Date.parse(pending.expires_at);
78 if (Number.isFinite(exp)) return exp;
79 }
80 for (const key of ['last_sync_at', 'revoked_at']) {
81 const raw = connector[key];
82 if (typeof raw === 'string') {
83 const ms = Date.parse(raw);
84 if (Number.isFinite(ms)) return ms;
85 }
86 }
87 return 0;
88 }
89
90 /**
91 * @param {Record<string, unknown>} a
92 * @param {Record<string, unknown>} b
93 * @returns {Record<string, unknown>}
94 */
95 function pickConnector(a, b) {
96 const scoreA = connectorStatusScore(a);
97 const scoreB = connectorStatusScore(b);
98 if (scoreA !== scoreB) return scoreA > scoreB ? a : b;
99 const timeA = connectorRecencyMs(a);
100 const timeB = connectorRecencyMs(b);
101 return timeB >= timeA ? b : a;
102 }
103
104 /**
105 * Merge local + blob docs stores so warm-Lambda pending OAuth is not wiped.
106 * @param {string} localRaw
107 * @param {string} blobRaw
108 * @returns {string}
109 */
110 export function mergeDocsStoreJson(localRaw, blobRaw) {
111 const parse = (raw) => {
112 if (typeof raw !== 'string' || !raw.trim()) return null;
113 try {
114 return JSON.parse(raw);
115 } catch {
116 return null;
117 }
118 };
119
120 const local = parse(localRaw);
121 const blob = parse(blobRaw);
122 if (!local && !blob) return blobRaw || localRaw || '';
123 if (!local) return blobRaw;
124 if (!blob) return localRaw;
125
126 if (!local.vaults || typeof local.vaults !== 'object') local.vaults = {};
127 if (!blob.vaults || typeof blob.vaults !== 'object') blob.vaults = {};
128
129 const vaultIds = new Set([...Object.keys(local.vaults), ...Object.keys(blob.vaults)]);
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 if (!localVault) {
137 mergedVaults[vaultId] = blobVault;
138 continue;
139 }
140 if (!blobVault) {
141 mergedVaults[vaultId] = localVault;
142 continue;
143 }
144
145 /** @type {Map<string, Record<string, unknown>>} */
146 const byId = new Map();
147 for (const connector of [
148 ...(Array.isArray(blobVault.connectors) ? blobVault.connectors : []),
149 ...(Array.isArray(localVault.connectors) ? localVault.connectors : []),
150 ]) {
151 if (!connector || typeof connector !== 'object') continue;
152 const id = typeof connector.connector_id === 'string' ? connector.connector_id.trim() : '';
153 if (!id) continue;
154 const existing = byId.get(id);
155 byId.set(id, existing ? pickConnector(existing, connector) : connector);
156 }
157
158 mergedVaults[vaultId] = {
159 ...blobVault,
160 ...localVault,
161 connectors: [...byId.values()],
162 };
163 }
164
165 return JSON.stringify({
166 ...blob,
167 ...local,
168 vaults: mergedVaults,
169 });
170 }
171
172 /**
173 * @param {string} dataDir
174 * @returns {string[]}
175 */
176 export function listDocsConnectorIdsForBlobSync(dataDir) {
177 const store = loadDocsStore(dataDir);
178 /** @type {Set<string>} */
179 const ids = new Set();
180 for (const vault of Object.values(store.vaults ?? {})) {
181 for (const connector of vault.connectors ?? []) {
182 if (
183 connector.status === 'connected'
184 && typeof connector.connector_id === 'string'
185 && connector.connector_id.trim()
186 && connector.provider === 'google-drive'
187 ) {
188 ids.add(connector.connector_id.trim());
189 }
190 }
191 }
192 const oauthDir = docsOAuthDir(dataDir);
193 if (fs.existsSync(oauthDir)) {
194 for (const name of fs.readdirSync(oauthDir)) {
195 if (name.endsWith('.enc')) ids.add(name.slice(0, -4));
196 }
197 }
198 return [...ids];
199 }
200
201 /**
202 * @param {BlobStore|null|undefined} blobStore
203 * @param {string} dataDir
204 */
205 export async function hydrateDocsStoresFromBlob(blobStore, dataDir) {
206 if (!blobStore || typeof blobStore.get !== 'function') return;
207 fs.mkdirSync(dataDir, { recursive: true });
208 fs.mkdirSync(docsOAuthDir(dataDir), { recursive: true });
209
210 const storePath = getDocsStorePath(dataDir);
211 let localRaw = '';
212 if (fs.existsSync(storePath)) {
213 try {
214 localRaw = fs.readFileSync(storePath, 'utf8');
215 } catch {
216 localRaw = '';
217 }
218 }
219
220 try {
221 const storeRaw = await blobStore.get(DOCS_STORE_BLOB_KEY, {
222 type: 'text',
223 consistency: 'strong',
224 });
225 if (typeof storeRaw === 'string' && storeRaw.trim()) {
226 const merged = mergeDocsStoreJson(localRaw, storeRaw);
227 if (merged.trim()) fs.writeFileSync(storePath, merged, 'utf8');
228 }
229 } catch {
230 /* keep existing */
231 }
232
233 for (const connectorId of listDocsConnectorIdsForBlobSync(dataDir)) {
234 try {
235 const raw = await blobStore.get(docsOAuthBlobKey(connectorId), {
236 type: 'text',
237 consistency: 'strong',
238 });
239 if (typeof raw === 'string' && raw.trim()) {
240 const dest = oauthTokenVaultPath(dataDir, connectorId);
241 fs.mkdirSync(path.dirname(dest), { recursive: true });
242 fs.writeFileSync(dest, raw, 'utf8');
243 }
244 } catch {
245 /* skip */
246 }
247 }
248 }
249
250 /**
251 * @param {BlobStore|null|undefined} blobStore
252 * @param {string} dataDir
253 */
254 export async function persistDocsStoresToBlob(blobStore, dataDir) {
255 if (!blobStore || typeof blobStore.set !== 'function') return;
256
257 const storePath = getDocsStorePath(dataDir);
258 if (fs.existsSync(storePath)) {
259 try {
260 const raw = fs.readFileSync(storePath, 'utf8');
261 if (raw.trim()) await blobStore.set(DOCS_STORE_BLOB_KEY, raw);
262 } catch {
263 /* non-fatal */
264 }
265 }
266
267 const oauthDir = docsOAuthDir(dataDir);
268 if (!fs.existsSync(oauthDir)) return;
269 for (const name of fs.readdirSync(oauthDir)) {
270 if (!name.endsWith('.enc')) continue;
271 const connectorId = name.slice(0, -4);
272 try {
273 const raw = fs.readFileSync(path.join(oauthDir, name), 'utf8');
274 if (raw.trim()) await blobStore.set(docsOAuthBlobKey(connectorId), raw);
275 } catch {
276 /* non-fatal */
277 }
278 }
279 }
280
281 /**
282 * @template T
283 * @param {{
284 * blobStore: BlobStore|null|undefined,
285 * dataDir: string,
286 * persist?: boolean,
287 * run: () => T | Promise<T>,
288 * }} opts
289 * @returns {Promise<T>}
290 */
291 export async function withDocsBlobSync(opts) {
292 await hydrateDocsStoresFromBlob(opts.blobStore, opts.dataDir);
293 const result = await opts.run();
294 if (opts.persist !== false) {
295 await persistDocsStoresToBlob(opts.blobStore, opts.dataDir);
296 }
297 return result;
298 }
File History 1 commit
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 11 days ago