aaronrene / knowtation public
media-blob-store.mjs
248 lines 8.4 KB
Raw
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 9 days ago
1 /**
2 * Hosted bridge: persist media store files in Netlify Blobs (SEC-SEAM-MEDIA-b / SM-C6).
3 *
4 * Self-hosted Hub uses DATA_DIR files only. On Netlify, DATA_DIR is ephemeral; this
5 * module hydrates the media store files from Blobs before media propose/apply/list
6 * reads and persists them after mutating writes (capture blob-sync parity).
7 *
8 * Merge strategy (documented per SM-C6, CAPTURE-STORE-STALE-MERGE lessons):
9 * - `hub_attachment_external_refs.json`: id-keyed union per vault by attachment_id;
10 * on collision the newer `updated` wins (local wins ties). A warm lambda's stale
11 * local file must never mask a ref another instance persisted to Blobs.
12 * - `hub_media_import_consent.json`: id-keyed union per vault by consent_id; on
13 * collision a `revoked` record wins over `active` (fail-closed — a revoke on one
14 * instance must never be resurrected), otherwise newer `granted_at` wins.
15 * - `hub_media_connector_policy.json` / `hub_media_write_policy.json`: ops-managed
16 * documents — blob copy replaces local on hydrate (all mutations happen inside
17 * `withMediaBlobSync`, which persists after run, so blob ≥ local at hydrate time).
18 */
19
20 import fs from 'fs';
21 import path from 'path';
22
23 /** @typedef {{ get: (key: string, opts?: { type?: string }) => Promise<string|ArrayBuffer|null>, set: (key: string, value: string) => Promise<void> }} BlobStore */
24
25 export const MEDIA_EXTERNAL_REFS_FILENAME = 'hub_attachment_external_refs.json';
26 export const MEDIA_IMPORT_CONSENT_FILENAME = 'hub_media_import_consent.json';
27 export const MEDIA_CONNECTOR_POLICY_FILENAME = 'hub_media_connector_policy.json';
28 export const MEDIA_WRITE_POLICY_FILENAME = 'hub_media_write_policy.json';
29
30 export const MEDIA_BLOB_FILES = [
31 MEDIA_EXTERNAL_REFS_FILENAME,
32 MEDIA_IMPORT_CONSENT_FILENAME,
33 MEDIA_CONNECTOR_POLICY_FILENAME,
34 MEDIA_WRITE_POLICY_FILENAME,
35 ];
36
37 /**
38 * @param {string} filename
39 * @returns {string}
40 */
41 export function mediaBlobKey(filename) {
42 return `media/${filename}`;
43 }
44
45 /**
46 * @param {string} raw
47 * @returns {Record<string, unknown>|null}
48 */
49 function parseStore(raw) {
50 if (typeof raw !== 'string' || !raw.trim()) return null;
51 try {
52 const parsed = JSON.parse(raw);
53 return parsed && typeof parsed === 'object' && !Array.isArray(parsed) ? parsed : null;
54 } catch {
55 return null;
56 }
57 }
58
59 /**
60 * Union `vaults.<vid>.<collection>` maps from local and blob copies of a store file.
61 *
62 * @param {string} localRaw
63 * @param {string} blobRaw
64 * @param {string} collection - e.g. 'refs' | 'consents'
65 * @param {(localRec: Record<string, unknown>, blobRec: Record<string, unknown>) => Record<string, unknown>} pick
66 * @returns {string}
67 */
68 function mergeVaultKeyedStoreJson(localRaw, blobRaw, collection, pick) {
69 const local = parseStore(localRaw);
70 const blob = parseStore(blobRaw);
71 if (!local && !blob) return blobRaw || localRaw || '';
72 if (!local) return blobRaw;
73 if (!blob) return localRaw;
74
75 const localVaults =
76 local.vaults && typeof local.vaults === 'object' ? /** @type {Record<string, any>} */ (local.vaults) : {};
77 const blobVaults =
78 blob.vaults && typeof blob.vaults === 'object' ? /** @type {Record<string, any>} */ (blob.vaults) : {};
79
80 const vaultIds = new Set([...Object.keys(blobVaults), ...Object.keys(localVaults)]);
81 /** @type {Record<string, any>} */
82 const outVaults = {};
83 for (const vid of vaultIds) {
84 const localRows =
85 localVaults[vid]?.[collection] && typeof localVaults[vid][collection] === 'object'
86 ? localVaults[vid][collection]
87 : {};
88 const blobRows =
89 blobVaults[vid]?.[collection] && typeof blobVaults[vid][collection] === 'object'
90 ? blobVaults[vid][collection]
91 : {};
92 /** @type {Record<string, unknown>} */
93 const merged = { ...blobRows };
94 for (const [id, rec] of Object.entries(localRows)) {
95 if (!rec || typeof rec !== 'object') continue;
96 const existing = merged[id];
97 if (!existing || typeof existing !== 'object') {
98 merged[id] = rec;
99 continue;
100 }
101 merged[id] = pick(
102 /** @type {Record<string, unknown>} */ (rec),
103 /** @type {Record<string, unknown>} */ (existing),
104 );
105 }
106 outVaults[vid] = { ...(blobVaults[vid] ?? {}), ...(localVaults[vid] ?? {}), [collection]: merged };
107 }
108
109 return JSON.stringify({ ...blob, ...local, vaults: outVaults });
110 }
111
112 /**
113 * External refs: newest `updated` wins; local wins ties.
114 *
115 * @param {string} localRaw
116 * @param {string} blobRaw
117 * @returns {string}
118 */
119 export function mergeExternalRefStoreJson(localRaw, blobRaw) {
120 return mergeVaultKeyedStoreJson(localRaw, blobRaw, 'refs', (localRec, blobRec) => {
121 const tLocal = Date.parse(String(localRec.updated || localRec.created || '')) || 0;
122 const tBlob = Date.parse(String(blobRec.updated || blobRec.created || '')) || 0;
123 return tLocal >= tBlob ? localRec : blobRec;
124 });
125 }
126
127 /**
128 * Import consents: `revoked` wins over `active` (fail-closed); else newest granted_at
129 * wins; local wins ties.
130 *
131 * @param {string} localRaw
132 * @param {string} blobRaw
133 * @returns {string}
134 */
135 export function mergeImportConsentStoreJson(localRaw, blobRaw) {
136 return mergeVaultKeyedStoreJson(localRaw, blobRaw, 'consents', (localRec, blobRec) => {
137 const localRevoked = localRec.status === 'revoked';
138 const blobRevoked = blobRec.status === 'revoked';
139 if (localRevoked !== blobRevoked) return localRevoked ? localRec : blobRec;
140 const tLocal = Date.parse(String(localRec.granted_at || '')) || 0;
141 const tBlob = Date.parse(String(blobRec.granted_at || '')) || 0;
142 return tLocal >= tBlob ? localRec : blobRec;
143 });
144 }
145
146 const MERGERS = {
147 [MEDIA_EXTERNAL_REFS_FILENAME]: mergeExternalRefStoreJson,
148 [MEDIA_IMPORT_CONSENT_FILENAME]: mergeImportConsentStoreJson,
149 };
150
151 /**
152 * Load media store files from Blobs into DATA_DIR (hosted cold-start hydration).
153 *
154 * @param {BlobStore|null|undefined} blobStore
155 * @param {string} dataDir
156 */
157 export async function hydrateMediaStoresFromBlob(blobStore, dataDir) {
158 if (!blobStore || typeof blobStore.get !== 'function') return;
159 fs.mkdirSync(dataDir, { recursive: true });
160 for (const filename of MEDIA_BLOB_FILES) {
161 const fp = path.join(dataDir, filename);
162 try {
163 const raw = await blobStore.get(mediaBlobKey(filename), { type: 'text' });
164 if (typeof raw === 'string' && raw.trim()) {
165 const merger = MERGERS[filename];
166 if (merger && fs.existsSync(fp)) {
167 const merged = merger(fs.readFileSync(fp, 'utf8'), raw);
168 if (merged.trim()) fs.writeFileSync(fp, merged, 'utf8');
169 } else {
170 fs.writeFileSync(fp, raw, 'utf8');
171 }
172 }
173 } catch {
174 /* keep existing file or empty */
175 }
176 }
177 }
178
179 /**
180 * Write media store files from DATA_DIR to Blobs after a mutation.
181 *
182 * @param {BlobStore|null|undefined} blobStore
183 * @param {string} dataDir
184 */
185 export async function persistMediaStoresToBlob(blobStore, dataDir) {
186 if (!blobStore || typeof blobStore.set !== 'function') return;
187 for (const filename of MEDIA_BLOB_FILES) {
188 const fp = path.join(dataDir, filename);
189 if (!fs.existsSync(fp)) continue;
190 try {
191 const raw = fs.readFileSync(fp, 'utf8');
192 if (raw.trim()) {
193 await blobStore.set(mediaBlobKey(filename), raw);
194 }
195 } catch {
196 /* non-fatal */
197 }
198 }
199 }
200
201 /**
202 * Run a media store mutation with hosted Blob hydrate/persist when available.
203 * Persists only files whose content changed during the run.
204 *
205 * @template T
206 * @param {{
207 * blobStore: BlobStore|null|undefined,
208 * dataDir: string,
209 * run: () => T | Promise<T>,
210 * }} opts
211 * @returns {Promise<T>}
212 */
213 export async function withMediaBlobSync(opts) {
214 if (!opts.blobStore || typeof opts.blobStore.get !== 'function') {
215 return opts.run();
216 }
217
218 await hydrateMediaStoresFromBlob(opts.blobStore, opts.dataDir);
219
220 const before = new Map();
221 for (const filename of MEDIA_BLOB_FILES) {
222 const fp = path.join(opts.dataDir, filename);
223 if (fs.existsSync(fp)) {
224 try {
225 before.set(filename, fs.readFileSync(fp, 'utf8'));
226 } catch {
227 /* treat as absent */
228 }
229 }
230 }
231
232 const result = await opts.run();
233
234 for (const filename of MEDIA_BLOB_FILES) {
235 const fp = path.join(opts.dataDir, filename);
236 if (!fs.existsSync(fp)) continue;
237 try {
238 const raw = fs.readFileSync(fp, 'utf8');
239 if (raw.trim() && raw !== before.get(filename)) {
240 await opts.blobStore.set(mediaBlobKey(filename), raw);
241 }
242 } catch {
243 /* non-fatal */
244 }
245 }
246
247 return result;
248 }
File History 1 commit
sha256:700fafdd1afa490919f9515d660ca6e75456bcd5bb67513abcd8757a634c01f6 docs: record AIP-b SD-21 land (KN #308) Human 9 days ago