aaronrene / knowtation public
external-agent-blob-store.mjs
204 lines 6.5 KB
Raw
sha256:b5f647cb9c409f563d4671fe3fc05ddea01fabfed9b41fc11cb923588e1c1baf mirror: GitHub Phase A durable MCP OAuth (#270) Human minor ⚠ breaking 10 days ago
1 /**
2 * Hosted bridge: persist external protocol state in Netlify Blobs.
3 *
4 * Self-hosted bridge uses DATA_DIR files only. On Netlify, DATA_DIR is ephemeral;
5 * this module hydrates files from Blobs before protocol handlers and persists after writes.
6 */
7
8 import fs from 'fs';
9 import path from 'path';
10 import { FLOW_STORE_FILENAME } from '../../lib/flow/flow-store.mjs';
11 import { hydrateDelegationStoresFromBlob } from './delegation-blob-store.mjs';
12
13 /** @typedef {{ get: (key: string, opts?: { type?: string }) => Promise<string|ArrayBuffer|null>, set: (key: string, value: string) => Promise<void> }} BlobStore */
14
15 export const EXTERNAL_PROTOCOL_BLOB_FILES = [
16 FLOW_STORE_FILENAME,
17 'hub_external_protocol_idempotency.json',
18 'hub_delegation_audit.json',
19 ];
20
21 /**
22 * @param {string} filename
23 * @returns {string}
24 */
25 export function externalProtocolBlobKey(filename) {
26 return `external-protocol/${filename}`;
27 }
28
29 /**
30 * Merge local and blob flow stores without losing fresher task writes.
31 * Task apply may land on a warm lambda before the external-protocol blob reflects it.
32 *
33 * @param {string} localRaw
34 * @param {string} blobRaw
35 * @returns {string}
36 */
37 export function mergeFlowStoreJson(localRaw, blobRaw) {
38 const parse = (raw) => {
39 if (typeof raw !== 'string' || !raw.trim()) return null;
40 try {
41 return JSON.parse(raw);
42 } catch {
43 return null;
44 }
45 };
46
47 const local = parse(localRaw);
48 const blob = parse(blobRaw);
49 if (!local && !blob) return blobRaw || localRaw || '';
50 if (!local) return blobRaw;
51 if (!blob) return localRaw;
52
53 /** @param {unknown[]} localArr @param {unknown[]} blobArr @param {string} idField */
54 const mergeById = (localArr, blobArr, idField) => {
55 /** @type {Map<string, Record<string, unknown>>} */
56 const byId = new Map();
57 for (const rec of blobArr || []) {
58 if (rec && typeof rec === 'object' && typeof rec[idField] === 'string') {
59 byId.set(rec[idField], /** @type {Record<string, unknown>} */ (rec));
60 }
61 }
62 for (const rec of localArr || []) {
63 if (!rec || typeof rec !== 'object' || typeof rec[idField] !== 'string') continue;
64 const row = /** @type {Record<string, unknown>} */ (rec);
65 const existing = byId.get(row[idField]);
66 if (!existing) {
67 byId.set(row[idField], row);
68 continue;
69 }
70 const tLocal = Date.parse(String(row.updated || row.created || '')) || 0;
71 const tBlob = Date.parse(String(existing.updated || existing.created || '')) || 0;
72 byId.set(row[idField], tLocal >= tBlob ? row : existing);
73 }
74 return [...byId.values()];
75 };
76
77 if (!local.vaults) local.vaults = {};
78 for (const [vaultId, blobVault] of Object.entries(blob.vaults || {})) {
79 const localVault =
80 local.vaults[vaultId] && typeof local.vaults[vaultId] === 'object'
81 ? /** @type {Record<string, unknown>} */ (local.vaults[vaultId])
82 : {};
83 const blobVaultObj =
84 blobVault && typeof blobVault === 'object'
85 ? /** @type {Record<string, unknown>} */ (blobVault)
86 : {};
87 local.vaults[vaultId] = {
88 ...blobVaultObj,
89 ...localVault,
90 tasks: mergeById(
91 /** @type {unknown[]} */ (localVault.tasks),
92 /** @type {unknown[]} */ (blobVaultObj.tasks),
93 'task_id',
94 ),
95 task_loops: mergeById(
96 /** @type {unknown[]} */ (localVault.task_loops),
97 /** @type {unknown[]} */ (blobVaultObj.task_loops),
98 'loop_id',
99 ),
100 };
101 }
102
103 return JSON.stringify(local);
104 }
105
106 /**
107 * Load external protocol store files from Blobs into DATA_DIR (hosted cold-start hydration).
108 *
109 * @param {BlobStore|null|undefined} blobStore
110 * @param {string} dataDir
111 */
112 export async function hydrateExternalProtocolStoresFromBlob(blobStore, dataDir) {
113 if (!blobStore || typeof blobStore.get !== 'function') return;
114 fs.mkdirSync(dataDir, { recursive: true });
115 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
116 const fp = path.join(dataDir, filename);
117 try {
118 const raw = await blobStore.get(externalProtocolBlobKey(filename), { type: 'text' });
119 if (typeof raw === 'string' && raw.trim()) {
120 if (filename === FLOW_STORE_FILENAME && fs.existsSync(fp)) {
121 const localRaw = fs.readFileSync(fp, 'utf8');
122 const merged = mergeFlowStoreJson(localRaw, raw);
123 if (merged.trim()) {
124 fs.writeFileSync(fp, merged, 'utf8');
125 }
126 } else {
127 fs.writeFileSync(fp, raw, 'utf8');
128 }
129 }
130 } catch {
131 /* keep existing file or empty */
132 }
133 }
134 }
135
136 /**
137 * Write external protocol store files from DATA_DIR to Blobs after a mutation.
138 *
139 * @param {BlobStore|null|undefined} blobStore
140 * @param {string} dataDir
141 */
142 export async function persistExternalProtocolStoresToBlob(blobStore, dataDir) {
143 if (!blobStore || typeof blobStore.set !== 'function') return;
144 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
145 const fp = path.join(dataDir, filename);
146 if (!fs.existsSync(fp)) continue;
147 try {
148 const raw = fs.readFileSync(fp, 'utf8');
149 if (raw.trim()) {
150 await blobStore.set(externalProtocolBlobKey(filename), raw);
151 }
152 } catch {
153 /* non-fatal */
154 }
155 }
156 }
157
158 /**
159 * Run an external protocol mutation with hosted Blob hydrate/persist when available.
160 * Pushes them back if changed.
161 *
162 * @template T
163 * @param {{
164 * blobStore: BlobStore|null|undefined,
165 * dataDir: string,
166 * run: () => T | Promise<T>,
167 * }} opts
168 * @returns {Promise<T>}
169 */
170 export async function withExternalProtocolBlobSync(opts) {
171 if (!opts.blobStore || typeof opts.blobStore.get !== 'function') {
172 return opts.run();
173 }
174
175 await hydrateExternalProtocolStoresFromBlob(opts.blobStore, opts.dataDir);
176 await hydrateDelegationStoresFromBlob(opts.blobStore, opts.dataDir);
177
178 // Snapshot before run to avoid unnecessary blob writes if unmodified
179 const before = new Map();
180 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
181 const fp = path.join(opts.dataDir, filename);
182 if (fs.existsSync(fp)) {
183 try { before.set(filename, fs.readFileSync(fp, 'utf8')); } catch {}
184 }
185 }
186
187 const result = await opts.run();
188
189 // Persist if changed
190 for (const filename of EXTERNAL_PROTOCOL_BLOB_FILES) {
191 const fp = path.join(opts.dataDir, filename);
192 if (!fs.existsSync(fp)) continue;
193 try {
194 const raw = fs.readFileSync(fp, 'utf8');
195 if (raw.trim() && raw !== before.get(filename)) {
196 await opts.blobStore.set(externalProtocolBlobKey(filename), raw);
197 }
198 } catch {
199 /* non-fatal */
200 }
201 }
202
203 return result;
204 }
File History 1 commit
sha256:b5f647cb9c409f563d4671fe3fc05ddea01fabfed9b41fc11cb923588e1c1baf mirror: GitHub Phase A durable MCP OAuth (#270) Human minor ⚠ 10 days ago