fix: index many-to-many query invalidation

src/contracts/query-invalidation.ts

@@ -1,4 +1,8 @@
 import { isCacheInvalidationTopic } from "./cache-invalidation.ts";
+import {
+  queryNamespaceIdentityKey,
+  type QueryNamespaceIdentity,
+} from "./query-keys.ts";
 
 declare const queryInvalidationTopicBrand: unique symbol;
 
@@ -37,20 +41,28 @@ export const INVALIDATION_REGISTRY_BOUNDS = Object.freeze({
 
 export type InvalidationRegistryEdge = Readonly<{
   topicId: string;
-  namespace: string;
+  namespace: QueryNamespaceIdentity;
 }>;
 
 export interface InvalidationRegistry {
   readonly topics: readonly string[];
-  readonly namespaces: readonly string[];
+  readonly namespaces: readonly QueryNamespaceIdentity[];
   readonly edges: readonly InvalidationRegistryEdge[];
 }
 
 export type InvalidationRegistryIndex = Readonly<{
-  namespacesForTopic: ReadonlyMap<string, readonly string[]>;
+  namespacesForTopic: ReadonlyMap<
+    string,
+    readonly QueryNamespaceIdentity[]
+  >;
   topicsForNamespace: ReadonlyMap<string, readonly string[]>;
 }>;
 
+export type InvalidationTopicVersionDefinition = Readonly<{
+  topicId: string;
+  topicVersion: number;
+}>;
+
 function hasControlCharacter(value: string): boolean {
   for (const character of value) {
     const codePoint = character.codePointAt(0) ?? 0;
@@ -96,44 +108,60 @@ export function indexInvalidationRegistry(
     }
     topics.add(topic);
   }
-  const namespaces = new Set<string>();
+  const namespaces = new Map<string, QueryNamespaceIdentity>();
   for (const namespace of registry.namespaces) {
-    assertId(namespace, "namespace");
-    if (namespaces.has(namespace)) {
-      throw new TypeError(`Duplicate invalidation namespace: ${namespace}`);
+    let namespaceKey: string;
+    try {
+      namespaceKey = queryNamespaceIdentityKey(namespace);
+    } catch (error) {
+      throw new TypeError("Invalidation registry namespace is invalid.", {
+        cause: error,
+      });
     }
-    namespaces.add(namespace);
+    if (namespaces.has(namespaceKey)) {
+      throw new TypeError(`Duplicate invalidation namespace: ${namespaceKey}`);
+    }
+    namespaces.set(namespaceKey, namespace);
   }
 
-  const namespacesForTopic = new Map<string, string[]>();
+  const namespacesForTopic = new Map<string, QueryNamespaceIdentity[]>();
   const topicsForNamespace = new Map<string, string[]>();
   const seenEdges = new Map<string, Set<string>>();
   for (const edge of registry.edges) {
-    if (!topics.has(edge.topicId) || !namespaces.has(edge.namespace)) {
+    let namespaceKey: string;
+    try {
+      namespaceKey = queryNamespaceIdentityKey(edge.namespace);
+    } catch (error) {
+      throw new TypeError("Invalidation registry namespace is invalid.", {
+        cause: error,
+      });
+    }
+    const registeredNamespace = namespaces.get(namespaceKey);
+    if (!topics.has(edge.topicId) || !registeredNamespace) {
       throw new TypeError("Invalidation edge references an unknown endpoint.");
     }
     const seenNamespaces = seenEdges.get(edge.topicId) ?? new Set<string>();
-    if (seenNamespaces.has(edge.namespace)) {
+    if (seenNamespaces.has(namespaceKey)) {
       throw new TypeError("Duplicate invalidation edge.");
     }
-    seenNamespaces.add(edge.namespace);
+    seenNamespaces.add(namespaceKey);
     seenEdges.set(edge.topicId, seenNamespaces);
 
     const fanOut = namespacesForTopic.get(edge.topicId) ?? [];
-    fanOut.push(edge.namespace);
+    fanOut.push(registeredNamespace);
     if (fanOut.length > bounds.maxTopicFanOut) {
       throw new TypeError(`Invalidation topic fan-out exceeded: ${edge.topicId}`);
     }
     namespacesForTopic.set(edge.topicId, fanOut);
 
-    const fanIn = topicsForNamespace.get(edge.namespace) ?? [];
+    const fanIn = topicsForNamespace.get(namespaceKey) ?? [];
     fanIn.push(edge.topicId);
     if (fanIn.length > bounds.maxNamespaceFanIn) {
       throw new TypeError(
-        `Invalidation namespace fan-in exceeded: ${edge.namespace}`,
+        `Invalidation namespace fan-in exceeded: ${namespaceKey}`,
       );
     }
-    topicsForNamespace.set(edge.namespace, fanIn);
+    topicsForNamespace.set(namespaceKey, fanIn);
   }
 
   for (const topic of topics) {
@@ -141,9 +169,9 @@ export function indexInvalidationRegistry(
       throw new TypeError(`Orphan invalidation topic: ${topic}`);
     }
   }
-  for (const namespace of namespaces) {
-    if (!topicsForNamespace.has(namespace)) {
-      throw new TypeError(`Orphan invalidation namespace: ${namespace}`);
+  for (const namespaceKey of namespaces.keys()) {
+    if (!topicsForNamespace.has(namespaceKey)) {
+      throw new TypeError(`Orphan invalidation namespace: ${namespaceKey}`);
     }
   }
 
@@ -151,7 +179,7 @@ export function indexInvalidationRegistry(
     namespacesForTopic: new Map(
       [...namespacesForTopic].map(([key, value]) => [
         key,
-        Object.freeze([...value]) as readonly string[],
+        Object.freeze([...value]) as readonly QueryNamespaceIdentity[],
       ]),
     ),
     topicsForNamespace: new Map(
@@ -163,6 +191,41 @@ export function indexInvalidationRegistry(
   });
 }
 
+/** Projects the graph's wire-only topic versions and rejects composition drift. */
+export function indexInvalidationTopicVersions(
+  registry: Pick<InvalidationRegistry, "topics">,
+  definitions: readonly InvalidationTopicVersionDefinition[],
+): ReadonlyMap<string, number> {
+  if (
+    registry.topics.length > INVALIDATION_REGISTRY_BOUNDS.maxTopics ||
+    definitions.length > INVALIDATION_REGISTRY_BOUNDS.maxTopics
+  ) {
+    throw new TypeError("Invalidation topic version registry exceeds its bounds.");
+  }
+  const registeredTopics = new Set(registry.topics);
+  if (
+    registeredTopics.size !== registry.topics.length ||
+    definitions.length !== registeredTopics.size
+  ) {
+    throw new TypeError("Invalidation topic version registry is inconsistent.");
+  }
+
+  const versions = new Map<string, number>();
+  for (const definition of definitions) {
+    if (
+      !isCacheInvalidationTopic(definition.topicId) ||
+      !registeredTopics.has(definition.topicId) ||
+      !Number.isSafeInteger(definition.topicVersion) ||
+      definition.topicVersion < 1 ||
+      versions.has(definition.topicId)
+    ) {
+      throw new TypeError("Invalidation topic version registry is invalid.");
+    }
+    versions.set(definition.topicId, definition.topicVersion);
+  }
+  return versions;
+}
+
 export type QueryMutationLease = Readonly<{
   /**
    * Releases one local mutation fence. Remote hints coalesced while the fence