diff --git a/packages/core/src/kv/hexastore/KvTripleStore.ts b/packages/core/src/kv/hexastore/KvTripleStore.ts index 3f2cd4a..7eeb409 100644 --- a/packages/core/src/kv/hexastore/KvTripleStore.ts +++ b/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 { + return this.getByEntityTypeWhereAsync(entityType, (datom) => datom.retractedAt === null); + } + + /** Async version of getByEntityTypeTemporal using batched META fetches. */ + getByEntityTypeTemporalAsync( + entityType: string, + basis: ResolvedTemporalBasis, + ): Effect.Effect { + return this.getByEntityTypeWhereAsync(entityType, (datom) => visibleAt(datom, basis)); + } + + private getByEntityTypeWhereAsync( + entityType: string, + keep: (datom: Datom) => boolean, + ): Effect.Effect { 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; diff --git a/packages/core/src/kv/layers/KvTriplesLive.ts b/packages/core/src/kv/layers/KvTriplesLive.ts index d617ebf..2ebd18c 100644 --- a/packages/core/src/kv/layers/KvTriplesLive.ts +++ b/packages/core/src/kv/layers/KvTriplesLive.ts @@ -281,16 +281,21 @@ const makeKvTriplesService = Effect.gen(function* () { basis?: ResolvedTemporalBasis, ): Effect.Effect => 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); diff --git a/packages/core/test/kv/layers/KvTriples.store.test.ts b/packages/core/test/kv/layers/KvTriples.store.test.ts index 3acd921..e781965 100644 --- a/packages/core/test/kv/layers/KvTriples.store.test.ts +++ b/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[0]) => + backend.getRange(options).pipe(Stream.tap(() => Effect.sync(() => scanned++))), + getRangeSync: (options: Parameters>[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", () => {