/**
 * musehub-pack-receiver — Cloudflare Worker
 *
 * Receives Muse Wire Protocol (MWP) object packs from muse CLI clients.
 * Objects are written directly to R2 using the native R2 binding (~1 ms per PUT,
 * no EC2→R2 hop) then the object IDs are registered in the MuseHub Postgres DB
 * via a lightweight internal callback to the MuseHub API.
 *
 * Latency profile (per pack of 250 objects):
 *   Previous path  EC2 receives → asyncio gather → boto3 PUT×250 → respond    ~875 ms/PUT
 *   This Worker:   Worker receives → R2 PUT×250 via Promise.all → notify MuseHub  ~3–8 ms/PUT
 *
 * Bindings required in Cloudflare dashboard:
 *   R2:      BUCKET                    (the musehub-objects R2 bucket)
 *   Secret:  MUSEHUB_INTERNAL_KEY      (shared secret for Worker→MuseHub calls)
 *   Var:     MUSEHUB_ORIGIN            (e.g. "https://staging.musehub.ai")
 *
 * Routes:
 *   POST /{owner}/{slug}/push/object-pack   mirrors the MuseHub wire endpoint
 *
 * The client sends the exact same MWP msgpack body it would send to MuseHub.
 * The Authorization header (MSign) is forwarded verbatim to MuseHub for
 * identity verification — the Worker itself only checks MUSEHUB_INTERNAL_KEY
 * on the callback leg, not on the inbound client request.
 */

// ── msgpack decode (subset sufficient for MWP pack bodies) ────────────────────
// We only decode: map, str, bin, uint, bool.  No floats, no ext.

function msgpackDecode(buf) {
  const view = new DataView(buf);
  let pos = 0;

  function readStr(len) {
    const bytes = new Uint8Array(buf, pos, len);
    pos += len;
    return new TextDecoder().decode(bytes);
  }

  function readBin(len) {
    const slice = buf.slice(pos, pos + len);
    pos += len;
    return slice;
  }

  function read() {
    const b = view.getUint8(pos++);

    // positive fixint
    if ((b & 0x80) === 0) return b;
    // fixstr
    if ((b & 0xe0) === 0xa0) return readStr(b & 0x1f);
    // fixmap
    if ((b & 0xf0) === 0x80) {
      const n = b & 0x0f;
      const m = {};
      for (let i = 0; i < n; i++) { const k = read(); m[k] = read(); }
      return m;
    }
    // fixarray
    if ((b & 0xf0) === 0x90) {
      const n = b & 0x0f;
      const a = [];
      for (let i = 0; i < n; i++) a.push(read());
      return a;
    }
    // negative fixint
    if ((b & 0xe0) === 0xe0) return b - 256;

    switch (b) {
      case 0xc0: return null;
      case 0xc2: return false;
      case 0xc3: return true;

      // bin8 / bin16 / bin32
      case 0xc4: { const n = view.getUint8(pos++); return readBin(n); }
      case 0xc5: { const n = view.getUint16(pos); pos += 2; return readBin(n); }
      case 0xc6: { const n = view.getUint32(pos); pos += 4; return readBin(n); }

      // uint8 / uint16 / uint32 / uint64
      case 0xcc: { const v = view.getUint8(pos++); return v; }
      case 0xcd: { const v = view.getUint16(pos); pos += 2; return v; }
      case 0xce: { const v = view.getUint32(pos); pos += 4; return v; }
      case 0xcf: { const hi = view.getUint32(pos); const lo = view.getUint32(pos + 4); pos += 8; return hi * 2**32 + lo; }

      // int8 / int16 / int32
      case 0xd0: { const v = view.getInt8(pos++); return v; }
      case 0xd1: { const v = view.getInt16(pos); pos += 2; return v; }
      case 0xd2: { const v = view.getInt32(pos); pos += 4; return v; }

      // str8 / str16 / str32
      case 0xd9: { const n = view.getUint8(pos++); return readStr(n); }
      case 0xda: { const n = view.getUint16(pos); pos += 2; return readStr(n); }
      case 0xdb: { const n = view.getUint32(pos); pos += 4; return readStr(n); }

      // array16 / array32
      case 0xdc: { const n = view.getUint16(pos); pos += 2; const a = []; for (let i=0;i<n;i++) a.push(read()); return a; }
      case 0xdd: { const n = view.getUint32(pos); pos += 4; const a = []; for (let i=0;i<n;i++) a.push(read()); return a; }

      // map16 / map32
      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; }
      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; }

      default:
        throw new Error(`msgpack: unsupported byte 0x${b.toString(16)} at pos ${pos-1}`);
    }
  }

  return read();
}

// ── msgpack encode (only what we need for responses) ──────────────────────────

function msgpackEncode(value) {
  const bufs = [];
  function enc(v) {
    if (v === null || v === undefined) { bufs.push(new Uint8Array([0xc0])); return; }
    if (typeof v === 'boolean') { bufs.push(new Uint8Array([v ? 0xc3 : 0xc2])); return; }
    if (typeof v === 'number') {
      if (Number.isInteger(v) && v >= 0 && v <= 0xffffffff) {
        if (v <= 0x7f) { bufs.push(new Uint8Array([v])); return; }
        if (v <= 0xff) { bufs.push(new Uint8Array([0xcc, v])); return; }
        if (v <= 0xffff) { const b=new Uint8Array(3); b[0]=0xcd; new DataView(b.buffer).setUint16(1,v); bufs.push(b); return; }
        const b=new Uint8Array(5); b[0]=0xce; new DataView(b.buffer).setUint32(1,v); bufs.push(b); return;
      }
    }
    if (typeof v === 'string') {
      const encoded = new TextEncoder().encode(v);
      const len = encoded.length;
      if (len <= 31) { bufs.push(new Uint8Array([0xa0 | len])); bufs.push(encoded); }
      else if (len <= 0xff) { bufs.push(new Uint8Array([0xd9, len])); bufs.push(encoded); }
      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); }
      else { const h=new Uint8Array(5); h[0]=0xdb; new DataView(h.buffer).setUint32(1,len); bufs.push(h); bufs.push(encoded); }
      return;
    }
    if (v instanceof ArrayBuffer || ArrayBuffer.isView(v)) {
      const bytes = v instanceof ArrayBuffer ? new Uint8Array(v) : new Uint8Array(v.buffer, v.byteOffset, v.byteLength);
      const len = bytes.length;
      if (len <= 0xff) bufs.push(new Uint8Array([0xc4, len]));
      else if (len <= 0xffff) { const h=new Uint8Array(3); h[0]=0xc5; new DataView(h.buffer).setUint16(1,len); bufs.push(h); }
      else { const h=new Uint8Array(5); h[0]=0xc6; new DataView(h.buffer).setUint32(1,len); bufs.push(h); }
      bufs.push(bytes); return;
    }
    if (Array.isArray(v)) {
      const len = v.length;
      if (len <= 15) bufs.push(new Uint8Array([0x90 | len]));
      else if (len <= 0xffff) { const h=new Uint8Array(3); h[0]=0xdc; new DataView(h.buffer).setUint16(1,len); bufs.push(h); }
      else { const h=new Uint8Array(5); h[0]=0xdd; new DataView(h.buffer).setUint32(1,len); bufs.push(h); }
      for (const item of v) enc(item);
      return;
    }
    if (typeof v === 'object') {
      const keys = Object.keys(v);
      const len = keys.length;
      if (len <= 15) bufs.push(new Uint8Array([0x80 | len]));
      else if (len <= 0xffff) { const h=new Uint8Array(3); h[0]=0xde; new DataView(h.buffer).setUint16(1,len); bufs.push(h); }
      else { const h=new Uint8Array(5); h[0]=0xdf; new DataView(h.buffer).setUint32(1,len); bufs.push(h); }
      for (const k of keys) { enc(k); enc(v[k]); }
      return;
    }
    throw new Error(`msgpack: cannot encode ${typeof v}`);
  }
  enc(value);
  const total = bufs.reduce((s, b) => s + b.length, 0);
  const out = new Uint8Array(total);
  let off = 0;
  for (const b of bufs) { out.set(b, off); off += b.length; }
  return out.buffer;
}

// ── R2 key derivation — mirrors S3Backend._key() in backends.py ───────────────

function r2Key(objectId) {
  // Strip "sha256:" prefix; store as "objects/<hex>"
  const hex = objectId.startsWith('sha256:') ? objectId.slice(7) : objectId;
  return `objects/${hex}`;
}

// ── Worker entry point ────────────────────────────────────────────────────────

export default {
  async fetch(request, env) {
    // Only accept POST to /{owner}/{slug}/push/object-pack
    if (request.method !== 'POST') {
      return new Response('Method Not Allowed', { status: 405 });
    }

    const url = new URL(request.url);
    const parts = url.pathname.split('/').filter(Boolean);
    // Expect: ["<owner>", "<slug>", "push", "object-pack"]
    if (parts.length < 4 || parts[2] !== 'push' || parts[3] !== 'object-pack') {
      return new Response('Not Found', { status: 404 });
    }
    const owner = parts[0];
    const slug  = parts[1];

    // Read and decode the msgpack body
    const bodyBuf = await request.arrayBuffer();
    let pack;
    try {
      pack = msgpackDecode(bodyBuf);
    } catch (e) {
      return new Response(`Bad Request: ${e.message}`, { status: 400 });
    }

    if (!pack || !Array.isArray(pack.objects)) {
      return new Response('Bad Request: missing objects array', { status: 400 });
    }

    // ── Step 1: write all objects to R2 in parallel (native binding, ~1 ms each) ──
    const stored = [];
    const skipped = [];

    await Promise.all(pack.objects.map(async (obj) => {
      if (!obj.object_id || obj.content == null) return;
      const key = r2Key(obj.object_id);
      // R2.put is idempotent — safe to re-push existing objects.
      // head() first to avoid unnecessary writes for large repos with many
      // shared objects.  Skip head() for small packs (≤50) to keep it fast.
      let needsWrite = true;
      if (pack.objects.length > 50) {
        const existing = await env.BUCKET.head(key);
        if (existing) { skipped.push(obj.object_id); needsWrite = false; }
      }
      if (needsWrite) {
        const content = obj.content instanceof ArrayBuffer
          ? obj.content
          : obj.content.buffer.slice(obj.content.byteOffset, obj.content.byteOffset + obj.content.byteLength);
        await env.BUCKET.put(key, content, {
          httpMetadata: { contentType: 'application/octet-stream' },
          customMetadata: { object_id: obj.object_id, path: obj.path || '' },
        });
        stored.push({ object_id: obj.object_id, path: obj.path || '', size: content.byteLength });
      }
    }));

    // ── Step 2: register stored objects in MuseHub DB (one batch HTTP call) ────
    // Forward the original MSign Authorization header so MuseHub can verify
    // the pusher's identity and check repo write permission.
    if (stored.length > 0) {
      const registerUrl = `${env.MUSEHUB_ORIGIN}/internal/register-objects`;
      const registerBody = JSON.stringify({
        owner,
        slug,
        objects: stored,
      });
      const regResp = await fetch(registerUrl, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': request.headers.get('Authorization') || '',
          'X-Musehub-Worker-Key': env.MUSEHUB_INTERNAL_KEY,
        },
        body: registerBody,
      });
      if (!regResp.ok) {
        const errText = await regResp.text();
        console.error(`register-objects failed: ${regResp.status} ${errText}`);
        return new Response(`Internal Error: DB registration failed`, { status: 502 });
      }
      const regJson = await regResp.json();
      // regJson.stored / regJson.skipped are the DB-level counts (dedup)
      const dbStored  = regJson.stored  ?? stored.length;
      const dbSkipped = regJson.skipped ?? skipped.length;
      const accept = request.headers.get('Accept') || '';
      if (accept.includes('application/x-msgpack')) {
        return new Response(msgpackEncode({ stored: dbStored, skipped: dbSkipped }), {
          headers: { 'Content-Type': 'application/x-msgpack' },
        });
      }
      return Response.json({ stored: dbStored, skipped: dbSkipped });
    }

    // Nothing new was stored — all objects were already in R2.
    const accept = request.headers.get('Accept') || '';
    const totalSkipped = pack.objects.filter(o => o.object_id && o.content != null).length;
    if (accept.includes('application/x-msgpack')) {
      return new Response(msgpackEncode({ stored: 0, skipped: totalSkipped }), {
        headers: { 'Content-Type': 'application/x-msgpack' },
      });
    }
    return Response.json({ stored: 0, skipped: totalSkipped });
  },
};