fix(core-ts): scope asset_refs by site and skip rebuilds on unchanged imports (#196)

.changeset/asset-refs-by-site.md

@@ -0,0 +1,11 @@
+---
+"@cloudhal/core-ts": patch
+---
+
+`asset_refs` is scoped by site (RFC-0045 §2.5, rollout step 0), and an unchanged import no longer rebuilds it (#196).
+
+- An import, push or pull planned `unchanged` used to rebuild the site's `asset_refs` from rows read at plan time. An edit that committed in between could lose its new row, so the asset it started using could be deleted and showed as unused. The rebuild now runs only in a batch that also writes the node set.
+- The rebuild records only references to assets of the same site, and splits its `INSERT` so each statement stays below 90 KB (large sites could exceed D1's statement limit). Its statements count against the import's 1,000-statement cap.
+- Deleting an asset and the media library's "unused" filter look only at references from the asset's own site, so a foreign archive naming another site's asset no longer blocks its delete or hides it.
+- A publish that changes an asset now also purges the pages that show it through a typed `image` or `richText` field (RFC-0041), not only through covers, bodies and sections.
+- While a field rename runs, entries not rewritten yet still hold the old key. Both the rebuild and the purge planner now read those entries through the definition's `migration.fromKey`, as the renderer does. An archive exported mid-rename keeps those entries' rows, and an asset change purges their pages.

packages/core-ts/src/assets.ts

@@ -31,7 +31,7 @@ import { ID_PREFIX } from "./ids.ts";
 import { RASTER_TYPES } from "./media/image.ts";
 import { type InspectOptions, inspectUpload } from "./media/ingest.ts";
 import { type AssetRefUse, assetRefsOf, canonicalImageRefs } from "./media/refs.ts";
-import { refsRebuild } from "./media/refs-rebuild.ts";
+import { refsRebuild, UNUSED_ASSET_SQL } from "./media/refs-rebuild.ts";
 import { sniff } from "./media/sniff.ts";
 import {
   assetKindFor,
@@ -319,7 +319,7 @@ export class AssetEngine {
       where.push("json_extract(body, '$.uploadedBy.kind') = ?");
       params.push(q.uploadedBy);
     }
-    if (q.unused) where.push("NOT EXISTS (SELECT 1 FROM asset_refs r WHERE r.asset_id = nodes.id)");
+    if (q.unused) where.push(UNUSED_ASSET_SQL);
     if (q.missingAlt)
       where.push(
         "json_extract(body, '$.kind') = 'image' AND json_extract(body, '$.alt') IS NULL AND json_extract(body, '$.decorative') IS NULL",
@@ -864,8 +864,8 @@ export class AssetEngine {
    */
   async deleteAsset(a: Asset): Promise<void> {
     const r = await this.#sql.run(
-      "DELETE FROM nodes WHERE id = ? AND rev = ? AND kind = 'asset' AND site_id = ? AND NOT EXISTS (SELECT 1 FROM asset_refs WHERE asset_id = ?)",
-      [a.id, a.rev, a.siteId, a.id],
+      "DELETE FROM nodes WHERE id = ? AND rev = ? AND kind = 'asset' AND site_id = ? AND NOT EXISTS (SELECT 1 FROM asset_refs WHERE site_id = ? AND asset_id = ?)",
+      [a.id, a.rev, a.siteId, a.siteId, a.id],
     );
     if (r.changes === 1) return;
     const usage = (await this.assetUsage(a.siteId, [a.id])).get(a.id);

packages/core-ts/src/cache-purge-assets.ts

@@ -0,0 +1,74 @@
+/**
+ * The purge planner's asset rule (RFC-0004 §7.4): an asset that left the live snapshot or changed
+ * purges its media URLs (`cloudhal-asset-<id>`) and, when it changed, the responses that show it,
+ * since they resolve its URL, alt, caption and size when they render: the node tag of every entry
+ * and page that refers to it, and the term tag (with ancestors) of every term that does. An
+ * entry's typed fields count too (RFC-0045 §2.5): an `image` field's value and a `richText`
+ * field's `asset:` links, read through the target snapshot's own field definitions (and, while a
+ * rename runs, through its `migration.fromKey`, as the renderer reads them). A newly added
+ * asset purges nothing (404s are never cached, and whatever starts showing it changed itself).
+ * Part of `planCachePurge` (`cache-purge.ts`); reads through its counted, hash-keyed `read`.
+ */
+import type { Asset, ContentEntry, NodeRef, Site, Snapshot } from "@cloudhal/core-contracts";
+import { assetCacheTag, nodeCacheTag, termCacheTag } from "./cache-tags.ts";
+import { fieldAssetRefs, migratedFieldValues, siteFieldDefinitions } from "./media/field-refs.ts";
+import { type AssetRefUse, assetRefsOf } from "./media/refs.ts";
+
+export interface AssetPurgeInput {
+  /** Every node that differs between the two snapshots. */
+  readonly changed: readonly { before?: NodeRef; after?: NodeRef }[];
+  /** The snapshot going live. */
+  readonly to: Snapshot;
+  /** The planner's node reader (counted against `maxReads`, cached by hash). */
+  readonly read: <T>(ref: NodeRef) => Promise<T>;
+  /** A term and its ancestors in either snapshot. */
+  readonly withAncestors: (id: string) => string[];
+  /** Collected tags and log lines, added to in place. */
+  readonly tags: Set<string>;
+  readonly why: string[];
+}
+
+/** Adds the asset rule's tags and log lines to `input.tags` and `input.why`. */
+export async function planAssetPurge(input: AssetPurgeInput): Promise<void> {
+  const { tags, why, read } = input;
+  const shown = new Set<string>();
+  for (const c of input.changed) {
+    if (!c.before || c.before.kind !== "asset") continue;
+    const id = c.before.id;
+    tags.add(assetCacheTag(id));
+    if (!c.after) {
+      why.push(`asset ${id} was removed`);
+      continue;
+    }
+    const [a, b] = await Promise.all([read<Asset>(c.before), read<Asset>(c.after)]);
+    if (a.slug !== b.slug || a.status !== b.status) why.push(`asset ${id}: slug or status changed`);
+    else why.push(`asset ${id} changed`);
+    shown.add(id);
+  }
+  if (shown.size === 0) return;
+  const refsOf = await referencesIn(input.to, read);
+  for (const ref of input.to.nodes) {
+    if (ref.kind !== "contentEntry" && ref.kind !== "page" && ref.kind !== "term") continue;
+    const uses = refsOf(ref.kind, await read<unknown>(ref));
+    if (!uses.some((r) => shown.has(r.assetId))) continue;
+    if (ref.kind === "term")
+      for (const id of input.withAncestors(ref.id)) tags.add(termCacheTag(id));
+    else tags.add(nodeCacheTag(ref.id));
+  }
+}
+
+/** A node's asset references in `snapshot`: its own, and an entry's typed fields' by its defs. */
+async function referencesIn(
+  snapshot: Snapshot,
+  read: AssetPurgeInput["read"],
+): Promise<(kind: string, body: unknown) => AssetRefUse[]> {
+  const siteRef = snapshot.nodes.find((n) => n.kind === "site" && n.id === snapshot.siteId);
+  const defs = siteRef ? siteFieldDefinitions(await read<Site>(siteRef)) : {};
+  return (kind, body) => {
+    const refs = assetRefsOf(kind, body);
+    if (kind !== "contentEntry") return refs;
+    const entry = body as ContentEntry;
+    const own = defs[entry.collection] ?? [];
+    return [...refs, ...fieldAssetRefs(own, migratedFieldValues(own, entry.fields))];
+  };
+}

packages/core-ts/src/cache-purge.ts

@@ -19,32 +19,26 @@
  *
  * - an **asset** (RFC-0004 §7.4) that left the live snapshot or changed: its media URLs
  *   (`cloudhal-asset-<id>`), and — when it changed — the responses that show it: the node tag of
- *   every entry and page that refers to it, and the term tag of every term (archives, chips) that
- *   does. A newly added asset purges nothing (404s are never cached, and whatever starts showing
- *   it changed itself). Media never carry the site tag, so the site-wide purges above leave them.
+ *   every entry and page that refers to it (typed fields included, RFC-0045 §2.5), and the term tag
+ *   of every term (archives, chips) that does (`cache-purge-assets.ts`). A newly added asset purges
+ *   nothing (404s are never cached, and whatever starts showing it changed itself). Media never
+ *   carry the site tag, so the site-wide purges above leave them.
  *
  * Falls back to purging everything when there is no previous snapshot, a snapshot or node cannot
  * be read, reading the diff would take more than `maxReads` node reads, or the result has more
  * than `maxTags` tags.
  */
 import type {
-  Asset,
   ContentEntry,
   NodeRef,
   Site,
   Snapshot,
   Taxonomy,
   Term,
 } from "@cloudhal/core-contracts";
-import {
-  assetCacheTag,
-  listCacheTag,
-  nodeCacheTag,
-  siteCacheTag,
-  termCacheTag,
-} from "./cache-tags.ts";
+import { planAssetPurge } from "./cache-purge-assets.ts";
+import { listCacheTag, nodeCacheTag, siteCacheTag, termCacheTag } from "./cache-tags.ts";
 import { canonicalJson } from "./hash.ts";
-import { assetRefsOf } from "./media/refs.ts";
 
 export interface SnapshotReader {
   getSnapshot(id: string): Promise<Snapshot | undefined>;
@@ -202,34 +196,8 @@ export async function planCachePurge(
     const tags = new Set<string>();
     const why: string[] = [];
 
-    // Media: an asset that left the live snapshot or changed (§7.4). Its own URLs, and the
-    // responses that show it (they resolve its URL, alt, caption and size when they render).
-    const assetChanges = changed.filter((c) => (c.after ?? c.before)?.kind === "asset" && c.before);
-    if (assetChanges.length > 0) {
-      const shown = new Set<string>();
-      for (const c of assetChanges) {
-        const id = (c.before as NodeRef).id;
-        tags.add(assetCacheTag(id));
-        if (!c.after) {
-          why.push(`asset ${id} was removed`);
-          continue;
-        }
-        const [a, b] = await Promise.all([read<Asset>(c.before as NodeRef), read<Asset>(c.after)]);
-        if (a.slug !== b.slug || a.status !== b.status)
-          why.push(`asset ${id}: slug or status changed`);
-        else why.push(`asset ${id} changed`);
-        shown.add(id);
-      }
-      if (shown.size > 0)
-        for (const ref of (to as Snapshot).nodes) {
-          if (ref.kind !== "contentEntry" && ref.kind !== "page" && ref.kind !== "term") continue;
-          const body = await read<unknown>(ref);
-          if (!assetRefsOf(ref.kind, body).some((r) => shown.has(r.assetId))) continue;
-          if (ref.kind === "term")
-            for (const id of withAncestors(ref.id)) tags.add(termCacheTag(id));
-          else tags.add(nodeCacheTag(ref.id));
-        }
-    }
+    // Media (§7.4): an asset that left the live snapshot or changed (`cache-purge-assets.ts`).
+    await planAssetPurge({ changed, to: to as Snapshot, read, withAncestors, tags, why });
 
     for (const c of termChanges) {
       const a = c.before ? await read<Term>(c.before) : undefined;

packages/core-ts/src/media/field-refs.ts

@@ -35,6 +35,26 @@ export function fieldAssetRefs(
   );
 }
 
+/**
+ * `values` as readers see them while a rename runs (RFC-0041 §6, `resolveFields`' `read`): a
+ * defined key the entry does not hold yet reads its definition's `migration.fromKey`. For the
+ * whole-site readers of stored bodies (the `asset_refs` rebuild, the purge planner's asset rule);
+ * a write's own checks read the values it writes.
+ */
+export function migratedFieldValues(
+  defs: readonly FieldDefinition[],
+  values: Readonly<Record<string, unknown>> | undefined,
+): Readonly<Record<string, unknown>> | undefined {
+  if (!values || !defs.some((d) => d.migration?.fromKey)) return values;
+  const out: Record<string, unknown> = { ...values };
+  for (const def of defs) {
+    const from = def.migration?.fromKey;
+    if (from && !Object.hasOwn(values, def.key) && Object.hasOwn(values, from))
+      out[def.key] = values[from];
+  }
+  return out;
+}
+
 /** Field definitions per collection of a site node (`{}` when it has none). */
 export function siteFieldDefinitions(site: {
   collections?: Record<string, { fields?: FieldDefinition[] | undefined } | undefined> | undefined;

packages/core-ts/src/media/refs-rebuild.ts

@@ -1,38 +1,76 @@
 /**
  * `asset_refs` for a whole site from its node rows (RFC-0004 §4.4): sync import and repair. Entry
  * rows also count their typed fields' references (RFC-0041 §2) when the site's own row is among
- * them. Worker-safe.
+ * them, read through a running rename's `migration.fromKey` like `resolveFields` (an archive
+ * exported mid-operation keeps the rows of entries not rewritten yet). Only references to assets
+ * of the same site are recorded (RFC-0045 §2.5): the `INSERT` joins `nodes`, so a foreign archive
+ * naming another site's asset neither blocks that asset's delete nor hides it from that site's
+ * unused list. Run it only in a batch that also writes the site's node set (after the node
+ * `INSERT`s): a rebuild beside an unchanged node set would replace rows a concurrent write
+ * committed with plan-time ones (#196). Worker-safe.
  */
 import type { ContentEntry, Site } from "@cloudhal/core-contracts";
 import type { SqlStatement } from "@cloudhal/storage";
-import { fieldAssetRefs, siteFieldDefinitions } from "./field-refs.ts";
+import { fieldAssetRefs, migratedFieldValues, siteFieldDefinitions } from "./field-refs.ts";
 import { assetRefsOf } from "./refs.ts";
 
-/** `asset_refs` for a whole site from node rows (DELETE + one INSERT). */
+/** Each rebuild `INSERT` stays below this many bytes, SQL and parameters together (D1: 100 KB). */
+export const REFS_STATEMENT_BYTES = 90_000;
+
+/** `listAssets`' `unused` filter: no reference from a node of the asset's own site (RFC-0045 §2.5). */
+export const UNUSED_ASSET_SQL =
+  "NOT EXISTS (SELECT 1 FROM asset_refs r WHERE r.site_id = nodes.site_id AND r.asset_id = nodes.id)";
+
+const INSERT_SQL = `INSERT INTO asset_refs (site_id, asset_id, node_id, node_kind, field)
+  SELECT ?, a.id, json_extract(r.value, '$.n'), json_extract(r.value, '$.k'), json_extract(r.value, '$.f')
+  FROM json_each(?) r JOIN nodes a ON a.id = json_extract(r.value, '$.a') AND a.site_id = ? AND a.kind = 'asset'`;
+
+const encoder = new TextEncoder();
+
+/** The JSON-encoded rows, in `json_each` arrays of at most `budget` bytes each. */
+function chunks(items: readonly string[], budget: number): string[] {
+  const out: string[] = [];
+  let open: string[] = [];
+  let bytes = 2;
+  for (const item of items) {
+    const size = encoder.encode(item).length + 1;
+    if (open.length > 0 && bytes + size > budget) {
+      out.push(`[${open.join(",")}]`);
+      open = [];
+      bytes = 2;
+    }
+    open.push(item);
+    bytes += size;
+  }
+  if (open.length > 0) out.push(`[${open.join(",")}]`);
+  return out;
+}
+
+/** `asset_refs` for a whole site from node rows: one `DELETE`, then `INSERT`s below 90 KB each. */
 export function refsRebuild(
   siteId: string,
   rows: readonly { id: string; kind: string; body: string }[],
 ): SqlStatement[] {
   const siteRow = rows.find((r) => r.kind === "site" && r.id === siteId);
   const defs = siteRow ? siteFieldDefinitions(JSON.parse(siteRow.body) as Site) : {};
-  const all: { a: string; n: string; k: string; f: string }[] = [];
+  const all = new Set<string>();
   for (const r of rows) {
     const node = JSON.parse(r.body);
     const refs = assetRefsOf(r.kind, node);
     if (r.kind === "contentEntry") {
       const entry = node as ContentEntry;
-      refs.push(...fieldAssetRefs(defs[entry.collection] ?? [], entry.fields));
+      const own = defs[entry.collection] ?? [];
+      refs.push(...fieldAssetRefs(own, migratedFieldValues(own, entry.fields)));
     }
-    for (const ref of refs) all.push({ a: ref.assetId, n: r.id, k: r.kind, f: ref.field });
+    for (const ref of refs)
+      all.add(JSON.stringify({ a: ref.assetId, n: r.id, k: r.kind, f: ref.field }));
   }
-  const out: SqlStatement[] = [
+  const fixed = encoder.encode(INSERT_SQL + siteId + siteId).length;
+  return [
     { sql: "DELETE FROM asset_refs WHERE site_id = ?", params: [siteId] },
+    ...chunks([...all], REFS_STATEMENT_BYTES - fixed).map((rows) => ({
+      sql: INSERT_SQL,
+      params: [siteId, rows, siteId],
+    })),
   ];
-  if (all.length > 0)
-    out.push({
-      sql: `INSERT OR IGNORE INTO asset_refs (site_id, asset_id, node_id, node_kind, field)
-            SELECT ?, json_extract(value, '$.a'), json_extract(value, '$.n'), json_extract(value, '$.k'), json_extract(value, '$.f') FROM json_each(?)`,
-      params: [siteId, JSON.stringify(all)],
-    });
-  return out;
 }

packages/core-ts/src/sync-plan.ts

@@ -303,9 +303,11 @@ function nodeBatch(
       inserted++;
     }
     batch.push(...extra.lists);
+    // The derived reference index follows the node set in the same batch (RFC-0004 §4.4), and
+    // only then: beside an unchanged node set its plan-time rows could replace a concurrent
+    // write's fresher ones (#196). Its chunked `INSERT`s count against the cap below.
+    batch.push(...refsRebuild(siteId, changes.rows));
   }
-  // The derived reference index follows the node set in the same batch (RFC-0004 §4.4).
-  batch.push(...refsRebuild(siteId, changes.rows));
   const written = changes.rows.find((n) => n.kind === "site" && n.id === siteId) ?? siteRow;
   batch.push(...hostStatements(extra.hosts, { siteId, hash: written.hash }));
   if (batch.length > 1000)