query

func Get ¶

func Get[V any](
	ctx context.Context,
	store state.Store[V],
	entity string,
) (V, bool, error)

Get returns the all-time aggregation value for entity. Used for non-windowed pipelines (Bucket = 0).

func GetRange ¶

func GetRange[V any](
	ctx context.Context,
	store state.Store[V],
	m monoid.Monoid[V],
	w windowed.Config,
	entity string,
	start, end time.Time,
) (V, error)

GetRange returns the aggregation merged over the bucket range covering [start, end] under the given Windowed config.

func GetRangeMany ¶

func GetRangeMany[V any](
	ctx context.Context,
	store state.Store[V],
	m monoid.Monoid[V],
	w windowed.Config,
	entities []string,
	start, end time.Time,
) ([]V, error)

GetRangeMany is the absolute-range counterpart to GetWindowMany.

func GetTrailing ¶

func GetTrailing[V any](
	ctx context.Context,
	store state.Store[V],
	m monoid.Monoid[V],
	w windowed.Config,
	entity string,
	duration time.Duration,
	now time.Time,
) (V, error)

GetTrailing returns the aggregation merged across the trailing window covering [now-duration, now] under the given Windowed config. A thin wrapper over GetWindow that documents the "trailing window" intent.

Use this for "last 7d unique viewers", "trailing 30d likes", "last 24h impressions" — the dominant shape for counter queries.

func GetTrailingMany ¶

func GetTrailingMany[V any](
	ctx context.Context,
	store state.Store[V],
	m monoid.Monoid[V],
	w windowed.Config,
	entities []string,
	duration time.Duration,
	now time.Time,
) ([]V, error)

GetTrailingMany batches GetTrailing across many entities in a single fanned-out fetch. Same single-store-call contract as GetWindowMany.

func GetWindow ¶

func GetWindow[V any](
	ctx context.Context,
	store state.Store[V],
	m monoid.Monoid[V],
	w windowed.Config,
	entity string,
	duration time.Duration,
	now time.Time,
) (V, error)

GetWindow returns the aggregation merged over the bucket range covering [now-duration, now] under the given Windowed config.

Implementation: compute the inclusive bucket-ID range, BatchGetItem all buckets in one or more requests, then fold via the monoid Combine in stable order. Missing buckets are treated as the monoid Identity.

Cost note: for fine-grained windows the bucket count can be large (e.g., a "last 30 days" query on hourly buckets reads 720 items). For sketches this means 720 sketch merges in process. If query latency becomes a bottleneck, the Phase 2 plan calls for pre-rolled rollup buckets or a Valkey-cached pre-merged window.

func GetWindowMany ¶

func GetWindowMany[V any](
	ctx context.Context,
	store state.Store[V],
	m monoid.Monoid[V],
	w windowed.Config,
	entities []string,
	duration time.Duration,
	now time.Time,
) ([]V, error)

GetWindowMany returns the windowed merge for each of `entities` in a single fanned-out fetch. Equivalent to calling GetWindow per entity, but uses ONE underlying store.GetMany over (N entities × M buckets) keys instead of N separate GetMany calls of M keys each.

For ML rerank with windowed counter features over N=200 candidates with M=7 daily buckets, this is 14 BatchGetItem calls in parallel (~20ms p99) instead of 200 sequential GetWindow calls (~4 seconds).

Returns one value per entity in input order. Missing entities (no bucket data within the range) return the monoid Identity rather than an error — callers branch on Identity, not on a separate "present" flag, since "merged-empty" is a legitimate windowed result.

func WarmupNonWindowed ¶

func WarmupNonWindowed[V any](
	ctx context.Context,
	cache state.Cache[V],
	store state.Store[V],
	entities []string,
) (int, error)

WarmupNonWindowed prefetches all-time entries for the given entities into the cache. Same shape as WarmupWindowed but for pipelines without a time-bucket dimension.

func WarmupWindowed ¶

func WarmupWindowed[V any](
	ctx context.Context,
	cache state.Cache[V],
	store state.Store[V],
	w windowed.Config,
	entities []string,
	duration time.Duration,
	now time.Time,
) (int, error)

WarmupWindowed prefetches the (entity × bucket) keys covering the most-recent `duration` window for each entity into the cache, sourcing fresh values from the authoritative store via cache.Repopulate.

Use case: cold-cache p99 inflation. After a Valkey restart or a fresh query-service deployment, the first batch of GetWindow / GetWindowMany queries falls through to DDB and pays full latency. A pre-flight WarmupWindowed at process start (or on a known traffic pattern) loads the hot keys before user requests arrive.

Cost is `len(entities) × bucketCount` cache writes plus one batched store.GetMany over the same key set. For ML-rerank-style workloads where the candidate set is bounded (typically N=200 popular content items × M=7 daily buckets = 1400 keys), the warmup completes in a few hundred milliseconds against typical DDB capacity and is a straightforward step in the service-start path.

Returns the count of successfully warmed entries (entities × buckets that the store had data for) so callers can log meaningful progress.