gabriel / musehub public
worker.js javascript
284 lines 12.1 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 153 days ago
1 /**
2 * musehub-pack-receiver — Cloudflare Worker
3 *
4 * Receives Muse Wire Protocol (MWP) object packs from muse CLI clients.
5 * Objects are written directly to R2 using the native R2 binding (~1 ms per PUT,
6 * no EC2→R2 hop) then the object IDs are registered in the MuseHub Postgres DB
7 * via a lightweight internal callback to the MuseHub API.
8 *
9 * Latency profile (per pack of 250 objects):
10 * Previous path EC2 receives → asyncio gather → boto3 PUT×250 → respond ~875 ms/PUT
11 * This Worker: Worker receives → R2 PUT×250 via Promise.all → notify MuseHub ~3–8 ms/PUT
12 *
13 * Bindings required in Cloudflare dashboard:
14 * R2: BUCKET (the musehub-objects R2 bucket)
15 * Secret: MUSEHUB_INTERNAL_KEY (shared secret for Worker→MuseHub calls)
16 * Var: MUSEHUB_ORIGIN (e.g. "https://staging.musehub.ai")
17 *
18 * Routes:
19 * POST /{owner}/{slug}/push/object-pack mirrors the MuseHub wire endpoint
20 *
21 * The client sends the exact same MWP msgpack body it would send to MuseHub.
22 * The Authorization header (MSign) is forwarded verbatim to MuseHub for
23 * identity verification — the Worker itself only checks MUSEHUB_INTERNAL_KEY
24 * on the callback leg, not on the inbound client request.
25 */
26
27 // ── msgpack decode (subset sufficient for MWP pack bodies) ────────────────────
28 // We only decode: map, str, bin, uint, bool. No floats, no ext.
29
30 function msgpackDecode(buf) {
31 const view = new DataView(buf);
32 let pos = 0;
33
34 function readStr(len) {
35 const bytes = new Uint8Array(buf, pos, len);
36 pos += len;
37 return new TextDecoder().decode(bytes);
38 }
39
40 function readBin(len) {
41 const slice = buf.slice(pos, pos + len);
42 pos += len;
43 return slice;
44 }
45
46 function read() {
47 const b = view.getUint8(pos++);
48
49 // positive fixint
50 if ((b & 0x80) === 0) return b;
51 // fixstr
52 if ((b & 0xe0) === 0xa0) return readStr(b & 0x1f);
53 // fixmap
54 if ((b & 0xf0) === 0x80) {
55 const n = b & 0x0f;
56 const m = {};
57 for (let i = 0; i < n; i++) { const k = read(); m[k] = read(); }
58 return m;
59 }
60 // fixarray
61 if ((b & 0xf0) === 0x90) {
62 const n = b & 0x0f;
63 const a = [];
64 for (let i = 0; i < n; i++) a.push(read());
65 return a;
66 }
67 // negative fixint
68 if ((b & 0xe0) === 0xe0) return b - 256;
69
70 switch (b) {
71 case 0xc0: return null;
72 case 0xc2: return false;
73 case 0xc3: return true;
74
75 // bin8 / bin16 / bin32
76 case 0xc4: { const n = view.getUint8(pos++); return readBin(n); }
77 case 0xc5: { const n = view.getUint16(pos); pos += 2; return readBin(n); }
78 case 0xc6: { const n = view.getUint32(pos); pos += 4; return readBin(n); }
79
80 // uint8 / uint16 / uint32 / uint64
81 case 0xcc: { const v = view.getUint8(pos++); return v; }
82 case 0xcd: { const v = view.getUint16(pos); pos += 2; return v; }
83 case 0xce: { const v = view.getUint32(pos); pos += 4; return v; }
84 case 0xcf: { const hi = view.getUint32(pos); const lo = view.getUint32(pos + 4); pos += 8; return hi * 2**32 + lo; }
85
86 // int8 / int16 / int32
87 case 0xd0: { const v = view.getInt8(pos++); return v; }
88 case 0xd1: { const v = view.getInt16(pos); pos += 2; return v; }
89 case 0xd2: { const v = view.getInt32(pos); pos += 4; return v; }
90
91 // str8 / str16 / str32
92 case 0xd9: { const n = view.getUint8(pos++); return readStr(n); }
93 case 0xda: { const n = view.getUint16(pos); pos += 2; return readStr(n); }
94 case 0xdb: { const n = view.getUint32(pos); pos += 4; return readStr(n); }
95
96 // array16 / array32
97 case 0xdc: { const n = view.getUint16(pos); pos += 2; const a = []; for (let i=0;i<n;i++) a.push(read()); return a; }
98 case 0xdd: { const n = view.getUint32(pos); pos += 4; const a = []; for (let i=0;i<n;i++) a.push(read()); return a; }
99
100 // map16 / map32
101 case 0xde: { const n = view.getUint16(pos); pos += 2; const m = {}; for (let i=0;i<n;i++){const k=read();m[k]=read();} return m; }
102 case 0xdf: { const n = view.getUint32(pos); pos += 4; const m = {}; for (let i=0;i<n;i++){const k=read();m[k]=read();} return m; }
103
104 default:
105 throw new Error(`msgpack: unsupported byte 0x${b.toString(16)} at pos ${pos-1}`);
106 }
107 }
108
109 return read();
110 }
111
112 // ── msgpack encode (only what we need for responses) ──────────────────────────
113
114 function msgpackEncode(value) {
115 const bufs = [];
116 function enc(v) {
117 if (v === null || v === undefined) { bufs.push(new Uint8Array([0xc0])); return; }
118 if (typeof v === 'boolean') { bufs.push(new Uint8Array([v ? 0xc3 : 0xc2])); return; }
119 if (typeof v === 'number') {
120 if (Number.isInteger(v) && v >= 0 && v <= 0xffffffff) {
121 if (v <= 0x7f) { bufs.push(new Uint8Array([v])); return; }
122 if (v <= 0xff) { bufs.push(new Uint8Array([0xcc, v])); return; }
123 if (v <= 0xffff) { const b=new Uint8Array(3); b[0]=0xcd; new DataView(b.buffer).setUint16(1,v); bufs.push(b); return; }
124 const b=new Uint8Array(5); b[0]=0xce; new DataView(b.buffer).setUint32(1,v); bufs.push(b); return;
125 }
126 }
127 if (typeof v === 'string') {
128 const encoded = new TextEncoder().encode(v);
129 const len = encoded.length;
130 if (len <= 31) { bufs.push(new Uint8Array([0xa0 | len])); bufs.push(encoded); }
131 else if (len <= 0xff) { bufs.push(new Uint8Array([0xd9, len])); bufs.push(encoded); }
132 else if (len <= 0xffff) { const h=new Uint8Array(3); h[0]=0xda; new DataView(h.buffer).setUint16(1,len); bufs.push(h); bufs.push(encoded); }
133 else { const h=new Uint8Array(5); h[0]=0xdb; new DataView(h.buffer).setUint32(1,len); bufs.push(h); bufs.push(encoded); }
134 return;
135 }
136 if (v instanceof ArrayBuffer || ArrayBuffer.isView(v)) {
137 const bytes = v instanceof ArrayBuffer ? new Uint8Array(v) : new Uint8Array(v.buffer, v.byteOffset, v.byteLength);
138 const len = bytes.length;
139 if (len <= 0xff) bufs.push(new Uint8Array([0xc4, len]));
140 else if (len <= 0xffff) { const h=new Uint8Array(3); h[0]=0xc5; new DataView(h.buffer).setUint16(1,len); bufs.push(h); }
141 else { const h=new Uint8Array(5); h[0]=0xc6; new DataView(h.buffer).setUint32(1,len); bufs.push(h); }
142 bufs.push(bytes); return;
143 }
144 if (Array.isArray(v)) {
145 const len = v.length;
146 if (len <= 15) bufs.push(new Uint8Array([0x90 | len]));
147 else if (len <= 0xffff) { const h=new Uint8Array(3); h[0]=0xdc; new DataView(h.buffer).setUint16(1,len); bufs.push(h); }
148 else { const h=new Uint8Array(5); h[0]=0xdd; new DataView(h.buffer).setUint32(1,len); bufs.push(h); }
149 for (const item of v) enc(item);
150 return;
151 }
152 if (typeof v === 'object') {
153 const keys = Object.keys(v);
154 const len = keys.length;
155 if (len <= 15) bufs.push(new Uint8Array([0x80 | len]));
156 else if (len <= 0xffff) { const h=new Uint8Array(3); h[0]=0xde; new DataView(h.buffer).setUint16(1,len); bufs.push(h); }
157 else { const h=new Uint8Array(5); h[0]=0xdf; new DataView(h.buffer).setUint32(1,len); bufs.push(h); }
158 for (const k of keys) { enc(k); enc(v[k]); }
159 return;
160 }
161 throw new Error(`msgpack: cannot encode ${typeof v}`);
162 }
163 enc(value);
164 const total = bufs.reduce((s, b) => s + b.length, 0);
165 const out = new Uint8Array(total);
166 let off = 0;
167 for (const b of bufs) { out.set(b, off); off += b.length; }
168 return out.buffer;
169 }
170
171 // ── R2 key derivation — mirrors S3Backend._key() in backends.py ───────────────
172
173 function r2Key(objectId) {
174 // Strip "sha256:" prefix; store as "objects/<hex>"
175 const hex = objectId.startsWith('sha256:') ? objectId.slice(7) : objectId;
176 return `objects/${hex}`;
177 }
178
179 // ── Worker entry point ────────────────────────────────────────────────────────
180
181 export default {
182 async fetch(request, env) {
183 // Only accept POST to /{owner}/{slug}/push/object-pack
184 if (request.method !== 'POST') {
185 return new Response('Method Not Allowed', { status: 405 });
186 }
187
188 const url = new URL(request.url);
189 const parts = url.pathname.split('/').filter(Boolean);
190 // Expect: ["<owner>", "<slug>", "push", "object-pack"]
191 if (parts.length < 4 || parts[2] !== 'push' || parts[3] !== 'object-pack') {
192 return new Response('Not Found', { status: 404 });
193 }
194 const owner = parts[0];
195 const slug = parts[1];
196
197 // Read and decode the msgpack body
198 const bodyBuf = await request.arrayBuffer();
199 let pack;
200 try {
201 pack = msgpackDecode(bodyBuf);
202 } catch (e) {
203 return new Response(`Bad Request: ${e.message}`, { status: 400 });
204 }
205
206 if (!pack || !Array.isArray(pack.objects)) {
207 return new Response('Bad Request: missing objects array', { status: 400 });
208 }
209
210 // ── Step 1: write all objects to R2 in parallel (native binding, ~1 ms each) ──
211 const stored = [];
212 const skipped = [];
213
214 await Promise.all(pack.objects.map(async (obj) => {
215 if (!obj.object_id || obj.content == null) return;
216 const key = r2Key(obj.object_id);
217 // R2.put is idempotent — safe to re-push existing objects.
218 // head() first to avoid unnecessary writes for large repos with many
219 // shared objects. Skip head() for small packs (≤50) to keep it fast.
220 let needsWrite = true;
221 if (pack.objects.length > 50) {
222 const existing = await env.BUCKET.head(key);
223 if (existing) { skipped.push(obj.object_id); needsWrite = false; }
224 }
225 if (needsWrite) {
226 const content = obj.content instanceof ArrayBuffer
227 ? obj.content
228 : obj.content.buffer.slice(obj.content.byteOffset, obj.content.byteOffset + obj.content.byteLength);
229 await env.BUCKET.put(key, content, {
230 httpMetadata: { contentType: 'application/octet-stream' },
231 customMetadata: { object_id: obj.object_id, path: obj.path || '' },
232 });
233 stored.push({ object_id: obj.object_id, path: obj.path || '', size: content.byteLength });
234 }
235 }));
236
237 // ── Step 2: register stored objects in MuseHub DB (one batch HTTP call) ────
238 // Forward the original MSign Authorization header so MuseHub can verify
239 // the pusher's identity and check repo write permission.
240 if (stored.length > 0) {
241 const registerUrl = `${env.MUSEHUB_ORIGIN}/internal/register-objects`;
242 const registerBody = JSON.stringify({
243 owner,
244 slug,
245 objects: stored,
246 });
247 const regResp = await fetch(registerUrl, {
248 method: 'POST',
249 headers: {
250 'Content-Type': 'application/json',
251 'Authorization': request.headers.get('Authorization') || '',
252 'X-Musehub-Worker-Key': env.MUSEHUB_INTERNAL_KEY,
253 },
254 body: registerBody,
255 });
256 if (!regResp.ok) {
257 const errText = await regResp.text();
258 console.error(`register-objects failed: ${regResp.status} ${errText}`);
259 return new Response(`Internal Error: DB registration failed`, { status: 502 });
260 }
261 const regJson = await regResp.json();
262 // regJson.stored / regJson.skipped are the DB-level counts (dedup)
263 const dbStored = regJson.stored ?? stored.length;
264 const dbSkipped = regJson.skipped ?? skipped.length;
265 const accept = request.headers.get('Accept') || '';
266 if (accept.includes('application/x-msgpack')) {
267 return new Response(msgpackEncode({ stored: dbStored, skipped: dbSkipped }), {
268 headers: { 'Content-Type': 'application/x-msgpack' },
269 });
270 }
271 return Response.json({ stored: dbStored, skipped: dbSkipped });
272 }
273
274 // Nothing new was stored — all objects were already in R2.
275 const accept = request.headers.get('Accept') || '';
276 const totalSkipped = pack.objects.filter(o => o.object_id && o.content != null).length;
277 if (accept.includes('application/x-msgpack')) {
278 return new Response(msgpackEncode({ stored: 0, skipped: totalSkipped }), {
279 headers: { 'Content-Type': 'application/x-msgpack' },
280 });
281 }
282 return Response.json({ stored: 0, skipped: totalSkipped });
283 },
284 };
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 153 days ago