Simplify query policies, fetch commits, and graph lifetimes

lib/core/fetchCommit.ts

@@ -0,0 +1,173 @@
+import type { PageResponse, QueryResponse } from '../adapters/adapter.js'
+import { rebaseResponseData, type FetchRebasePlan } from './fetchRebase.js'
+import { queryPolicy } from './queryMaintenance.js'
+import { commitQuery } from './queryResults.js'
+import {
+  entityKey,
+  type EntityKey,
+  type ItemId,
+  type ProcessedCacheEvent,
+  type ProcessedProjectionEvent,
+  type Query,
+  type QueuedEvent,
+  type ServiceState,
+  type TraceCause,
+} from './queryTypes.js'
+import { applyEventsToService, diffCompleteSet, isCompleteSetQuery } from './windowMaintenance.js'
+
+export type StoreResponse<TMeta> =
+  | QueryResponse<unknown, TMeta | undefined>
+  | PageResponse<unknown[], TMeta>
+
+interface CompleteSetCommit {
+  previousEntities: Map<EntityKey, unknown>
+  nextItemIds: Set<EntityKey>
+  ignoredItemIds: ReadonlySet<EntityKey>
+}
+
+interface EntityAccess {
+  getId: (item: unknown) => ItemId | undefined
+  isItemStale: (current: unknown, next: unknown) => boolean
+}
+
+export interface FetchCommitPlan<TMeta> {
+  query: Omit<Query<unknown, TMeta>, 'rows'> & { fetchedAt: number }
+  entityEvents: QueuedEvent[]
+  completeSet: CompleteSetCommit | null
+  replayEvents: readonly ProcessedCacheEvent[]
+  deferredMutationLaneKeys: readonly string[]
+}
+
+/** Plan against the confirmed lane bases and current overlays without mutating the cache. */
+export function buildFetchCommit<TMeta>({
+  service,
+  query,
+  result,
+  source,
+  rebasePlan,
+  pendingMutationEvents,
+  getId,
+  isItemStale,
+  meta,
+  fetchedAt,
+  cause,
+}: EntityAccess & {
+  service: ServiceState<TMeta>
+  query: Query<unknown, TMeta>
+  result: StoreResponse<TMeta>
+  source: 'cache' | 'server'
+  rebasePlan: FetchRebasePlan
+  pendingMutationEvents: readonly ProcessedProjectionEvent[]
+  meta: TMeta
+  fetchedAt: number
+  cause?: TraceCause
+}): FetchCommitPlan<TMeta> {
+  const policy = queryPolicy(query.maintenance)
+  const mode = policy.responseMode
+  const queryMutationEvents = mode === 'snapshot' ? [] : pendingMutationEvents
+  const latestEventById = new Map(rebasePlan.latestEventById)
+  for (const event of queryMutationEvents) latestEventById.set(event.itemId, event)
+  const ignoredItemIds = new Set(rebasePlan.itemIds)
+  // Every pending mutation protects its canonical entity from a fetch overwrite,
+  // including snapshots, whose returned rows deliberately ignore those mutations.
+  for (const event of pendingMutationEvents) ignoredItemIds.add(event.itemId)
+
+  const rebased = rebaseResponseData({
+    data: result.data,
+    mode,
+    latestEventById,
+    entities: service.entities,
+    getId,
+    isItemStale,
+    canKeepCurrentItem: item =>
+      !(
+        query.desc.method === 'find' &&
+        policy.rebaseMembership &&
+        query.maintenance.matches(item) === false
+      ),
+  })
+  const entityEvents: QueuedEvent[] = []
+  // Projections never replace complete canonical entities, even at the same revision.
+  if (source === 'server' && !query.maintenance.isProjection) {
+    for (const item of rebased.items) {
+      const id = getId(item)
+      if (id === undefined || ignoredItemIds.has(entityKey(id))) continue
+      entityEvents.push({
+        mode: 'server',
+        source: 'fetch',
+        serviceName: query.desc.serviceName,
+        type: service.entities.has(entityKey(id)) ? 'updated' : 'created',
+        item,
+        ...(cause === undefined ? {} : { cause }),
+      })
+    }
+  }
+
+  let completeSet: CompleteSetCommit | null = null
+  if (isCompleteSetQuery(query)) {
+    const previousEntities = new Map<EntityKey, unknown>()
+    for (const itemId of query.rows.ids) {
+      const entity = service.entities.get(itemId)
+      if (entity !== undefined) previousEntities.set(itemId, entity)
+    }
+    const nextItemIds = new Set(rebased.itemIds)
+    // A stale complete response cannot delete a newly created row or resurrect a removal.
+    for (const [itemId, event] of latestEventById) {
+      if (event.type === 'removed') nextItemIds.delete(itemId)
+      else nextItemIds.add(itemId)
+    }
+    completeSet = { previousEntities, nextItemIds, ignoredItemIds }
+  }
+
+  const events = [...rebasePlan.events, ...queryMutationEvents]
+  const replayEvents =
+    mode === 'entity'
+      ? events
+      : mode === 'fetch-owned'
+        ? events.filter(event => event.mode !== 'server' || event.source !== 'realtime')
+        : []
+  const pageInfo = 'pageInfo' in result ? result.pageInfo : undefined
+  return {
+    query: {
+      ...query,
+      fetchedAt,
+      state: {
+        status: 'success',
+        data: rebased.data,
+        meta,
+        ...(pageInfo ? { pageInfo } : {}),
+        isFetching: false,
+        error: null,
+      },
+    },
+    entityEvents,
+    completeSet,
+    replayEvents,
+    deferredMutationLaneKeys: queryMutationEvents.map(event => event.mutationLaneKey),
+  }
+}
+
+/** Apply entities and query membership before the store publishes any effects. */
+export function applyFetchCommit<TMeta>(
+  service: ServiceState<TMeta>,
+  plan: FetchCommitPlan<TMeta>,
+  access: EntityAccess,
+): ProcessedCacheEvent[] {
+  const { query, completeSet } = plan
+  const processedEvents: ProcessedCacheEvent[] = []
+  applyEventsToService({
+    service,
+    serviceName: query.desc.serviceName,
+    events: plan.entityEvents,
+    ...access,
+    processedEvents,
+  })
+  if (completeSet) {
+    service.materialized = { queryId: query.queryId, fetchedAt: query.fetchedAt }
+  }
+  commitQuery(service, query)
+  // Complete-set diffs establish membership; ordinary fetches only establish row values.
+  return completeSet
+    ? diffCompleteSet({ service, serviceName: query.desc.serviceName, ...completeSet })
+    : processedEvents
+}

lib/core/fetchRebase.ts

@@ -1,3 +1,4 @@
+import type { FetchResponseMode } from './queryMaintenance.js'
 import { CappedBuffer } from './cappedBuffer.js'
 import { entityKey, type EntityKey, type ItemId, type ProcessedCacheEvent } from './queryTypes.js'
 
@@ -155,8 +156,6 @@ export interface RebasedResponse {
   readonly itemIds: ReadonlySet<EntityKey>
 }
 
-export type FetchResponseMode = 'entity' | 'projection' | 'snapshot' | 'fetch-owned'
-
 function overlayProjectionItem(
   responseItem: unknown,
   event: ProcessedCacheEvent,

lib/core/figbird.ts

@@ -34,12 +34,12 @@ import {
 import { resolveQueryInput, type PreparedQuery, type QueryInput } from './queryDefinition.js'
 import {
   explainQueryNode,
-  isServerMaintained,
   type ClassificationReason,
   type ExplainReport,
   type QueryNodeClass,
 } from './queryClassification.js'
 export type { ExplainNode, ExplainReport } from './queryClassification.js'
+import { queryPolicy } from './queryMaintenance.js'
 import { QueryRef } from './queryRef.js'
 import {
   DEFAULT_GC_TIME,
@@ -79,7 +79,7 @@ import type {
 } from './schema.js'
 import { foreignKeys, resolveServicePath } from './schema.js'
 import type { ValueComparator } from './sort.js'
-import { isWithinStaleTime, validatePrefetchStaleTime, validateStaleTime } from './staleTime.js'
+import { validatePrefetchStaleTime, validateStaleTime } from './staleTime.js'
 import { createTransactionContext, type TransactionContext } from './transactions.js'
 
 type DescriptorWriteProjection<TItem> =
@@ -328,11 +328,6 @@ export class Figbird<
   dispose(): void {
     if (this.#disposed) return
     this.#disposed = true
-    for (const prefetch of this.#prefetches.values()) {
-      prefetch.timer.cancel()
-      prefetch.release()
-    }
-    this.#prefetches.clear()
     for (const ref of this.#windowQueryCache.values()) ref.dispose()
     for (const ref of this.#relationalQueryCache.values()) ref.dispose()
     this.#windowQueryCache.clear()
@@ -608,9 +603,6 @@ export class Figbird<
     }
   }
 
-  // Active speculative pins, keyed by query hash (see prefetch()).
-  #prefetches: Map<string, { at: number; release: () => void; timer: ClockTimer }> = new Map()
-
   /**
    * Speculatively warm a query — the idempotent, fire-and-forget sibling of `prepare()`.
    *
@@ -635,28 +627,7 @@ export class Figbird<
     options?: { staleTime?: number },
   ): void {
     const staleTime = validatePrefetchStaleTime(options?.staleTime ?? 30_000)
-    const ref = this.query(query)
-    const hash = ref.hash()
-
-    const now = this.clock.now()
-    const existing = this.#prefetches.get(hash)
-    if (existing && isWithinStaleTime(existing.at, staleTime, now)) return
-    if (existing) {
-      existing.timer.cancel()
-      existing.release()
-      this.#prefetches.delete(hash)
-    }
-
-    // The pin also carries the staleTime so a warm-in-store read within the window
-    // skips the SWR revalidation instead of re-fetching.
-    const release = ref.subscribe(() => {}, { staleTime, source: 'prefetch' })
-    const timer = this.clock.setTimeout(() => {
-      this.#prefetches.delete(hash)
-      release()
-    }, staleTime)
-    // Never keep a Node process alive for a speculative pin (browsers ignore this).
-    timer.unref()
-    this.#prefetches.set(hash, { at: now, release, timer })
+    this.query(query).prefetch(staleTime)
   }
 
   // Descriptor layer — the primitive the relational engine (and the deprecated
@@ -1053,7 +1024,7 @@ export class Figbird<
                 server: query.config.server,
                 allPages: 'allPages' in query.config && query.config.allPages === true,
                 localOperators: locallySupportedOperators(this.adapter, serviceName),
-                snapshot: query.config.realtime === 'disabled',
+                snapshot: queryPolicy(query.maintenance).responseMode === 'snapshot',
               })
             : null
         rows.push({
@@ -1073,13 +1044,7 @@ export class Figbird<
             : {}),
           classification: query.maintenance.classification,
           classificationReasons: explanation?.reasons ?? [],
-          realtimeStrategy:
-            query.config.realtime === 'disabled'
-              ? 'manual'
-              : query.config.realtime === 'refetch' ||
-                  isServerMaintained(query.maintenance.classification)
-                ? 'refetch'
-                : 'merge',
+          realtimeStrategy: queryPolicy(query.maintenance).realtimeStrategy,
           skipped: query.config.skip === true,
           status: query.state.status,
           isFetching: query.state.isFetching,

lib/core/queryMaintenance.ts

@@ -1,6 +1,8 @@
 import {
   classifyStoredQuery,
   isProjectionQuery,
+  isServerMaintained,
+  usesFetchOwnedRows,
   type StoredQueryClass,
 } from './queryClassification.js'
 import { buildComparator, type ValueComparator } from './sort.js'
@@ -11,6 +13,8 @@ import {
   type QueryConfig,
 } from './queryTypes.js'
 
+export type FetchResponseMode = 'entity' | 'projection' | 'snapshot' | 'fetch-owned'
+
 export interface QueryMaintenance {
   classification: StoredQueryClass
   matches: (item: unknown) => MatchResult
@@ -21,6 +25,29 @@ export interface QueryMaintenance {
   isProjection: boolean
 }
 
+/** Immutable execution rules kept out of subscriber-visible query state. */
+interface QueryPolicy {
+  responseMode: FetchResponseMode
+  /** Preferred row ownership; missing IDs still force a query to retain its own values. */
+  rowSource: 'entities' | 'values'
+  /** Whether event replay may change membership through the local matcher. */
+  mergeEvents: boolean
+  /** Keep visible row values current when a membership merge cannot be proved. */
+  replayVisibleEvents: boolean
+  /** Whether a newer entity can be rejected by the fetched query's local predicate. */
+  rebaseMembership: boolean
+  refetchEvents: boolean
+  realtimeStrategy: 'manual' | 'refetch' | 'merge'
+}
+
+const policies = new WeakMap<QueryMaintenance, Readonly<QueryPolicy>>()
+
+export function queryPolicy(maintenance: QueryMaintenance): Readonly<QueryPolicy> {
+  const policy = policies.get(maintenance)
+  if (!policy) throw new Error('figbird: query maintenance has not been compiled')
+  return policy
+}
+
 /** Split window operators off a query so the rest can feed the local matcher. */
 function splitWindow(q: Record<string, unknown> | undefined): {
   filters: Record<string, unknown> | undefined
@@ -60,12 +87,21 @@ export function compileQueryMaintenance({
     allPages: 'allPages' in config && config.allPages === true,
     localOperators,
   })
+  const isProjection = isProjectionQuery(query)
+  const responseMode: FetchResponseMode =
+    config.realtime === 'disabled'
+      ? 'snapshot'
+      : isProjection
+        ? 'projection'
+        : usesFetchOwnedRows(classification, config.realtime)
+          ? 'fetch-owned'
+          : 'entity'
   const { filters, sort, limit, skip } = splitWindow(query)
   const effectiveSort = sort ?? defaultSort
   // Local reads also serve queries with realtime disabled. Find matchers receive
   // only predicates here, while realtime matchers retain the original query input.
   let localMatcher: ((item: unknown) => MatchResult) | undefined
-  return {
+  const maintenance: QueryMaintenance = {
     classification,
     matches:
       config.realtime === 'merge' && classification !== 'server-authoritative'
@@ -76,6 +112,27 @@ export function compileQueryMaintenance({
     compare: effectiveSort ? buildComparator(effectiveSort, compare) : undefined,
     limit,
     skip,
-    isProjection: isProjectionQuery(query),
+    isProjection,
   }
+  policies.set(maintenance, {
+    responseMode,
+    rowSource:
+      responseMode !== 'entity' || config.fetchPolicy === 'network-only' ? 'values' : 'entities',
+    mergeEvents:
+      config.realtime === 'merge' &&
+      !(desc.method === 'find' && config.fetchPolicy === 'network-only'),
+    replayVisibleEvents:
+      config.realtime === 'refetch' ||
+      config.fetchPolicy === 'network-only' ||
+      isServerMaintained(classification),
+    rebaseMembership: config.realtime === 'merge' && !isServerMaintained(classification),
+    refetchEvents: config.realtime === 'refetch',
+    realtimeStrategy:
+      config.realtime === 'disabled'
+        ? 'manual'
+        : config.realtime === 'refetch' || isServerMaintained(classification)
+          ? 'refetch'
+          : 'merge',
+  })
+  return maintenance
 }