/**
 * Edge-cache middleware — collapses duplicate SSR work for unauth GET pages.
 *
 * Problem
 * -------
 * Cloudflare Bundled CPU tier (~10 ms ceiling) gets exceeded by SSR-heavy
 * routes (`/deal/{UUID}`, `/deals/all`, etc.) causing 503 `exceededCpu`
 * isolate kills under modest load (~5 req/s repro). Each unauth visitor
 * triggers identical work — same Neon HTTP queries, same renderToString —
 * even though the output is byte-identical.
 *
 * Solution
 * --------
 * Use the Workers Cache API (`caches.default`) to serve unauth GET responses
 * for designated public route prefixes from edge cache for 3600 s (1 h). One SSR
 * execution amortises across 1 h of duplicate traffic in the same colo.
 *
 * This is the canonical Cloudflare-recommended pattern for SSR amortisation
 * and is NOT a tier change — it stays on Bundled.
 *
 * Correctness invariants
 * ----------------------
 *  1. ONLY cache when both auth cookies absent (`multideal_at`, `multideal_rt`).
 *  2. ONLY cache GET (HEAD) — never mutating methods.
 *  3. ONLY cache 2xx responses (status 200).
 *  4. ONLY cache routes in `CACHEABLE_PREFIXES` + homepage `/`.
 *  5. NEVER cache a response that carries `Set-Cookie` — defensive against
 *     anything downstream issuing per-request cookies.
 *  6. Cache key is a synthetic GET Request whose URL has marketing/cb query
 *     params stripped and remaining params sorted (stable key). Locale is
 *     embedded in a header on the synthetic Request because Workers Cache
 *     API does NOT respect `Vary` for `cache.match` — the key must encode
 *     every dimension the response varies on.
 *  7. `cache.put` is scheduled on the execution context (waitUntil) so the
 *     response is not blocked by the write.
 *
 * Placement
 * ---------
 * Register FIRST in the middleware sequence so that on a cache hit we skip
 * the expensive session/CSRF/locale work entirely. On a miss the response
 * unwinds back through this middleware after all downstream middleware
 * (security-headers, locale, session, csrf, services) have applied their
 * mutations — so what we cache is the fully-headered final response.
 *
 * Invalidation
 * ------------
 * Catalog epoch in stored entry metadata enables soft-miss: stale entries are
 * served immediately while async regeneration writes fresh content to the same
 * key. BUILD_ID in the cache key still hard-invalidates on deploy (CSS-hash safety).
 * TTL is 3600 s (1 h) as a secondary bound.
 */

import { defineMiddleware } from 'astro:middleware';
import { cacheHeaders } from '@platform-modules/util/http-cache-headers';
import { captureCaught } from '@/lib/observability';
import { isTransientDoError, captureTransient } from '../observability/transient.js';
import { getCachedEpochSnapshot, getEpoch } from '@/server/cache/epoch.js';
import { env } from '@/server/env';

// Injected at build time by Vite define — orphans stale cache entries after each deploy.
declare const __BUILD_ID__: string;
const BUILD_ID = typeof __BUILD_ID__ !== 'undefined' ? __BUILD_ID__ : 'dev';

// Coalesces concurrent cold-start SSRs within one isolate: prevents N parallel
// cache misses from each spawning their own SSR (N× CPU burst → 1102).
// Keys are cache-key URL strings; values resolve with a cloneable Response.
const pendingSSR = new Map<string, Promise<Response>>();

// Coalesces concurrent stale-hit regens within one isolate: at most one background
// SSR per cache key while a regen is in flight.
const pendingRegen = new Set<string>();

// ---------------------------------------------------------------------------
// Configuration
// ---------------------------------------------------------------------------

/** Route prefixes for authed RenderDO offload (identity-scoped doKey, never edge-cached). */
const AUTHED_OFFLOAD_PREFIXES = ['/admin/', '/purchases/', '/vendor/'] as const;

/** Route prefixes whose unauth GET responses are safe to cache for 60 s. */
const CACHEABLE_PREFIXES = [
  '/deals',
  '/category/',
  '/business/',
  '/tag/',
  '/tags',
  '/search',
  '/stores',
  '/whats-left',
  '/he/whats-left',
  '/en/whats-left',
  '/he/deals',
  '/en/deals',
  '/group-deal/',
] as const;

/**
 * Regex-based allowlist for the 5 core public pages covered by Phase 5
 * personalization. Exported for use by other middleware / tests.
 */
type EdgeCacheRouteClass = 'home' | 'category' | 'tag' | 'deal' | 'search';

const EDGE_CACHED_PAGES: ReadonlyArray<readonly [EdgeCacheRouteClass, RegExp]> = [
  ['home', /^\/$/],
  ['category', /^\/category\/[^/]+\/?$/],
  ['tag', /^\/tag\/[^/]+\/?$/],
  ['deal', /^\/(?:[a-z]{2}\/)?deals?\/[^/]+\/?$/],
  ['search', /^\/search\/?$/],
];

export function isEdgeCacheablePage(pathname: string): boolean {
  return EDGE_CACHED_PAGES.some(([, re]) => re.test(pathname));
}

function getEdgeCacheRouteClass(pathname: string): EdgeCacheRouteClass | null {
  const match = EDGE_CACHED_PAGES.find(([, re]) => re.test(pathname));
  return match ? match[0] : null;
}

/** True when a stored entry's epoch is behind the current catalog epoch. */
export function isStale(storedEpoch: number, currentEpoch: number): boolean {
  return storedEpoch < currentEpoch;
}

function bakeStoreHeaders(responseHeaders: Headers, isAnonShell: boolean): Headers {
  const storeHeaders = new Headers(responseHeaders);
  if (isAnonShell) {
    storeHeaders.set(
      'Cache-Control',
      `public, max-age=0, s-maxage=${TTL_SECONDS}, stale-while-revalidate=${TTL_SECONDS}`,
    );
    if (!storeHeaders.has('Vary')) storeHeaders.set('Vary', 'Accept-Encoding');
  } else {
    storeHeaders.set(
      'Cache-Control',
      `public, max-age=0, s-maxage=${TTL_SECONDS}, stale-while-revalidate=${TTL_SECONDS}`,
    );
    if (!storeHeaders.has('Vary')) storeHeaders.set('Vary', 'Cookie, Accept-Encoding');
    else storeHeaders.append('Vary', 'Cookie');
  }
  return storeHeaders;
}

function applyWorkersCacheHeaders(
  responseHeaders: Headers,
  routeClass: EdgeCacheRouteClass,
  epoch: number,
): void {
  const headers = cacheHeaders({
    cdn: { maxAge: TTL_SECONDS, staleWhileRevalidate: TTL_SECONDS },
    tags: [`route:${routeClass}`, `epoch:${epoch}`],
  });
  // CDN headers only — `Cache-Control` stays owned by bakeStoreHeaders (its
  // s-maxage drives the internal Cache API TTL; the leaf's browser default
  // `max-age=0, must-revalidate` would kill stored-entry freshness).
  for (const key of ['Cloudflare-CDN-Cache-Control', 'Cache-Tag'] as const) {
    if (headers[key]) responseHeaders.set(key, headers[key]);
  }
}

/** Auth cookies — presence of either disables caching for the request. */
const AUTH_COOKIES = ['multideal_at', 'multideal_rt'] as const;

/** Query params that don't change content — stripped from cache key. */
const STRIPPED_QUERY_PARAMS = new Set([
  'cb',
  'fbclid',
  'gclid',
  'gbraid',
  'wbraid',
  'msclkid',
  'mc_eid',
  'mc_cid',
  'ref',
  'referrer',
  // UTM tracking (all variants)
  'utm_source',
  'utm_medium',
  'utm_campaign',
  'utm_content',
  'utm_term',
  'utm_id',
  // Additional ad-network click IDs
  'gad_source',
  'ttclid',
  'li_fat_id',
  '_gl',
]);

/** TTL for cached responses, in seconds. BUILD_ID-keyed cache invalidates on every deploy,
 * so stale-after-deploy is impossible — the only churn `s-maxage` controls is the per-shard
 * revalidation rate within a single build. 3600s = 1 h keeps each colo's cache warm between
 * deploy warmup runs; deploy.sh primes the cache post-deploy so the full hour is served. */
const TTL_SECONDS = 3600;
const DO_INNER_HEADER = 'x-md-do-inner';
const DO_INNER_HEADER_VALUE = '1';

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

export function isCacheablePath(pathname: string): boolean {
  // Homepage is cacheable — anonymous visitors get identical SSR output.
  if (pathname === '/') return true;
  return CACHEABLE_PREFIXES.some((p) =>
    p.endsWith('/') ? pathname.startsWith(p) : pathname === p || pathname.startsWith(`${p}/`),
  );
}

function isAuthedOffloadable(pathname: string): boolean {
  return AUTHED_OFFLOAD_PREFIXES.some((p) => pathname === p.slice(0, -1) || pathname.startsWith(p));
}

export function isRenderDoEnabled(
  enabled: string | undefined,
  secret: string | undefined,
): boolean {
  return enabled !== 'false' && Boolean(secret);
}

async function sha256hex(input: string): Promise<string> {
  const buf = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(input));
  return [...new Uint8Array(buf)].map((b) => b.toString(16).padStart(2, '0')).join('');
}

/** First non-empty string arg, else undefined. Treats '' as missing (never use `??` for cookie fallback). */
function firstNonEmpty(...vals: Array<string | undefined>): string | undefined {
  for (const v of vals) {
    if (typeof v === 'string' && v.length > 0) return v;
  }
  return undefined;
}

function hasAuthCookie(cookieHeader: string | null): boolean {
  if (!cookieHeader) return false;
  // Lightweight scan — full cookie parse is downstream's job (session.ts).
  for (const name of AUTH_COOKIES) {
    // Match `name=` at start-of-cookie boundary (after `;` or string start).
    if (
      cookieHeader.startsWith(`${name}=`) ||
      cookieHeader.includes(`; ${name}=`) ||
      cookieHeader.includes(`;${name}=`)
    ) {
      return true;
    }
  }
  return false;
}

function readCookieValue(cookieHeader: string | null, name: string): string | undefined {
  if (!cookieHeader) return undefined;
  const parts = cookieHeader.split(';');
  for (const raw of parts) {
    const eq = raw.indexOf('=');
    if (eq === -1) continue;
    const key = raw.slice(0, eq).trim();
    if (key === name) return raw.slice(eq + 1).trim();
  }
  return undefined;
}

/**
 * Resolve the effective locale BEFORE downstream localeMiddleware runs.
 * Precedence matches locale.ts: `?lang` query → `multideal_locale` cookie
 * → default `he`.
 */
function resolveLocale(url: URL, cookieHeader: string | null): 'he' | 'en' {
  const queryLang = url.searchParams.get('lang')?.toLowerCase();
  if (queryLang === 'he' || queryLang === 'en') return queryLang;
  const cookieLang = readCookieValue(cookieHeader, 'multideal_locale')?.toLowerCase();
  if (cookieLang === 'he' || cookieLang === 'en') return cookieLang;
  return 'he';
}

/**
 * Build a stable, normalised cache-key URL:
 *  - Drop tracking / cache-buster params.
 *  - Sort remaining params alphabetically for deterministic ordering.
 *  - Preserve pathname exactly (matters for /deals/all vs /deals).
 */
function normaliseCacheUrl(url: URL): URL {
  const out = new URL(url.toString());
  // Collect retained pairs.
  const retained: Array<[string, string]> = [];
  for (const [k, v] of out.searchParams.entries()) {
    if (STRIPPED_QUERY_PARAMS.has(k)) continue;
    retained.push([k, v]);
  }
  retained.sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
  // Wipe and re-add in sorted order.
  out.search = '';
  for (const [k, v] of retained) out.searchParams.append(k, v);
  // Drop fragment defensively (shouldn't be present server-side, but be safe).
  out.hash = '';
  return out;
}

function buildCacheKey(url: URL, locale: 'he' | 'en'): Request {
  const keyUrl = normaliseCacheUrl(url);
  // CF `caches.default` keys on the URL ONLY — custom request headers and the
  // response `Vary` are NOT honoured by `cache.match`. Every varying dimension
  // must therefore live in the URL itself:
  //  - `_v` = BUILD_ID → each deploy orphans stale entries (prevents FOUC from
  //    edge HTML referencing CSS hashes deleted by rebuild).
  //  - `_l` = locale → he/en never share a key. (A prior `x-edge-cache-locale`
  //    request header did NOT isolate them, so bare `/` with locale resolved
  //    from the cookie cross-served the wrong language to the other locale.)
  keyUrl.searchParams.append('_v', BUILD_ID);
  keyUrl.searchParams.append('_l', locale);
  return new Request(keyUrl.toString(), { method: 'GET' });
}

/**
 * Derive a stable R2 key for an HTML page response.
 * Only covers clean paths (no non-tracking query params) — filter/sort pages
 * with params miss R2 and fall through to SSR as normal.
 */
function buildR2HtmlKey(url: URL, locale: 'he' | 'en'): string | null {
  // Require clean path — no remaining query params after stripping tracking ones.
  const normalised = normaliseCacheUrl(url);
  if (normalised.search !== '' && normalised.search !== '?') return null;

  let path = normalised.pathname.replace(/^\/+/, '').replace(/\/+$/, '');
  // Strip locale prefix when it matches (e.g. /en/deals → deals under en/ key).
  if (path.startsWith(`${locale}/`)) path = path.slice(locale.length + 1);
  if (!path) path = 'index';
  return `html-cache/${locale}/${path}.html`;
}

export function shouldBypassCacheEpochRead(request: Request): boolean {
  return request.headers.get(DO_INNER_HEADER) === DO_INNER_HEADER_VALUE;
}

// ---------------------------------------------------------------------------
// Middleware
// ---------------------------------------------------------------------------

export const edgeCacheMiddleware = defineMiddleware(async (context, next) => {
  const { request } = context;

  // Inner pass of a /he/* → /en/* internal rewrite. The outer pass already
  // owns the cache key for the original URL; participating here a second
  // time would store the localized response under the rewritten URL's key
  // and pollute future direct visits to /en/* with /he/* content.
  if (context.locals.localeFixed) return next();

  // -- Filter: method + path ------------------------------------------------
  const method = request.method.toUpperCase();
  if (method !== 'GET' && method !== 'HEAD') return next();

  const url = new URL(request.url);

  // -- Authed RenderDO offload (identity-scoped, no shared cache) ------------
  // Every authed page view hits the DO (never edge-cached). Free tier = 100k DO req/day;
  // at scale, authed-page-views/day must stay below that or revisit.
  const cookieHeader = request.headers.get('cookie');
  const insideRenderDo = shouldBypassCacheEpochRead(request);
  const renderDoSecret = env.RENDER_DO_SECRET;
  const doEnabled = isRenderDoEnabled(env.RENDER_DO_ENABLED, renderDoSecret);
  const renderDo = (env as { RENDER_DO?: DurableObjectNamespace }).RENDER_DO;
  const idVal = firstNonEmpty(
    readCookieValue(cookieHeader, 'multideal_at'),
    readCookieValue(cookieHeader, 'multideal_rt'),
  );

  if (
    !insideRenderDo &&
    doEnabled &&
    renderDo &&
    hasAuthCookie(cookieHeader) &&
    idVal &&
    isAuthedOffloadable(url.pathname) &&
    !isEdgeCacheablePage(url.pathname)
  ) {
    const locale = resolveLocale(url, cookieHeader);
    const keyReq = buildCacheKey(url, locale);
    const idHash = await sha256hex(idVal);
    const doKey = `auth:${idHash}:${keyReq.url}`;
    const stub = renderDo.get(renderDo.idFromName(doKey));
    const internalReq = new Request(request.url, {
      method: request.method,
      headers: (() => {
        const h = new Headers(request.headers);
        h.set('x-md-render-do', renderDoSecret ?? '');
        return h;
      })(),
    });
    const inlineFallbackAuthed = async (doErr: unknown): Promise<Response> => {
      if (isTransientDoError(doErr)) {
        captureTransient(doErr, {
          scope: 'server.middleware.edge-cache.do-render-authed',
        });
      } else {
        captureCaught(doErr, {
          scope: 'server.middleware.edge-cache.do-render-authed',
          severity: 'error',
        });
      }
      try {
        const response = await next();
        response.headers.set('x-render-path', 'inline-fallback-authed');
        return response;
      } catch (inlineErr) {
        if (isTransientDoError(inlineErr)) {
          captureTransient(inlineErr, {
            scope: 'server.middleware.edge-cache.inline-fallback-authed',
          });
        } else {
          captureCaught(inlineErr, {
            scope: 'server.middleware.edge-cache.inline-fallback-authed',
            severity: 'error',
          });
        }
        return new Response('Service temporarily unavailable', { status: 503 });
      }
    };
    try {
      const doRes = await stub.fetch(internalReq);
      if (doRes.status >= 500) {
        return await inlineFallbackAuthed(new Error(`DO render returned ${doRes.status}`));
      }
      const setCookies = (doRes.headers as Headers & { getSetCookie(): string[] }).getSetCookie();
      const out = new Response(doRes.body, doRes);
      // Reassert every Set-Cookie explicitly — session re-mint + csrf must not collapse.
      out.headers.delete('Set-Cookie');
      for (const c of setCookies) out.headers.append('Set-Cookie', c);
      out.headers.set('x-render-path', 'do-render-authed');
      return out;
    } catch (doErr) {
      return await inlineFallbackAuthed(doErr);
    }
  }

  // Two cache modes:
  //  - Anon-shell pages (5 routes): identical HTML for guest + logged-in,
  //    cache regardless of auth.
  //  - Legacy cacheable prefixes (business/, tags, deals listing): unauth-only.
  const isAnonShell = isEdgeCacheablePage(url.pathname);
  const isLegacyCacheable = isCacheablePath(url.pathname);
  if (!isAnonShell && !isLegacyCacheable) return next();

  // -- Filter: auth presence (legacy paths only) ----------------------------
  if (!isAnonShell && hasAuthCookie(cookieHeader)) return next();

  // -- Workers Cache API availability --------------------------------------
  // `caches.default` exists only on Cloudflare Workers; absent in local
  // `astro dev`. Bail safely so dev never hits stale entries.
  const cacheGlobal = (globalThis as { caches?: { default?: Cache } }).caches;
  const cache = cacheGlobal?.default;
  if (!cache) return next();

  const locale = resolveLocale(url, cookieHeader);
  const cacheKey = buildCacheKey(url, locale);
  const currentEpoch = insideRenderDo ? getCachedEpochSnapshot() : await getEpoch();
  const routeClass =
    isAnonShell && !hasAuthCookie(cookieHeader) ? getEdgeCacheRouteClass(url.pathname) : null;
  const cfContext = (
    context.locals as unknown as { cfContext?: { waitUntil(p: Promise<unknown>): void } }
  ).cfContext;

  const scheduleRegen = () => {
    const regenKey = cacheKey.url;
    if (pendingRegen.has(regenKey)) return;
    pendingRegen.add(regenKey);

    const work = (async () => {
      try {
        const fresh = await next();
        if (fresh.status !== 200) return;
        const setCookie = fresh.headers.get('Set-Cookie') ?? '';
        if (setCookie.includes('multideal_at=') || setCookie.includes('multideal_rt=')) return;
        const freshCacheControl = (fresh.headers.get('Cache-Control') ?? '').toLowerCase();
        if (freshCacheControl.includes('no-store') || freshCacheControl.includes('private')) return;

        const shareableClone = new Response(fresh.clone().body, fresh);
        shareableClone.headers.delete('Set-Cookie');

        const storeHeaders = bakeStoreHeaders(shareableClone.headers, isAnonShell);
        storeHeaders.set('x-cache-epoch', String(currentEpoch));
        if (routeClass) applyWorkersCacheHeaders(storeHeaders, routeClass, currentEpoch);

        // Clone before reading body for cache.put — mirror MISS path.
        const cacheable = shareableClone.clone();

        const toStore = new Response(cacheable.body, {
          status: cacheable.status,
          statusText: cacheable.statusText,
          headers: storeHeaders,
        });

        await cache.put(cacheKey, toStore).catch((putErr) => {
          captureCaught(putErr, {
            scope: 'server.middleware.edge-cache.regen-put',
            severity: 'info',
          });
        });

        if (env.R2_BUCKET) {
          const r2WriteKey = buildR2HtmlKey(url, locale);
          if (r2WriteKey) {
            const htmlBuffer = await shareableClone.clone().arrayBuffer();
            const envelope = JSON.stringify({
              buildId: BUILD_ID,
              epoch: currentEpoch,
              headers: Object.fromEntries(storeHeaders.entries()),
            });
            const envBytes = new TextEncoder().encode(envelope + '\n');
            const combined = new Uint8Array(envBytes.byteLength + htmlBuffer.byteLength);
            combined.set(envBytes, 0);
            combined.set(new Uint8Array(htmlBuffer), envBytes.byteLength);
            await env.R2_BUCKET.put(r2WriteKey, combined.buffer as ArrayBuffer, {
              httpMetadata: { contentType: 'application/octet-stream' },
            });
          }
        }
      } catch (regenErr) {
        captureCaught(regenErr, { scope: 'server.middleware.edge-cache.regen', severity: 'info' });
      } finally {
        pendingRegen.delete(regenKey);
      }
    })();
    if (cfContext && typeof cfContext.waitUntil === 'function') {
      cfContext.waitUntil(work);
    } else {
      void work;
    }
  };

  // -- Try cache ------------------------------------------------------------
  // Always cache-first. (No force-ssr bypass: the old x-edge-cache-force-ssr
  // header was removed — it only skipped THIS isolate-local layer, never R2 or
  // the RenderDO render, so it never forced a fresh render and was honored from
  // any caller. Freshness is handled by epoch isStale→scheduleRegen below,
  // BUILD_ID keying, and event-driven prewarm — not a per-request bypass.)
  try {
    const hit = await cache.match(cacheKey);
    if (hit) {
      const hitEpoch = Number(hit.headers.get('x-cache-epoch') ?? '0');
      const cloned = new Response(hit.body, hit);
      if (isStale(hitEpoch, currentEpoch)) {
        cloned.headers.set('x-edge-cache', 'HIT-STALE');
        cloned.headers.set('x-render-path', 'edge-hit');
        scheduleRegen();
      } else {
        cloned.headers.set('x-edge-cache', 'HIT');
        cloned.headers.set('x-render-path', 'edge-hit');
      }
      return cloned;
    }
  } catch (readErr) {
    captureCaught(readErr, { scope: 'server.middleware.edge-cache.read', severity: 'info' });
  }

  // -- R2 HTML persistent cache (survives isolate eviction + deploys) --------
  if (env.R2_BUCKET) {
    const r2Key = buildR2HtmlKey(url, locale);
    if (r2Key) {
      try {
        const r2Obj = await env.R2_BUCKET.get(r2Key);
        if (r2Obj) {
          const raw = await r2Obj.arrayBuffer();
          const bytes = new Uint8Array(raw);
          let sep = -1;
          for (let i = 0; i < bytes.length; i++) {
            if (bytes[i] === 10) {
              sep = i;
              break;
            }
          }
          if (sep === -1) {
            // Corrupt entry (old format without header envelope) — delete + fall through.
            const cfCtxCorrupt = (
              context.locals as unknown as { cfContext?: { waitUntil(p: Promise<unknown>): void } }
            ).cfContext;
            const delCorrupt = env.R2_BUCKET.delete(r2Key).catch((delErr) => {
              captureCaught(delErr, {
                scope: 'server.middleware.edge-cache.r2-delete',
                severity: 'info',
              });
            });
            if (cfCtxCorrupt) cfCtxCorrupt.waitUntil(delCorrupt);
            else void delCorrupt;
          } else {
            try {
              const envelope = JSON.parse(new TextDecoder().decode(bytes.slice(0, sep))) as {
                buildId: string;
                epoch?: number;
                headers: Record<string, string>;
              };
              if (envelope.buildId !== BUILD_ID) {
                // Stale build. Delete + fall through.
                const del = env.R2_BUCKET.delete(r2Key).catch((delErr) => {
                  captureCaught(delErr, {
                    scope: 'server.middleware.edge-cache.r2-delete',
                    severity: 'info',
                  });
                });
                if (cfContext) cfContext.waitUntil(del);
                else void del;
              } else if (isStale(envelope.epoch ?? 0, currentEpoch)) {
                const htmlBytes = bytes.slice(sep + 1);
                const responseHeaders = new Headers(envelope.headers);
                responseHeaders.set('x-edge-cache', 'R2-HIT-STALE');
                responseHeaders.set('x-render-path', 'r2-hit');
                scheduleRegen();
                return new Response(htmlBytes, { status: 200, headers: responseHeaders });
              } else {
                const htmlBytes = bytes.slice(sep + 1);
                const responseHeaders = new Headers(envelope.headers);
                responseHeaders.set('x-edge-cache', 'R2-HIT');
                responseHeaders.set('x-render-path', 'r2-hit');
                return new Response(htmlBytes, { status: 200, headers: responseHeaders });
              }
            } catch (envelopeErr) {
              // Corrupt JSON envelope — delete + fall through.
              captureCaught(envelopeErr, {
                scope: 'server.middleware.edge-cache.r2-envelope',
                severity: 'info',
              });
              void env.R2_BUCKET.delete(r2Key).catch((delErr) => {
                captureCaught(delErr, {
                  scope: 'server.middleware.edge-cache.r2-delete',
                  severity: 'info',
                });
              });
            }
          }
        }
      } catch (r2ReadErr) {
        captureCaught(r2ReadErr, {
          scope: 'server.middleware.edge-cache.r2-read',
          severity: 'info',
        });
      }
    }
  }

  // -- Request coalescing ---------------------------------------------------
  // Multiple simultaneous cold-start requests for the same URL would each
  // trigger SSR independently — N× CPU burst on a fresh isolate reliably
  // exceeds the 10 ms free-tier ceiling (1102). Serialize them: the first
  // request renders and the rest clone its result.
  //
  // Security: only coalesce anonymous (no-cookie) requests. Authenticated
  // requests bypass the map entirely — they share no promises and never
  // receive another user's render.
  const canCoalesce = !hasAuthCookie(cookieHeader);
  const coalescingKey = cacheKey.url;

  if (canCoalesce) {
    const existingSSR = pendingSSR.get(coalescingKey);
    if (existingSSR !== undefined) {
      try {
        const shared = await existingSSR;
        const cloned = shared.clone();
        cloned.headers.set('x-edge-cache', 'HIT-COALESCED');
        cloned.headers.set('x-render-path', 'coalesced');
        return cloned;
      } catch (coalesceErr) {
        // First SSR errored — fall through to run our own below.
        captureCaught(coalesceErr, {
          scope: 'server.middleware.edge-cache.coalesce',
          severity: 'info',
        });
      }
    }
  }

  let resolveSSR!: (r: Response) => void;
  let rejectSSR!: (e: unknown) => void;
  let ssrPromise: Promise<Response> | undefined;
  if (canCoalesce) {
    ssrPromise = new Promise<Response>((res, rej) => {
      resolveSSR = res;
      rejectSSR = rej;
    });
    pendingSSR.set(coalescingKey, ssrPromise);
  }

  // -- Render origin --------------------------------------------------------
  let response: Response;
  // Recursion guard: x-md-do-inner is set by RenderDO.#render() before it re-enters
  // handle(). When present, we ARE the middleware pass running INSIDE the DO (30s
  // budget) — dispatching to the DO again would deadlock on its #inflight promise.
  // Render inline instead. (Sentinel is not a secret: an external caller fabricating
  // it merely opts out of DO offload for their own request — risks 1102 on their
  // request only, no privilege gained.)

  if (
    !insideRenderDo &&
    doEnabled &&
    renderDo &&
    (request.method === 'GET' || request.method === 'HEAD')
  ) {
    // Offload SSR to RenderDO (30s CPU budget). GET/HEAD only — this block is already
    // inside the cacheable-route miss path, which never handles mutating methods.
    // doKey = normalised cache key (path + sorted surviving params + locale).
    // Per-key DO instance = global request coalescing across isolates for this URL.
    // NOTE: cacheKey.url embeds _v=${BUILD_ID}, so each deploy maps URLs to FRESH DO
    // instances — expected; idle DO instances cost nothing.
    const doKey = coalescingKey; // same as coalescingKey (cacheKey.url)
    const doId = renderDo.idFromName(doKey);
    const stub = renderDo.get(doId);
    const internalReq = new Request(request.url, {
      method: request.method,
      headers: (() => {
        const h = new Headers(request.headers);
        h.set('x-md-render-do', env.RENDER_DO_SECRET ?? '');
        return h;
      })(),
    });
    try {
      const doRes = await stub.fetch(internalReq);
      // stub.fetch() returns a platform Response with IMMUTABLE headers — every
      // header mutation in the post-render block below (x-edge-cache, x-render-path,
      // bake/coalesce clones) would throw TypeError. Wrap into a mutable copy
      // (same pattern as `new Response(hit.body, hit)` used elsewhere in this file).
      response = new Response(doRes.body, doRes);
      response.headers.set('x-render-path', 'do-render');
    } catch (doErr) {
      // DO failure: unblock isolate-local coalesced waiters FIRST — otherwise
      // concurrent requests parked on pendingSSR.get(coalescingKey) hang forever.
      if (canCoalesce) {
        rejectSSR(doErr instanceof Error ? doErr : new Error(String(doErr)));
        pendingSSR.delete(coalescingKey);
      }
      if (isTransientDoError(doErr)) {
        captureTransient(doErr, { scope: 'server.middleware.edge-cache.do-render' });
      } else {
        captureCaught(doErr, {
          scope: 'server.middleware.edge-cache.do-render',
          severity: 'error',
        });
      }

      // Failure ladder: stale cache → stale R2 envelope → inline best-effort → 503.
      try {
        const staleHit = await cache.match(cacheKey);
        if (staleHit) {
          const staleClone = new Response(staleHit.body, staleHit);
          staleClone.headers.set('x-edge-cache', 'HIT-STALE-FALLBACK');
          staleClone.headers.set('x-render-path', 'stale-fallback');
          return staleClone;
        }
      } catch (staleErr) {
        captureCaught(staleErr, {
          scope: 'server.middleware.edge-cache.stale-fallback',
          severity: 'info',
        });
      }
      // (Stale R2 envelope step: reuse existing R2 read helper ignoring epoch — clean paths only.)

      // Inline best-effort (may 1102, but stale > nothing)
      try {
        response = await next();
        response.headers.set('x-render-path', 'inline-fallback');
      } catch (inlineErr) {
        captureCaught(inlineErr, {
          scope: 'server.middleware.edge-cache.inline-fallback',
          severity: 'error',
        });
        return new Response('Service temporarily unavailable', { status: 503 });
      }
    }
  } else {
    // Kill-switch / inside-DO path: inline SSR (original behaviour)
    try {
      response = await next();
      response.headers.set('x-render-path', insideRenderDo ? 'do-inner' : 'inline-fallback');
    } catch (renderErr) {
      if (canCoalesce) {
        rejectSSR(renderErr);
        pendingSSR.delete(coalescingKey);
      }
      throw renderErr;
    }
  }

  // Only cache fresh 200 responses. Non-200s are errors or redirects.
  if (response.status !== 200) {
    if (canCoalesce) {
      rejectSSR(new Error(`non-200: ${response.status}`));
      pendingSSR.delete(coalescingKey);
    }
    return response;
  }

  // Block if response sets AUTH cookies — these are per-user sessions.
  // Tracking cookies (multideal_anon_sid) are allowed: they're stripped from
  // cached/coalesced copies so each user still gets their own on first SSR.
  const setCookieHeader = response.headers.get('Set-Cookie') ?? '';
  const setsAuthCookie =
    setCookieHeader.includes('multideal_at=') || setCookieHeader.includes('multideal_rt=');
  if (setsAuthCookie) {
    if (canCoalesce) {
      rejectSSR(new Error('set-auth-cookie'));
      pendingSSR.delete(coalescingKey);
    }
    return response;
  }

  // Honour the response's own cacheability directive (RFC 9111). edge-cache
  // otherwise overrides Cache-Control with its baked s-maxage, so a `no-store` /
  // `private` render would still be stored on a shared cache. Veto it. This is
  // the SOLE guard for the admin `?__preview=1` draft (index.astro sets
  // `no-store` for previewDraft): the render still flows through RenderDO (CPU-
  // protected) but is never stored, so anon visitors never read a cached draft.
  // Also catches any future uncacheable HTML.
  const responseCacheControl = (response.headers.get('Cache-Control') ?? '').toLowerCase();
  if (responseCacheControl.includes('no-store') || responseCacheControl.includes('private')) {
    if (canCoalesce) {
      rejectSSR(new Error('no-store'));
      pendingSSR.delete(coalescingKey);
    }
    return response;
  }

  // Build a shareable clone with Set-Cookie stripped — coalesced waiters get
  // page content without inheriting the first requester's tracking cookie.
  const shareableClone = new Response(response.clone().body, response);
  shareableClone.headers.delete('Set-Cookie');

  // Resolve coalesced waiters with the stripped clone before consuming body.
  if (canCoalesce) {
    resolveSSR(shareableClone.clone());
    pendingSSR.delete(coalescingKey);
  }

  // Clone before reading body for cache.put — we MUST return the original
  // response unconsumed so the client gets the body (including their Set-Cookie).
  const cacheable = shareableClone.clone();

  // Bake Cache-Control headers onto the *stored* copy.
  const storeHeaders = bakeStoreHeaders(cacheable.headers, isAnonShell);
  storeHeaders.set('x-cache-epoch', String(currentEpoch));
  if (routeClass) applyWorkersCacheHeaders(storeHeaders, routeClass, currentEpoch);

  const toStore = new Response(cacheable.body, {
    status: cacheable.status,
    statusText: cacheable.statusText,
    headers: storeHeaders,
  });

  // Tag the response actually returned to the client so we can see misses.
  response.headers.set('x-edge-cache', 'MISS');
  if (routeClass) applyWorkersCacheHeaders(response.headers, routeClass, currentEpoch);

  // `cache.put` may run after response is sent — schedule on waitUntil if
  // Astro's CF runtime hook is available, else fire-and-forget.
  const putPromise = cache.put(cacheKey, toStore).catch((putErr) => {
    captureCaught(putErr, { scope: 'server.middleware.edge-cache.put', severity: 'info' });
  });
  if (cfContext && typeof cfContext.waitUntil === 'function') {
    cfContext.waitUntil(putPromise);
  } else {
    // Detach so an unhandled rejection on a missing-runtime fallback path
    // doesn't crash the isolate.
    void putPromise;
  }

  // R2 write — persist HTML for cold-isolate resilience.
  // Only for cacheable paths with no query params (warm-routes fetches are always clean).
  if (env.R2_BUCKET) {
    const r2WriteKey = buildR2HtmlKey(url, locale);
    if (r2WriteKey) {
      const r2PutPromise = (async () => {
        const htmlBuffer = await shareableClone.clone().arrayBuffer();
        const envelope = JSON.stringify({
          buildId: BUILD_ID,
          epoch: currentEpoch,
          headers: Object.fromEntries(storeHeaders.entries()),
        });
        const envBytes = new TextEncoder().encode(envelope + '\n');
        const combined = new Uint8Array(envBytes.byteLength + htmlBuffer.byteLength);
        combined.set(envBytes, 0);
        combined.set(new Uint8Array(htmlBuffer), envBytes.byteLength);
        await env.R2_BUCKET.put(r2WriteKey, combined.buffer as ArrayBuffer, {
          httpMetadata: { contentType: 'application/octet-stream' },
        });
      })().catch((r2PutErr) => {
        captureCaught(r2PutErr, { scope: 'server.middleware.edge-cache.r2-put', severity: 'info' });
      });
      if (cfContext && typeof cfContext.waitUntil === 'function') {
        cfContext.waitUntil(r2PutPromise);
      } else {
        void r2PutPromise;
      }
    }
  }

  return response;
});