kv: answer entity-type matches at a temporal basis from the TYPE index

packages/core/src/kv/hexastore/KvTripleStore.ts

@@ -755,6 +755,22 @@ export class KvTripleStore {
    * Returns null if sync KV operations are not available (caller should fall back).
    */
   getByEntityType(entityType: string): Datom[] | null {
+    return this.getByEntityTypeWhere(entityType, (datom) => datom.retractedAt === null);
+  }
+
+  /**
+   * Datoms of one entity type visible at a bitemporal basis, read from the TYPE
+   * index. Retraction only rewrites META, so the index still names every datom
+   * ever asserted with this type and the basis decides which are visible.
+   */
+  getByEntityTypeTemporal(entityType: string, basis: ResolvedTemporalBasis): Datom[] | null {
+    return this.getByEntityTypeWhere(entityType, (datom) => visibleAt(datom, basis));
+  }
+
+  private getByEntityTypeWhere(
+    entityType: string,
+    keep: (datom: Datom) => boolean,
+  ): Datom[] | null {
     if (!this.kv.getRangeSync) return null;
 
     const { start, end } = typePrefixRange(entityType);
@@ -764,8 +780,7 @@ export class KvTripleStore {
     for (let i = 0; i < entries.length; i++) {
       const tripleId = textDecoder.decode(entries[i]![1]);
       const datom = this.getByIdSync(tripleId);
-      if (datom === null) continue;
-      if (datom.retractedAt !== null) continue;
+      if (datom === null || !keep(datom)) continue;
       results.push(datom);
     }
     return results;
@@ -874,6 +889,21 @@ export class KvTripleStore {
    * Scans the TYPE index, then batch-fetches all META records in one call.
    */
   getByEntityTypeAsync(entityType: string): Effect.Effect<Datom[]> {
+    return this.getByEntityTypeWhereAsync(entityType, (datom) => datom.retractedAt === null);
+  }
+
+  /** Async version of getByEntityTypeTemporal using batched META fetches. */
+  getByEntityTypeTemporalAsync(
+    entityType: string,
+    basis: ResolvedTemporalBasis,
+  ): Effect.Effect<Datom[]> {
+    return this.getByEntityTypeWhereAsync(entityType, (datom) => visibleAt(datom, basis));
+  }
+
+  private getByEntityTypeWhereAsync(
+    entityType: string,
+    keep: (datom: Datom) => boolean,
+  ): Effect.Effect<Datom[]> {
     const { start, end } = typePrefixRange(entityType);
 
     return Effect.gen({ self: this }, function* () {
@@ -889,8 +919,7 @@ export class KvTripleStore {
 
       const results: Datom[] = [];
       for (const datom of datoms) {
-        if (datom === null) continue;
-        if (datom.retractedAt !== null) continue;
+        if (datom === null || !keep(datom)) continue;
         results.push(datom);
       }
       return results;

packages/core/src/kv/layers/KvTriplesLive.ts

@@ -281,16 +281,21 @@ const makeKvTriplesService = Effect.gen(function* () {
     basis?: ResolvedTemporalBasis,
   ): Effect.Effect<readonly Triple[], ReadError> =>
     Effect.gen(function* () {
-      // Fast path: entityType-only query uses the KV TYPE index
+      // Fast path: entityType-only query uses the KV TYPE index, at any basis.
+      // Public reads always arrive with a resolved basis, so without the
+      // temporal variant every entity-type read scanned the whole store.
       if (pattern.entityType && !pattern.entityId && !pattern.attribute && !pattern.value) {
-        const datoms = basis === undefined ? hexaStore.getByEntityType(pattern.entityType) : null;
+        const datoms =
+          basis === undefined
+            ? hexaStore.getByEntityType(pattern.entityType)
+            : hexaStore.getByEntityTypeTemporal(pattern.entityType, basis);
         if (datoms !== null) {
           return datoms.map(datomToTriple);
         }
         const asyncDatoms =
           basis === undefined
             ? yield* hexaStore.getByEntityTypeAsync(pattern.entityType)
-            : yield* hexaStore.scanCollectTemporalAsync({}, basis);
+            : yield* hexaStore.getByEntityTypeTemporalAsync(pattern.entityType, basis);
         return asyncDatoms
           .filter((datom) => datom.entityType === pattern.entityType)
           .map(datomToTriple);

packages/core/test/kv/layers/KvTriples.store.test.ts

@@ -6,7 +6,7 @@
  */
 
 import { describe, it, expect } from "vitest";
-import { Effect, Layer, Option } from "effect";
+import { Effect, Layer, Option, Stream } from "effect";
 import { Triples } from "../../../src/store/Triples.js";
 import type { TripleId, EntityId } from "@triplex-build/triplex";
 import { KvTriplesLive } from "../../../src/kv/layers/KvTriplesLive.js";
@@ -349,6 +349,75 @@ describe("KvTriplesLive - temporal match", () => {
   });
 });
 
+describe("KvTriplesLive - entity type at a temporal basis", () => {
+  it("answers from the TYPE index, honouring recorded and valid time", async () => {
+    // Count the KV entries each range read returns, to see which index served the match.
+    let scanned = 0;
+    const backend = makeTestKvBackend();
+    const counting = {
+      ...backend,
+      getRange: (options: Parameters<typeof backend.getRange>[0]) =>
+        backend.getRange(options).pipe(Stream.tap(() => Effect.sync(() => scanned++))),
+      getRangeSync: (options: Parameters<NonNullable<typeof backend.getRangeSync>>[0]) => {
+        const entries = backend.getRangeSync!(options);
+        scanned += entries.length;
+        return entries;
+      },
+    };
+    const layer = KvTriplesLive.pipe(
+      Layer.provide(TripleStoreRuntimeLayer),
+      Layer.provide(Layer.succeed(KvBackend, counting)),
+    );
+    await Effect.runPromise(
+      Effect.gen(function* () {
+        const store = yield* Triples;
+        // Plenty of other facts, so a full scan would be visible in the count.
+        yield* store.transact(
+          Array.from({ length: 500 }, (_, i) => ({
+            op: "assert" as const,
+            entityId: `other:${i}` as EntityId,
+            attribute: ":other/n",
+            value: num(i),
+          })),
+        );
+        const early = yield* store.assert({
+          entityId: "w:1",
+          attribute: ":widget/name",
+          value: str("first"),
+          entityType: "Widget",
+          validFrom: 1_000,
+          validTo: 2_000,
+        });
+        yield* store.assert({
+          entityId: "w:1",
+          attribute: ":widget/name",
+          value: str("second"),
+          entityType: "Widget",
+          validFrom: 2_000,
+        });
+        const recorded = Date.now();
+        yield* Effect.sleep("5 millis");
+        yield* store.retract(early.id);
+
+        const names = (triples: readonly { value: { value: unknown } }[]) =>
+          triples.map((t) => t.value.value).sort();
+        scanned = 0;
+        expect(names(yield* store.match({ entityType: "Widget" }, { validAt: 1_500 }))).toEqual([]);
+        expect(
+          names(
+            yield* store.match({ entityType: "Widget" }, { validAt: 1_500, recordedAt: recorded }),
+          ),
+        ).toEqual(["first"]);
+        expect(names(yield* store.match({ entityType: "Widget" }, { validAt: 2_500 }))).toEqual([
+          "second",
+        ]);
+        // Three reads of a two-entry type index, not three scans of 500+ facts and the journal.
+        expect(scanned).toBe(6);
+      }).pipe(Effect.provide(layer)),
+    );
+  });
+});
+
 // ─── History ───────────────────────────────────────────────────────────────
 
 describe("KvTriplesLive - history", () => {